feat(e2b): add sandbox lifecycle provider

README.md

@@ -6,41 +6,37 @@
 
 ## Quick Install
 
-After the package is published:
-
 ```bash
 pip install langchain-e2b
 ```
 
 ```python
-from e2b import Sandbox
-
-from langchain_e2b import E2BSandbox
+from langchain_e2b import E2BProvider
 
-sandbox = Sandbox.create()
-backend = E2BSandbox(sandbox=sandbox)
+provider = E2BProvider()
+sandbox = provider.get_or_create()
 
 try:
-    result = backend.execute("echo hello")
+    result = sandbox.execute("echo hello")
     print(result.output)
 finally:
-    sandbox.kill()
+    provider.delete(sandbox_id=sandbox.id)
 ```
 
 ## What is this?
 
-`langchain-e2b` adapts an existing E2B sandbox to the Deep Agents sandbox
-protocol. It uses the low-level `e2b` SDK so Deep Agents can run shell commands
-and move files through the standard Deep Agents backend interface.
+`langchain-e2b` adapts E2B sandboxes to the Deep Agents sandbox protocol. It
+uses the low-level `e2b` SDK so Deep Agents can create or reconnect to
+sandboxes, run shell commands, and move files through the standard Deep Agents
+sandbox interface.
 
-This package intentionally does not hide E2B sandbox lifecycle management. Use
-the E2B SDK to create, connect to, configure, and kill sandboxes, then pass the
-connected sandbox object to `E2BSandbox`.
+Use `E2BProvider` when you want the package to manage sandbox lifecycle. Use
+`E2BSandbox` directly when you already have an E2B SDK sandbox object.
 
 ## Contributing
 
 Contributions are welcome. Keep the adapter focused on implementing the Deep
-Agents sandbox backend protocol over the official E2B SDK.
+Agents sandbox protocol over the official E2B SDK.
 
 ## Development
 

langchain_e2b/__init__.py

@@ -1,5 +1,6 @@
 """E2B sandbox integration for Deep Agents."""
 
+from langchain_e2b.provider import E2BProvider
 from langchain_e2b.sandbox import E2BSandbox
 
-__all__ = ["E2BSandbox"]
+__all__ = ["E2BProvider", "E2BSandbox"]

langchain_e2b/_version.py

@@ -2,4 +2,4 @@
 
 # Keep the `x-release-please-version` annotation — release-please uses it to
 # bump `__version__` in sync with `pyproject.toml` on every release PR.
-__version__ = "0.0.1"  # x-release-please-version
+__version__ = "0.0.2"  # x-release-please-version

langchain_e2b/provider.py

