v0.1.0pip install cua-sandbox

cua-sandbox — ephemeral and persistent sandboxed computer environments.

Usage::

import cua_sandbox as cua

Configure API access

cua.configure(api_key="sk-...")

Local sandbox

async with cua.sandbox(local=True) as sb: await sb.screenshot()

Localhost (no sandbox, direct host control)

async with cua.localhost() as host: await host.mouse.click(100, 200)

Classes

Class	Description
`Image`	Immutable, chainable image specification.
`Localhost`	Direct host control via cua_auto — no sandboxing.
`Sandbox`	A sandboxed computer environment.
`SandboxInfo`	Metadata for a local or cloud sandbox.
`CloudTransport`	Transport that provisions / connects to a CUA cloud VM.

Functions

Function	Description
`login`	Open the CUA login page in a browser and store credentials.
`whoami`	Return info about the authenticated user.
`configure`	Set global configuration for the CUA SDK.

Image

Immutable, chainable image specification.

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

Constructor

Image(self, os_type: str, distro: str, version: str, kind: Optional[str] = None, _layers: Tuple[Dict[str, Any], ...] = (), _env: Tuple[Tuple[str, str], ...] = (), _ports: Tuple[int, ...] = (), _files: Tuple[Tuple[str, str], ...] = (), _registry: Optional[str] = None, _disk_path: Optional[str] = None, _agent_type: Optional[str] = None) -> None

Attributes

Name	Type	Description
`os_type`	`str`
`distro`	`str`
`version`	`str`
`kind`	`Optional[str]`

Methods

Image.linux

def linux(cls, distro: str = 'ubuntu', version: str = '24.04', kind: str = 'container') -> Image

Linux image. Defaults to 'container' (Docker/XFCE). Use kind='vm' for QEMU.

Image.macos

def macos(cls, version: str = '26', kind: str = 'vm') -> Image

macOS image. Always a VM (Apple Virtualization / Lume).

Supported versions: "15" / "sequoia", "26" / "tahoe".

Image.windows

def windows(cls, version: str = '11', kind: str = 'vm') -> Image

Windows image. Always a VM (QEMU or Hyper-V).

Image.android

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

Android image. Always a VM (QEMU emulator).

Image.from_registry

def from_registry(cls, ref: str) -> Image

Create an image from a registry reference. kind is resolved after pull.

Image.from_file

