Agent Sandbox Registry

python-sdk

SDK Usage Guide

docs/en/guide/sdk-usage.md

SDK Usage Guide

This document covers the complete usage of the Agent Sandbox Python SDK.


Installation

pip install agent-sandbox

Import

from agent_sandbox.api.sandbox import Sandbox

Create and Destroy

Async Creation

sandbox = await Sandbox.create(
    template="base",      # Template name (default "base")
    timeout=300,           # Timeout in seconds (1–86400, default 300)
    envs={"KEY": "val"},   # Injected environment variables
    metadata={"team": "x"},# Arbitrary metadata key-value pairs
    cpu=2,                 # CPU cores (optional)
    memory=4096,           # Memory in MB (optional)
    disk=10240,            # Disk in MB (optional)
    gpu="A10",             # GPU spec (optional)
)

Sync Creation

sandbox = Sandbox.create_sync(template="base")

Automatically calls kill() to destroy the sandbox when exiting the with block:

async with await Sandbox.create(template="base") as sandbox:
    result = await sandbox.run_code("print(1+1)")
    print(result.text)  # "2"
# Automatically destroyed

Connect to an Existing Sandbox

sandbox = await Sandbox.connect("sbx-xxxx")

# Sync version
sandbox = Sandbox.connect_sync("sbx-xxxx")

Destroy a Sandbox

# Instance method
await sandbox.kill()

# Class method (destroy by ID, no need to connect first)
await Sandbox.kill_by_id("sbx-xxxx")

# Sync version
Sandbox.kill_by_id_sync("sbx-xxxx")

List Sandboxes

sandboxes = await Sandbox.list(status=SandboxStatus.RUNNING, limit=20)
for sb in sandboxes:
    print(f"{sb.sandbox_id} - {sb.template} - {sb.status.value}")

Execute Code — run_code

Execute code via the Code Interpreter:

result = await sandbox.run_code(
    "print('hello')",
    language="python",   # Default "python"
    timeout=30,          # Timeout in seconds (default 30)
    envs={"KEY": "val"}, # Temporary env vars for this execution (envd scope)
)

print(result.text)           # Output text (stdout with trailing whitespace stripped)
print(result.stdout)         # Raw stdout
print(result.stderr)         # Raw stderr
print(result.exit_code)      # Exit code
print(result.execution_time) # Execution time (seconds)
print(result.output_files)   # Output file list [OutputFile]

Callback support:

result = await sandbox.run_code(
    "for i in range(3): print(i)",
    on_stdout=lambda s: print(f"[OUT] {s}"),
    on_stderr=lambda s: print(f"[ERR] {s}"),
)

Capability requirement: Requires the code capability (included in the default capability set {shell, files, code}).


Execute Commands — commands

The sandbox.commands submodule provides shell command execution capabilities.

Execute and Wait

result = await sandbox.commands.run(
    "ls -la /home/user",
    timeout=60,                # Timeout in seconds (default 60)
    env={"MY_VAR": "value"},   # Additional env vars for this command (envd scope)
    cwd="/app",                # Working directory
)

print(result.stdout)
print(result.stderr)
print(result.exit_code)
print(result.execution_time)
print(result.success)  # exit_code == 0

Streaming Output

async for chunk in sandbox.commands.stream("pip install flask"):
    if chunk.type.value == "stdout":
        print(chunk.data, end="")
    elif chunk.type.value == "stderr":
        print(chunk.data, end="", file=sys.stderr)
    elif chunk.type.value == "exit":
        print(f"\nExit code: {chunk.exit_code}")

Background Execution

reader = await sandbox.commands.start("python server.py")
# reader is a StreamReader that can be consumed asynchronously

You can also use the background=True parameter of run:

reader = await sandbox.commands.run("sleep 60", background=True)

Process Management

# List processes
processes = await sandbox.commands.list()

# Kill a process
await sandbox.commands.kill(pid=12345)

# Send a signal: envd delivers only SIGTERM (15) and SIGKILL (9);
# any other number (for example 2, SIGINT) raises ValueError
await sandbox.commands.send_signal(pid=12345, signal=15)  # SIGTERM
await sandbox.commands.send_signal(pid=12345, signal=9)   # SIGKILL

# Send stdin data
await sandbox.commands.send_input(pid=12345, data="yes\n")

Sync Versions

All methods have synchronous counterparts with a _sync suffix:

result = sandbox.commands.run_sync("echo hello")

File Operations — files