@@ -0,0 +1,136 @@
+"""E2B sandbox lifecycle provider."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Callable
+from typing import NoReturn
+
+import e2b
+
+from langchain_e2b.sandbox import E2BSandbox
+
+DEFAULT_SANDBOX_TIMEOUT = 30 * 60
+EnvResolver = Callable[[str], str | None]
+
+
+def _default_resolve_env_var(name: str) -> str | None:
+    return os.environ.get(name)
+
+
+def _resolve_sandbox_timeout(resolve_env_var: EnvResolver) -> int:
+    value = resolve_env_var("E2B_SANDBOX_TIMEOUT")
+    if not value:
+        return DEFAULT_SANDBOX_TIMEOUT
+
+    try:
+        timeout = int(value)
+    except ValueError as exc:
+        msg = "E2B_SANDBOX_TIMEOUT must be an integer number of seconds"
+        raise ValueError(msg) from exc
+
+    if timeout <= 0:
+        msg = "E2B_SANDBOX_TIMEOUT must be positive"
+        raise ValueError(msg)
+    return timeout
+
+
+def _raise_unsupported_kwargs(kwargs: dict[str, object]) -> NoReturn:
+    names = ", ".join(sorted(kwargs))
+    msg = f"Received unsupported arguments: {names}"
+    raise TypeError(msg)
+
+
+class E2BProvider:
+    """Manage E2B sandbox lifecycle for Deep Agents."""
+
+    def __init__(
+        self,
+        *,
+        api_key: str | None = None,
+        resolve_env_var: EnvResolver | None = None,
+    ) -> None:
+        """Initialize the provider.
+
+        Args:
+            api_key: E2B API key. If omitted, `E2B_API_KEY` is resolved through
+                `resolve_env_var`.
+            resolve_env_var: Environment resolver used for provider settings.
+
+        Raises:
+            ValueError: If no API key is available or timeout configuration is
+                invalid.
+        """
+        self._resolve_env_var = resolve_env_var or _default_resolve_env_var
+
+        resolved_api_key = api_key or self._resolve_env_var("E2B_API_KEY")
+        if not resolved_api_key:
+            msg = "No E2B API key found. Set E2B_API_KEY."
+            raise ValueError(msg)
+
+        self._api_key = resolved_api_key
+        self._template = self._resolve_env_var("E2B_TEMPLATE") or None
+        self._sandbox_timeout = _resolve_sandbox_timeout(self._resolve_env_var)
+
+    def get_or_create(
+        self,
+        *,
+        sandbox_id: str | None = None,
+        timeout: int = 180,
+        **kwargs: object,
+    ) -> E2BSandbox:
+        """Get or create an E2B sandbox backend.
+
+        Args:
+            sandbox_id: Existing E2B sandbox ID, or None to create a sandbox.
+            timeout: Accepted for provider-interface compatibility. E2B sandbox
+                lifetime is configured with `E2B_SANDBOX_TIMEOUT`.
+            **kwargs: Unsupported provider options.
+
+        Returns:
+            E2B sandbox backend.
+
+        Raises:
+            KeyError: If `sandbox_id` does not exist.
+            TypeError: If unsupported provider options are passed.
+        """
+        del timeout
+        if kwargs:
+            _raise_unsupported_kwargs(kwargs)
+
+        if sandbox_id is not None:
+            try:
+                sandbox = e2b.Sandbox.connect(
+                    sandbox_id,
+                    timeout=self._sandbox_timeout,
+                    api_key=self._api_key,
+                )
+            except e2b.SandboxNotFoundException as exc:
+                raise KeyError(sandbox_id) from exc
+        elif self._template:
+            sandbox = e2b.Sandbox.create(
+                template=self._template,
+                timeout=self._sandbox_timeout,
+                api_key=self._api_key,
+            )
+        else:
+            sandbox = e2b.Sandbox.create(
+                timeout=self._sandbox_timeout,
+                api_key=self._api_key,
+            )
+
+        return E2BSandbox(sandbox=sandbox)
+
+    def delete(self, *, sandbox_id: str, **kwargs: object) -> None:
+        """Kill an E2B sandbox.
+
+        Args:
+            sandbox_id: E2B sandbox ID.
+            **kwargs: Unsupported provider options.
+
+        Raises:
+            TypeError: If unsupported provider options are passed.
+        """
+        if kwargs:
+            _raise_unsupported_kwargs(kwargs)
+        e2b.Sandbox.kill(sandbox_id, api_key=self._api_key)

pyproject.toml

@@ -19,7 +19,7 @@ classifiers = [
     "Topic :: Scientific/Engineering :: Artificial Intelligence",
 ]
 
-version = "0.0.1"
+version = "0.0.2"
 requires-python = ">=3.11,<4.0"
 dependencies = [
     "deepagents>=0.6.0,<0.7.0",

tests/unit_tests/test_provider.py

@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+from unittest.mock import MagicMock, patch
+
+import e2b
+import pytest
+
+from langchain_e2b import E2BProvider, E2BSandbox
+from langchain_e2b.provider import DEFAULT_SANDBOX_TIMEOUT
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+
+def _resolver(values: dict[str, str]) -> Callable[[str], str | None]:
+    return values.get
+
+
+def _sandbox(sandbox_id: str = "sbx-test") -> e2b.Sandbox:
+    return cast("e2b.Sandbox", MagicMock(sandbox_id=sandbox_id))
+
+
+def test_provider_resolves_api_key_from_environment(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The provider can read `E2B_API_KEY` without an explicit constructor key."""
+    monkeypatch.setenv("E2B_API_KEY", "env-key")
+    sandbox = _sandbox()
+
+    with patch(
+        "langchain_e2b.provider.e2b.Sandbox.create",
+        return_value=sandbox,
+    ) as create:
+        provider = E2BProvider()
+        backend = provider.get_or_create()
+
+    create.assert_called_once_with(
+        timeout=DEFAULT_SANDBOX_TIMEOUT,
+        api_key="env-key",
+    )
+    assert isinstance(backend, E2BSandbox)
+    assert backend.id == "sbx-test"
+
+
+def test_provider_requires_api_key() -> None:
+    """A missing API key should fail before any SDK call is attempted."""
+    with pytest.raises(ValueError, match="No E2B API key found"):
+        E2BProvider(resolve_env_var=_resolver({}))
+
+
+def test_provider_creates_sandbox_with_default_options() -> None:
+    """Creating a sandbox uses the package-owned lifecycle implementation."""
+    sandbox = _sandbox("sbx-new")
+
+    with patch(
+        "langchain_e2b.provider.e2b.Sandbox.create",
+        return_value=sandbox,
+    ) as create:
+        provider = E2BProvider(api_key="fake-key", resolve_env_var=_resolver({}))
+        backend = provider.get_or_create(timeout=2)
+
+    create.assert_called_once_with(
+        timeout=DEFAULT_SANDBOX_TIMEOUT,
+        api_key="fake-key",
+    )
+    assert backend.id == "sbx-new"
+
+
+def test_provider_creates_sandbox_with_template_and_timeout() -> None:
+    """Template and sandbox lifetime are configured by E2B environment values."""
+    sandbox = _sandbox("sbx-template")
+    resolve_env_var = _resolver(
+        {
+            "E2B_TEMPLATE": "custom-template",
+            "E2B_SANDBOX_TIMEOUT": "7200",
+        }
+    )
+
+    with patch(
+        "langchain_e2b.provider.e2b.Sandbox.create",
+        return_value=sandbox,
+    ) as create:
+        provider = E2BProvider(api_key="fake-key", resolve_env_var=resolve_env_var)
+        backend = provider.get_or_create()
+
+    create.assert_called_once_with(
+        template="custom-template",
+        timeout=7200,
+        api_key="fake-key",
+    )
+    assert backend.id == "sbx-template"
+
+
+@pytest.mark.parametrize("value", ["not-a-number", "0", "-1"])
+def test_provider_rejects_invalid_sandbox_timeout(value: str) -> None:
+    """Timeout configuration should fail with a clear error."""
+    resolve_env_var = _resolver({"E2B_SANDBOX_TIMEOUT": value})
+
+    with pytest.raises(ValueError, match="E2B_SANDBOX_TIMEOUT"):
+        E2BProvider(api_key="fake-key", resolve_env_var=resolve_env_var)
+
+
+def test_provider_connects_existing_sandbox() -> None:
+    """Providing `sandbox_id` reconnects instead of creating a new sandbox."""
+    sandbox = _sandbox("sbx-existing")
+
+    with patch(
+        "langchain_e2b.provider.e2b.Sandbox.connect",
+        return_value=sandbox,
+    ) as connect:
+        provider = E2BProvider(api_key="fake-key", resolve_env_var=_resolver({}))
+        backend = provider.get_or_create(sandbox_id="sbx-existing")
+
+    connect.assert_called_once_with(
+        "sbx-existing",
+        timeout=DEFAULT_SANDBOX_TIMEOUT,
+        api_key="fake-key",
+    )
+    assert backend.id == "sbx-existing"
+
+
+def test_provider_translates_missing_sandbox_to_key_error() -> None:
+    """Deep Agents adapters can map missing sandboxes without importing E2B SDK."""
+    with patch(
+        "langchain_e2b.provider.e2b.Sandbox.connect",
+        side_effect=e2b.SandboxNotFoundException("missing"),
+    ):
+        provider = E2BProvider(api_key="fake-key", resolve_env_var=_resolver({}))
+
+        with pytest.raises(KeyError) as exc_info:
+            provider.get_or_create(sandbox_id="sbx-missing")
+
+    assert exc_info.value.args == ("sbx-missing",)
+
+
+def test_provider_rejects_unsupported_get_options() -> None:
+    """Unsupported lifecycle options fail in the package provider."""
+    provider = E2BProvider(api_key="fake-key", resolve_env_var=_resolver({}))
+
+    with pytest.raises(TypeError, match="Received unsupported arguments: metadata"):
+        provider.get_or_create(metadata={"purpose": "test"})
+
+
+def test_provider_deletes_sandbox() -> None:
+    """Delete delegates to the E2B SDK lifecycle operation."""
+    with patch("langchain_e2b.provider.e2b.Sandbox.kill") as kill:
+        provider = E2BProvider(api_key="fake-key", resolve_env_var=_resolver({}))
+        provider.delete(sandbox_id="sbx-delete")
+
+    kill.assert_called_once_with("sbx-delete", api_key="fake-key")
+
+
+def test_provider_rejects_unsupported_delete_options() -> None:
+    """Delete only accepts the sandbox ID."""
+    provider = E2BProvider(api_key="fake-key", resolve_env_var=_resolver({}))
+
+    with pytest.raises(TypeError, match="Received unsupported arguments: force"):
+        provider.delete(sandbox_id="sbx-delete", force=True)