Sandbox

The Sandbox class is the entry point of cua-sandbox: create, connect to, or reattach a sandbox, then use its interfaces.

Local is the default: Sandbox.create(image) runs on this machine, local=False or cloud=CloudOptions(...) in the cloud, and every other parameter means the same on both. Sandbox.connect(name) and the class methods take local=None: the local sandbox of that name when one exists, else the cloud.

Sandboxes#

Sandbox#

class Sandbox
 
Sandbox(transport: Transport, name: Optional[str] = None)

A sandboxed computer environment.

Provides programmatic control of a VM or container through a unified interface: .mouse, .keyboard, .screen, .clipboard, .shell, .window, and .terminal.

Sandboxes are always isolated: they never control the host machine directly. To control the local machine, use cua-driver.

There are three ways to obtain a Sandbox:

1. Persistent: provision and keep alive after the script exits:

sb = await Sandbox.create(Image.linux())
await sb.shell.run("whoami")
await sb.disconnect()

1. Connect: attach to an already-running sandbox by name:

sb = await Sandbox.connect("my-sandbox")
await sb.screenshot()
await sb.disconnect()

1. Ephemeral: auto-destroyed when the async with block exits:

async with Sandbox.ephemeral(Image.linux()) as sb:
    await sb.shell.run("whoami")

Sandbox.disconnect#

async def disconnect() -> None

Drop the transport connection. The sandbox keeps running.

A claim on a managed Fleet pool stops being renewed and lives until its shutdown time (claim_ttl, or later if keep_alive was called).

Sandbox.overlay#

async def overlay(
    overlay: 'dict[str, Any] | list',
    timeout: Optional[float] = None,
) -> list

Injects binaries into this sandbox (see create(overlay=...)): {"cua-driver": "./target/release/cua-driver"}. Returns one cua.OverlayResult per binary (name, target, sha256, previous_sha256, restarted); pass NAME=sha256:<hex> to cua doctor --expect. timeout (seconds) bounds a cua-spacesd restart.

Sandbox.info#

async def info() -> SandboxInfo

Portable info: status (provisioning, starting, ready, stopped), location (local, cloud or direct), services (name -> guest port), expires_at (cloud) and provider_details.

Sandbox.service#

def service(name: str) -> 'ServiceHandle'

A named service (services={"mcp": 8765} at create): request, url and public_url, the same local and in the cloud.

Sandbox.public_url#

async def public_url(
    service: str,
    *,
    ttl: float = 3600,
    label: Optional[str] = None,
) -> PublicUrl

A shareable URL for service that stops working after ttl seconds (60 s to 24 h). Cloud: a signed service URL. Local: a loopback URL with its own token, served by the cua daemon (started if needed).

Sandbox.revoke_public_url#

async def revoke_public_url(url: 'PublicUrl | str') -> None

Revokes a URL from public_url (or its id).

Sandbox.to_dict#

def to_dict() -> dict[str, Any]

Serialize a durable Fleet sandbox reference.

Sandbox.from_dict#

@classmethod
def from_dict(data: dict[str, Any]) -> '_ConnectResult'

Reconnect to a serialized Fleet claim reference.

Sandbox.keep_alive#

async def keep_alive(*, minutes: float) -> None

Push a Fleet claim's controller-enforced shutdown time forward.

Sandbox.close#

async def close() -> None

Release this Fleet claim; repeated calls are safe.

Sandbox.snapshot#

async def snapshot(name: str | None = None, stateful: bool = False) -> 'Image'

Snapshot this sandbox's current state. Returns an Image.

The returned Image can be passed to Sandbox.create() or Sandbox.ephemeral() to boot a new sandbox from the snapshot (COW fork: instant on btrfs).

Returns: An Image with _snapshot_source set, ready to pass to Sandbox.ephemeral().

Sandbox.destroy#

async def destroy() -> None

Disconnect and permanently delete the sandbox (VM/container).

Sandbox.screenshot#

async def screenshot(
    text: Optional[str] = None,
    format: str = 'png',
    quality: int = 95,
) -> bytes

Sandbox.screenshot_base64#

async def screenshot_base64(
    text: Optional[str] = None,
    format: str = 'png',
    quality: int = 95,
) -> str

Sandbox.get_environment#

async def get_environment() -> str

Sandbox.get_display_url#

async def get_display_url(*, share: bool = False) -> str

Return a URL to view this sandbox's display.