def from_file(cls, 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.

Parameters:

Name	Type	Description
`path`	`Any`	Local file path or URL (http/https).
`os_type`	`Any`	OS type hint ("linux", "windows", "macos", "android").
`kind`	`Any`	"vm" or "container".
`agent_type`	`Any`	Agent type hint (e.g. "osworld" for OSWorld Flask server).

Image.from_dict

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

Reconstruct an Image from a serialized spec dict.

Image.apt_install

def apt_install(self, packages: str = ()) -> Image

Install packages via apt (Linux only).

Image.brew_install

def brew_install(self, packages: str = ()) -> Image

Install packages via Homebrew (macOS).

Image.choco_install

def choco_install(self, packages: str = ()) -> Image

Install packages via Chocolatey (Windows).

Image.winget_install

def winget_install(self, packages: str = ()) -> Image

Install packages via winget (Windows).

Image.apk_install

def apk_install(self, apk_paths: str = ()) -> Image

Install APK files via adb (Android only).

Image.pwa_install

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

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

Bubblewrap reads the Web App Manifest at manifest_url, generates a Trusted Web Activity (TWA) APK signed with keystore, and installs it via adb. The keystore's SHA-256 fingerprint must match what the server serves from /.well-known/assetlinks.json so Chrome trusts the TWA and skips the browser UI entirely.

Parameters:

Name	Type	Description
`manifest_url`	`Any`	Full URL to the PWA's `manifest.json`.

Example:

package_name: Android package ID.  Defaults to a reversed-hostname
              derivation (e.g. ``"com.example.app"``).
keystore:     Path to a ``*.keystore`` / ``*.jks`` file.  When
              omitted a fresh keystore is generated and cached under
              ``~/.cua/cua-sandbox/pwa-cache/``.  Pass the keystore
              bundled with your PWA repo so the fingerprint is
              deterministic and pre-loaded into ``assetlinks.json``.
keystore_alias:    Key alias inside the keystore (default ``"android"``).
keystore_password: Password for both the store and the key
                   (default ``"android"``).

Image.uv_install

def uv_install(self, packages: str = ()) -> Image

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

Image.pip_install

def pip_install(self, packages: str = ()) -> Image

Install Python packages via pip.

Image.run

def run(self, command: str) -> Image

Run a shell command during image build.

Image.env

def env(self, variables: str = {}) -> Image

Set environment variables.

Image.copy

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

Copy a file into the image.

Image.expose

def expose(self, port: int) -> Image

Expose a port.

Image.to_dict

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

Serialize to a plain dict suitable for JSON or cloud API.

Image.to_cloud_init

def to_cloud_init(self) -> str

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

Localhost

Direct host control via cua_auto — no sandboxing.

Constructor

Localhost(self) -> None

Attributes

Name	Type	Description
`screen`	`Any`
`mouse`	`Any`
`keyboard`	`Any`
`clipboard`	`Any`
`shell`	`Any`
`window`	`Any`
`terminal`	`Any`

Methods

Localhost.disconnect

async def disconnect(self) -> None

Localhost.screenshot

async def screenshot(self, text: Optional[str] = None) -> bytes

Localhost.screenshot_base64

async def screenshot_base64(self, text: Optional[str] = None) -> str

Localhost.get_environment

async def get_environment(self) -> str

Localhost.get_dimensions

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

Localhost.connect

def connect(cls) -> '_ConnectResult'

Connect to the local machine.

Supports both await and async with.

Example:

# plain await
host = await Localhost.connect()
await host.shell.run("echo hello")
await host.disconnect()

# context manager
async with Localhost.connect() as host:
    await host.shell.run("echo hello")

Sandbox

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. For unsandboxed host control, use :func:cua_sandbox.localhost.

There are three ways to obtain a Sandbox:

Persistent — provision and keep alive after the script exits::

sb = await Sandbox.create(Image.desktop("ubuntu")) await sb.shell.run("whoami") await sb.disconnect()
Connect — attach to an already-running sandbox by name::

sb = await Sandbox.connect("my-sandbox") await sb.screenshot() await sb.disconnect()
Ephemeral — auto-destroyed when the async with block exits::

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

Constructor

Sandbox(self, transport: Transport, name: Optional[str] = None, _runtime: Optional[Runtime] = None, _runtime_info: Optional[RuntimeInfo] = None, _ephemeral: Optional[bool] = None)

Attributes

Name	Type	Description
`name`	`Any`
`screen`	`Any`
`mouse`	`Any`
`keyboard`	`Any`
`clipboard`	`Any`
`shell`	`Any`
`window`	`Any`
`terminal`	`Any`
`mobile`	`Any`
`tunnel`	`Any`

Methods

Sandbox.disconnect

async def disconnect(self) -> None

Drop the transport connection. The sandbox keeps running.

Sandbox.destroy

async def destroy(self) -> None

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

Sandbox.screenshot

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

Sandbox.screenshot_base64

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

Sandbox.get_environment

async def get_environment(self) -> str

Sandbox.get_display_url

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

Return a URL to view this sandbox's display.

Parameters:

Name	Type	Description
`share`	`Any`	If True, return a public link with embedded credentials (cloud only). If False, return a direct connection URL.

Sandbox.get_dimensions

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

Sandbox.create

async def create(cls, image: Image, name: Optional[str] = None, api_key: Optional[str] = None, local: bool = False, runtime: Optional['Runtime'] = None, cpu: Optional[int] = None, memory_mb: Optional[int] = None, disk_gb: Optional[int] = None, region: str = 'us-east-1') -> 'Sandbox'

Provision a new persistent sandbox and return it connected.

The sandbox is kept alive after your script exits — call close() when you are done, or use :meth:ephemeral if you want it destroyed automatically.

Parameters:

Name	Type	Description
`image`	`Any`	Image to run (e.g. `Image.desktop("ubuntu")`).
`name`	`Any`	Optional name to assign to the sandbox.
`api_key`	`Any`	CUA API key for cloud sandboxes.
`local`	`Any`	Use a local runtime instead of cloud.
`runtime`	`Any`	Explicit runtime backend (DockerRuntime, QEMURuntime, etc.).
`cpu`	`Any`	Number of CPUs for the cloud sandbox.
`memory_mb`	`Any`	Memory in MB for the cloud sandbox.
`disk_gb`	`Any`	Disk size in GB for the cloud sandbox.
`region`	`Any`	Cloud region (default `"us-east-1"`).

Example:

sb = await Sandbox.create(Image.desktop("ubuntu"))
await sb.shell.run("uname -a")
print(sb.name)  # save to reconnect later
await sb.disconnect()

Sandbox.connect

def connect(cls, name: str, api_key: Optional[str] = None, local: bool = False, 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') -> '_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.

Parameters:

Name	Type	Description
`name`	`Any`	Name of the existing sandbox.
`api_key`	`Any`	CUA API key for cloud sandboxes.
`ws_url`	`Any`	WebSocket URL for a remote computer-server.
`http_url`	`Any`	HTTP base URL for a remote computer-server.
`container_name`	`Any`	Container name for cloud auth (HTTP transport).
`region`	`Any`	Cloud region (default `"us-east-1"`).

Example:

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

Sandbox.ephemeral

async def ephemeral(cls, image: Image, name: Optional[str] = None, api_key: Optional[str] = None, local: bool = False, runtime: Optional['Runtime'] = None, cpu: Optional[int] = None, memory_mb: Optional[int] = None, disk_gb: Optional[int] = None, region: str = 'us-east-1') -> AsyncIterator['Sandbox']

Create an ephemeral sandbox that is automatically destroyed on exit.

Parameters:

Name	Type	Description
`image`	`Any`	Image to run (e.g. `Image.desktop("ubuntu")`).
`name`	`Any`	Optional name to assign to the sandbox.
`api_key`	`Any`	CUA API key for cloud sandboxes.
`local`	`Any`	Use a local runtime instead of cloud.
`runtime`	`Any`	Explicit runtime backend (DockerRuntime, QEMURuntime, etc.).
`cpu`	`Any`	Number of CPUs for the cloud sandbox.
`memory_mb`	`Any`	Memory in MB for the cloud sandbox.
`disk_gb`	`Any`	Disk size in GB for the cloud sandbox.
`region`	`Any`	Cloud region (default `"us-east-1"`).

Example:

async with Sandbox.ephemeral(Image.desktop("ubuntu")) as sb:
    await sb.shell.run("whoami")
# sandbox is destroyed here

Sandbox.list

async def list(cls, local: bool = False, api_key: Optional[str] = None) -> 'list[SandboxInfo]'

List running and suspended sandboxes.

Parameters:

Name	Type	Description
`local`	`Any`	If True, list local sandboxes (Lume, Docker, QEMU). If False, list cloud sandboxes.
`api_key`	`Any`	CUA API key for cloud sandboxes.

Sandbox.get_info

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

Get metadata for a specific sandbox.

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, look up in local runtimes.
`api_key`	`Any`	CUA API key for cloud.

Sandbox.suspend

async def suspend(cls, name: str, local: bool = False, api_key: Optional[str] = None) -> None

Suspend a running sandbox (save state).

For local QEMU bare-metal: saves a QMP snapshot then quits the process. For local Docker/QEMU-docker: pauses the container. For local Lume: stops the Lume VM (Lume persists state). For cloud: calls POST /v1/vms/{name}/stop.

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, operate on a local sandbox.
`api_key`	`Any`	CUA API key for cloud.

Sandbox.resume

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

Resume a suspended sandbox and return a connected Sandbox.

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, resume a local sandbox.
`api_key`	`Any`	CUA API key for cloud.

Returns: A connected Sandbox ready to use.

Sandbox.restart

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

Restart a sandbox (suspend then resume) and return a connected Sandbox.

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, restart a local sandbox.
`api_key`	`Any`	CUA API key for cloud.

Returns: A connected Sandbox ready to use.

Sandbox.delete

async def delete(cls, name: str, local: bool = False, 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}.

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, delete a local sandbox.
`api_key`	`Any`	CUA API key for cloud.

SandboxInfo

Metadata for a local or cloud sandbox.

Constructor

SandboxInfo(self, 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) -> None

Attributes

Name	Type	Description
`name`	`str`
`status`	`str`
`source`	`str`
`os_type`	`Optional[str]`
`host`	`Optional[str]`
`vnc_url`	`Optional[str]`
`api_url`	`Optional[str]`
`created_at`	`Optional[str]`

CloudTransport

Inherits from: Transport

Transport that provisions / connects to a CUA cloud VM.

Constructor

CloudTransport(self, name: Optional[str] = None, api_key: Optional[str] = None, base_url: Optional[str] = None, image: Optional[Any] = None, cpu: Optional[int] = None, memory_mb: Optional[int] = None, disk_gb: Optional[int] = None, region: str = 'us-east-1')

Attributes

Name	Type	Description
`name`	`Optional[str]`

Methods

CloudTransport.connect

async def connect(self) -> None

CloudTransport.disconnect

async def disconnect(self) -> None

CloudTransport.delete_vm

async def delete_vm(self) -> None

Delete the cloud VM via the platform API.

CloudTransport.suspend_vm

async def suspend_vm(self) -> None

Stop (suspend) the cloud VM.

CloudTransport.resume_vm

async def resume_vm(self) -> None

Start (resume) the cloud VM.

CloudTransport.restart_vm

async def restart_vm(self) -> None

Restart the cloud VM.

CloudTransport.send

async def send(self, action: str, params: Any = {}) -> Any

CloudTransport.screenshot

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

CloudTransport.get_screen_size

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

CloudTransport.get_environment

async def get_environment(self) -> str

CloudTransport.get_display_url

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

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.

configure

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

Set global configuration for the CUA SDK.

Parameters:

Name	Type	Description
`api_key`	`Any`	API key for cloud sandboxes.
`base_url`	`Any`	Base URL for the CUA cloud API.

localhost

Localhost — wraps cua_auto directly. No sandbox, no computer-server.

All cua_auto calls are sync; async wrappers use asyncio.to_thread().

Usage::

from cua_sandbox import localhost

async with localhost() as host: await host.mouse.click(100, 200) img = await host.screen.screenshot()

LocalTransport

Inherits from: Transport

Transport that routes commands directly to cua_auto modules.

Methods

LocalTransport.connect

async def connect(self) -> None

LocalTransport.disconnect

async def disconnect(self) -> None

LocalTransport.send

async def send(self, action: str, params: Any = {}) -> Any

Dispatch a named command to the appropriate cua_auto module.

LocalTransport.screenshot

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

LocalTransport.get_screen_size

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

LocalTransport.get_environment

async def get_environment(self) -> str

Localhost

Direct host control via cua_auto — no sandboxing.

Constructor

Localhost(self) -> None

Attributes

Name	Type	Description
`screen`	`Any`
`mouse`	`Any`
`keyboard`	`Any`
`clipboard`	`Any`
`shell`	`Any`
`window`	`Any`
`terminal`	`Any`

Methods

Localhost.disconnect

async def disconnect(self) -> None

Localhost.screenshot

async def screenshot(self, text: Optional[str] = None) -> bytes

Localhost.screenshot_base64

async def screenshot_base64(self, text: Optional[str] = None) -> str

Localhost.get_environment

async def get_environment(self) -> str

Localhost.get_dimensions

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

Localhost.connect

def connect(cls) -> '_ConnectResult'

Connect to the local machine.

Supports both await and async with.

Example:

# plain await
host = await Localhost.connect()
await host.shell.run("echo hello")
await host.disconnect()

# context manager
async with Localhost.connect() as host:
    await host.shell.run("echo hello")

localhost

async def localhost() -> AsyncIterator[Localhost]

Async context manager yielding a Localhost instance.

.. deprecated:: Prefer Localhost.connect() instead.

sandbox

Sandbox class — the primary entry point for sandboxed environments.

Exposes .mouse, .keyboard, .screen, .clipboard, .shell, .window, .terminal as interface objects backed by a Transport.

Usage::

from cua_sandbox import Sandbox, Image

Provision a new persistent sandbox

sb = await Sandbox.create(Image.desktop("ubuntu")) await sb.shell.run("uname -a") await sb.disconnect()

Connect to an existing sandbox by name (plain await or async with)

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

async with Sandbox.connect("my-sandbox") as sb: # disconnects on exit await sb.screenshot()

Ephemeral — auto-destroyed on exit

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

Image

Immutable, chainable image specification.

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

Constructor

Image(self, os_type: str, distro: str, version: str, kind: Optional[str] = None, _layers: Tuple[Dict[str, Any], ...] = (), _env: Tuple[Tuple[str, str], ...] = (), _ports: Tuple[int, ...] = (), _files: Tuple[Tuple[str, str], ...] = (), _registry: Optional[str] = None, _disk_path: Optional[str] = None, _agent_type: Optional[str] = None) -> None

Attributes

Name	Type	Description
`os_type`	`str`
`distro`	`str`
`version`	`str`
`kind`	`Optional[str]`

Methods

Image.linux

def linux(cls, distro: str = 'ubuntu', version: str = '24.04', kind: str = 'container') -> Image

Linux image. Defaults to 'container' (Docker/XFCE). Use kind='vm' for QEMU.

Image.macos

def macos(cls, version: str = '26', kind: str = 'vm') -> Image

macOS image. Always a VM (Apple Virtualization / Lume).

Supported versions: "15" / "sequoia", "26" / "tahoe".

Image.windows

def windows(cls, version: str = '11', kind: str = 'vm') -> Image

Windows image. Always a VM (QEMU or Hyper-V).

Image.android

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

Android image. Always a VM (QEMU emulator).

Image.from_registry

def from_registry(cls, ref: str) -> Image

Create an image from a registry reference. kind is resolved after pull.

Image.from_file

def from_file(cls, 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.

Parameters:

Name	Type	Description
`path`	`Any`	Local file path or URL (http/https).
`os_type`	`Any`	OS type hint ("linux", "windows", "macos", "android").
`kind`	`Any`	"vm" or "container".
`agent_type`	`Any`	Agent type hint (e.g. "osworld" for OSWorld Flask server).

Image.from_dict

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

Reconstruct an Image from a serialized spec dict.

Image.apt_install

def apt_install(self, packages: str = ()) -> Image

Install packages via apt (Linux only).

Image.brew_install

def brew_install(self, packages: str = ()) -> Image

Install packages via Homebrew (macOS).

Image.choco_install

def choco_install(self, packages: str = ()) -> Image

Install packages via Chocolatey (Windows).

Image.winget_install

def winget_install(self, packages: str = ()) -> Image

Install packages via winget (Windows).

Image.apk_install

def apk_install(self, apk_paths: str = ()) -> Image

Install APK files via adb (Android only).

Image.pwa_install

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

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

Parameters:

Name	Type	Description
`manifest_url`	`Any`	Full URL to the PWA's `manifest.json`.

Example:

package_name: Android package ID.  Defaults to a reversed-hostname
              derivation (e.g. ``"com.example.app"``).
keystore:     Path to a ``*.keystore`` / ``*.jks`` file.  When
              omitted a fresh keystore is generated and cached under
              ``~/.cua/cua-sandbox/pwa-cache/``.  Pass the keystore
              bundled with your PWA repo so the fingerprint is
              deterministic and pre-loaded into ``assetlinks.json``.
keystore_alias:    Key alias inside the keystore (default ``"android"``).
keystore_password: Password for both the store and the key
                   (default ``"android"``).

Image.uv_install

def uv_install(self, packages: str = ()) -> Image

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

Image.pip_install

def pip_install(self, packages: str = ()) -> Image

Install Python packages via pip.

Image.run

def run(self, command: str) -> Image

Run a shell command during image build.

Image.env

def env(self, variables: str = {}) -> Image

Set environment variables.

Image.copy

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

Copy a file into the image.

Image.expose

def expose(self, port: int) -> Image

Expose a port.

Image.to_dict

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

Serialize to a plain dict suitable for JSON or cloud API.

Image.to_cloud_init

def to_cloud_init(self) -> str

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

Transport

Inherits from: ABC

Base class for all transports.

Methods

Transport.connect

async def connect(self) -> None

Establish the transport connection.

Transport.disconnect

async def disconnect(self) -> None

Tear down the transport connection.

Transport.send

async def send(self, action: str, params: Any = {}) -> Any

Send a command and return the result.

Transport.screenshot

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

Capture a screenshot and return raw image bytes.

Parameters:

Name	Type	Description
`format`	`Any`	"png" (lossless, default) or "jpeg" (lossy, ~5-10x smaller).
`quality`	`Any`	JPEG quality 1-95, ignored for PNG.

Transport.get_screen_size

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

Return {"width": ..., "height": ...}.

Transport.get_environment

async def get_environment(self) -> str

Return 'windows', 'mac', 'linux', or 'browser'.

Transport.forward_tunnel

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

Forward sandbox_port to an available host port and return info.

Subclasses that support tunnelling must override this method. The returned :class:~cua_sandbox.interfaces.tunnel.TunnelInfo must have host and port set to the host-side address.

Transport.close_tunnel

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

Release a previously forwarded port or socket. No-op by default.

Transport.get_display_url

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

Return a URL to view this sandbox's display.

Parameters:

Name	Type	Description
`share`	`Any`	If True, return a public link with embedded credentials (cloud only). If False, return a direct connection URL (localhost VNC for local runtimes; auth-gated URL for cloud). Raises NotImplementedError for transports that don't expose a display (e.g. HTTP, ADB).

CloudTransport

Inherits from: Transport

Transport that provisions / connects to a CUA cloud VM.

Constructor

CloudTransport(self, name: Optional[str] = None, api_key: Optional[str] = None, base_url: Optional[str] = None, image: Optional[Any] = None, cpu: Optional[int] = None, memory_mb: Optional[int] = None, disk_gb: Optional[int] = None, region: str = 'us-east-1')

Attributes

Name	Type	Description
`name`	`Optional[str]`

Methods

CloudTransport.connect

async def connect(self) -> None

CloudTransport.disconnect

async def disconnect(self) -> None

CloudTransport.delete_vm

async def delete_vm(self) -> None

Delete the cloud VM via the platform API.

CloudTransport.suspend_vm

async def suspend_vm(self) -> None

Stop (suspend) the cloud VM.

CloudTransport.resume_vm

async def resume_vm(self) -> None

Start (resume) the cloud VM.

CloudTransport.restart_vm

async def restart_vm(self) -> None

Restart the cloud VM.

CloudTransport.send

async def send(self, action: str, params: Any = {}) -> Any

CloudTransport.screenshot

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

CloudTransport.get_screen_size

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

CloudTransport.get_environment

async def get_environment(self) -> str

CloudTransport.get_display_url

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

HTTPTransport

Inherits from: Transport

Transport that communicates with computer-server over HTTP POST /cmd (SSE).

Constructor

HTTPTransport(self, base_url: str, api_key: Optional[str] = None, container_name: Optional[str] = None, timeout: float = 30.0)

Methods

HTTPTransport.connect

async def connect(self) -> None

HTTPTransport.disconnect

async def disconnect(self) -> None

HTTPTransport.send

async def send(self, action: str, params: Any = {}) -> Any

HTTPTransport.screenshot

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

HTTPTransport.get_screen_size

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

HTTPTransport.get_environment

async def get_environment(self) -> str

WebSocketTransport

Inherits from: Transport

Transport that communicates with computer-server over WebSocket.

Constructor

WebSocketTransport(self, url: str, api_key: Optional[str] = None)

Methods

WebSocketTransport.connect

async def connect(self) -> None

WebSocketTransport.disconnect

async def disconnect(self) -> None

WebSocketTransport.send

async def send(self, action: str, params: Any = {}) -> Any

WebSocketTransport.screenshot

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

WebSocketTransport.get_screen_size

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

WebSocketTransport.get_environment

async def get_environment(self) -> str

Runtime

Inherits from: ABC

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

Methods

Runtime.start

async def start(self, image: Image, name: str, opts = {}) -> RuntimeInfo

Start a VM/container and return connection info.

Runtime.stop

async def stop(self, name: str) -> None

Stop and clean up a VM/container.

Runtime.is_ready

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

Wait until the computer-server inside is reachable.

Runtime.suspend

async def suspend(self, name: str) -> None

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

Runtime.resume

async def resume(self, image: 'Image', name: str, opts = {}) -> RuntimeInfo

Resume a suspended VM and return its RuntimeInfo.

Runtime.list

async def list(self) -> list[dict]

List known VMs managed by this runtime.

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

RuntimeInfo

Connection info returned after a runtime spins up.

Constructor

RuntimeInfo(self, 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 = 8000, 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) -> None

Attributes

Name	Type	Description
`host`	`str`
`api_port`	`int`
`vnc_port`	`Optional[int]`
`api_key`	`Optional[str]`
`container_id`	`Optional[str]`
`name`	`Optional[str]`
`qmp_port`	`Optional[int]`
`environment`	`Optional[str]`
`agent_type`	`Optional[str]`
`guest_server_port`	`int`
`ssh_port`	`Optional[int]`
`ssh_username`	`Optional[str]`
`ssh_password`	`Optional[str]`
`ssh_key_filename`	`Optional[str]`
`vnc_host`	`Optional[str]`
`vnc_password`	`Optional[str]`
`grpc_port`	`Optional[int]`

SandboxInfo

Metadata for a local or cloud sandbox.

Constructor

SandboxInfo(self, 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) -> None

Attributes

Name	Type	Description
`name`	`str`
`status`	`str`
`source`	`str`
`os_type`	`Optional[str]`
`host`	`Optional[str]`
`vnc_url`	`Optional[str]`
`api_url`	`Optional[str]`
`created_at`	`Optional[str]`

Sandbox

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. For unsandboxed host control, use :func:cua_sandbox.localhost.

There are three ways to obtain a Sandbox:

Persistent — provision and keep alive after the script exits::

sb = await Sandbox.create(Image.desktop("ubuntu")) await sb.shell.run("whoami") await sb.disconnect()
Connect — attach to an already-running sandbox by name::

sb = await Sandbox.connect("my-sandbox") await sb.screenshot() await sb.disconnect()
Ephemeral — auto-destroyed when the async with block exits::

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

Constructor

Sandbox(self, transport: Transport, name: Optional[str] = None, _runtime: Optional[Runtime] = None, _runtime_info: Optional[RuntimeInfo] = None, _ephemeral: Optional[bool] = None)

Attributes

Name	Type	Description
`name`	`Any`
`screen`	`Any`
`mouse`	`Any`
`keyboard`	`Any`
`clipboard`	`Any`
`shell`	`Any`
`window`	`Any`
`terminal`	`Any`
`mobile`	`Any`
`tunnel`	`Any`

Methods

Sandbox.disconnect

async def disconnect(self) -> None

Drop the transport connection. The sandbox keeps running.

Sandbox.destroy

async def destroy(self) -> None

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

Sandbox.screenshot

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

Sandbox.screenshot_base64

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

Sandbox.get_environment

async def get_environment(self) -> str

Sandbox.get_display_url

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

Return a URL to view this sandbox's display.

Parameters:

Name	Type	Description
`share`	`Any`	If True, return a public link with embedded credentials (cloud only). If False, return a direct connection URL.

Sandbox.get_dimensions

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

Sandbox.create

async def create(cls, image: Image, name: Optional[str] = None, api_key: Optional[str] = None, local: bool = False, runtime: Optional['Runtime'] = None, cpu: Optional[int] = None, memory_mb: Optional[int] = None, disk_gb: Optional[int] = None, region: str = 'us-east-1') -> 'Sandbox'

Provision a new persistent sandbox and return it connected.

The sandbox is kept alive after your script exits — call close() when you are done, or use :meth:ephemeral if you want it destroyed automatically.

Parameters:

Name	Type	Description
`image`	`Any`	Image to run (e.g. `Image.desktop("ubuntu")`).
`name`	`Any`	Optional name to assign to the sandbox.
`api_key`	`Any`	CUA API key for cloud sandboxes.
`local`	`Any`	Use a local runtime instead of cloud.
`runtime`	`Any`	Explicit runtime backend (DockerRuntime, QEMURuntime, etc.).
`cpu`	`Any`	Number of CPUs for the cloud sandbox.
`memory_mb`	`Any`	Memory in MB for the cloud sandbox.
`disk_gb`	`Any`	Disk size in GB for the cloud sandbox.
`region`	`Any`	Cloud region (default `"us-east-1"`).

Example:

sb = await Sandbox.create(Image.desktop("ubuntu"))
await sb.shell.run("uname -a")
print(sb.name)  # save to reconnect later
await sb.disconnect()

Sandbox.connect

def connect(cls, name: str, api_key: Optional[str] = None, local: bool = False, 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') -> '_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.

Parameters:

Name	Type	Description
`name`	`Any`	Name of the existing sandbox.
`api_key`	`Any`	CUA API key for cloud sandboxes.
`ws_url`	`Any`	WebSocket URL for a remote computer-server.
`http_url`	`Any`	HTTP base URL for a remote computer-server.
`container_name`	`Any`	Container name for cloud auth (HTTP transport).
`region`	`Any`	Cloud region (default `"us-east-1"`).

Example:

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

Sandbox.ephemeral

async def ephemeral(cls, image: Image, name: Optional[str] = None, api_key: Optional[str] = None, local: bool = False, runtime: Optional['Runtime'] = None, cpu: Optional[int] = None, memory_mb: Optional[int] = None, disk_gb: Optional[int] = None, region: str = 'us-east-1') -> AsyncIterator['Sandbox']

Create an ephemeral sandbox that is automatically destroyed on exit.

Parameters:

Name	Type	Description
`image`	`Any`	Image to run (e.g. `Image.desktop("ubuntu")`).
`name`	`Any`	Optional name to assign to the sandbox.
`api_key`	`Any`	CUA API key for cloud sandboxes.
`local`	`Any`	Use a local runtime instead of cloud.
`runtime`	`Any`	Explicit runtime backend (DockerRuntime, QEMURuntime, etc.).
`cpu`	`Any`	Number of CPUs for the cloud sandbox.
`memory_mb`	`Any`	Memory in MB for the cloud sandbox.
`disk_gb`	`Any`	Disk size in GB for the cloud sandbox.
`region`	`Any`	Cloud region (default `"us-east-1"`).

Example:

async with Sandbox.ephemeral(Image.desktop("ubuntu")) as sb:
    await sb.shell.run("whoami")
# sandbox is destroyed here

Sandbox.list

async def list(cls, local: bool = False, api_key: Optional[str] = None) -> 'list[SandboxInfo]'

List running and suspended sandboxes.

Parameters:

Name	Type	Description
`local`	`Any`	If True, list local sandboxes (Lume, Docker, QEMU). If False, list cloud sandboxes.
`api_key`	`Any`	CUA API key for cloud sandboxes.

Sandbox.get_info

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

Get metadata for a specific sandbox.

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, look up in local runtimes.
`api_key`	`Any`	CUA API key for cloud.

Sandbox.suspend

async def suspend(cls, name: str, local: bool = False, api_key: Optional[str] = None) -> None

Suspend a running sandbox (save state).

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, operate on a local sandbox.
`api_key`	`Any`	CUA API key for cloud.

Sandbox.resume

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

Resume a suspended sandbox and return a connected Sandbox.

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, resume a local sandbox.
`api_key`	`Any`	CUA API key for cloud.

Returns: A connected Sandbox ready to use.

Sandbox.restart

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

Restart a sandbox (suspend then resume) and return a connected Sandbox.

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, restart a local sandbox.
`api_key`	`Any`	CUA API key for cloud.

Returns: A connected Sandbox ready to use.

Sandbox.delete

async def delete(cls, name: str, local: bool = False, 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}.

Parameters:

Name	Type	Description
`name`	`Any`	Sandbox name.
`local`	`Any`	If True, delete a local sandbox.
`api_key`	`Any`	CUA API key for cloud.

sandbox

async def sandbox(local: bool = False, ws_url: Optional[str] = None, http_url: Optional[str] = None, api_key: Optional[str] = None, container_name: Optional[str] = None, image: Optional[Image] = None, runtime: Optional['Runtime'] = 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') -> AsyncIterator[Sandbox]

Async context manager for a sandboxed environment.

.. deprecated:: Prefer Sandbox.create(), Sandbox.connect(), or Sandbox.ephemeral() instead.

image

Image builder — pure-data immutable chained builder for sandbox images.

Supports Linux, macOS, and Windows constructors. Serializes to a spec dict for cloud API or cloud-init consumption.

Usage::

from cua_sandbox import Image

img = ( Image.linux("ubuntu", "24.04") .apt_install("curl", "git", "build-essential") .pip_install("numpy", "pandas") .env(MY_VAR="hello") .run("echo 'setup complete'") .expose(8080) )

spec = img.to_dict()

Image

Immutable, chainable image specification.

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

Constructor

Image(self, os_type: str, distro: str, version: str, kind: Optional[str] = None, _layers: Tuple[Dict[str, Any], ...] = (), _env: Tuple[Tuple[str, str], ...] = (), _ports: Tuple[int, ...] = (), _files: Tuple[Tuple[str, str], ...] = (), _registry: Optional[str] = None, _disk_path: Optional[str] = None, _agent_type: Optional[str] = None) -> None

Attributes

Name	Type	Description
`os_type`	`str`
`distro`	`str`
`version`	`str`
`kind`	`Optional[str]`

Methods

Image.linux

def linux(cls, distro: str = 'ubuntu', version: str = '24.04', kind: str = 'container') -> Image

Linux image. Defaults to 'container' (Docker/XFCE). Use kind='vm' for QEMU.

Image.macos

def macos(cls, version: str = '26', kind: str = 'vm') -> Image

macOS image. Always a VM (Apple Virtualization / Lume).

Supported versions: "15" / "sequoia", "26" / "tahoe".

Image.windows

def windows(cls, version: str = '11', kind: str = 'vm') -> Image

Windows image. Always a VM (QEMU or Hyper-V).

Image.android

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

Android image. Always a VM (QEMU emulator).

Image.from_registry

def from_registry(cls, ref: str) -> Image

Create an image from a registry reference. kind is resolved after pull.

Image.from_file

def from_file(cls, 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.

Parameters:

Name	Type	Description
`path`	`Any`	Local file path or URL (http/https).
`os_type`	`Any`	OS type hint ("linux", "windows", "macos", "android").
`kind`	`Any`	"vm" or "container".
`agent_type`	`Any`	Agent type hint (e.g. "osworld" for OSWorld Flask server).

Image.from_dict

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

Reconstruct an Image from a serialized spec dict.

Image.apt_install

def apt_install(self, packages: str = ()) -> Image

Install packages via apt (Linux only).

Image.brew_install

def brew_install(self, packages: str = ()) -> Image

Install packages via Homebrew (macOS).

Image.choco_install

def choco_install(self, packages: str = ()) -> Image

Install packages via Chocolatey (Windows).

Image.winget_install

def winget_install(self, packages: str = ()) -> Image

Install packages via winget (Windows).

Image.apk_install

def apk_install(self, apk_paths: str = ()) -> Image

Install APK files via adb (Android only).

Image.pwa_install

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

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

Parameters:

Name	Type	Description
`manifest_url`	`Any`	Full URL to the PWA's `manifest.json`.

Example:

package_name: Android package ID.  Defaults to a reversed-hostname
              derivation (e.g. ``"com.example.app"``).
keystore:     Path to a ``*.keystore`` / ``*.jks`` file.  When
              omitted a fresh keystore is generated and cached under
              ``~/.cua/cua-sandbox/pwa-cache/``.  Pass the keystore
              bundled with your PWA repo so the fingerprint is
              deterministic and pre-loaded into ``assetlinks.json``.
keystore_alias:    Key alias inside the keystore (default ``"android"``).
keystore_password: Password for both the store and the key
                   (default ``"android"``).

Image.uv_install

def uv_install(self, packages: str = ()) -> Image

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

Image.pip_install

def pip_install(self, packages: str = ()) -> Image

Install Python packages via pip.

Image.run

def run(self, command: str) -> Image

Run a shell command during image build.

Image.env

def env(self, variables: str = {}) -> Image

Set environment variables.

Image.copy

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

Copy a file into the image.

Image.expose

def expose(self, port: int) -> Image

Expose a port.

Image.to_dict

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

Serialize to a plain dict suitable for JSON or cloud API.

Image.to_cloud_init

def to_cloud_init(self) -> str

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

builder

Image builder — build QEMU VM images from Image layer specs.

LayerExecutor

Execute Image layer specs against a running computer-server.

Constructor

LayerExecutor(self, base_url: str, timeout: float = 600)

Attributes

Name	Type	Description
`base_url`	`Any`
`timeout`	`Any`

Methods

LayerExecutor.run_command

async def run_command(self, command: str, timeout: float | None = None) -> dict

Run a shell command via computer-server and return the result.

LayerExecutor.execute_layer

async def execute_layer(self, layer: dict) -> dict

Execute a single Image layer and return the result.

LayerExecutor.execute_layers

async def execute_layers(self, layers: list[dict]) -> list[dict]

Execute all layers sequentially. Raises on first failure.

create_overlay

def create_overlay(backing: Path, overlay: Path) -> Path

Create a qcow2 overlay with the given backing file.

interfaces

Clipboard

Clipboard read/write.

Constructor

Clipboard(self, transport: Transport)

Methods

Clipboard.get

async def get(self) -> str

Return the current clipboard text.

Clipboard.set

async def set(self, text: str) -> None

Set the clipboard text.

Keyboard

Keyboard control.

Constructor

Keyboard(self, transport: Transport)

Methods

Keyboard.type

async def type(self, text: str) -> None

Type a string of text.

Keyboard.keypress

async def keypress(self, keys: Union[List[str], str]) -> None

Press a key combination (e.g. ["ctrl", "c"] or "enter").

Keyboard.key_down

async def key_down(self, key: str) -> None

Keyboard.key_up

async def key_up(self, key: str) -> None

Mobile

Mobile (Android) touch and hardware-key control.

All touch coordinates are in screen pixels. Single-touch methods use input tap/swipe via adb shell. Multi-touch gestures delegate to the multitouch_gesture transport action, which uses adb root + MT Protocol B sendevent for reliable injection on both local and cloud transports.

Constructor

Mobile(self, transport: Transport)

Methods

Mobile.tap

async def tap(self, x: int, y: int) -> None

Mobile.long_press

async def long_press(self, x: int, y: int, duration_ms: int = 1000) -> None

Mobile.double_tap

async def double_tap(self, x: int, y: int, delay: float = 0.1) -> None

Mobile.type_text

async def type_text(self, text: str) -> None

Mobile.swipe

async def swipe(self, x1: int, y1: int, x2: int, y2: int, duration_ms: int = 300) -> None

Mobile.scroll_up

async def scroll_up(self, x: int, y: int, distance: int = 600, duration_ms: int = 400) -> None

Mobile.scroll_down

async def scroll_down(self, x: int, y: int, distance: int = 600, duration_ms: int = 400) -> None

Mobile.scroll_left

async def scroll_left(self, x: int, y: int, distance: int = 400, duration_ms: int = 300) -> None

Mobile.scroll_right

async def scroll_right(self, x: int, y: int, distance: int = 400, duration_ms: int = 300) -> None

Mobile.fling

async def fling(self, x1: int, y1: int, x2: int, y2: int) -> None

Mobile.gesture

async def gesture(self, finger_paths: tuple[int, int] = (), duration_ms: int = 400, steps: int = 0) -> None

Inject an arbitrary N-finger gesture via MT Protocol B sendevent.

Each positional argument is a sequence of (x, y) waypoints for one finger. Pass an even number of (x, y) tuples and they are paired into start/end positions per finger::

two-finger pinch-out

await mobile.gesture( (cx - 20, cy), (cx - 200, cy), # finger 0 (cx + 20, cy), (cx + 200, cy), # finger 1 )

Delegates to the multitouch_gesture transport action, which uses adb root + MT Protocol B sendevent for reliable injection on both local ADB and cloud HTTP transports.

Parameters:

Name	Type	Description
`duration_ms`	`Any`	Total gesture duration in milliseconds.
`steps`	`Any`	Interpolation steps (0 = auto: duration_ms // 20, min 5).

Mobile.pinch_in

async def pinch_in(self, cx: int, cy: int, spread: int = 300, duration_ms: int = 400) -> None

Pinch-in (zoom out) with two real simultaneous fingers.

Mobile.pinch_out

async def pinch_out(self, cx: int, cy: int, spread: int = 300, duration_ms: int = 400) -> None

Pinch-out (zoom in) with two real simultaneous fingers.

Mobile.key

async def key(self, keycode: int) -> None

Mobile.home

async def home(self) -> None

Mobile.back

async def back(self) -> None

Mobile.recents

async def recents(self) -> None

Mobile.power

async def power(self) -> None

Mobile.volume_up

async def volume_up(self) -> None

Mobile.volume_down

async def volume_down(self) -> None

Mobile.enter

async def enter(self) -> None

Mobile.backspace

async def backspace(self) -> None

Mobile.wake

async def wake(self) -> None

Mobile.notifications

async def notifications(self) -> None

Mobile.close_notifications

async def close_notifications(self) -> None

Mouse

Mouse control.

Constructor

Mouse(self, transport: Transport)

Methods

Mouse.click

async def click(self, x: int, y: int, button: str = 'left') -> None

Mouse.right_click

async def right_click(self, x: int, y: int) -> None

Mouse.double_click

async def double_click(self, x: int, y: int) -> None

Mouse.move

async def move(self, x: int, y: int) -> None

Mouse.scroll

async def scroll(self, x: int, y: int, scroll_x: int = 0, scroll_y: int = 3) -> None

Mouse.mouse_down

async def mouse_down(self, x: int, y: int, button: str = 'left') -> None

Mouse.mouse_up

async def mouse_up(self, x: int, y: int, button: str = 'left') -> None

Mouse.drag

async def drag(self, start_x: int, start_y: int, end_x: int, end_y: int, button: str = 'left') -> None

Screen

Screen capture and info.

Constructor

Screen(self, transport: Transport)

Methods

Screen.screenshot

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

Capture a screenshot and return raw image bytes.

Parameters:

Name	Type	Description
`format`	`Any`	"png" (lossless, default) or "jpeg" (lossy, ~5-10x smaller).
`quality`	`Any`	JPEG quality 1-95, ignored for PNG.

Screen.screenshot_base64

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

Capture a screenshot and return as a base64-encoded string.

Screen.size

async def size(self) -> Tuple[int, int]

Return (width, height) of the screen.

Shell

Shell command execution.

Constructor

Shell(self, transport: Transport)

Methods

Shell.run

async def run(self, command: str, timeout: int = 30) -> CommandResult

Run a shell command and return the result.

Terminal

PTY terminal sessions.

Constructor

Terminal(self, transport: Transport)

Methods

Terminal.create

async def create(self, shell: str = 'bash', cols: int = 80, rows: int = 24) -> dict

Create a new PTY session. Returns {"pid": int, "cols": int, "rows": int}.

Terminal.send_input

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

Send input to a PTY session.

Terminal.resize

async def resize(self, pid: int, cols: int, rows: int) -> None

Resize a PTY session.

Terminal.close

async def close(self, pid: int) -> Optional[int]

Close a PTY session. Returns exit code.

Tunnel

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

Constructor

Tunnel(self, transport: Transport)

Methods

Tunnel.forward

def forward(self, 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 :class: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

A single forwarded port.

Constructor

TunnelInfo(self, host: str, port: int, sandbox_port: int) -> None

Attributes

Name	Type	Description
`host`	`Any`
`port`	`Any`
`sandbox_port`	`Any`
`url`	`str`

Methods

TunnelInfo.close

async def close(self) -> None

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

Window

Window management.

Constructor

Window(self, transport: Transport)

Methods

Window.get_active_title

async def get_active_title(self) -> str

Return the title of the currently focused window.

Was this page helpful?

Sandbox SDK API Reference