Cua client and local runtimes

The Cua handle (embedded or daemon client, configuration, SDK version) and the local container and VM runtimes.

Every program starts from one Cua handle, embedded in the process or connected to a running cua daemon, with the same API either way. Its accessors return the other objects. The config_* functions read and write the user defaults in $CUA_HOME/config.toml (cua config).

connect() without an address reads ~/.cua/daemon.json, then the socket ~/.cua/cua.sock. The connection is lazy: the first call fails with Transport when no daemon runs (cua daemon start).

Cua#

The entry point of the SDK: create one with Cua::embedded or Cua::connect, then reach sandboxes, Spaces and the rest through its accessors.

Cua.auto#

The default topology: the cua daemon this machine runs when one accepts connections (so the CLI, MCP clients and apps share sandboxes and Spaces), else the runtime in this process with config. A discovery file (~/.cua/daemon.json) or socket left by a daemon that exited is not a running daemon: the files are removed when their pid is provably dead, and this falls back to embedded. A daemon that is starting (~/.cua/daemon.starting) is waited for, up to 30 s. Probes with a local connect only (no RPC), except in a process inside an app bundle that ships its own cua: it uses only that build's daemon, and fails with CuaError.DaemonNotRunning naming the running one (another app's, or its own before a rebuild or update) rather than silently use it; <bundle>/Contents/MacOS/cua daemon start replaces it.

@classmethod
def auto(cls, config: CuaConfig) -> Cua

Returns Cua · Raises CuaError (DaemonNotRunning)

Cua.connect#

Connects to a running cua daemon. address is a socket path (unix: prefix optional) or a loopback URL (then token is required unless the discovery file has it); None uses ~/.cua/daemon.json, then ~/.cua/cua.sock. Connects lazily: the first call fails with DaemonNotRunning when no daemon runs; use Cua.info to check, or Cua.auto to fall back to the runtime in this process.

@classmethod
def connect(cls, address: Optional[str], token: Optional[str]) -> Cua

Returns Cua · Raises CuaError (DaemonNotRunning)

Cua.embedded#

Runs the SDK runtime in this process. Performs no I/O beyond reading the credential store when account_from_session is set (for the relay account).

@classmethod
def embedded(cls, config: CuaConfig) -> Cua

Returns Cua · Raises CuaError

Example

import cua
 
c = cua.embedded()  # the SDK runs in this process; cua.connect() uses a running `cua daemon`
print(c.mode())

Cua.info#

SDK (and daemon) identity. In daemon mode this is the connectivity check.

async def info(self) -> CuaInfo

Returns CuaInfo · Async · Raises CuaError

Cua.shutdown_daemon#

Asks a connected daemon to stop (daemon mode only).

async def shutdown_daemon(self) -> None

Async · Raises CuaError

Cua.spacesd#

Connects to cua-spacesd at url (host:port, http(s)://… or a relay URL) without a sandbox.

async def spacesd(self, url: str, token: Optional[str]) -> SpacesdClient

Returns SpacesdClient · Async · Raises CuaError

CuaConfig record#

Configuration of an embedded SDK runtime.

CuaInfo record#

Identity of the SDK (and daemon, when connected).

Returned by Cua.info.

CuaMode enum#

Where the SDK runtime lives.

CuaMode.EMBEDDED
CuaMode.DAEMON

cua_sdk_version#

Version of the SDK library.

def cua_sdk_version() -> str

Returns String

config_list#

Every setting with its effective value and source.

def config_list() -> List[ConfigEntry]

Returns Vec<ConfigEntry> · Raises CuaError

config_get#

One setting's effective value and source.

def config_get(key: str) -> ConfigEntry

Returns ConfigEntry · Raises CuaError

config_set#

Writes a setting to the config file (checked and normalised: default.on takes local or a provider). Returns its effective value, which an environment variable may still override.

def config_set(key: str, value: str) -> ConfigEntry

Returns ConfigEntry · Raises CuaError

config_unset#

Removes a setting from the config file. Returns its effective value afterwards (the environment variable or the built-in default).

def config_unset(key: str) -> ConfigEntry

Returns ConfigEntry · Raises CuaError

config_path#

The config file's path ($CUA_HOME/config.toml).

def config_path() -> str

Returns String

ConfigEntry record#

One setting's effective value.

Returned by config_get, config_list, config_set, config_unset.

locations#

Every location with the kinds and runtimes it offers (for pickers and help text).

def locations() -> List[LocationInfo]

Returns Vec<LocationInfo>

LocationInfo record#

The locations on accepts: local, then registered providers.

Returned by locations.

check_placement#

Checks a location, kind and runtime against each other without creating anything: InvalidPlacement (listing the valid values) when the combination does not exist. Empty strings are auto; an empty on is the user default.

def check_placement(on: str, kind: str, runtime: str) -> None

Raises CuaError (InvalidPlacement)

error_doc_url#

A link to the errors-reference entry of a CuaError variant, by name (InvalidArgument, case-insensitive). An unknown name links to the page itself. The bindings' doc_url / docUrl on CuaError use it.

def error_doc_url(variant: str) -> str

Returns String

Local runtimes#

cua.local() checks and sets up the local runtimes and moves images in and out of the local engines.

Python programs usually use the high-level API instead: cua_sandbox runtimes.

Local#

Local runtimes and images.

Returned by Cua.local.

Local.build_image#

Builds an image from an images.cua.ai/v1alpha1 Image resource (JSON) on base; pushes to push when set. Returns reference@digest or the local output directory.

async def build_image(self, spec_json: str, base: str, push: Optional[str]) -> str

Returns String · Async · Raises CuaError

Local.doctor#

Inspects local runtimes (read-only: never installs or starts anything).

async def doctor(self) -> DoctorReport

Returns DoctorReport · Async · Raises CuaError

Example

import asyncio
 
import cua
 
async def main():
    report = await cua.embedded().local().doctor()
    for check in report.checks:
        print(check.name, check.status)
 
asyncio.run(main())

Local.pull_image#

Pulls an image (container engine for container refs, the image cache for vm: containerDisks).

async def pull_image(self, reference: str) -> LocalImage

Returns LocalImage · Async · Raises CuaError

Local.push_image#

Copies an image (host architecture) to destination; returns the pushed manifest digest.

async def push_image(self, reference: str, destination: str) -> str

Returns String · Async · Raises CuaError

Local.setup#

Provisions components (qemu, lume; runsc reports its plan). Empty = everything provisionable. dry_run only reports.

async def setup(self, components: List[str], dry_run: bool) -> List[SetupStep]

Returns Vec<SetupStep> · Async · Raises CuaError

Local.storage#

Free space where local Spaces are written (Lume's VM storage, the cua home for QEMU, the container engine's data disk) and which catalog images are already pulled. Read-only and bounded; probed in this process (the daemon runs on the same host).

async def storage(self) -> LocalStorage

Returns LocalStorage · Async · Raises CuaError

DoctorReport record#

Doctor output.

Returned by Local.doctor.

LocalImage record#

A local image.

Returned by Local.pull_image.

LocalStorage record#

What local Spaces have room for (see Local.storage).

Returned by Local.storage.

RuntimeCheck record#

One doctor check (one backend).

SetupStep record#

One setup step.

Returned by Local.setup.

StorageVolume record#

Space on the volume a local engine writes to.

RuntimeCheckStatus enum#

Status of one local runtime component.

RuntimeCheckStatus.OK
RuntimeCheckStatus.INSTALLABLE
RuntimeCheckStatus.ERROR
RuntimeCheckStatus.NOT_APPLICABLE
RuntimeCheckStatus.UNKNOWN