With cua-spacesd this is a link to its HTML5 viewer, carrying a scoped viewer ticket that expires. Images without cua-spacesd return the page of a declared web display service, or a VNC address.

Sandbox.viewer_url#

async def viewer_url() -> str

A browser link to this sandbox's desktop in the cua-spacesd HTML5 viewer (video, audio, input, clipboard, files). The link carries a scoped viewer ticket that expires after an hour. Same as get_display_url() for sandboxes with cua-spacesd.

Sandbox.get_dimensions#

async def get_dimensions() -> tuple[int, int]

Sandbox.mcp#

def mcp(service: str, *, path: str = '/mcp') -> Any

An official MCP SDK client for the MCP server behind service.

An async context manager (pip install "cua-sandbox[mcp]"):

async with sb.mcp("mcp") as client:
    tools = await client.list_tools()
    result = await client.call_tool("add", {"a": 2, "b": 3})

Requests go through the sandbox's service route (loopback locally, the Fleet gateway in the cloud), streamed both ways, so every protocol feature and content block works unchanged. No cua-spacesd needed.

Sandbox.mcp_config#

async def mcp_config(service: str, *, path: str = '/mcp') -> dict

{"url": ..., "headers": {...}} of service's MCP endpoint, for any MCP client (Claude Code, Cursor, other SDKs). The headers can carry a short-lived Fleet bearer: fetch a fresh config per connection.

Sandbox.spacesd#

async def spacesd() -> Any

The raw cua.SpacesdClient for this sandbox's cua-spacesd.

The typed escape hatch to every cua.env.v1 RPC (call_json), process streaming, chunked transfers and media sessions. Raises cua_sandbox._sdk.SpacesdNotAvailable when the sandbox has no spacesd (for example a plain VNC-only image).

Sandbox.create#

@classmethod
async def create(
    image: Image | None = None,
    *,
    pool: 'Pool | str | None' = None,
    name: Optional[str] = None,
    replicas: int = 1,
    service: str = ENV_SERVICE,
    claim_spec: Any = None,
    keep_alive_minutes: float | None = None,
    api_key: Optional[str] = None,
    on: Optional[str] = None,
    local: Optional[bool] = None,
    kind: Optional[str] = None,
    runtime: 'Runtime | str | None' = None,
    cpu: Optional[int] = None,
    memory_mb: Optional[int] = None,
    disk_gb: Optional[int] = None,
    region: str = 'us-east-1',
    time_to_start: Optional[float] = None,
    request_timeout: Optional[float] = None,
    server_port: Optional[int] = None,
    telemetry_enabled: bool = True,
    warm: Optional[bool] = None,
    max_pool_size: Optional[int] = None,
    claim_ttl: 'float | timedelta | None' = None,
    progress: Optional[Callable[[Any], Any]] = None,
    command: Optional[list[str]] = None,
    sidecars: Optional[list] = None,
    env: Optional[dict[str, str]] = None,
    services: Optional[dict[str, int]] = None,
    wait_for: WaitFor = None,
    cloud: Optional[CloudOptions] = None,
    memory: 'str | int | None' = None,
    network: Optional[str] = None,
    overlay: 'dict[str, Any] | list | None' = None,
) -> 'Sandbox'

Provision or claim a persistent sandbox and return it connected.

Where, what kind and which engine are three separate choices:

• on: "local" or "cloud". local=True/False is the same switch; on and a contradicting local is an InvalidArgument, and so is local=True with cloud=. Unset, cloud= implies the cloud; a cloud-only argument (a Fleet pool, warm, max_pool_size, claim_ttl, claim_spec, api_key, ...) still implies it with a DeprecationWarning; otherwise the user default applies (CUA_DEFAULT_ON, else cua config set default.on cloud in $CUA_HOME/config.toml, else local, with a one-time notice that CUA_QUIET_DEFAULT=1 hides).
• kind: "auto", "container" or "vm". Unset: the image's kind when it has one, else default.kind, else auto (macOS and Windows images are VMs, a container rootfs is a container, a disk-only image is a VM).
• runtime: the engine, "auto" or one the location offers for the kind: locally "gvisor"/"runc" (containers) and "qemu"/"lume" (VMs); in the cloud "gvisor" and "kubevirt". Unset: default.runtime, else auto. A local sandbox with sidecars needs "runc" where gVisor would run (separate gVisor containers cannot share a network namespace). A Runtime object (DockerRuntime(), TartRuntime(), ...) still picks a local adapter, as before.

