Runtimes and configuration

Local runtimes (Docker, QEMU, Lume, Hyper-V, Tart, Android emulator), support checks, configuration, sign-in and errors.

Local sandboxes run on a runtime the SDK picks for the image. Pass one explicitly to Sandbox.create(..., runtime=...) to override it.

Support checks#

RuntimeSupport#

class RuntimeSupport
 
RuntimeSupport(
    supported: bool,
    hw_accel: bool,
    runtime_installed: bool,
    auto_installable: bool,
    runtime_name: str,
    host_os: str,
    host_arch: str,
    guest_os: str,
    reason: str,
)

Result of a local runtime compatibility check for a given Image.

check_local_support#

def check_local_support(image: 'Image') -> RuntimeSupport

Return a RuntimeSupport describing whether image can run locally.

Returns: RuntimeSupport with .supported, .hw_accel, .reason, and more.

skip_if_unsupported#

def skip_if_unsupported(image: 'Image', *, require_hw_accel: bool = False) -> None

Call at the top of a pytest test/fixture to skip when the host can't run image.

Example:

async def test_macos_brew_install():
    skip_if_unsupported(Image.macos())
    async with Sandbox.ephemeral(Image.macos().brew_install("wget"), local=True) as sb:
        assert (await sb.shell.run("wget --version")).success

Runtimes#

Runtime#

class Runtime(ABC)

Base class for local runtimes that create VMs/containers from an Image.

Runtime.start#

async def start(image: Image, name: str, **opts) -> RuntimeInfo

Start a VM/container and return connection info.

Runtime.stop#

async def stop(name: str) -> None

Stop and clean up a VM/container.

Runtime.is_ready#

async def is_ready(info: RuntimeInfo, timeout: float = 120) -> bool

Wait until the sandbox is ready.

Daemon-agnostic: "ready" is the runtime's own notion (instance running, plus any probe the caller declared), never a specific guest agent.

Runtime.suspend#

async def suspend(name: str) -> None

Suspend (pause/save state of) a running VM.

Runtime.resume#

async def resume(image: 'Image', name: str, **opts) -> RuntimeInfo

Resume a suspended VM and return its RuntimeInfo.

Runtime.list#

async def list() -> list[dict]

List known VMs managed by this runtime.

Returns a list of dicts with at minimum: name, status.

Runtime.ensure_base#

async def ensure_base(image: 'Image', base_name: str) -> CheckpointInfo

Pull/build image into a stopped base sandbox if not already present.

Idempotent. First call does the full pull; subsequent calls return immediately. The base sandbox is never started: it only serves as a fork source for ephemeral VMs.

Runtime.fork#

async def fork(base_name: str, new_name: str, **opts) -> None

Create a new sandbox from a base/checkpoint using CoW where possible.

Runtime.checkpoint#

async def checkpoint(name: str, checkpoint_name: str, **opts) -> CheckpointInfo

Capture the current state of a running sandbox as a named checkpoint.

Runtime.list_checkpoints#

async def list_checkpoints() -> list[CheckpointInfo]

Runtime.delete_checkpoint#

async def delete_checkpoint(checkpoint_name: str) -> None

RuntimeInfo#

class RuntimeInfo
 
RuntimeInfo(
    host: str,
    api_port: int,
    vnc_port: Optional[int] = None,
    api_key: Optional[str] = None,
    container_id: Optional[str] = None,
    name: Optional[str] = None,
    qmp_port: Optional[int] = None,
    environment: Optional[str] = None,
    agent_type: Optional[str] = None,
    guest_server_port: int = 3211,
    exposed_ports: Optional[dict] = None,
    ssh_port: Optional[int] = None,
    ssh_username: Optional[str] = None,
    ssh_password: Optional[str] = None,
    ssh_key_filename: Optional[str] = None,
    vnc_host: Optional[str] = None,
    vnc_password: Optional[str] = None,
    grpc_port: Optional[int] = None,
    native: Any = None,
    env_ready_timeout: Optional[float] = None,
    env_token: Optional[str] = None,
)

Connection info returned after a runtime spins up.

DockerRuntime#

class DockerRuntime(NativeRuntime)
 
DockerRuntime(
    *,
    api_port: int = DEFAULT_API_PORT,
    vnc_port: int = DEFAULT_VNC_PORT,
    ephemeral: bool = True,
    volumes: Optional[list[str]] = None,
    environment: Optional[dict[str, str]] = None,
    devices: Optional[list[str]] = None,
    platform: Optional[str] = None,
    privileged: bool = False,
    stop_timeout: int = 120,
    cpus: Optional[int] = None,
    memory_mb: Optional[int] = None,
    server_port: Optional[int] = None,
)