The sandbox.files submodule provides filesystem operations.

Read/Write

# Write text or bytes
await sandbox.files.write("/home/user/hello.txt", "Hello World")
await sandbox.files.write("/home/user/data.bin", b"\x00\x01\x02")

# Read text
text = await sandbox.files.read("/home/user/hello.txt", encoding="utf-8")

# Read bytes
data = await sandbox.files.read_bytes("/home/user/data.bin")

Upload / Download

# Upload a local file to the sandbox
await sandbox.files.upload("./local_file.py", "/home/user/remote.py")

# Download a sandbox file to local
await sandbox.files.download("/home/user/output.csv", "./output.csv")

Directory Operations

# List directory contents
entries = await sandbox.files.list("/home/user")
for entry in entries:
    print(f"{entry.name} - {entry.type.value} - {entry.size}")

# Create a directory (including parents)
await sandbox.files.make_dir("/home/user/new/nested/dir")

# Check if a path exists
exists = await sandbox.files.exists("/home/user/hello.txt")

# Get file info
info = await sandbox.files.get_info("/home/user/hello.txt")

Move / Delete

# Move/rename
await sandbox.files.move("/home/user/old.txt", "/home/user/new.txt")

# Delete a file or directory
await sandbox.files.remove("/home/user/temp.txt")

Watch File Changes

watcher = await sandbox.files.watch("/home/user")
async for event in watcher:
    print(f"{event.type}: {event.path}")

Network — network

The sandbox.network submodule computes port access URLs (purely local computation, no network requests involved).

Capability requirement: Requires the ports capability. The default capability set does not include ports; it must be explicitly declared in the template's template.yaml.

# Get port URL
url = sandbox.network.get_url(8080)
# => "https://8080-sbx-xxxx.cn-hangzhou.e2b.fc.aliyuncs.com"

# Get host
host = sandbox.network.get_host(8080)
# => "8080-sbx-xxxx.cn-hangzhou.e2b.fc.aliyuncs.com"

# Get access headers (includes X-Access-Token in secure mode)
headers = sandbox.network.get_access_headers()

Sandbox Lifecycle

# Check if running (makes an API call)
running = await sandbox.is_running()

# Refresh sandbox info
info = await sandbox.refresh_info()

# Set timeout
await sandbox.set_timeout(600)

# Pause/resume
await sandbox.pause()
await sandbox.resume()

Properties

sandbox.id           # Sandbox ID (str)
sandbox.status       # Current status (SandboxStatus)
sandbox.url          # envd URL (str)
sandbox.info         # Full info (SandboxInfo)
sandbox.capabilities # Capability set (frozenset[str])

Sync vs Async

The SDK uses an async-first design — all core methods are asynchronous. Each async method has a corresponding synchronous wrapper (generated via make_sync):

Async Method Sync Method
Sandbox.create() Sandbox.create_sync()
Sandbox.connect() Sandbox.connect_sync()
sandbox.kill() sandbox.kill_sync()
sandbox.run_code() sandbox.run_code_sync()
sandbox.commands.run() sandbox.commands.run_sync()
sandbox.files.read() sandbox.files.read_sync()
sandbox.files.write() sandbox.files.write_sync()

Sync methods run an event loop in the background and are suitable for scripts and non-async environments.


Error Handling

All SDK errors inherit from SandboxError and carry code, message, and suggestion attributes:

from agent_sandbox.models.errors import (
    SandboxError,
    InvalidAPIKeyError,       # E1001
    TemplateNotFoundError,    # E2001
    SandboxNotFoundError,     # E2009
    CapabilityNotSupportedError,  # E3004
    FileNotFoundError_,       # E4001
    ConnectionError_,         # E5001
)

try:
    sandbox = await Sandbox.create(template="nonexistent")
except TemplateNotFoundError as e:
    print(f"[{e.code}] {e.message}")
    print(f"Suggestion: {e.suggestion}")
except SandboxError as e:
    print(f"SDK error: {e}")

See Error Codes Reference for details.


Named Commands

