flash-sandbox 0.1.6

A Python SDK for interacting with the Sandbox Orchestrator.

Sandbox SDK

A Python SDK for interacting with the Sandbox Orchestrator. Supports both gRPC and HTTP transports with an identical public interface, so you can swap protocols without changing application logic.

The SDK offers two layers of abstraction:

• Low-level clients (HTTPClient, AsyncHTTPClient, SandboxClient) – manage sandboxes via client methods. HTTPClient and AsyncHTTPClient 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 grpcio, protobuf, 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
sandbox = client.start_sandbox(
    type="docker",
    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.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()}")

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

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

gRPC transport

The gRPC SandboxClient returns raw string IDs (not Sandbox objects):

from flash_sandbox import SandboxClient

client = SandboxClient(host="localhost", port=50051)

# Start a Docker sandbox
docker_id = client.start_sandbox(
    type="docker",
    image="alpine:latest",
    command=["tail", "-f", "/dev/null"],
    memory_mb=128,
    cpu_cores=0.5,
)
print(f"Docker Sandbox ID: {docker_id}")

# Start a Firecracker sandbox
fc_id = client.start_sandbox(
    type="firecracker",
    image="alpine:latest",
    command=["tail", "-f", "/dev/null"],
    memory_mb=512,
    cpu_cores=1.0,
)
print(f"Firecracker Sandbox ID: {fc_id}")

# Start a gVisor sandbox
gvisor_id = client.start_sandbox(
    type="gvisor",
    image="alpine:latest",
    command=["tail", "-f", "/dev/null"],
    memory_mb=256,
    cpu_cores=1.0,
)
print(f"gVisor Sandbox ID: {gvisor_id}")

# Check status
status = client.get_status(fc_id)
print(f"Firecracker Status: {status}")

# Snapshot and Resume
snap_res = client.snapshot_sandbox(fc_id)
print(f"Snapshot Data: {snap_res}")
client.resume_sandbox(fc_id)

# Execute commands
exec_res = client.exec_command(gvisor_id, ["echo", "Hello from gVisor!"])
print(f"gVisor Output: {exec_res.stdout.strip()}")
print(f"gVisor Exit Code: {exec_res.exit_code}")

# Stop sandboxes
client.stop_sandbox(docker_id)
client.stop_sandbox(fc_id)
client.stop_sandbox(gvisor_id)
client.close()

Tip: You can wrap a gRPC client's string IDs with Sandbox / AsyncSandbox manually:

from flash_sandbox import SandboxClient, Sandbox
client = SandboxClient(host="localhost", port=50051)
sid = client.start_sandbox(type="docker", image="alpine:latest")
sandbox = Sandbox(sid, client)
sandbox.exec_command(["echo", "hello"])

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 are transport-agnostic: they work with HTTPClient, AsyncHTTPClient, or SandboxClient.

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.
snapshot()	Create a snapshot. Returns {"snapshot_path": ..., "mem_file_path": ...}.
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, AsyncHTTPClient, and SandboxClient expose the same set of methods. AsyncHTTPClient methods must be awaited.

Note: HTTPClient.start_sandbox returns a Sandbox and AsyncHTTPClient.start_sandbox returns an AsyncSandbox. The gRPC SandboxClient.start_sandbox returns a plain string ID.

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(type, image, command, memory_mb, cpu_cores, ...)	Start a new sandbox. HTTP clients return a Sandbox/AsyncSandbox; gRPC returns a string ID.
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 {"snapshot_path": ..., "mem_file_path": ...}. 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.
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
)

SandboxClient (gRPC)

SandboxClient(
    host="localhost",       # Orchestrator hostname
    port=50051,             # gRPC port (default 50051)
    address=None,           # Full address for proxy routing (e.g. "localhost:8092/v1/service/sandbox")
)

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.

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

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 through a proxy
http_client = HTTPClient(address="http://proxy.example.com:8092/v1/service/sandbox")

# gRPC through a proxy
grpc_client = SandboxClient(address="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/