Callback Lifecycle

on_run_start()           # Agent run begins
  │
  ├─► on_llm_start()       # Before API call (preprocess messages)
  │     └─► on_api_start() # API request begins
  │         └─► on_api_end() # API response received
  │     └─► on_llm_end()   # After API call (postprocess)
  │
  ├─► on_computer_call_start()  # Before executing action
  │     └─► Execute action (click, type, etc.)
  │     └─► on_screenshot()     # Screenshot captured
  │     └─► on_computer_call_end()
  │
  ├─► on_usage()           # Token usage tracked
  ├─► on_run_continue()    # Check if should continue
  │
  └─► Loop until done
      │
on_run_end()             # Agent run complete

Switching Models

# Full computer-use models (API-based)
agent = ComputerAgent(model="anthropic/claude-sonnet-4-5-20250929")
agent = ComputerAgent(model="openai/computer-use-preview")
agent = ComputerAgent(model="gemini-2.5-computer-use-preview")

# Via OpenRouter (access many models)
agent = ComputerAgent(model="openrouter/qwen/qwen3-vl-235b-a22b-instruct")

# Local inference with HuggingFace Transformers
agent = ComputerAgent(model="huggingface-local/ByteDance-Seed/UI-TARS-1.5-7B")

# Local inference with Ollama
agent = ComputerAgent(model="ollama/llava")

# MLX (Apple Silicon optimized)
agent = ComputerAgent(model="mlx/llava-v1.6-mistral-7b")

Model Composition (Grounding + Planning)

Combine a specialized grounding model with a general LLM for planning:

# Format: {grounding-model}+{planning-llm}

# UI-TARS for clicking + Claude for reasoning
agent = ComputerAgent(
    model="huggingface-local/ByteDance-Seed/UI-TARS-1.5-7B+anthropic/claude-sonnet-4-5"
)

# Moondream3 grounding + GPT-4o planning
agent = ComputerAgent(model="moondream3+openai/gpt-4o")

# OmniParser grounding + any LLM
agent = ComputerAgent(model="omniparser+openai/gpt-4o")

# Human-in-the-loop for click verification
agent = ComputerAgent(model="openai/computer-use-preview+human/human")

from computer import Computer

computer = Computer(
    # Display & Resources
    display="1024x768",       # Screen resolution (WIDTHxHEIGHT)
    memory="8GB",             # RAM allocation
    cpu="4",                  # CPU cores

    # OS Selection
    os_type="macos",          # "macos", "linux", or "windows"
    image=None,               # VM image (auto-selected if None)
    name="my-vm",             # VM identifier

    # Provider
    provider_type="lume",     # lume, docker, cloud, winsandbox
    host="localhost",         # Provider host
    port=7777,                # Provider API port
    noVNC_port=8006,          # Web VNC interface port

    # Storage
    storage="/path/to/storage",  # Persistent storage path
    ephemeral=False,             # True = destroy on stop
    shared_directories=[         # Mount host directories
        "/Users/me/shared"
    ],

    # Cloud/Remote
    api_key=None,             # CUA_API_KEY (from env if not set)
    use_host_computer_server=False,  # Use localhost instead of VM

    # Logging
    verbosity=20,             # logging.INFO
    telemetry_enabled=True,   # Anonymous usage tracking

    # Experiments
    experiments=["app-use"]   # macOS app-specific automation
)

from agent import ComputerAgent

agent = ComputerAgent(
    # Core
    model="anthropic/claude-sonnet-4-5-20250929",
    tools=[computer],                    # Computer + function tools

    # Control
    custom_loop=None,                    # Override default agent loop
    max_retries=3,                       # API retry attempts
    screenshot_delay=0.5,                # Delay before screenshots (seconds)

    # Memory Optimization
    only_n_most_recent_images=3,         # Keep only N screenshots in context
    use_prompt_caching=False,            # Anthropic prompt caching

    # Callbacks
    callbacks=[                          # Built-in and custom callbacks
        TrajectorySaverCallback(path="./trajectories"),
        BudgetManagerCallback(max_usd=5.0),
        LoggingCallback(level=logging.DEBUG),
    ],
    instructions="Custom system prompt here",

    # Tracking
    trajectory_dir="./trajectories",     # Save actions to disk
    max_trajectory_budget=5.0,           # Cost budget in USD
    telemetry_enabled=True,

    # Logging
    verbosity=10,                        # logging.DEBUG

    # Model Provider Overrides
    api_key=None,                        # Override provider API key
    api_base=None,                       # Override provider URL
    trust_remote_code=False,             # For HuggingFace models

    # Additional LLM kwargs
    temperature=0.7,
    max_tokens=4096,
)

