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).

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 requests and aiohttp automatically.

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")

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)

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.

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 (calls start_sandbox, then wraps if needed)
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 requests.Session for custom TLS / auth / retries
    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 aiohttp.ClientSession
    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 requests.Session for connection pooling, mutual TLS, retry policies, or authentication headers.
• Typed exceptions: SandboxHTTPError (with .status_code and .detail) and the more specific SandboxNotFoundError for 404 responses.
• 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.

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.