sandbox-agents

Python client for the Sandbox Agents control-plane — agents that run in sandboxed containers, conversations against them, and the event log everything is observed through.

pip install sandbox-agents

Requires Python 3.11 or newer. Sync and async clients, typed models, resumable event streaming, client-tool dispatch and webhook verification are all included; httpx and pydantic are the only dependencies.

The model in three sentences

An agent is a reusable definition: instructions, a model, the manifest its sandboxes start from. A session is one conversation and the container it owns — the workspace, the ports, the snapshots. A run is one turn, and it is asynchronous: sending a message returns a run id, and the answer arrives over the session's event log.

Quickstart

from sandbox_agents import Client

client = Client(base_url="http://localhost:8000", api_key="ak_…")

agent = client.agents.create(
    slug="renewal-analyst",
    name="Renewal analyst",
    instructions="Inspect the files before answering. Cite the source of every claim.",
    model="gpt-5.4-mini",
    manifest={
        "entries": {
            "brief.md": {"kind": "file", "content": "# Northwind\n- Renewal: 2026-04-15\n"},
            "output": {"kind": "dir"},
        }
    },
)

session = client.sessions.create_and_wait(agent.slug, title="Northwind renewal")
result = client.sessions.run(session.id, "Write output/report.md listing every blocker.")

print(result.text)                                   # the model's answer
print(client.sessions.read_text(session.id, "output/report.md"))  # what it wrote
client.sessions.stop(session.id)

create_and_wait and run are the two waiting helpers: the first returns when the container is up, the second when the turn is over. Both follow the event log rather than holding a request open, so neither is affected by a proxy's idle timeout.

Configuration

client = Client()                       # entirely from the environment
client = Client(project="acme")          # X-Project, by id or slug

project can be left out where the answer is unambiguous — a deployment with one project, or an API key bound to one. With several, a request that does not name one is refused rather than guessed at.

Keep one client for the process. It owns a connection pool, and it remembers where it is in each session's log — which is what lets run start from now instead of paging the whole conversation.

Async

The same surface, awaited. Prefer it for anything following more than one conversation: a turn is minutes of mostly waiting.

import asyncio
from sandbox_agents import AsyncClient

async def main() -> None:
    async with AsyncClient() as client:
        session = await client.sessions.create_and_wait("renewal-analyst")
        async for event in client.sessions.stream(session.id):
            if event.type == "agent.text_delta":
                print(event.text, end="", flush=True)

asyncio.run(main())

Streaming

sessions.stream yields the log from a cursor and then live events. It resumes on its own: a dropped connection reconnects from the last seq seen, and because that cursor is a database sequence it survives a server restart too.

for event in client.sessions.stream(session.id, after=cursor):
    if event.type == "agent.message":
        print(event.text)
    elif event.type == "agent.tool_use":
        print(f"→ {event.tool}")
    elif event.type in ("session.status_idle", "session.status_error"):
        break

Token deltas (agent.text_delta) are streamed but never stored, so they carry seq == 0 and never move the cursor — event.is_persistent is the check. sandbox_agents.events has every type name as a constant, plus TERMINAL_TYPES and EPHEMERAL_TYPES.

To render a transcript first and then follow it, page the log and stream from where the page ended:

history = list(client.sessions.history(session.id))
for event in client.sessions.stream(session.id, after=history[-1].seq):
    ...

Client tools

An agent can declare tools it does not implement. When the model calls one, the turn stops and waits for the caller to answer — that is how an agent in a container reaches the page a user is looking at.

from sandbox_agents import Tool

def open_account(args: dict) -> dict:
    return {"id": args["id"], "status": "active"}

result = client.sessions.run(
    session.id,
    "Look up account 42 and summarise it.",
    tools=[
        Tool(
            name="open_account",
            handler=open_account,
            description="Read an account by id",
            parameters={"type": "object", "properties": {"id": {"type": "string"}}},
        )
    ],
)

A Tool is declared for that turn and answered by it. For tools already declared on the agent, pass handlers by name instead: tools={"open_account": open_account}.

A handler that raises does not fail the run: the exception is reported to the model as the tool's output, because the turn is blocked on this answer and losing the conversation over one failed lookup is worse. Handlers may be async def on the async client.

Files, shell, ports

client.sessions.upload_file(session.id, "brief.pdf")            # → uploads/brief.pdf
client.sessions.list_files(session.id, "output")
client.sessions.download_file(session.id, "output/report.md", "./report.md")
client.sessions.exec(session.id, "ls -la output").check()
client.sessions.port(session.id, 8000).url                       # a preview URL

Snapshots

snapshot = client.sessions.create_snapshot(session.id, label="before-refactor")
fork = client.sessions.create_and_wait("renewal-analyst", from_snapshot_id=snapshot.id)

Webhooks

For consumers that are not connected when something happens. Bodies are signed with HMAC-SHA256 over "<timestamp>.<body>"; verify against the raw bytes, before anything parses the JSON.

from fastapi import FastAPI, Request, Response
from sandbox_agents import webhooks

app = FastAPI()

@app.post("/hooks/agents")
async def receive(request: Request) -> Response:
    raw = await request.body()
    if not webhooks.verify(SECRET, raw, request.headers.get(webhooks.SIGNATURE_HEADER)):
        return Response(status_code=401)
    delivery = webhooks.parse(raw)
    print(delivery.type, delivery.event.text if delivery.event else "")
    return Response(status_code=204)

Delivery is at-least-once; X-Anecdote-Delivery is stable across retries and is what makes a consumer idempotent.

Errors

from sandbox_agents import ConflictError, NotFoundError, RunFailedError

try:
    agent = client.agents.update(agent.id, version=agent.version, name="New name")
except ConflictError:
    agent = client.agents.get(agent.id)   # somebody edited it first; re-apply

APIStatusError and its subclasses (BadRequestError, AuthenticationError, PermissionDeniedError, NotFoundError, ConflictError, UnprocessableEntityError, RateLimitError, InternalServerError) carry status_code and detail. APIConnectionError and APITimeoutError mean no answer arrived. RunFailedError is a turn that ended in session.status_error; TimeoutExpiredError is a waiting helper giving up on something still running, and carries last_seq so the wait can be resumed.

Safe methods and 429s are retried with jittered backoff (max_retries=2). A POST is not: one that timed out may already have created the session.

Partial updates: None means clear

The control-plane distinguishes a field that was not sent from one sent as null, so this client does too. Anything you do not pass is left alone; None clears.

client.agents.update(ref, version=7, image=None)   # clears the image override
client.agents.update(ref, version=7, name="x")     # touches nothing else

Development

poetry install
poetry run pytest
poetry run ruff check . && poetry run mypy src

The full API reference — every resource, with the event catalogue and the webhook payloads — is in the control-plane UI under Developers → Python SDK, alongside the OpenAPI schema at /docs on the API itself.