Sandbox, images and services

Sandbox, the sandbox() helper and its creation options, the Image builder, and the services, tunnels and MCP endpoints of a 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 computer, on="aws", "gcp" or "modal" in your own cloud. local=False meant Cua Cloud, which ended on 18 October 2026: it fails with a clear error. Sandbox.connect(name) and the class methods take local=None: the local sandbox of that name when one exists.

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.

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 or direct), services (name -> guest port), expires_at and provider_details.

Sandbox.service#

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

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

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): 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.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), 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 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: 'SandboxPool | str | None' = None,
    name: 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,
    time_to_start: Optional[float] = None,
    request_timeout: Optional[float] = None,
    server_port: Optional[int] = None,
    telemetry_enabled: bool = True,
    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,
    memory: 'str | int | None' = None,
    network: Optional[str] = None,
    overlay: 'dict[str, Any] | list | None' = None,
) -> 'Sandbox'

Provision a persistent local sandbox and return it connected.

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

• on: "local". local=True is the same switch; on and a contradicting local is an InvalidArgument. Unset, the user default applies (CUA_DEFAULT_ON, else cua config set default.on in $CUA_HOME/config.toml, else local, with a one-time notice that CUA_QUIET_DEFAULT=1 hides). Cua Cloud has ended: on="cloud", local=False and pool= naming no pool of this machine raise CloudEnded.
• 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: "gvisor"/"runc" (containers) and "qemu"/"lume" (VMs). Unset: default.runtime, else auto. A 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") raises InvalidPlacement, listing the valid values.

Options:

• 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".
• 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 and Lume 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.

pool: a pool of this machine (a SandboxPool from create_pool, or its name) gives the sandbox the pool's image and size and takes one of its ready sandboxes (about 0.1 s).

server_port names a port your image serves itself: it 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 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", "/"),
        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,
    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: 'SandboxPool | str | None' = None,
    name: 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,
    time_to_start: Optional[float] = None,
    request_timeout: Optional[float] = None,
    server_port: Optional[int] = None,
    telemetry_enabled: bool = True,
    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,
    memory: 'str | int | None' = None,
    network: Optional[str] = None,
) -> AsyncIterator['Sandbox']

A local sandbox that is deleted when the async with block exits.

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

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 the local sandboxes (containers, QEMU, Lume, Android), each with its location, kind and runtime.

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 local sandbox name, keeping its state; resume continues it. Only that sandbox is affected.

The SDK suspends it on its backend (a container is paused or checkpointed, a QEMU VM snapshotted, a Lume VM stopped with its disk).

Sandbox.resume#

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

Resume the suspended local sandbox name and return it connected. The backend resumes it.

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 local sandbox name (suspend then resume) and return it connected (see suspend).

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 local sandbox: stops the VM or container and removes the state file.

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

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

Creation options#

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: 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, relay:<id>); pass one of them. The native SDK raises cua.CuaError.AmbiguousSandbox for the same case.

Images and registries#

Image is an immutable, chainable description of what a sandbox boots. Builder methods return a new Image.

Builder methods add layers and return a new Image. Locally, a container image is built into your container engine before boot; it is cached by content. Layers a build cannot run (brew_install, choco_install, ...) are applied after boot through cua-spacesd, so they need an image that runs it.

Image builder#

Image#

class Image
 
Image(os_type: str, distro: str, version: str, kind: Optional[str] = None)

Immutable, chainable image specification.

Each mutation method returns a new Image instance so that builders can be forked at any point.

Example

from cua_sandbox import Image
 
img = Image.linux().apt_install("curl").pip_install("requests").env(DEBUG="1").expose(8080)
print(img.to_dict())
# {'os_type': 'linux', 'distro': 'ubuntu', 'version': '24.04', 'kind': None,
#  'layers': [{'type': 'apt_install', 'packages': ['curl']},
#             {'type': 'pip_install', 'packages': ['requests']}],
#  'env': {'DEBUG': '1'}, 'ports': [8080]}

Image.linux#

@classmethod
def linux(
    distro: str = 'ubuntu',
    version: str = '24.04',
    kind: Optional[str] = None,
    tier: Optional[str] = None,
) -> Image

Linux: the canonical ghcr.io/trycua/linux:<version> image.

kind=None (or "auto") runs what the image is (the rootfs: a gVisor container); kind="vm" its -disk containerDisk (QEMU); kind="container" the rootfs. Sandbox.create(kind=...) overrides it.

tier: "full" (the default: dev tooling) or "slim" (24.04-slim: cua-spacesd and Chromium only; what CI runs). A tier CI has not published yet raises CuaError.ImageNotPublished; use Image.from_registry(ref) to run it anyway.