There is one mechanism for named commands: functions registered in the sandbox server (@sandbox.register, or the server's CommandRegistry). sandbox.run(name, **kwargs) calls one of them (POST /commands/{name}), and sandbox.list_commands() lists them:

commands = await sandbox.list_commands()
for cmd in commands:
    print(f"{cmd.name}: {cmd.description}")

result = await sandbox.run("dev", port=8080)   # the registered function's return value

An unknown name raises CommandNotFoundError; argument or execution errors raise ServerAPIError. The sandbox image must run agent-sandbox-server (see below). For a bare shell command use sandbox.commands.run(cmd); it never goes through the registry.


Template Server — server

Templates that ship an agent-sandbox-server expose a second, template-authored data plane (port 9000). sandbox.server is its typed client. Use it when the template restricts what is reachable (per-endpoint policy) or when you need endpoints envd does not have (search, archive, file watching, process output/stdin, git, browser).

async with await Sandbox.create(template="my-server-template") as sandbox:
    server = sandbox.server                        # or sandbox.server_at(9000, token="...")
    caps = await server.discover()                 # explicit; enables fail-fast pre-checks
    if caps.can("files_download_stream"):
        data = await server.files.download("/home/user/out/report.csv")
    await server.files.upload("/home/user/in/data.json", b"{}")

    result = await server.process.shell("ls -la", cwd="/home/user")
    async for event in server.process.shell_stream("make test"):
        print(event.event, event.data)

    print(server.files.list_sync("/home/user"))    # every coroutine has a *_sync twin

Starting a server in a running sandbox (the image must contain agent-sandbox-server):

server = await sandbox.start_server(policy="read-only", token="s3cret", commands="/home/user/app/commands.py")

Restricting endpoints (template author)

from agent_sandbox.server import SandboxServer

policy = {
    "preset": "transfer-only",                      # only upload + download endpoints, size-capped
    "endpoints": {
        "files_download_stream": {"paths": {"allow": ["/home/user/out"], "deny": ["**/.env"]},
                                  "limits": {"max_bytes": "5MB"}},
        "files_upload_stream":   {"paths": {"allow": ["/home/user/in"]}, "params": {"overwrite": False}},
    },
}
SandboxServer(policy=policy).serve(9000)   # a dict or a ServerPolicy

or declaratively in template.yaml (server: block, see the template spec), or agent-sandbox-server serve --policy server-policy.json. Validate with agent-sandbox-server check-config.

Handling denials

from agent_sandbox.models.errors import (
    EndpointNotAvailableError, FilePolicyDeniedError, PayloadTooLargeError, RateLimitedError, ServerAPIError,
)

try:
    await server.files.download("/etc/passwd")
except FilePolicyDeniedError as exc:            # also a PermissionDeniedError
    print(exc.error_code, exc.endpoint)         # path_not_allowed files_download_stream
except EndpointNotAvailableError:               # disabled by policy (or unknown to an older server)
    ...
except RateLimitedError as exc:
    await asyncio.sleep(exc.retry_after or 1)
except ServerAPIError as exc:                   # base class: status_code, error_type, error_code, endpoint
    ...

The server is always authoritative: after discover() the client merely fails fast with the same exception without a round trip. Policy changes are picked up with server.forget_discovery() + discover().


Environment Variables

Environment variables can be injected at creation (envs=) or per-call (env=/envs=). Commands containing unquoted shell operators (pipes |, semicolons ;, logical operators &&/||, redirects >/<, subshells (...), command substitution $(...)) are automatically wrapped in sh -c by commands.run(), commands.stream(), and commands.start(). Variable expansion ($VAR), backtick substitution (`cmd`), and glob patterns (*, ?) still require explicit sh -c '...' wrapping, or use printenv VAR to read a single variable.

For full details on scopes (envd vs SandboxServer), injection patterns, and security considerations, see the dedicated Environment Variables Guide.


Agent workflow pattern

This is the canonical pattern for AI Agent scenarios — create a sandbox, run task code, collect the result, then destroy it:

import asyncio
from agent_sandbox.api.sandbox import Sandbox

async def agent_workflow(task_code: str) -> str:
    """Agent → create sandbox → exec code → read result → kill sandbox"""
    async with await Sandbox.create(template="my-app") as sandbox:
        # Write the task code
        await sandbox.files.write("/app/task.py", task_code)

        # Execute
        result = await sandbox.commands.run("python3 /app/task.py")

        # Read the output
        if result.exit_code == 0:
            try:
                output = await sandbox.files.read("/app/output.txt")
                return output
            except Exception:
                return result.stdout
        else:
            return f"Error (exit {result.exit_code}): {result.stderr}"

# Usage
result = asyncio.run(agent_workflow("""
import json
data = {"status": "done", "items": [1, 2, 3]}
with open("/app/output.txt", "w") as f:
    json.dump(data, f)
print("Task completed")
"""))
print(result)