Sandbox SDK

A Python SDK for interacting with Flash Sandbox over HTTP. It talks to the coordinator (cluster binary) — point host/port (or address) at your coordinator, not at a bare node (the SDK uses the coordinator's /sandboxes API). Opt-in scheduler mode routes sandbox operations through the external scheduler.

The SDK offers two layers of abstraction:

• Low-level clients (HTTPClient, AsyncHTTPClient) – manage sandboxes via client methods and return Sandbox / AsyncSandbox objects from start_sandbox, so you can call methods directly on the result.
• High-level wrappers (Sandbox, AsyncSandbox) – bind a single sandbox instance so you never need to pass the ID around.

Installation

pip install flash-sandbox

The package pulls in httpx2 automatically (a single library that provides both the sync and async HTTP transports).

HTTP transport

The SDK requires Python 3.10 or later. Both clients use httpx2; callers may supply their own sync or async session for TLS, authentication, and pooling. Caller-supplied sessions remain caller-owned and are not closed by the SDK.

Requests carry an explicit RequestPolicy: operation lane, replayability, admission deadline, and observation context. Reads and fenced creates can retry transient failures. An unfenced mutation is retried only when connection establishment proves no request was dispatched. A lost response or interrupted stream never causes an exec to run again. Streaming requires a terminal exit frame with an integer exit code. Empty, truncated, or malformed streams raise SandboxHTTPError with code execution_uncertain; server error frames preserve their typed error envelope and raise instead of appearing as successful EOF.

All HTTP and transport failures expose SandboxHTTPError.code, .origin, and .outcome alongside .status_code and .detail. SandboxNotFoundError means worker-confirmed sandbox absence; a coordinator's missing route is a distinct route_not_found error. Unknown outcomes remain uncertain. Transport timeouts also retain the SandboxTimeoutError type.

Use client.configure_transport(request_hooks=..., response_hooks=...) for observation hooks. Hooks follow the sync or async httpx2 callback contract and apply to every request pool. Advanced calls use client.request(method, path, policy=RequestPolicy(...)), which returns an httpx2.Response in both clients. Configure hooks before starting concurrent requests.

Scheduler routing

Both clients keep coordinator-direct routing by default. To use the scheduler, set with_scheduler=True and an explicit scheduler_address:

client = HTTPClient(
    address="http://127.0.0.1:8080",
    with_scheduler=True,
    scheduler_address="http://127.0.0.1:8090",
)

All sandbox operations use the scheduler endpoint, including creation, execution, snapshot, and restore. AsyncHTTPClient accepts the same routing options. The scheduler applies its configured admission, priority, and residency policies; the SDK retains its existing sandbox methods. There is no automatic coordinator fallback if the scheduler is unavailable.

Run the scheduler's live proxy as described in the trainer contract. Use RLVRClient at the same scheduler address to declare rollout groups and pass execution_metadata(group, rollout_id) to sandbox execution when needed.

Quick Start

Sync usage (recommended for scripts)

HTTPClient.start_sandbox() returns a Sandbox object. You can call methods directly on it, or access the raw ID via .id:

from flash_sandbox import HTTPClient

client = HTTPClient(host="localhost", port=8080)

# start_sandbox returns a Sandbox object.
# No `type` → the cluster picks any available backend (Docker, Kubernetes, …).
# Pass type="docker" (etc.) to force a specific backend.
sandbox = client.start_sandbox(
    image="alpine:latest",
    command=["tail", "-f", "/dev/null"],
    memory_mb=128,
    cpu_cores=0.5,
)
print(f"Sandbox ID: {sandbox.id}")

# Call methods directly on the sandbox — no need to pass an ID
print(f"Status: {sandbox.get_status()}")

result = sandbox.exec_command(["echo", "Hello!"])
print(f"Output: {result.stdout.strip()}")
print(f"Exit: {result.exit_code}, server time: {result.duration_ms:.1f}ms")

# Optional `metadata` is forwarded uninterpreted to the exec trace /
# event log so execs can be classified downstream (see "Classifying
# execs" below). The backend ignores it; it doesn't change execution.
result = sandbox.exec_command(
    ["echo", "verified"],
    metadata={"query_type": "verification", "task_id": 7},
)

py_out = sandbox.run_python("print('Hello Python in Sandbox')")
print(f"Python Output: {py_out.strip()}")

details = sandbox.get_info()
print(f"Scheduled on: {details.node_id} ({details.backend})")

plat_info = sandbox.get_platform_info()
print(f"Platform: {plat_info['system']} {plat_info.get('release', '')}")

metrics = sandbox.get_metrics()
print(f"Memory: {metrics.memory_usage_bytes} / {metrics.memory_limit_bytes}")
print(f"CPU: {metrics.cpu_percent}%")

snap = sandbox.snapshot()
print(f"Snapshot: {snap.snapshot_path}")
sandbox.resume()

sandbox.write_file("/tmp/hello.txt", b"Hello from SDK!")
content = sandbox.read_file("/tmp/hello.txt")
print(f"File Content: {content.decode()}")

sandbox.stop()            # cleanup=True by default: removes the container
# sandbox.stop(cleanup=False)  # stop without removing resources
client.close()

Pre-warm a future batch with a claimable ticket:

ticket = client.demand(
    "trainer",
    "batch-43",
    image="python:3.11-slim",
    backend="libkrun",
    expect=next_batch_at,  # RFC3339 string or timezone-aware datetime
    count=8,
    cpu=2,
    memory=1024,
)

sandbox = client.claim(ticket)
cleanup = client.cancel(ticket)  # cleans up anything still unclaimed

claim(ticket) waits for one prepared sandbox, returns a normal Sandbox, and can be repeated while ticket.remaining_claims is nonzero. Retrying with the same live ticket safely replays an ambiguous claim. Use the lower-level claim_warm_demand(..., claim_id=...) API when the claim ID must survive a client process restart. The async client exposes the same methods with await. Cancelling a ticket does not stop sandboxes that were already claimed.

Use as a context manager for automatic cleanup:

from flash_sandbox import HTTPClient, Sandbox

client = HTTPClient(host="localhost", port=8080)

with Sandbox.create(
    client,
    type="docker",
    image="alpine:latest",
    command=["tail", "-f", "/dev/null"],
) as sandbox:
    result = sandbox.exec_command(["echo", "Hello!"])
    print(result.stdout.strip())

# sandbox.stop() is called automatically on exit
client.close()

You can also wrap an existing sandbox ID:

from flash_sandbox import HTTPClient, Sandbox

client = HTTPClient(host="localhost")
sandbox = Sandbox("existing-sandbox-id", client)
print(sandbox.get_status())

Async usage

AsyncHTTPClient.start_sandbox() returns an AsyncSandbox object. Call await on its methods:

import asyncio
from flash_sandbox import AsyncHTTPClient

async def main():
    client = AsyncHTTPClient(host="localhost", port=8080)

    # start_sandbox returns an AsyncSandbox object
    sandbox = await client.start_sandbox(
        type="docker",
        image="alpine:latest",
        command=["tail", "-f", "/dev/null"],
    )
    print(f"Sandbox ID: {sandbox.id}")

    result = await sandbox.exec_command(["echo", "Hello from Async!"])
    print(f"Output: {result.stdout.strip()}")

    await sandbox.write_file("/tmp/async.txt", b"Async hello!")
    content = await sandbox.read_file("/tmp/async.txt")
    print(f"Async File Content: {content.decode()}")

    await sandbox.stop()
    await client.close()

asyncio.run(main())

Class-based pattern

When the client lives as an instance attribute and sandbox start/stop happen across different methods:

from flash_sandbox import AsyncHTTPClient

class MyAgent:
    def __init__(self, orchestrator_host: str = "localhost"):
        self.sandbox_client = AsyncHTTPClient(host=orchestrator_host, port=8080)

    async def forward(self, prompt: str, metadata: dict):
        # start_sandbox returns an AsyncSandbox object
        sandbox = await self.sandbox_client.start_sandbox(
            type="docker",
            image=metadata["image_name"],
            command=["tail", "-f", "/dev/null"],
        )

        result = await sandbox.exec_command(["echo", "Hello!"])
        print(f"Output: {result.stdout.strip()}")

        platform_info = await sandbox.get_platform_info()
        print(f"Platform: {platform_info['system']} {platform_info.get('release', '')}")

        await sandbox.stop()

    async def close(self):
        await self.sandbox_client.close()

Context manager pattern

Use async with for automatic cleanup:

import asyncio
from flash_sandbox import AsyncHTTPClient, AsyncSandbox

async def main():
    async with AsyncHTTPClient(host="localhost", port=8080) as client:
        async with await AsyncSandbox.create(
            client,
            type="docker",
            image="alpine:latest",
            command=["tail", "-f", "/dev/null"],
        ) as sandbox:
            status = await sandbox.get_status()
            print(f"Status: {status}")

            result = await sandbox.exec_command(["echo", "Hello!"])
            print(f"Output: {result.stdout.strip()}")

        # sandbox.stop() is called automatically

asyncio.run(main())

API Reference

Sandbox / AsyncSandbox (high-level)

The Sandbox class wraps a sandbox ID and a client, exposing all operations as simple method calls. AsyncSandbox is the async equivalent — all I/O methods are coroutines.

Both work with HTTPClient and AsyncHTTPClient.

Construction

# HTTP clients return Sandbox / AsyncSandbox directly from start_sandbox
sandbox = client.start_sandbox(type="docker", image="alpine:latest", ...)
sandbox = await async_client.start_sandbox(type="docker", image="alpine:latest", ...)

# Or wrap an existing sandbox ID (works with any client)
sandbox = Sandbox(sandbox_id, client)
sandbox = AsyncSandbox(sandbox_id, async_client)

# Or use the create() factory (reuses the object returned by start_sandbox)
sandbox = Sandbox.create(client, type="docker", image="alpine:latest", ...)
sandbox = await AsyncSandbox.create(async_client, type="docker", image="alpine:latest", ...)

Methods

Properties

Context manager

Sandbox supports with and AsyncSandbox supports async with. The sandbox is automatically stopped when the block exits:

with Sandbox.create(client, type="docker", image="alpine:latest") as sandbox:
    sandbox.exec_command(["echo", "hi"])
# sandbox.stop(cleanup=True) is called automatically on exit

async with await AsyncSandbox.create(client, type="docker", image="alpine:latest") as sandbox:
    await sandbox.exec_command(["echo", "hi"])
# await sandbox.stop(cleanup=True) is called automatically on exit

Low-level client methods

HTTPClient and AsyncHTTPClient expose the same set of methods. AsyncHTTPClient methods must be awaited.

SandboxID — flexible sandbox identifiers

Every method that accepts a sandbox_id parameter uses the SandboxID type, which is Union[str, Sandbox, AsyncSandbox]. This means you can pass either a plain string ID or a Sandbox / AsyncSandbox object — they are fully interchangeable:

sandbox = client.start_sandbox(type="docker", image="alpine:latest")

# All three forms are equivalent:
client.get_status(sandbox)            # pass the Sandbox object
client.get_status(sandbox.id)         # pass the .id string
client.get_status("sb-abc123")        # pass any string ID

This makes it easy to mix high-level wrapper usage with low-level client calls during migration or when you need both styles in the same codebase.

Constructor options

HTTPClient

HTTPClient(
    host="localhost",       # Coordinator hostname — or a full URL (see below)
    port=8080,              # Coordinator HTTP port (default 8080)
    address=None,           # Full URL, overrides host/port (e.g. "http://proxy:9090/v1/service/sandbox")
    timeout=30.0,           # Request timeout in seconds (None = no timeout)
    session=None,           # Optional httpx2.Client for custom TLS / auth / retries
    pool_size=128,          # Max connections (and keepalive) per client
    api_key=None,           # Sent as "Authorization: Bearer <key>"; defaults to $FLASH_API_KEY
)

AsyncHTTPClient

AsyncHTTPClient(
    host="localhost",       # Coordinator hostname — or a full URL (see below)
    port=8080,              # Coordinator HTTP port (default 8080)
    address=None,           # Full URL, overrides host/port
    timeout=30.0,           # Request timeout in seconds
    session=None,           # Optional httpx2.AsyncClient
    pool_size=128,          # Max connections (and keepalive) per client
    api_key=None,           # Sent as "Authorization: Bearer <key>"; defaults to $FLASH_API_KEY
)

Connecting with just a URL

The first positional argument accepts a full URL, so a hosted coordinator needs nothing else:

client = HTTPClient("https://sandbox.example.com")
client = HTTPClient("localhost:8080")
client = HTTPClient("http://proxy:9090/v1/service/sandbox")

A value is treated as a URL when it carries a scheme, a path, a bracketed IPv6 authority, or an explicit port; a bare hostname (HTTPClient("localhost")) still pairs with port. When the first argument is a URL, port is ignored — the URL is the single source of truth. The explicit address= keyword keeps working and wins over host.

Authentication

Against a coordinator started with --api-key, pass the key as api_key (or set $FLASH_API_KEY); it is sent as Authorization: Bearer <key> on every request:

client = HTTPClient("https://sandbox.example.com", api_key="s3cret")

The orchestrator also accepts the key as an X-API-Key header, which is what the admin tooling under meta/tools/ uses.

HTTPClient-only extras

The HTTP transport includes a few additional features:

• Firecracker fields on start_sandbox: kernel_image, initrd_path, snapshot_path, mem_file_path.
• Custom timeouts per-client via the timeout parameter.
• Session injection – pass your own httpx2.Client / httpx2.AsyncClient for connection pooling, mutual TLS, retry policies, or authentication headers.
• Typed exceptions: SandboxHTTPError preserves .code, .origin, and .outcome; SandboxNotFoundError is reserved for worker-confirmed absence.
• Auto-expiry via ttl_seconds on start_sandbox — see below.

Auto-expiry (ttl_seconds)

Pass ttl_seconds=N to start_sandbox to cap the sandbox's wall-clock lifetime. The node-side reaper deletes the sandbox once the deadline passes, even if the caller exits without calling stop_sandbox. Useful as a safety net for batch jobs and benchmarks that can leak sandboxes on crash.

# Sandbox is auto-deleted ~10 minutes after creation, even if this
# process is killed before reaching stop().
sandbox = client.start_sandbox(
    type="kubernetes",
    image="alpine:latest",
    ttl_seconds=600,
)

Currently enforced by the kubernetes backend only; other backends accept the field and silently ignore it. The deadline is stamped on the pod as the flash-sandbox/ttl-expires-at annotation, so it survives orchestrator restarts.

Classifying execs (metadata)

exec_command (sync and async, on both HTTPClient/AsyncHTTPClient and Sandbox/AsyncSandbox) accepts an optional metadata mapping. It is opaque to the backend — backends ignore it and never change the command based on it — and is forwarded verbatim onto the exec trace (.jsonl, on the start and finish events), the structured event log (container_execs, under the client key), and the raw request body captured by FSB_DEBUG. This lets you classify execs downstream (e.g. interactive vs. verification, by task id) without encoding semantics into the command:

# verification query
sb.exec_command(["pytest", "-x"],
                metadata={"query_type": "verification", "task_id": 7})

# interactive query
sb.exec_command(["git", "status"],
                metadata={"query_type": "interactive", "task_id": 7})

# async is identical
await sb.exec_command(["pytest"],
                      metadata={"query_type": "verification", "task_id": 7})

# streaming variant honors metadata too
for frame in sb.exec_command(["pytest", "-v"], stream=True,
                             metadata={"query_type": "verification"}):
    ...

Values must be JSON-serializable. Omit metadata (the default) to keep the wire body minimal. To attribute CPU to a query type by bracketing an exec with get_metrics() samples, see docs/guides/exec-cpu-attribution.md.

Receiving metadata in Go

On the server side, metadata is a plain map[string]any that flows through four layers. None of them interpret it — the field is read from the request and forwarded onto the downstream captures.

To add a new exec-like endpoint that also receives metadata, decode the same field into a pebble.ExecOpts and pass it to Manager.Exec:

var req struct {
    Command  []string       `json:"command"`
    Metadata map[string]any `json:"metadata,omitempty"`
}
// ... decode req ...
res, err := m.Exec(ctx, id, req.Command, "my_op",
    pebble.ExecOpts{Metadata: req.Metadata})

For execs dispatched outside Manager.Exec (the streaming path), the coordinator records the same bag: Manager.RecordExecTerminal(id, metadata, valid) applies RLVR terminal-outcome accounting, and the per-exec metadata lands in the coordinator's api_calls.metadata column when FSB_DEBUG=1 (the former node-side TraceExecStart JSONL trace has been removed).

Response types (HTTP client)

All response dataclasses are frozen (immutable).

Context manager

All clients support the context-manager protocol (with for sync, async with for async):

from flash_sandbox import HTTPClient

with HTTPClient(host="localhost") as client:
    sandbox = client.start_sandbox(type="docker", image="alpine:latest")
    sandbox.exec_command(["echo", "hello"])
    sandbox.stop()
# Connection is automatically closed here.

Proxy / reverse-proxy support

Both clients support routing through a reverse proxy that uses path-based routing. Pass the full address including the path prefix:

http_client = HTTPClient(address="http://proxy.example.com:8092/v1/service/sandbox")

The Sandbox / AsyncSandbox wrappers inherit proxy support from whichever client you pass in.

Running tests

pip install -e ".[dev]"
pytest tests/

CLI (fsb)

Installing flash-sandbox also installs the fsb command.

pip install flash-sandbox

Point it at a coordinator with -H/--address or the FLASH_SANDBOX_ADDR env var (default localhost:8080). Like the SDK, fsb uses the coordinator's /sandboxes API, not a bare node's /native/*:

export FLASH_SANDBOX_ADDR=coordinator.example:8080

fsb ps                       # list running sandboxes
fsb inspect sb-a1b2c3        # status + platform info
fsb nodes                    # cluster nodes
fsb templates                # registered templates
fsb status                   # cluster summary

fsb create --template python # create from a template (or --image IMG)
fsb stop sb-a1b2c3
fsb snapshot sb-a1b2c3
fsb resume sb-a1b2c3

fsb exec sb-a1b2c3 -- ls -la /workspace   # exits with the remote exit code
fsb metrics sb-a1b2c3
fsb cp ./local.txt sb-a1b2c3:/tmp/x       # upload
fsb cp sb-a1b2c3:/tmp/x ./out.txt         # download

Add --json (before the subcommand) for machine-readable output:

fsb --json ps | jq '.[].id'

fsb speaks the coordinator's HTTP API only. For host-level containerd administration on a node (images/snapshots/content cleanup), use the containerd CLI (ctr) or the Docker CLI directly.