Image.macos#

@classmethod
def macos(version: str = '26', kind: str = 'vm', tier: Optional[str] = None) -> Image

macOS: the canonical ghcr.io/trycua/macos:<version> image (Lume locally).

Supported versions: "15" / "sequoia", "26" / "tahoe". tier: "full" (the default), "slim" or "xcode" / "xcode-<X.Y>" (full plus one pinned Xcode); unpublished tiers raise CuaError.ImageNotPublished.

Image.windows#

@classmethod
def windows(
    version: str = '2022',
    kind: str = 'vm',
    tier: Optional[str] = None,
) -> Image

Windows: the canonical ghcr.io/trycua/windows:2022 containerDisk.

Other versions, including "11", have no canonical image: they are installed from a downloaded evaluation ISO. tier="slim" names 2022-slim once published.

Image.omarchy#

@classmethod
def omarchy(channel: Optional[str] = None) -> Image

Omarchy (Arch Linux, Hyprland) with cua-spacesd: ghcr.io/trycua/omarchy:edge (or "rc"/"stable").

An amd64 VM: QEMU (emulated on arm64 hosts, so slow there). Until CI publishes it this raises CuaError.ImageNotPublished; use Image.from_registry("ghcr.io/trycua/omarchy:edge", kind="vm") to run it anyway.

Image.android#

@classmethod
def android(version: str = '14', kind: str = 'vm') -> Image

Android image. Always a VM (QEMU emulator).

Image.from_registry#

@classmethod
def from_registry(
    ref: str,
    *,
    os_type: str = 'linux',
    kind: Optional[str] = None,
    agent_type: Optional[str] = None,
    secret: Optional[Any] = None,
) -> Image

Create an image from a registry reference.

secret (a cua_sandbox.RegistrySecret) pulls a private image. It is never logged or saved.

os_type selects the firmware: Windows guest disks are built UEFI-only, so a Windows containerDisk pulled from a registry must say so or it is handed BIOS and will not boot. kind is resolved after pull when omitted. agent_type names the guest control server the disk already runs, the same hint as from_file: "osworld" selects the OSWorld Flask server on port 5000 instead of the computer-server on 8000, for local QEMU.

Image.from_file#

@classmethod
def from_file(
    path: str,
    *,
    os_type: str = 'windows',
    kind: str = 'vm',
    agent_type: Optional[str] = None,
) -> Image

Create an image from a local disk, ISO file, or URL.

Supported formats: qcow2, vhdx, raw, img, iso. URLs (http/https) are downloaded automatically. Zip files are extracted. For ISOs, the runtime will create a qcow2 disk and attach the ISO as a CD-ROM for installation/boot.

Image.from_dict#

@classmethod
def from_dict(data: Dict[str, Any]) -> Image

Reconstruct an Image from a serialized spec dict.

Example

from cua_sandbox import Image
 
base = Image.linux().apt_install("git")
dev = base.pip_install("rich")
same = Image.from_dict(dev.to_dict()).to_dict() == dev.to_dict()
print([same, len(base.to_dict()["layers"]), len(dev.to_dict()["layers"])])
# [True, 1, 2]

Image.apt_install#

def apt_install(*packages: str) -> Image

Install packages via apt (Linux only).

Image.brew_install#

def brew_install(*packages: str) -> Image

Install packages via Homebrew (macOS).

Image.choco_install#

def choco_install(*packages: str) -> Image

Install packages via Chocolatey (Windows).

Image.winget_install#

def winget_install(*packages: str) -> Image

Install packages via winget (Windows).

Image.apk_install#

def apk_install(*apk_paths: str) -> Image

Install APK files via adb (Android only).

Image.pwa_install#

def pwa_install(
    manifest_url: str,
    package_name: Optional[str] = None,
    keystore: Optional[str] = None,
    keystore_alias: str = 'android',
    keystore_password: str = 'android',
    builder: str = 'pwa2apk',
    push_timeout: Optional[float] = None,
) -> 'Image'

Build an APK from a PWA manifest URL and install it (Android only).

Two builder backends are supported:

• "pwa2apk" (default): generates a lightweight WebView-based APK. No Chrome dependency, no "Running in Chrome" banner, no Digital Asset Links needed. ~10 KB APK.
• "bubblewrap": generates a Chrome Trusted Web Activity (TWA) APK. Requires Chrome on the device and a matching /.well-known/assetlinks.json on the server. Shows a mandatory "Running in Chrome" privacy disclosure on every launch.

Image.uv_install#