Runs OCI container images through the SDK's container backend.

api_port/vnc_port are accepted for compatibility; the SDK picks free host ports and Sandbox.exposed_ports reports them. volumes, devices, platform and privileged are not supported by the sandboxed (gVisor) backend and are rejected.

DockerRuntime.start#

async def start(image: Image, name: str, **opts) -> RuntimeInfo

QEMURuntime#

def QEMURuntime(mode: str = 'docker', **kwargs) -> Runtime

Factory that returns the appropriate QEMU runtime.

QEMUDockerRuntime#

class QEMUDockerRuntime(NativeQEMURuntime)
 
QEMUDockerRuntime(
    *,
    api_port: int = DEFAULT_API_PORT,
    vnc_port: int = QEMU_VNC_PORT,
    ephemeral: bool = True,
    storage_dir: Optional[str | Path] = None,
    memory_mb: int = 8192,
    cpu_count: int = 4,
)

QEMURuntime(mode="docker"): kept for compatibility.

The trycua/cua-qemu-* wrapper images drove their guest through computer-server, which is gone; Linux and Windows guests now boot on the SDK's QEMU backend (the SDK provisions QEMU itself). Android guests use cua_sandbox.runtime.AndroidEmulatorRuntime or bare-metal QEMU.

QEMUDockerRuntime.start#

async def start(image: Image, name: str, **opts) -> RuntimeInfo

QEMUBaremetalRuntime#

class QEMUBaremetalRuntime(Runtime)
 
QEMUBaremetalRuntime(
    *,
    api_port: int = DEFAULT_API_PORT,
    vnc_display: int = 0,
    memory_mb: int = 4096,
    cpu_count: int = 2,
    arch: str = 'x86_64',
    qmp_port: int = 4444,
    use_qmp_transport: bool = False,
    extra_args: Optional[list[str]] = None,
    use_sdk: Optional[bool] = None,
)

Bare-metal QEMU: launches qemu-system-* directly on the host.

Requires: - qemu-system-x86_64 (or qemu-system-aarch64) on PATH

• A disk image (qcow2/vhdx/raw) with computer-server pre-installed, provided via Image.from_file('/path/to/disk.qcow2')

use_sdk=False keeps the legacy in-process launcher, which boots the given disk in place (the image builder needs that: the SDK always puts an instance overlay on top of the disk).

QEMUBaremetalRuntime.start#

async def start(image: Image, name: str, **opts) -> RuntimeInfo

QEMUBaremetalRuntime.stop#

async def stop(name: str) -> None

QEMUBaremetalRuntime.is_ready#

async def is_ready(info: RuntimeInfo, timeout: float = 120) -> bool

QEMUBaremetalRuntime.suspend#

async def suspend(name: str) -> None

Save VM state via QMP savevm then quit QEMU.

QEMUBaremetalRuntime.resume#

async def resume(image: 'Image', name: str, **opts) -> RuntimeInfo

Relaunch QEMU with -loadvm to restore saved state.

QEMUBaremetalRuntime.list#

async def list() -> list[dict]

List known bare-metal QEMU sandboxes from state files, checking if alive.

QEMUWSL2Runtime#

class QEMUWSL2Runtime(Runtime)
 
QEMUWSL2Runtime(
    *,
    api_port: int = DEFAULT_API_PORT,
    vnc_display: int = 0,
    memory_mb: int = 4096,
    cpu_count: int = 4,
    arch: str = 'x86_64',
    extra_args: Optional[list[str]] = None,
)

QEMU via WSL2 with KVM hardware acceleration.

Runs qemu-system-x86_64 inside WSL2 where /dev/kvm is available, while keeping disk images on the Windows filesystem. Windows paths are automatically converted to /mnt/c/... paths for WSL access.

This is the fastest way to run QEMU on Windows: native KVM speeds vs TCG software emulation (bare-metal Windows) or Hyper-V (needs admin).

QEMUWSL2Runtime.available#

@staticmethod
def available() -> bool

Check if WSL2 + QEMU + KVM are available.

QEMUWSL2Runtime.start#

async def start(image: Image, name: str, **opts) -> RuntimeInfo

QEMUWSL2Runtime.stop#

async def stop(name: str) -> None

QEMUWSL2Runtime.is_ready#

async def is_ready(info: RuntimeInfo, timeout: float = 120) -> bool

LumeRuntime#

class LumeRuntime(NativeRuntime)
 