A combination that does not exist (kind="container", runtime="qemu", or runtime="kubevirt" locally) raises InvalidPlacement, listing the valid values.

The same options mean the same thing local and in the cloud:

• command: argv replacing the image's entrypoint.

• env: environment variables.

• services: named guest ports, {"mcp": 8765}; reach them with sb.service("mcp"), sb.public_url("mcp") or sb.tunnel.forward(8765).

• wait_for: tcp("mcp") / http("mcp", "/health") (or a list); readiness otherwise is "the sandbox is running".

• cloud: CloudOptions (warm capacity, limits, TTL, a dedicated pool). The flat pool, warm, max_pool_size and claim_ttl keywords are deprecated aliases.

• memory: "4GB" / "512MB" (or memory_mb).

• network: "default" (outbound network, like a Docker container) or "none" (no egress; the SDK still reaches the guest's published ports). "none" needs a local QEMU VM; containers, Lume and cloud sandboxes raise Unsupported.

• overlay: binaries to inject once it is up, so tests run the build under test and not the copy the image bundles: {"cua-driver": "./target/release/cua-driver"} (cua-driver and cua-spacesd resolve their guest path; other names take (path, "/guest/path")). Each replaces the guest file atomically, is recorded with its sha256 (cua doctor --expect) and what runs it is restarted. If it fails, the sandbox is deleted and the error raised. See overlay.

Supplying pool claims from an existing Fleet pool without changing its configuration.

A Fleet registry image without pool is claimed from this account's managed pool for that image spec (cua-auto-*), created on first use and reused afterwards. The first start of an image can take a few minutes; later starts reuse its capacity. Fleet-only options:

• warm: seed one ready replica when the pool is first created.
• max_pool_size: autoscaling ceiling (default 10).
• claim_ttl: seconds (or a timedelta) the claim outlives this process; it is renewed while the process holds it (default 15 min). Use keep_alive(minutes=...) to keep a sandbox longer.
• progress: callback receiving AcquireProgress events.

Without on/local, passing one of them keeps the sandbox in the cloud (deprecated: pass local=False); with local=True they are ignored. An existing pool keeps its own kind and runtime (Pool.apply(..., runtime=)).

server_port names a port your image serves itself: it is exposed (Fleet: as the server service) and becomes the readiness probe. Leave it unset for the default, daemon-agnostic readiness (the provider reports the sandbox running; cua-spacesd on 3211 is used by the interfaces when the image has it).

Example

import asyncio
 
from cua_sandbox import CloudOptions, Image, Sandbox, http
 
async def main():
    sb = await Sandbox.create(
        Image.from_registry("python:3.12-slim"),
        command=["python", "-m", "http.server", "8000"],
        services={"web": 8000},
        wait_for=http("web", "/"),
        local=False,                     # the cloud; omit it to run locally
        cloud=CloudOptions(warm=False),  # optional; implies the cloud
        cpu=2, memory="4GB",
    )
    print(sb.id)
    await sb.destroy()
 
asyncio.run(main())

Sandbox.connect#

@classmethod
def connect(
    name: Optional[str] = None,
    *,
    url: Optional[str] = None,
    token: Optional[str] = None,
    api_key: Optional[str] = None,
    local: Optional[bool] = None,
    ws_url: Optional[str] = None,
    http_url: Optional[str] = None,
    container_name: Optional[str] = None,
    cpu: Optional[int] = None,
    memory_mb: Optional[int] = None,
    disk_gb: Optional[int] = None,
    region: str = 'us-east-1',
    telemetry_enabled: bool = True,
) -> '_ConnectResult'

Connect to an existing sandbox by name.

Supports both await and async with. When used as a context manager, disconnect() is called on exit: the sandbox keeps running.

Examples:

# plain await
sb = await Sandbox.connect("my-sandbox")
await sb.screenshot()
await sb.disconnect()
 
# context manager: disconnects on exit, sandbox keeps running
async with Sandbox.connect("my-sandbox") as sb:
    await sb.screenshot()

Example

import asyncio
 
from cua_sandbox import Sandbox
 
async def main():
    async with Sandbox.connect(url=URL, token=TOKEN) as sb:  # or Sandbox.connect("my-sandbox")
        result = await sb.shell.run("echo hello")
        print([result.stdout, result.returncode, result.success])
 
asyncio.run(main())

Sandbox.ephemeral#

@classmethod
async def ephemeral(
    image: Image | None = None,
    *,
    pool: 'Pool | str | None' = None,
    name: Optional[str] = None,
    replicas: int = 1,
    service: str = ENV_SERVICE,
    claim_spec: Any = None,
    keep_alive_minutes: float | None = None,
    keep_pool: bool = False,
    api_key: Optional[str] = None,
    on: Optional[str] = None,
    local: Optional[bool] = None,
    kind: Optional[str] = None,
    runtime: 'Runtime | str | None' = None,
    cpu: Optional[int] = None,
    memory_mb: Optional[int] = None,
    disk_gb: Optional[int] = None,
    region: str = 'us-east-1',
    time_to_start: Optional[float] = None,
    request_timeout: Optional[float] = None,
    server_port: Optional[int] = None,
    telemetry_enabled: bool = True,
    warm: Optional[bool] = None,
    max_pool_size: Optional[int] = None,
    claim_ttl: 'float | timedelta | None' = None,
    progress: Optional[Callable[[Any], Any]] = None,
    command: Optional[list[str]] = None,
    sidecars: Optional[list] = None,
    env: Optional[dict[str, str]] = None,
    services: Optional[dict[str, int]] = None,
    wait_for: WaitFor = None,
    cloud: Optional[CloudOptions] = None,
    memory: 'str | int | None' = None,
    network: Optional[str] = None,
) -> AsyncIterator['Sandbox']

A sandbox that is released when the async with block exits.

Takes the same on/local, kind, runtime, command, env, services, wait_for, cloud, memory and network options as create.

A Fleet registry image without pool is claimed from this account's reusable managed pool (see create for warm, max_pool_size, claim_ttl and progress). Exiting releases the claim; the pool stays for the next run and idle GC deletes it once unused. If the process dies, the claim expires after claim_ttl. keep_pool is deprecated and has no effect.

Sandbox.list#

@classmethod
async def list(
    *,
    location: Optional[str] = None,
    local: Optional[bool] = None,
    all: Optional[bool] = None,
    api_key: Optional[str] = None,
) -> 'list[SandboxInfo]'

List sandboxes, each with its location, kind and runtime: local and cloud ones by default.

The cloud part never fails the default listing: without cloud credentials it is left out silently, and when the cloud fails or does not answer within 5 s the local rows come back with a logged warning ("cloud sandboxes not listed: ...").

Sandbox.get_info#

@classmethod
async def get_info(
    name: str,
    *,
    local: Optional[bool] = None,
    api_key: Optional[str] = None,
) -> 'SandboxInfo'

Get metadata for a specific sandbox.

Sandbox.suspend#

@classmethod
async def suspend(
    name: str,
    *,
    local: Optional[bool] = None,
    api_key: Optional[str] = None,
) -> None

Suspend the sandbox name, keeping its state; resume continues it. Only that sandbox is affected.

Local: the SDK suspends it on its backend (a container is paused or checkpointed, a QEMU VM snapshotted, a Lume VM stopped with its disk). Cloud: a Fleet sandbox cannot be suspended on its own, so this raises Unsupported; keep it with keep_alive() or release it with delete. A legacy api.cua.ai VM (an API key) stops.

Sandbox.resume#

@classmethod
async def resume(
    name: str,
    *,
    local: Optional[bool] = None,
    api_key: Optional[str] = None,
) -> 'Sandbox'

Resume the suspended sandbox name and return it connected.

Local: the backend resumes it. Cloud: a running sandbox just reconnects (a Fleet sandbox is never suspended); one that is gone raises Unsupported. A legacy api.cua.ai VM (an API key) starts.

Returns: A connected Sandbox ready to use.

Sandbox.restart#

@classmethod
async def restart(
    name: str,
    *,
    local: Optional[bool] = None,
    api_key: Optional[str] = None,
) -> 'Sandbox'

Restart the sandbox name (suspend then resume) and return it connected. Local only (see suspend): a Fleet sandbox raises Unsupported.

Returns: A connected Sandbox ready to use.

Sandbox.delete#

@classmethod
async def delete(
    name: str,
    *,
    local: Optional[bool] = None,
    api_key: Optional[str] = None,
) -> None

Permanently delete a sandbox.

For local sandboxes, stops the VM and removes the state file. For cloud sandboxes, calls DELETE /v1/vms/{name}.

sandbox#

async def sandbox(
    *,
    on: Optional[str] = None,
    local: Optional[bool] = None,
    kind: Optional[str] = None,
    ws_url: Optional[str] = None,
    http_url: Optional[str] = None,
    url: Optional[str] = None,
    token: Optional[str] = None,
    api_key: Optional[str] = None,
    container_name: Optional[str] = None,
    image: Optional[Image] = None,
    runtime: 'Runtime | str | None' = None,
    name: Optional[str] = None,
    ephemeral: Optional[bool] = None,
    cpu: Optional[int] = None,
    memory_mb: Optional[int] = None,
    disk_gb: Optional[int] = None,
    region: str = 'us-east-1',
    warm: Optional[bool] = None,
    max_pool_size: Optional[int] = None,
    claim_ttl: Any = None,
    progress: Optional[Callable[[Any], Any]] = None,
) -> AsyncIterator[Sandbox]

Async context manager for a sandboxed environment.

on/local, kind and runtime mean what they mean for Sandbox.create.

.. deprecated:

Prefer `Sandbox.create()`, `Sandbox.connect()`, or
`Sandbox.ephemeral()` instead.

SandboxInfo#

class SandboxInfo
 
SandboxInfo(
    name: str,
    status: str,
    source: str,
    os_type: Optional[str] = None,
    host: Optional[str] = None,
    vnc_url: Optional[str] = None,
    api_url: Optional[str] = None,
    created_at: Optional[str] = None,
    id: Optional[str] = None,
    location: Optional[str] = None,
    services: dict = dict(),
    expires_at: Optional[str] = None,
    provider_details: dict = dict(),
    kind: Optional[str] = None,
    runtime: Optional[str] = None,
)

Metadata for a local or cloud sandbox.

status from Sandbox.info is portable: provisioning, starting, ready or stopped. Listings keep the provider's word (running, suspended, ...). Provider internals (the cloud pool, namespace and claim; the local backend) are in provider_details.

Creation options#

CloudOptions#

class CloudOptions
 
CloudOptions(
    warm: Optional[bool] = None,
    max_pool_size: Optional[int] = None,
    claim_ttl: Union[float, timedelta, None] = None,
    pool: 'Pool | str | None' = None,
    apply: bool = False,
)

Advanced options for cloud sandboxes. Passing them implies the cloud; with on="local"/local=True they are an InvalidArgument. The engine is Sandbox.create(runtime="gvisor"|"kubevirt"), not a cloud option.

• warm: keep one sandbox of this image ready (faster next start).
• max_pool_size: most sandboxes of this image at once (default 10).
• claim_ttl: how long the sandbox outlives this process without a keep_alive (seconds or a timedelta; renewed while held; default 15 min).
• pool: dedicated capacity: claim from this existing pool. Sandbox fields given with it (image, command, env, services, sidecars, cpu, memory, a registry secret) must match the pool's template, else cua_sandbox.PoolSpecMismatch (with a readable diff) is raised; fields left unset are not compared.
• apply: with pool, update the pool's template to the given fields instead (the pool's capacity is kept; managed cua-auto-* pools refuse it).

Probe#

class Probe
 
Probe(service: str, path: Optional[str] = None)

A readiness probe on a declared service: TCP, or GET path returning 2xx when path is set. Build it with tcp or http.

Probe.native#

def native() -> Any

The cua.ReadinessProbe for this probe.

http#

def http(service: str, path: str = '/') -> Probe

Ready once GET path on the declared service returns 2xx.

tcp#

def tcp(service: str) -> Probe

Ready once a TCP connect to the declared service succeeds.

PublicUrl#

class PublicUrl
 
PublicUrl(
    url: str,
    expires_at: str,
    id: str,
    service: str,
    provider_details: dict = dict(),
)

A shareable URL of a sandbox service that stops working at expires_at. Cloud: a signed service URL; local: a loopback URL with its own token, served by the cua daemon.

Sandbox references#

AmbiguousSandbox#

class AmbiguousSandbox(LookupError)
 
AmbiguousSandbox(name: str, candidates: list[str])

A bare sandbox name matches sandboxes in more than one location.

candidates are the qualified refs it matches (local:box, cloud:box); pass one of them, or local=True/local=False. The native SDK raises cua.CuaError.AmbiguousSandbox for the same case.