def uv_install(*packages: str) -> Image

Install Python packages via uv add into the cua-server project.

Image.pip_install#

def pip_install(*packages: str) -> Image

Install Python packages via pip.

Image.app_install#

def app_install(app_id: str) -> Image

Install an app from the cua-sandbox-apps catalog.

Requires cua-sandbox-apps to be installed. The app's install script for this image's OS is executed as a build layer.

Image.run#

def run(command: str) -> Image

Run a shell command during image build.

Image.env#

def env(**variables: str) -> Image

Set environment variables.

Image.copy#

def copy(src: str, dst: str) -> Image

Copy a file into the image.

Image.expose#

def expose(port: int) -> Image

Expose a port.

Image.to_dict#

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

Serialize to a plain dict suitable for JSON.

Image.to_cloud_init#

def to_cloud_init() -> str

Generate a cloud-init user-data script from the image layers.

Image.local_support#

def local_support()

Check whether this image can run locally on the current host.

Returns a cua_sandbox.runtime.compat.RuntimeSupport describing:

• supported : runtime is available or auto-installable on this OS
• hw_accel : hardware acceleration (HVF / KVM / Hyper-V) is available
• runtime_installed: runtime binary found right now (no install needed)
• auto_installable: SDK can install the runtime automatically
• reason : human-readable explanation

In tests, use the bundled helper instead of checking manually:

from cua_sandbox.runtime.compat import skip_if_unsupported
 
async def test_something():
    skip_if_unsupported(Image.macos())
    async with Sandbox.ephemeral(Image.macos(), local=True) as sb:
        ...

ImageInfo#

class ImageInfo
 
ImageInfo(
    reference: str,
    pinned_ref: str,
    digest: str,
    variant: str,
    arch: Optional[str],
    os: str,
    emulated: bool,
)

The image a sandbox runs, as resolved and pinned at create time.

reference is the reference as requested (normalised, e.g. docker.io/library/python:3.12-slim), pinned_ref the registry/repo@sha256:... that ran, variant one of rootfs, containerdisk or lume, arch the architecture that ran (amd64/arm64, when known), os the guest OS and emulated whether it ran under emulation.

Containers and registries#

Container#

class Container
 
Container(
    image: str,
    command: Optional[Sequence[str]] = None,
    env: Optional[Mapping[str, str]] = None,
    ports: Optional[Sequence[int]] = None,
    name: Optional[str] = None,
)

A sidecar container, reachable from the sandbox at its name.

name (a DNS label, not main) defaults from the image (redis for redis:7-alpine); the sidecar reaches the sandbox at main. command replaces the image's entrypoint. env holds plain values, not secrets.

Container.native#

def native() -> Any

RegistrySecret#

class RegistrySecret
 
RegistrySecret(
    username: str,
    password: str,
    *,
    registry: Optional[str] = None,
)

Credentials for a private registry image.

• RegistrySecret(username, password): explicit values (a token works as the password).
• RegistrySecret.from_env(): read CUA_REGISTRY_USERNAME / CUA_REGISTRY_PASSWORD (or the variables you name) at create time.
• RegistrySecret.aws_ecr(region=None): a private Amazon ECR image, with a login token from the AWS CLI.

registry scopes them to one host; by default they apply to the image's registry. The password never appears in repr or logs.

Example

from cua_sandbox import Image, RegistrySecret
 
print(Image.from_registry("python:3.12-slim").to_dict()["registry"])
print(repr(RegistrySecret("robot", "s3cr3t", registry="registry.example.com")))  # the password never shows

RegistrySecret.from_env#

@classmethod
def from_env(
    username_var: str = 'CUA_REGISTRY_USERNAME',
    password_var: str = 'CUA_REGISTRY_PASSWORD',
    *,
    registry: Optional[str] = None,
) -> 'RegistrySecret'

RegistrySecret.aws_ecr#

@classmethod
def aws_ecr(region: Optional[str] = None) -> 'RegistrySecret'

RegistrySecret.native#

def native() -> Any

Pools on this machine#

create_pool keeps sandboxes of one shape ready on this machine (the cua daemon holds them); Sandbox.create(pool=...) takes one in about 0.1 s.

Pools#

create_pool#

async def create_pool(
    image: 'Image | str',
    *,
    name: Optional[str] = None,
    min: int = 1,
    max: Optional[int] = None,
    idle: Seconds = None,
    on: str = 'local',
    kind: Optional[str] = None,
    runtime: Optional[str] = None,
    cpu: Optional[int] = None,
    memory: 'str | int | None' = None,
    memory_mb: Optional[int] = None,
    env: Optional[dict[str, str]] = None,
    services: Optional[dict[str, int]] = None,
) -> SandboxPool