LumeRuntime(
    *,
    lume_host: str = 'localhost',
    lume_port: int = LUME_PROVIDER_PORT,
    api_port: int = LUME_API_PORT,
    ephemeral: bool = True,
    cpus: Optional[int] = None,
    memory_mb: Optional[int] = None,
)

Runs macOS VMs via Lume, through the cua SDK.

LumeRuntime.start#

async def start(image: Image, name: str, **opts) -> RuntimeInfo

LumeRuntime.ensure_base#

async def ensure_base(image: Image, base_name: str)

Pull an OCI image into a stopped golden base VM, if it doesn't exist yet.

Idempotent: if base_name already exists and is stopped, returns immediately. Raises if the base VM is running, because cloning a live VM can produce dirty or inconsistent clones. The base VM is never started; it only serves as a fork source.

LumeRuntime.fork#

async def fork(base_name: str, new_name: str, **opts) -> None

Clone a stopped VM base_name → new_name via APFS clonefile (instant).

The source VM must be stopped before calling this: cloning a running VM risks dirty state or clone failures.

LumeRuntime.checkpoint#

async def checkpoint(name: str, checkpoint_name: str, **opts)

Stop name, clone it into a checkpoint, then restart name.

Stops, clones and restarts, so clones are always taken from a clean stopped state.

LumeRuntime.list_checkpoints#

async def list_checkpoints()

Return VMs that are checkpoints (cua-base-* or cua-ckpt-* prefix).

LumeRuntime.delete_checkpoint#

async def delete_checkpoint(checkpoint_name: str) -> None

Delete a checkpoint VM. Only operates on cua-base-* / cua-ckpt-* names.

LumeRuntime.list#

async def list() -> list[dict]

List all Lume VMs.

HyperVRuntime#

class HyperVRuntime(Runtime)
 
HyperVRuntime(
    *,
    api_port: int = DEFAULT_API_PORT,
    switch_name: str = 'Default Switch',
    memory_mb: int = 4096,
    cpu_count: int = 4,
    disk_size_gb: int = 64,
    cache_dir: Optional[Path] = None,
    windows_iso: Optional[str] = None,
    product_key: Optional[str] = None,
)

Runs Windows VMs via Hyper-V with layered VHDX caching.

On first use, builds a base Windows image from ISO using unattended install (same Autounattend.xml as the QEMU builder). The base image includes CUA computer-server pre-installed. This is cached and reused.

Each sandbox session gets its own ephemeral differencing disk on top.

HyperVRuntime.start#

async def start(image: Image, name: str, **opts) -> RuntimeInfo

HyperVRuntime.stop#

async def stop(name: str) -> None

Stop the VM, remove it, and delete the ephemeral session disk.

HyperVRuntime.is_ready#

async def is_ready(info: RuntimeInfo, timeout: float = 120) -> bool

Daemon-agnostic readiness: the VM is running and has an address.

start() already waited for the guest IP; the guest agent port is a real TCP endpoint on the Hyper-V switch, so wait until it accepts a connection (any daemon listening there), bounded by timeout.

TartRuntime#

class TartRuntime(Runtime)
 
TartRuntime(
    *,
    ephemeral: bool = True,
    cpu_count: Optional[int] = None,
    memory_mb: Optional[int] = None,
    disk_gb: Optional[int] = None,
    display: str = '1024x768',
    ssh_username: str = TART_DEFAULT_USERNAME,
    ssh_password: str = TART_DEFAULT_PASSWORD,
)

Apple Virtualization.framework runtime via Tart CLI.

Uses VNC (host-side, random port) for screenshots and SSH for shell commands.

TartRuntime.start#

async def start(image: Image, name: str, **opts) -> RuntimeInfo

TartRuntime.is_ready#

async def is_ready(info: RuntimeInfo, timeout: float = 120) -> bool

Wait until SSH is reachable.

TartRuntime.stop#

async def stop(name: str) -> None

AndroidEmulatorRuntime#

class AndroidEmulatorRuntime(Runtime)
 
AndroidEmulatorRuntime(
    *,
    api_level: int = 34,
    img_type: str = 'google_apis',
    device_id: str = 'pixel',
    memory_mb: int = 4096,
    cpu_count: int = 4,
    adb_port: Optional[int] = None,
    no_boot_anim: bool = True,
    gpu_mode: str = 'swiftshader_indirect',
    headless: bool = True,
    grpc_port: Optional[int] = None,
)

Android emulator runtime using the official Android SDK.

Creates an AVD and launches the Android emulator, following the docker-android pattern. Uses ADB for boot detection, APK install, and shell commands.

