flash-sandbox 0.2.8

A Python SDK for interacting with the Sandbox Orchestrator.

Sandbox SDK

A Python SDK for interacting with the Sandbox Orchestrator over HTTP.

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

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

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

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

Method	Description
get_status()	Return the status string ("running", "stopped", etc.).
exec_command(command)	Execute a command. Returns an object with stdout, stderr, exit_code.
run_python(code)	Execute Python code inside the sandbox. Returns stdout.
get_platform_info()	Get platform information. Returns a dict with keys like system, node, release, etc.
get_metrics()	Return point-in-time resource-usage metrics.
read_file(path)	Read a file from the sandbox as bytes.
write_file(path, content)	Write bytes to a path in the sandbox.
snapshot()	Create a snapshot. Returns a SnapshotResult.
resume()	Resume a paused / snapshotted sandbox.
stop(cleanup=True)	Stop the sandbox. When cleanup=True (default), backing resources are removed. Pass cleanup=False to stop without removing resources.

Properties

Property	Description
id	The raw sandbox ID string.

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.

Method	Description
start_sandbox(image, command, memory_mb, cpu_cores, type="", ...)	Start a new sandbox. Returns a Sandbox/AsyncSandbox. type is optional — omit it (or pass "auto") to let the cluster pick any available backend; set it ("docker", "kubernetes", …) to force one.
stop_sandbox(sandbox_id, cleanup=True)	Stop a sandbox. When cleanup=True (default), backing resources are removed. Pass cleanup=False to keep resources for inspection. Accepts SandboxID.
exec_command(sandbox_id, command)	Execute a command in a sandbox. Returns an object with stdout, stderr, and exit_code. Accepts SandboxID.
get_status(sandbox_id)	Return the status string ("running", "stopped", etc.). Accepts SandboxID.
get_metrics(sandbox_id)	Return point-in-time resource-usage metrics (memory, CPU, network, block I/O). Accepts SandboxID.
snapshot_sandbox(sandbox_id)	Create a snapshot. Returns a SnapshotResult. Accepts SandboxID.
resume_sandbox(sandbox_id)	Resume a paused / snapshotted sandbox. Accepts SandboxID.
run_python(sandbox_id, code)	Execute arbitrary Python code inside the sandbox. Returns the stdout output. Accepts SandboxID.
get_platform_info(sandbox_id)	Get platform information from the sandbox. Returns a dict with platform data. Accepts SandboxID.
read_file(sandbox_id, path)	Read a file from the sandbox as bytes. Accepts SandboxID.
write_file(sandbox_id, path, content)	Write bytes to a path in the sandbox. Accepts SandboxID.
close()	Release the underlying connection / session.

Constructor options

HTTPClient

HTTPClient(
    host="localhost",       # Orchestrator hostname
    port=8080,              # 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
)

AsyncHTTPClient

AsyncHTTPClient(
    host="localhost",       # Orchestrator hostname
    port=8080,              # HTTP port (default 8080)
    address=None,           # Full URL, overrides host/port
    timeout=30.0,           # Request timeout in seconds
    session=None,           # Optional aiohttp.ClientSession
)

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)

Class	Fields
ExecResult	stdout: str, stderr: str, exit_code: int
MetricsResult	memory_usage_bytes, memory_limit_bytes, memory_percent, cpu_percent, pids_current, net_rx_bytes, net_tx_bytes, block_read_bytes, block_write_bytes
SnapshotResult	snapshot_path: str, mem_file_path: str, node_id: str

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 (or standalone node) with -H/--address or the FLASH_SANDBOX_ADDR env var (default localhost:8080):

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 orchestrator's HTTP API only. For host-level containerd administration on a node (images/snapshots/content cleanup), use the flash-ctr script.