Custom Callback

from agent.callbacks import AsyncCallbackHandler

class MyCallback(AsyncCallbackHandler):
    async def on_run_start(self, messages):
        print("Agent starting...")

    async def on_computer_call_start(self, action, call_id):
        print(f"Executing: {action}")

    async def on_computer_call_end(self, action, call_id, result):
        print(f"Completed: {action}")

    async def on_screenshot(self, screenshot_bytes):
        # Save or process screenshot
        pass

    async def on_usage(self, usage):
        print(f"Tokens: {usage['total_tokens']}")

    async def on_run_end(self, result):
        print(f"Agent finished. Cost: ${result.get('cost', 0):.4f}")

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

Basic macOS Automation

import asyncio
from agent import ComputerAgent
from computer import Computer

async def main():
    computer = Computer(
        os_type="macos",
        display="1024x768",
        memory="8GB",
        provider_type="lume"
    )

    try:
        await computer.run()

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

        messages = [{
            "role": "user",
            "content": "Open Safari, go to github.com, and take a screenshot"
        }]

        async for result in agent.run(messages):
            for item in result.get("output", []):
                if item.get("type") == "message":
                    print(f"Agent: {item['content'][0]['text']}")
                elif item.get("type") == "computer_call":
                    print(f"Action: {item['action']}")

    finally:
        await computer.stop()

asyncio.run(main())

Using Local Model (UI-TARS)

import asyncio
from agent import ComputerAgent
from computer import Computer

async def main():
    computer = Computer(
        os_type="macos",
        provider_type="lume"
    )

    await computer.run()

    # Use UI-TARS locally (requires GPU with enough VRAM)
    agent = ComputerAgent(
        model="huggingface-local/ByteDance-Seed/UI-TARS-1.5-7B",
        tools=[computer],
        trust_remote_code=True  # Required for HuggingFace models
    )

    async for result in agent.run([{
        "role": "user",
        "content": "Click on the Safari icon in the dock"
    }]):
        print(result)

    await computer.stop()

asyncio.run(main())

With Budget & Trajectory Saving

import asyncio
import logging
from agent import ComputerAgent
from agent.callbacks import TrajectorySaverCallback, BudgetManagerCallback
from computer import Computer

async def main():
    computer = Computer(
        os_type="macos",
        provider_type="lume",
        storage="./vm-storage"  # Persistent storage
    )

    await computer.run()

    agent = ComputerAgent(
        model="anthropic/claude-sonnet-4-5-20250929",
        tools=[computer],
        callbacks=[
            TrajectorySaverCallback(path="./trajectories"),
            BudgetManagerCallback(max_usd=2.0),
        ],
        only_n_most_recent_images=5,
        verbosity=logging.INFO,
        instructions="Be concise. Take direct actions."
    )

    async for result in agent.run([{
        "role": "user",
        "content": """
        1. Open System Settings
        2. Navigate to General
        3. Check the About section
        """
    }]):
        usage = result.get("usage", {})
        if usage:
            print(f"Tokens used: {usage.get('total_tokens', 0)}")

    await computer.stop()

asyncio.run(main())

Composed Grounding + Planning

import asyncio
from agent import ComputerAgent
from computer import Computer

async def main():
    computer = Computer(
        os_type="macos",
        provider_type="lume"
    )

    await computer.run()

    # Moondream3 for click prediction + Claude for reasoning
    agent = ComputerAgent(
        model="moondream3+anthropic/claude-sonnet-4-5-20250929",
        tools=[computer]
    )

    async for result in agent.run([{
        "role": "user",
        "content": "Find and click on the Finder icon"
    }]):
        print(result)

    await computer.stop()

asyncio.run(main())

# List available images
lume images

# Pull an image
lume pull macos-sequoia-cua:latest

# Run a VM (interactive)
lume run macos-sequoia-cua:latest

# List running VMs
lume list

# Stop a VM
lume stop <vm-name>

# Delete a VM
lume delete <vm-name>

# Get VM info
lume info <vm-name>

# Clone a VM (for snapshots)
lume clone <source> <destination>