AndroidEmulatorRuntime.start#

async def start(image: Image, name: str, **opts) -> RuntimeInfo

AndroidEmulatorRuntime.is_ready#

async def is_ready(info: RuntimeInfo, timeout: float = 300) -> bool

Wait for sys.boot_completed == 1 via ADB, like docker-android.

AndroidEmulatorRuntime.list#

async def list() -> list[dict]

List known Android emulator sandboxes from state files, checking if alive.

AndroidEmulatorRuntime.stop#

async def stop(name: str) -> None

AndroidEmulatorRuntime.take_screenshot#

async def take_screenshot() -> bytes

Take a screenshot via ADB screencap.

Configuration and errors#

Process-wide settings, sign-in helpers, transports and the typed errors of cua-sandbox.

Configuration and sign-in#

configure#

def configure(*, api_key: Optional[str] = None, base_url: Optional[str] = None) -> None

Set global configuration for the legacy login() and whoami() helpers.

login#

def login(*, base_url: Optional[str] = None) -> None

Open the CUA login page in a browser and store credentials.

This initiates a Clerk-based browser authentication flow. The user completes login in their browser, and the resulting API key is stored in ~/.cua/credentials.

whoami#

def whoami(*, api_key: Optional[str] = None) -> Dict[str, Any]

Return info about the authenticated user.

Returns: Dict with user info (id, email, etc.) from the CUA API.

Errors#

SpacesdNotAvailable#

class SpacesdNotAvailable(RuntimeError)

The sandbox has no reachable cua-spacesd.

Sandboxes are daemon-agnostic: creating, listing and deleting them never needs a guest agent. The computer interfaces (screen, mouse, shell, files ...) need cua-spacesd, or one of the agentless transports (QMP, VNC, SSH, ADB) the runtime exposes.

InvalidArgument#

class InvalidArgument(ValueError)

Contradictory or invalid options (the native SDK's CuaError.InvalidArgument), for example on="local" together with local=False.

InvalidPlacement#

class InvalidPlacement(InvalidArgument)

A location, kind and runtime that do not go together (the native SDK's CuaError.InvalidPlacement), for example kind="container" with runtime="qemu". The message lists the valid values.

Unsupported#

class Unsupported(NotImplementedError)

The operation is not supported for this sandbox, image or provider (the native SDK's CuaError.Unsupported). A NotImplementedError, so older except NotImplementedError handlers still catch it.

Transports#

EnvTransport#

class EnvTransport(Transport)

Drives a sandbox through cua-spacesd.

EnvTransport.connect#

async def connect() -> None

EnvTransport.disconnect#

async def disconnect() -> None

EnvTransport.spacesd#

async def spacesd() -> Any

The raw cua.SpacesdClient (typed client escape hatch).

EnvTransport.send#

async def send(action: str, **params: Any) -> Any

EnvTransport.screenshot#

async def screenshot(format: str = 'png', quality: int = 95) -> bytes

EnvTransport.get_screen_size#

async def get_screen_size() -> Dict[str, int]

EnvTransport.get_environment#

async def get_environment() -> str

EnvTransport.get_display_url#

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

A browser URL for the sandbox's display.

With cua-spacesd (an env or viewer service), a link to its HTML5 viewer carrying a scoped, expiring viewer ticket; the same link serves share=True. For images without cua-spacesd, the page of the first declared legacy web display service (novnc, display, web; a raw RFB port 59xx is skipped, and share=True returns its expiring public link), then the agentless fallback's VNC address.

EnvTransport.request_service#

async def request_service(
    name: str,
    *,
    method: str,
    path: str,
    json_body: Any = None,
    headers: Any = None,
) -> Any

EnvTransport.native_handle#

async def native_handle() -> Any

The cua.Sandbox handle behind this transport (services, forwards, public URLs).

EnvTransport.native_service#

async def native_service(name: str) -> Any

EnvTransport.forward_tunnel#

async def forward_tunnel(sandbox_port: int | str) -> 'TunnelInfo'

EnvTransport.close_tunnel#

async def close_tunnel(info: 'TunnelInfo') -> None

EnvTransport.pty_create#

async def pty_create(
    command: Optional[str] = None,
    cols: int = 120,
    rows: int = 40,
    cwd: Optional[str] = None,
    envs: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]

EnvTransport.pty_send#

async def pty_send(pid: int, data: str) -> None

EnvTransport.pty_kill#

async def pty_kill(pid: int) -> bool

EnvTransport.pty_info#

async def pty_info(pid: int) -> Optional[Dict[str, Any]]