Agent SDK

pip install cua-agent

from agent import ComputerAgent

agent = ComputerAgent(
    model="anthropic/claude-sonnet-4-5-20250929",
    tools=[computer]
)

async for result in agent.run("Open Firefox and search for Cua"):
    print(result.text)

async for result in agent.run("Click the submit button"):
    if result.text:
        print(result.text)
    if result.action:
        print(f"Action: {result.action}")

result = await agent.run_to_completion("Open the calculator app")
print(result.text)

async for result in agent.run("Search for documentation"):
    # Check what happened this iteration
    if result.text:
        print(f"Agent said: {result.text}")

    if result.action:
        print(f"Action type: {result.action.type}")

    if result.usage:
        print(f"Tokens used: {result.usage.total_tokens}")

result.action.type          # "click", "type", "key", "scroll", etc.
result.action.coordinate    # [x, y] for click actions
result.action.text          # Text for type actions
result.action.key           # Key for key press actions

async for result in agent.run("Complete this form"):
    if result.usage:
        print(f"Input tokens: {result.usage.input_tokens}")
        print(f"Output tokens: {result.usage.output_tokens}")
        print(f"Total tokens: {result.usage.total_tokens}")
        print(f"Cost: ${result.usage.response_cost:.4f}")

from agent.callbacks import LoggingCallback
import logging

agent = ComputerAgent(
    model="...",
    tools=[computer],
    callbacks=[LoggingCallback(level=logging.DEBUG)]
)

from agent.callbacks import BudgetManagerCallback

agent = ComputerAgent(
    model="...",
    tools=[computer],
    callbacks=[BudgetManagerCallback(
        max_budget=10.0,
        reset_after_each_run=True,
        raise_error=False
    )]
)

from agent.callbacks import ImageRetentionCallback

agent = ComputerAgent(
    model="...",
    tools=[computer],
    callbacks=[ImageRetentionCallback(only_n_most_recent_images=3)]
)

from agent.callbacks import TrajectorySaverCallback

agent = ComputerAgent(
    model="...",
    tools=[computer],
    callbacks=[TrajectorySaverCallback(
        trajectory_dir="trajectories",
        reset_on_run=True,
        screenshot_dir="screenshots"
    )]
)

from agent.callbacks import PromptInstructionsCallback

agent = ComputerAgent(
    model="...",
    tools=[computer],
    callbacks=[PromptInstructionsCallback("Always confirm before clicking")]
)

from agent.callbacks.base import AsyncCallbackHandler

class MyCallback(AsyncCallbackHandler):
    async def on_run_start(self, kwargs, old_items):
        """Called when agent.run() begins"""
        print("Starting run...")

    async def on_run_continue(self, kwargs, old_items, new_items) -> bool:
        """Called before each iteration. Return False to stop."""
        return True

    async def on_llm_start(self, messages):
        """Preprocess messages before LLM call."""
        return messages

    async def on_llm_end(self, messages):
        """Postprocess messages after LLM call."""
        return messages

    async def on_usage(self, usage):
        """Called with usage stats after each LLM call."""
        print(f"Cost: ${usage.response_cost:.4f}")

    async def on_computer_call_start(self, item):
        """Called before a computer action."""
        pass

    async def on_computer_call_end(self, item, result):
        """Called after a computer action."""
        pass

    async def on_screenshot(self, screenshot, name):
        """Called when a screenshot is taken."""
        pass

    async def on_run_end(self, kwargs, old_items, new_items):
        """Called when agent.run() completes."""
        print("Run complete!")

from computer import Computer

computer = Computer(
    os_type="linux",
    provider_type="docker",
    image="trycua/cua-xfce:latest"
)
await computer.run()

agent = ComputerAgent(
    model="anthropic/claude-sonnet-4-5-20250929",
    tools=[computer]
)

from agent.tools import BrowserTool

browser = BrowserTool(interface=computer)

agent = ComputerAgent(
    model="google/gemini-2.5-flash",
    tools=[browser]
)

def calculate(a: int, b: int) -> int:
    """Calculate the sum of two integers"""
    return a + b

async def fetch_data(url: str) -> str:
    """Fetch data from a URL"""
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

agent = ComputerAgent(
    model="anthropic/claude-sonnet-4-5-20250929",
    tools=[computer, calculate, fetch_data]
)

from computer.helpers import sandboxed

@sandboxed()
def read_sandbox_file(path: str) -> str:
    """Read a file from inside the sandbox"""
    with open(path, 'r') as f:
        return f.read()

agent = ComputerAgent(
    model="anthropic/claude-sonnet-4-5-20250929",
    tools=[computer, read_sandbox_file]
)

from agent.tools import BaseTool, register_tool

@register_tool("database_query")
class DatabaseQueryTool(BaseTool):
    def __init__(self, connection_string: str):
        self.connection = connection_string

    @property
    def description(self) -> str:
        return "Execute a read-only SQL query"

    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "SQL SELECT query to execute"
                }
            },
            "required": ["query"]
        }

    def call(self, params, **kwargs):
        query = params["query"] if isinstance(params, dict) else params
        # Execute and return results
        return {"rows": [...]}

from agent.tools import ToolError

def divide(a: float, b: float) -> float:
    """Divide a by b"""
    if b == 0:
        raise ToolError("Cannot divide by zero")
    return a / b

# Anthropic
agent = ComputerAgent(model="anthropic/claude-sonnet-4-5-20250929", ...)

# OpenAI
agent = ComputerAgent(model="openai/computer-use-preview", ...)

# Google
agent = ComputerAgent(model="google/gemini-2.5-flash", ...)

# Local models via Ollama
agent = ComputerAgent(model="ollama/ui-tars:latest", ...)

# UI-TARS for grounding, Claude for planning
agent = ComputerAgent(
    model="ollama/ui-tars:latest+anthropic/claude-sonnet-4-5-20250929",
    tools=[computer]
)

# Qwen for grounding, GPT-4 for planning
agent = ComputerAgent(
    model="ollama/qwen2.5-vl:latest+openai/gpt-4o",
    tools=[computer]
)

from agent.types import Message

history = [
    Message(role="user", content="Open the browser"),
    Message(role="assistant", content="I've opened Firefox."),
]

async for result in agent.run(
    "Now search for Python tutorials",
    chat_history=history
):
    print(result.text)

from agent.types import Message

# User message
user_msg = Message(role="user", content="Click the button")

# Assistant message
assistant_msg = Message(role="assistant", content="I clicked the submit button.")

# System message (for instructions)
system_msg = Message(role="system", content="Be concise in responses.")

from agent import ComputerAgent
from agent.errors import AgentError, BudgetExceededError

try:
    async for result in agent.run("Complete the form"):
        if result.error:
            print(f"Action failed: {result.error}")
        print(result.text)
except BudgetExceededError:
    print("Budget limit reached")
except AgentError as e:
    print(f"Agent error: {e}")

agent = ComputerAgent(
    model="anthropic/claude-sonnet-4-5-20250929",
    tools=[computer],
    trajectory_dir={
        "trajectory_dir": "trajectories",
        "reset_on_run": False,         # Continue same trajectory across runs
        "screenshot_dir": "screenshots" # Save screenshots separately
    }
)