feat: add E2B sandbox provider with explicit recovery

This commit is contained in:
Ashpreet
2026-10-11 18:59:24 +01:00
parent 5d6c47446f
commit 9e28b44391
6 changed files with 442 additions and 0 deletions
+33
View File
@@ -156,3 +156,36 @@ customer-operated trusted teams; it is not a hostile multitenant security claim.
The provider adapter must never auto-resume a stopped execution during inspection.
Local acceptance exercises the helper with real Linux processes through a Docker
test bridge; each hosted provider still requires live acceptance.
### E2B (live acceptance required)
Install `agno[e2b-sandbox]` with Python 3.10+ (SDK pinned to 2.55.1). Set
`E2B_API_KEY` on the service and prepare an E2B template from a published image
containing the hosted runtime and native SDKs. For example, with the installed
E2B SDK: `Template.build(Template().from_image(image), alias=alias, cpu_count=2,
memory_mb=2048)`. Building a template and running the following command can incur
provider charges. Do not put service credentials in the image or template.
```bash
python cookbook/frameworks/sandbox/hosted_workload.py --provider e2b --image YOUR_TEMPLATE_ID
```
Set `AGNO_TEST_SANDBOX_DB_URL` to a dedicated Postgres database. The workload
creates two successive computers, clones/edits/tests/runs a local application,
checkpoints it, destroys compute, and verifies restoration/application restart.
It uses no model calls. Test allocation tables are retained for reconciliation
rather than deleting uncertain intents during cleanup.
For AgentOS wiring, create `E2BSandboxProvider(namespace, journal)` and
`HostedCheckpointTransfer(provider, object_store)`, then pass them to
`SandboxCoordinator` and `SandboxAgent` as in the Docker example. Set the agent's
`pids_limit=None`, `workspace_mb=None`, image to the template ID, and CPU/memory to
the template's actual resources. Set `network="internet"` explicitly for remote
model/MCP calls. The account/template must support the configured lifetime.
E2B transport uses a read-only SDK endpoint to retrieve the secure connection
information; it does not call `connect`, which can resume paused computers.
Allocation and control HTTP retries are disabled. Expiry kills compute and the
next turn restores from the last completed checkpoint. SDK contract tests run
without network; live account, template, filesystem permissions and both native
harnesses still need acceptance.
+10
View File
@@ -173,3 +173,13 @@ Python 3.9 follow-up: 76 sandbox unit tests, 40 real Postgres cases and 12 Agent
**Result:** 6 real Linux/Postgres cases passed across sync/async database adapters; hosted contract tests passed within the sandbox unit suite. No hosted platform or native-model calls were made.
---
### E2B installed-SDK contract
**Status:** PASS
**Description:** Five tests against E2B 2.55.1 with network endpoints replaced: read-only secure transport construction, paused-compute refusal, create settings, SDK inventory serialization/pagination, and bounded stream cleanup.
**Result:** 5 passed. Format/validation passed. `hosted_workload.py --provider e2b` is prepared but has not run against an E2B account; template permissions, quotas, networking, and native harness behavior remain live acceptance gates.
---
@@ -0,0 +1,90 @@
"""Opt-in hosted coding/checkpoint acceptance. This creates billable compute.
Run only after preparing a provider template and configuring a dedicated test
Postgres database. No model calls are made; native harness acceptance is separate.
"""
import argparse
import os
from dataclasses import replace
from pathlib import Path
from tempfile import TemporaryDirectory
from uuid import uuid4
from agno.db.postgres import PostgresDb
from agno.sandbox import SandboxSpec
from agno.sandbox.allocation import PostgresAllocationJournal
from agno.sandbox.checkpoint import FileCheckpointStore
from agno.sandbox.hosted_transfer import HostedCheckpointTransfer
from agno.sandbox.postgres import PostgresSandboxStore
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--provider", choices=("e2b",), required=True)
parser.add_argument(
"--image",
required=True,
help="Prepared provider template/snapshot/image identifier",
)
args = parser.parse_args()
from agno.sandbox.e2b import E2BSandboxProvider
namespace = "acceptance-" + uuid4().hex[:12]
db = PostgresDb(
db_url=os.environ["AGNO_TEST_SANDBOX_DB_URL"],
session_table=namespace.replace("-", "_") + "_sessions",
runs_table=namespace.replace("-", "_") + "_runs",
)
store = PostgresSandboxStore(
db, table_name=namespace.replace("-", "_") + "_bindings"
)
store.setup()
journal = PostgresAllocationJournal(
store, table_name=namespace.replace("-", "_") + "_allocations"
)
journal.setup()
provider = E2BSandboxProvider(namespace, journal, max_allocations=2, lifetime=600)
spec = SandboxSpec(
"coder", "acceptance", 1, args.image, pids_limit=None, workspace_mb=None
)
script = Path(__file__).with_name("workload_fixture.py").read_text()
handles = []
try:
with TemporaryDirectory() as directory:
transfer = HostedCheckpointTransfer(
provider, FileCheckpointStore(Path(directory) / "objects")
)
first = provider.create(spec)
handles.append(first)
result = provider.exec(
first, ["python", "-c", script, "prepare"], timeout=90
)
assert result.exit_code == 0, result.stderr
print(result.stdout.strip())
checkpoint = transfer.capture(
first, state={"fixture": "hosted-workload-v1"}
)
provider.destroy(first)
second = provider.create(replace(spec, generation=2))
handles.append(second)
assert transfer.restore(second, checkpoint) == {
"fixture": "hosted-workload-v1"
}
result = provider.exec(
second, ["python", "-c", script, "verify"], timeout=30
)
assert result.exit_code == 0, result.stderr
print(result.stdout.strip())
print("Verified checkpoint:", checkpoint.sha256)
finally:
for handle in handles:
provider.destroy(handle)
# Keep allocation records for ambiguous-create reconciliation. This
# cookbook never deletes an uncertain intent merely to tidy test data.
print("Inspect test allocation table before removal:", journal.table.name)
db.db_engine.dispose()
if __name__ == "__main__":
main()
+180
View File
@@ -0,0 +1,180 @@
"""E2B adapter for prepared templates, with no implicit resume or create retries."""
import shlex
from typing import Any, Dict, Optional, Sequence
from agno.sandbox.allocation import PostgresAllocationJournal
from agno.sandbox.base import ExecResult, SandboxError, SandboxSpec
from agno.sandbox.checkpoint import _thread
from agno.sandbox.hosted import HostedComputer, HostedSandboxProvider
class E2BSandboxProvider(HostedSandboxProvider):
"""Use an E2B template containing the root-owned Agno hosted runtime.
Requires Python 3.10+ and the optional ``e2b-sandbox`` extra. CPU and memory
must match the prepared template; disk/PID limits use provider defaults.
API credentials stay on the service. E2B lifetime expiry kills compute;
paused computers are never automatically resumed by this adapter.
"""
backend = "e2b"
def __init__(
self,
namespace: str,
journal: PostgresAllocationJournal,
*,
api_key: Optional[str] = None,
max_allocations: int = 10,
lifetime: int = 1800,
request_timeout: float = 30,
) -> None:
super().__init__(namespace, journal, max_allocations=max_allocations, lifetime=lifetime)
try:
import e2b
except ImportError as error:
raise ImportError("Install agno[e2b-sandbox] with Python 3.10+ to use E2B sandboxes") from error
if request_timeout <= 0:
raise ValueError("E2B request timeout must be positive")
self._sdk = e2b
self._options: Dict[str, Any] = dict(request_timeout=request_timeout, retries=0)
if api_key is not None:
self._options["api_key"] = api_key
def _detail(self, provider_ref: str) -> Any:
# The public connect operation resumes paused sandboxes. This read-only
# generated endpoint retains the envd token omitted by public get_info,
# allowing transport construction without any lifecycle mutation.
from e2b.api import handle_api_exception
from e2b.api.client.api.sandboxes import get_sandboxes_sandbox_id
from e2b.api.client.models import SandboxDetail
from e2b.api.client_sync import get_api_client
from e2b.connection_config import ConnectionConfig
response = get_sandboxes_sandbox_id.sync_detailed(
provider_ref, client=get_api_client(ConnectionConfig(**self._options))
)
if response.status_code == 404:
return None
if response.status_code >= 300:
raise handle_api_exception(response)
if not isinstance(response.parsed, SandboxDetail):
raise SandboxError("E2B returned an invalid sandbox detail")
return response.parsed
@staticmethod
def _computer(detail: Any) -> HostedComputer:
metadata = detail.metadata
if not isinstance(metadata, dict):
raise SandboxError("E2B sandbox is missing ownership metadata")
return HostedComputer(detail.sandbox_id, metadata, str(detail.state))
def _transport(self, provider_ref: str) -> Any:
from e2b.connection_config import ConnectionConfig
from packaging.version import Version
detail = self._detail(provider_ref)
if detail is None or str(detail.state) != "running":
raise SandboxError("E2B computer is not running; implicit resume is disabled")
token = detail.envd_access_token
if not isinstance(token, str) or not token:
raise SandboxError("E2B secure transport token is unavailable")
headers = {
"E2b-Sandbox-Id": provider_ref,
"E2b-Sandbox-Port": str(ConnectionConfig.envd_port),
"X-Access-Token": token,
}
return self._sdk.Sandbox(
sandbox_id=provider_ref,
sandbox_domain=detail.domain if isinstance(detail.domain, str) else None,
envd_version=Version(detail.envd_version),
envd_access_token=token,
connection_config=ConnectionConfig(extra_sandbox_headers=headers, **self._options),
)
async def _create(self, spec: SandboxSpec, metadata: Dict[str, str]) -> HostedComputer:
def create() -> HostedComputer:
sandbox = self._sdk.Sandbox.create(
template=spec.image,
timeout=self.lifetime,
metadata=metadata,
envs=dict(spec.env),
secure=True,
allow_internet_access=spec.network == "internet",
lifecycle={"on_timeout": "kill", "auto_resume": False},
**self._options,
)
# The creation response already supplies the authoritative immutable
# ID. Binding must survive failure of a subsequent detail lookup.
return HostedComputer(sandbox.sandbox_id, metadata, "running")
return await _thread(create)
async def _check_resources(self, spec: SandboxSpec, computer: HostedComputer) -> None:
detail = await _thread(self._detail, computer.provider_ref)
if detail is None or detail.cpu_count != spec.cpus or detail.memory_mb != spec.memory_mb:
raise SandboxError("E2B template CPU/memory differs from the requested specification")
if detail.allow_internet_access is not (spec.network == "internet"):
raise SandboxError("E2B network policy differs from the requested specification")
async def _get(self, provider_ref: str) -> Optional[HostedComputer]:
detail = await _thread(self._detail, provider_ref)
return None if detail is None else self._computer(detail)
async def _inventory(self, token: str) -> Sequence[HostedComputer]:
def inventory() -> Sequence[HostedComputer]:
from e2b.api.client.models import SandboxState
from e2b.sandbox.sandbox_api import SandboxQuery
paginator = self._sdk.Sandbox.list(
query=SandboxQuery(
metadata={"namespace": self.namespace, "token": token},
state=[SandboxState.RUNNING, SandboxState.PAUSED],
),
**self._options,
)
computers = []
while paginator.has_next:
for info in paginator.next_items():
computers.append(HostedComputer(info.sandbox_id, info.metadata, str(info.state)))
return computers
return await _thread(inventory)
async def _delete(self, provider_ref: str) -> None:
await _thread(self._sdk.Sandbox.kill, sandbox_id=provider_ref, **self._options)
async def _command(self, provider_ref: str, argv: Sequence[str], *, timeout: float) -> ExecResult:
def command() -> ExecResult:
from e2b.sandbox.commands.command_handle import CommandExitException
sandbox = self._transport(provider_ref)
try:
result = sandbox.commands.run(
shlex.join(argv), user="root", timeout=timeout, request_timeout=timeout + 5
)
except CommandExitException as error:
result = error
return ExecResult(result.exit_code, result.stdout, result.stderr)
return await _thread(command)
async def _write(self, provider_ref: str, path: str, payload: bytes) -> None:
await _thread(lambda: self._transport(provider_ref).files.write(path, payload, user="root"))
async def _read(self, provider_ref: str, path: str, *, max_bytes: int) -> bytes:
def read() -> bytes:
sandbox = self._transport(provider_ref)
if sandbox.files.get_info(path, user="root").size > max_bytes:
raise SandboxError("E2B transfer exceeds its byte limit")
chunks, size = [], 0
with sandbox.files.read(path, format="stream", user="root", request_timeout=120) as stream:
for chunk in stream:
size += len(chunk)
if size > max_bytes:
raise SandboxError("E2B transfer exceeds its byte limit")
chunks.append(chunk)
return b"".join(chunks)
return await _thread(read)
+2
View File
@@ -40,6 +40,7 @@ dependencies = [
]
[project.optional-dependencies]
e2b-sandbox = ["e2b==2.55.1"]
dev = [
"mypy==2.1.0",
"ruff==0.15.20",
@@ -586,6 +587,7 @@ module = [
"elasticsearch.*",
"email_validator.*",
"e2b_code_interpreter.*",
"e2b.*",
"ipykernel.*",
"jupyter_client.*",
"eth_account.*",
+127
View File
@@ -0,0 +1,127 @@
"""Installed-SDK contract checks; network endpoints are replaced explicitly."""
import datetime
from types import SimpleNamespace
import pytest
pytest.importorskip("e2b")
from e2b.api.client.models import SandboxDetail, SandboxState
from agno.sandbox.base import SandboxError, SandboxSpec
from agno.sandbox.e2b import E2BSandboxProvider
from tests.unit.sandbox.test_hosted import Journal
def provider():
return E2BSandboxProvider("tests", Journal(), api_key="e2b_test-key")
def detail(**kwargs):
values = dict(
template_id="template",
sandbox_id="computer",
client_id="client",
started_at=datetime.datetime.now(datetime.timezone.utc),
end_at=datetime.datetime.now(datetime.timezone.utc),
envd_version="0.5.6",
cpu_count=2,
memory_mb=2048,
disk_size_mb=1024,
state=SandboxState.RUNNING,
envd_access_token="transport-only",
metadata={},
allow_internet_access=False,
)
values.update(kwargs)
return SandboxDetail(**values)
def test_read_only_detail_and_transport_never_connect(monkeypatch):
from e2b.api.client.api.sandboxes import get_sandboxes_sandbox_id
calls = []
monkeypatch.setattr(
get_sandboxes_sandbox_id,
"sync_detailed",
lambda ref, **kwargs: calls.append(ref) or SimpleNamespace(status_code=200, parsed=detail()),
)
value = provider()
monkeypatch.setattr(value._sdk.Sandbox, "connect", lambda *a, **k: pytest.fail("connect can resume compute"))
sandbox = value._transport("computer")
assert sandbox.sandbox_id == "computer"
assert sandbox.connection_config.retries == 0
assert calls == ["computer"]
def test_paused_transport_is_rejected_without_resuming(monkeypatch):
value = provider()
monkeypatch.setattr(value, "_detail", lambda ref: detail(state=SandboxState.PAUSED))
with pytest.raises(SandboxError, match="implicit resume"):
value._transport("computer")
@pytest.mark.asyncio
async def test_create_config_disables_retry_resume_and_inherited_credentials(monkeypatch):
value = provider()
calls = []
monkeypatch.setenv("AGNO_DB_URL", "not-forwarded")
monkeypatch.setattr(
value._sdk.Sandbox, "create", lambda **kwargs: calls.append(kwargs) or SimpleNamespace(sandbox_id="computer")
)
spec = SandboxSpec("a", "s", 1, "template", pids_limit=None, workspace_mb=None, env={"MODEL_KEY": "explicit"})
result = await value._create(spec, {"token": "intent"})
assert result.provider_ref == "computer"
assert calls[0]["retries"] == 0 and calls[0]["secure"] is True
assert calls[0]["lifecycle"] == {"on_timeout": "kill", "auto_resume": False}
assert calls[0]["envs"] == {"MODEL_KEY": "explicit"}
assert calls[0]["allow_internet_access"] is False
@pytest.mark.asyncio
async def test_inventory_uses_sdk_query_and_visits_all_pages(monkeypatch):
from e2b.api.client.api.sandboxes import get_v2_sandboxes
from e2b.sandbox.sandbox_api import SandboxQuery
value = provider()
calls = []
class Pages:
has_next = True
index = 0
def next_items(self):
self.index += 1
self.has_next = self.index < 2
return [
SimpleNamespace(sandbox_id=str(self.index), metadata={"token": "intent"}, state=SandboxState.RUNNING)
]
def listing(*, query, **kwargs):
assert isinstance(query, SandboxQuery)
# Exercise the actual SDK request serializer, including enum states.
serialized = get_v2_sandboxes._get_kwargs(state=query.state)
assert serialized["params"]["state"] == "running,paused"
calls.append((query.metadata, kwargs))
return Pages()
monkeypatch.setattr(value._sdk.Sandbox, "list", listing)
assert len(await value._inventory("intent")) == 2
assert calls[0][0] == {"namespace": "e2b:tests", "token": "intent"}
assert calls[0][1]["retries"] == 0
@pytest.mark.asyncio
async def test_stream_bound_closes_reader(monkeypatch):
import httpx
from e2b.sandbox.filesystem.filesystem import FileStreamReader
value = provider()
response = httpx.Response(200, content=b"too-large")
stream = FileStreamReader(response)
files = SimpleNamespace(get_info=lambda *a, **k: SimpleNamespace(size=1), read=lambda *a, **k: stream)
monkeypatch.setattr(value, "_transport", lambda ref: SimpleNamespace(files=files))
with pytest.raises(SandboxError, match="byte limit"):
await value._read("computer", "root-object", max_bytes=2)
assert stream._closed