cua-driver Python SDK

Rust-backed Python SDK and bundled executable for Cua Driver.

Product boundary

This package is for client applications importing Cua Driver as an SDK:

from cua_driver import CuaDriver

It does not contain a Python MCP protocol implementation. The optional cua_driver.fleet module forwards service bytes to the shared Rust typed-MCP client. See the candidate Fleet connection guide for prerequisites, ownership, and release limits. Agents already have runtime-neutral MCP clients and should configure the bundled server directly:

cua-driver mcp

The removed pre-release MCP facade used CuaDriver.stdio(), AsyncCuaDriver, *Args, and transport classes. Application code imports the typed Rust-backed SDK shown below; agent code supplies cua-driver mcp to its agent SDK.

Installation

Install and usage docs live at https://cua.ai/docs/cua-driver/quickstart and https://cua.ai/docs/reference/cua-driver/mcp-tools.

The wheel contains generated UniFFI bindings, a platform-specific Rust SDK library, and the cua-driver executable. CuaDriver.create() loads the runtime in the importing process and does not require the executable or daemon.

SDK example

import asyncio

from cua_driver import (
    CuaDriver,
    CursorReducedMotion,
    EndSessionInput,
    GetDesktopStateInput,
    SetAgentCursorThemeInput,
    StartSessionInput,
)

async def main() -> None:
    driver = CuaDriver.create()
    await driver.start_session(
        StartSessionInput(session="demo", capture_scope=None, cursor_theme=None)
    )
    try:
        await driver.set_agent_cursor_theme(
            SetAgentCursorThemeInput(
                session="demo",
                theme_id="cua.default",
                reduced_motion=CursorReducedMotion.AUTO,
            )
        )
        desktop = await driver.get_desktop_state(
            GetDesktopStateInput(session="demo", screenshot_out_file=None)
        )
        print(desktop.images[0].mime_type)
    finally:
        await driver.end_session(EndSessionInput(session="demo"))
        await driver.shutdown()


asyncio.run(main())

SDK operations are asynchronous. Desktop observations return a typed ToolResult with text, images, verification/error metadata, and structured_json / raw_json for platform-extensible results. Session lifecycle calls return dedicated generated records.

start_session is optional for ordinary calls. The runtime creates one implicit session for this SDK transport and reuses it until shutdown, explicit end, or five minutes of inactivity. Use a named session when application code needs to configure or inspect that run explicitly.

The agent cursor is session-owned and initializes on the first cursor-bearing action, including move_cursor. Its default theme and custom dotLottie authoring workflow are documented in docs/cursor-themes.md. Custom source is compiled and installed with the local CLI; SDK and MCP tools select only an installed theme ID. The built-in cursor shows the sanitized public session name in a badge below the pointer.

Typed native-window migration

The next breaking release adds typed app and window discovery, window snapshots, and token-based clicks. Version 0.25 supports native-window operations through the generic tool surface; it does not expose this typed window API. Upgrade the bindings and native library together.

list_apps returns ListAppsOutput, list_windows returns ListWindowsOutput, and get_window_state returns WindowStateOutput. click takes a required exact target, a coordinate or element-token position, and an explicit delivery mode. It returns ActionResult directly and raises DriverError.Tool on refusal. Other action methods retain ToolResult.

Select a unique app and window, resolve an element from a fresh snapshot, request background delivery explicitly, then capture again to verify the intended UI change. An unsupported background route must not trigger an automatic foreground retry. Refresh stale tokens from the same exact window.

See the migration guide for input and return-type changes, and the complete Python and TypeScript examples for discovery, token selection, verification, and shutdown.

Authorization integrations

standard is promptless for normal automation. An application that needs to authorize attachment to an existing logged-in Chromium profile can construct a configured runtime with CuaDriver.create_configured_with_authorization_host(options, host). Implement DriverAuthorizationHost.authorize() in trusted application code and return the request's exact digest with ALLOW, DENY, or CANCEL.

CuaDriver.create_configured_with_activity_observer(options, observer) emits content-free action, refusal, grant, and session events. The observer cannot change authorization or tool results. Use create_configured_with_host_integrations when the application needs both.

See the SDK reference for complete examples and the callback trust rules.

CuaDriver.connect(socket_path) remains available while existing applications migrate. It exposes the same methods over the installed daemon, but it does not provide a second SDK contract.

shutdown() closes admission, waits for already admitted operations to finish, and is idempotent. Calls started after shutdown fail with DriverError.Shutdown. Destroying a binding handle releases native resources, but orderly applications should still await shutdown().

Daemon-backed MCP host

Applications that must also expose MCP to an external agent can own a private daemon child. The child provides a stable permission identity and session lifetime for short-lived or external clients:

import asyncio

from cua_driver import CuaDriver, EmbeddedCuaDriverHost, get_binary_path


async def main() -> None:
    host = EmbeddedCuaDriverHost(
        binary_path=str(get_binary_path()),
        host_bundle_id="com.example.your-app",
    )
    connection = await host.start()
    driver = CuaDriver.connect(connection.socket_path)
    try:
        # Application calls use driver. An agent runtime can launch
        # connection.mcp.command with connection.mcp.args and environment.
        print(await driver.metadata())
    finally:
        del driver
        await host.stop()


asyncio.run(main())

start() coalesces concurrent callers, stop() cancels startup and is idempotent, and restart() returns a new generation/PID/endpoint. Destroy SDK clients and MCP proxies before stopping or restarting, then reconnect from the new connection. wait_for_exit(connection.generation) observes unexpected termination. Dropping the host closes its parent-liveness pipe and kills the child as a fallback, but orderly applications should still await stop().

Binary wrapper

The package also exposes the bundled executable:

from cua_driver import get_binary_path, run_cua_driver

print(get_binary_path())
exit_code = run_cua_driver(["mcp"])

Platform support

License

MIT License: see LICENSE.