Creates pool name of image sandboxes (None: a generated name), or updates the pool of that name. kind, runtime, cpu, memory, env and services mean what they mean in Sandbox.create. Returns at once; its sandboxes start in the background (see the module docs for min, max and idle).

SandboxPool#

class SandboxPool
 
SandboxPool(
    name: str,
    image: str,
    location: str,
    kind: str,
    runtime: str,
    cpus: int,
    memory_mb: int,
    min: int,
    max: int,
    idle: Optional[timedelta],
    ready: int,
    starting: int,
    status: str,
)

A pool (see cua_sandbox.sandbox_pool), with what it held when read. Make one with create_pool.

SandboxPool.create#

async def create(name: Optional[str] = None, **kwargs: Any) -> 'Sandbox'

A sandbox from this pool (Sandbox.create(pool=self, ...)).

SandboxPool.ephemeral#

def ephemeral(**kwargs: Any) -> 'AsyncIterator[Sandbox]'

async with pool.ephemeral() as sb: a sandbox from this pool, deleted on exit.

SandboxPool.refresh#

async def refresh() -> 'SandboxPool'

This pool read again (what it holds now).

SandboxPool.delete#

async def delete() -> bool

Deletes this pool and its ready sandboxes (sandboxes created from it stay). Whether it existed.

get_pool#

async def get_pool(name: str) -> Optional[SandboxPool]

Pool name with what it holds now, or None when there is none.

list_pools#

async def list_pools() -> list[SandboxPool]

Every pool, with what each holds now.

delete_pool#

async def delete_pool(name: str) -> bool

Deletes pool name and its ready sandboxes (sandboxes created from it stay). Whether it existed.

Services, tunnels and MCP#

Reach ports inside a sandbox: named services, tunnels, shareable URLs and MCP servers.

Named services#

ServiceHandle#

class ServiceHandle

One named service of a sandbox (sb.service("mcp")).

Requests carry your headers, url() is usable from this machine and public_url() can be shared.

ServiceHandle.request#

async def request(
    method: str,
    path: str = '/',
    *,
    headers: Headers = None,
    body: bytes | str | None = None,
    json: Any = None,
    timeout: float = 30.0,
) -> Any

One HTTP request to path on the service; returns an httpx.Response.

ServiceHandle.url#

async def url() -> str

A URL for the service usable from this machine with no credentials (no trailing slash): the published loopback port.

ServiceHandle.public_url#

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

A shareable URL that stops working after ttl seconds (60 s to 24 h). See cua_sandbox.Sandbox.public_url.

Services#

class Services

Request a named service exposed by the active sandbox connection.

Services.request#

async def request(
    name: str,
    *,
    method: str,
    path: str,
    json: Any = None,
    headers: dict[str, str] | None = None,
) -> Any

Send a request to a named service on this sandbox.

Prefer sb.service(name).request(...).

Tunnels#

Tunnel#

class Tunnel

Port-forwarding interface: exposes sandbox ports on the host.

Tunnel.forward#

def forward(*ports: int | str) -> _TunnelContext

Forward one or more sandbox ports (or abstract sockets) to the host.

ports may be:

• int: TCP port inside the sandbox (e.g. 8080)
• str: Android abstract socket name (e.g. "chrome_devtools_remote")

Returns a context manager (or awaitable) that yields:

• a single TunnelInfo when one target is given
• a dict[sandbox_port, TunnelInfo] when multiple targets are given

Example: Chrome DevTools over ADB:

async with sb.tunnel.forward("chrome_devtools_remote") as t:
    # http://localhost:<random>/json lists CDP targets
    print(t.url)

TunnelInfo#

class TunnelInfo

A single forwarded port.

TunnelInfo.close#

async def close() -> None

Close this tunnel (no-op if already closed or inside a context manager).

MCP#

mcp_config#

async def mcp_config(
    service: Callable[[], Awaitable[Any]],
    path: str = '/mcp',
) -> Dict[str, Any]

{"url": ..., "headers": {...}} of the MCP endpoint at path.

connect#

async def connect(config: Dict[str, Any]) -> AsyncIterator[Any]

An initialized official-SDK client for config (url, headers).

With mcp 2.x this is an mcp.Client (protocol negotiated with server/discover, falling back to initialize); with 1.x an initialized mcp.ClientSession.

open_mcp#

async def open_mcp(
    service: Callable[[], Awaitable[Any]],
    path: str = '/mcp',
) -> AsyncIterator[Any]

connect(await mcp_config(service, path)).