Agent SDK API Reference

agent - Decorator-based Computer Use Agent with liteLLM integration

Classes

Functions

ComputerAgent

Main agent class that automatically selects the appropriate agent loop based on the model and executes tool calls.

Constructor

ComputerAgent(self, model: str, tools: Optional[List[Any]] = None, custom_loop: Optional[Callable] = None, only_n_most_recent_images: Optional[int] = None, callbacks: Optional[List[Any]] = None, instructions: Optional[str] = None, verbosity: Optional[int] = None, trajectory_dir: Optional[str | Path | dict] = None, max_retries: Optional[int] = 3, screenshot_delay: Optional[float | int] = 0.5, use_prompt_caching: Optional[bool] = False, max_trajectory_budget: Optional[float | dict] = None, telemetry_enabled: Optional[bool] = True, trust_remote_code: Optional[bool] = False, api_key: Optional[str] = None, api_base: Optional[str] = None, additional_generation_kwargs = {})

Attributes

Methods

ComputerAgent.run

async def run(self, messages: Messages, stream: bool = False, api_key: Optional[str] = None, api_base: Optional[str] = None, additional_generation_kwargs = {}) -> AsyncGenerator[Dict[str, Any], None]

Run the agent with the given messages using Computer protocol handler pattern.

Parameters:

Returns: AsyncGenerator that yields response chunks

ComputerAgent.predict_click

async def predict_click(self, instruction: str, image_b64: Optional[str] = None) -> Optional[Tuple[int, int]]

Predict click coordinates based on image and instruction.

Parameters:

Returns: None or tuple with (x, y) coordinates

ComputerAgent.get_capabilities

def get_capabilities(self) -> List[AgentCapability]

Get list of capabilities supported by the current agent config.

Returns: List of capability strings (e.g., ["step", "click"])

ComputerAgent.open

def open(self, port: Optional[int] = None)

Start the playground server and open it in the browser.

This method starts a local HTTP server that exposes the /responses endpoint and automatically opens the Cua playground interface in the default browser.

Parameters:

Example:

>>> agent = ComputerAgent(model="claude-sonnet-4")
>>> agent.open()  # Starts server and opens browser

register_agent

def register_agent(models: str, priority: int = 0, tool_type: Optional[str] = None)

Decorator to register an AsyncAgentConfig class.

Parameters:

types

Type definitions for agent

ToolError

Inherits from: RuntimeError

Base exception for tool-related errors

IllegalArgumentError

Inherits from: ToolError

Exception raised when function arguments are invalid

AgentConfigInfo

Inherits from: BaseModel

Information about a registered agent config

Attributes

Methods

AgentConfigInfo.matches_model

def matches_model(self, model: str) -> bool

Check if this agent config matches the given model

tools

Agent tools module. Provides base classes and registered tools for agent interactions.

BaseComputerTool

Inherits from: BaseTool

Base class for computer tools that can provide screenshots.

Computer tools must implement:

• All BaseTool requirements (name, description, parameters, call)
• screenshot() method that returns screenshot as base64 string

Methods

BaseComputerTool.screenshot

async def screenshot(self) -> str

Take a screenshot of the computer/browser.

Returns: Screenshot image data as base64-encoded string

BaseTool

Inherits from: ABC

Base class for all agent tools.

Tools must implement:

• name: str - The tool name (set by @register_tool decorator)
• description: property that returns str - Tool description
• parameters: property that returns dict - JSON schema for tool parameters
• call: method - Execute the tool with given parameters

Constructor

BaseTool(self, cfg: Optional[dict] = None)

Attributes

Methods

BaseTool.call

def call(self, params: Union[str, dict], kwargs = {}) -> Union[str, list, dict]

Execute the tool with the given parameters.

Parameters:

Returns: The result of the tool execution

BrowserTool

Inherits from: BaseComputerTool

Browser tool that uses the computer SDK's interface to control a browser. Implements a comprehensive computer_use action interface for browser control.

Constructor

BrowserTool(self, interface: GenericComputerInterface, cfg: Optional[dict] = None)

Attributes

Methods

BrowserTool.call

def call(self, params: Union[str, dict], kwargs = {}) -> Union[str, dict]

Execute a browser action.

Parameters:

Returns: Result of the action execution

BrowserTool.visit_url

async def visit_url(self, url: str) -> dict

Navigate to a URL.

BrowserTool.click

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

Click at coordinates. Supports both positional (x, y) and kwargs (button, x, y).

This is compatible with the normalized format from OperatorNormalizerCallback which transforms actions like {"type": "left_click", "coordinate": [x, y]} into {"type": "click", "button": "left", "x": x, "y": y}.

BrowserTool.type

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

Type text into the focused element.

BrowserTool.scroll

async def scroll(self, delta_x: int = None, delta_y: int = None, scroll_x: int = None, scroll_y: int = None, x: int = None, y: int = None, pixels: int = None, coordinate = None, kwargs = {}) -> dict

Scroll the page. Supports multiple formats:

• Legacy: scroll(delta_x, delta_y)
• Normalized: scroll(scroll_x=0, scroll_y=100, x=500, y=300)
• FARA: scroll(pixels=100, coordinate=[500, 300])

BrowserTool.web_search

async def web_search(self, query: str) -> dict

Navigate to a Google search for the query.

BrowserTool.screenshot

async def screenshot(self) -> str

Take a screenshot of the current browser page.

BrowserTool.get_current_url

async def get_current_url(self) -> str

Get the current URL of the browser page.

BrowserTool.left_click

async def left_click(self, coordinate = None, x: int = None, y: int = None, kwargs = {}) -> dict

Left click at coordinates. Supports coordinate array or x/y kwargs.

BrowserTool.right_click

async def right_click(self, coordinate = None, x: int = None, y: int = None, kwargs = {}) -> dict

Right click at coordinates. Supports coordinate array or x/y kwargs.

BrowserTool.middle_click

async def middle_click(self, coordinate = None, x: int = None, y: int = None, kwargs = {}) -> dict

Middle click at coordinates. Supports coordinate array or x/y kwargs.

BrowserTool.double_click

async def double_click(self, coordinate = None, x: int = None, y: int = None, kwargs = {}) -> dict

Double click at coordinates. Supports coordinate array or x/y kwargs.

BrowserTool.triple_click

async def triple_click(self, coordinate = None, x: int = None, y: int = None, button: str = None, kwargs = {}) -> dict

Triple click at coordinates. Supports coordinate array or x/y kwargs.

BrowserTool.mouse_move

async def mouse_move(self, coordinate = None, x: int = None, y: int = None, kwargs = {}) -> dict

Move mouse to coordinates. Supports coordinate array or x/y kwargs.

BrowserTool.move

async def move(self, x: int = None, y: int = None, kwargs = {}) -> dict

Move mouse to coordinates. Alias for mouse_move with x/y kwargs.

BrowserTool.left_click_drag

async def left_click_drag(self, coordinate = None, start_coordinate = None, end_coordinate = None, kwargs = {}) -> dict

Drag from start to end coordinates. FARA-compatible.

BrowserTool.key

async def key(self, keys = None, kwargs = {}) -> dict

Press keys. FARA-compatible.

BrowserTool.keypress

async def keypress(self, keys = None, kwargs = {}) -> dict

Press keys. Alias for key() - used by OperatorNormalizerCallback.

BrowserTool.hscroll

async def hscroll(self, pixels = None, coordinate = None, kwargs = {}) -> dict

Horizontal scroll. FARA-compatible.

BrowserTool.wait

async def wait(self, time = None, kwargs = {}) -> dict

Wait for specified seconds. FARA-compatible.

BrowserTool.history_back

async def history_back(self, kwargs = {}) -> dict

Go back in browser history. FARA-compatible.

BrowserTool.terminate

async def terminate(self, status = None, kwargs = {}) -> dict

Terminate and report status. FARA-compatible.

get_registered_tools

def get_registered_tools() -> Dict[str, type]

Get all registered tools.

Returns: Dictionary mapping tool names to tool classes

get_tool

def get_tool(name: str) -> Optional[type]

Get a registered tool by name.

Parameters:

Returns: The tool class, or None if not found

register_tool

def register_tool(name: str, allow_overwrite: bool = False)

Decorator to register a tool class with a given name.

Parameters:

Returns: Decorator function that registers the class

Example:

@register_tool("my_tool")
class MyTool(BaseTool):
    ...

callbacks

Callback system for ComputerAgent preprocessing and postprocessing hooks.

AsyncCallbackHandler

Inherits from: ABC

Base class for async callback handlers that can preprocess messages before the agent loop and postprocess output after the agent loop.

Methods

AsyncCallbackHandler.on_run_start

async def on_run_start(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]]) -> None

Called at the start of an agent run loop.

AsyncCallbackHandler.on_run_end

async def on_run_end(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]], new_items: List[Dict[str, Any]]) -> None

Called at the end of an agent run loop.

AsyncCallbackHandler.on_run_continue

async def on_run_continue(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]], new_items: List[Dict[str, Any]]) -> bool

Called during agent run loop to determine if execution should continue.

Parameters:

Returns: True to continue execution, False to stop

AsyncCallbackHandler.on_llm_start

async def on_llm_start(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]

Called before messages are sent to the agent loop.

Parameters:

Returns: List of preprocessed message dictionaries

AsyncCallbackHandler.on_llm_end

async def on_llm_end(self, output: List[Dict[str, Any]]) -> List[Dict[str, Any]]

Called after the agent loop returns output.

Parameters:

Returns: List of postprocessed output dictionaries

AsyncCallbackHandler.on_computer_call_start

async def on_computer_call_start(self, item: Dict[str, Any]) -> None

Called when a computer call is about to start.

Parameters:

AsyncCallbackHandler.on_computer_call_end

async def on_computer_call_end(self, item: Dict[str, Any], result: List[Dict[str, Any]]) -> None

Called when a computer call has completed.

Parameters:

AsyncCallbackHandler.on_function_call_start

async def on_function_call_start(self, item: Dict[str, Any]) -> None

Called when a function call is about to start.

Parameters:

AsyncCallbackHandler.on_function_call_end

async def on_function_call_end(self, item: Dict[str, Any], result: List[Dict[str, Any]]) -> None

Called when a function call has completed.

Parameters:

AsyncCallbackHandler.on_text

async def on_text(self, item: Dict[str, Any]) -> None

Called when a text message is encountered.

Parameters:

AsyncCallbackHandler.on_api_start

async def on_api_start(self, kwargs: Dict[str, Any]) -> None

Called when an API call is about to start.

Parameters:

AsyncCallbackHandler.on_api_end

async def on_api_end(self, kwargs: Dict[str, Any], result: Any) -> None

Called when an API call has completed.

Parameters:

AsyncCallbackHandler.on_usage

async def on_usage(self, usage: Dict[str, Any]) -> None

Called when usage information is received.

Parameters:

AsyncCallbackHandler.on_screenshot

async def on_screenshot(self, screenshot: Union[str, bytes], name: str = 'screenshot') -> None

Called when a screenshot is taken.

Parameters:

AsyncCallbackHandler.on_responses

async def on_responses(self, kwargs: Dict[str, Any], responses: Dict[str, Any]) -> None

Called when responses are received.

Parameters:

BudgetManagerCallback

Inherits from: AsyncCallbackHandler

Budget manager callback that tracks usage costs and can stop execution when budget is exceeded.

Constructor

BudgetManagerCallback(self, max_budget: float, reset_after_each_run: bool = True, raise_error: bool = False)

Attributes

Methods

BudgetManagerCallback.on_run_start

async def on_run_start(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]]) -> None

Reset budget if configured to do so.

BudgetManagerCallback.on_usage

async def on_usage(self, usage: Dict[str, Any]) -> None

Track usage costs.

BudgetManagerCallback.on_run_continue

async def on_run_continue(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]], new_items: List[Dict[str, Any]]) -> bool

Check if budget allows continuation.

ImageRetentionCallback

Inherits from: AsyncCallbackHandler

Callback handler that applies image retention policy to limit the number of recent images in message history to prevent context window overflow.

Constructor

ImageRetentionCallback(self, only_n_most_recent_images: Optional[int] = None)

Attributes

Methods

ImageRetentionCallback.on_llm_start

async def on_llm_start(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]

Apply image retention policy to messages before sending to agent loop.

Parameters:

Returns: List of messages with image retention policy applied

LoggingCallback

Inherits from: AsyncCallbackHandler

Callback handler that logs agent lifecycle events with configurable verbosity.

Logging levels:

• DEBUG: All events including API calls, message preprocessing, and detailed outputs
• INFO: Major lifecycle events (start/end, messages, outputs)
• WARNING: Only warnings and errors
• ERROR: Only errors

Constructor

LoggingCallback(self, logger: Optional[logging.Logger] = None, level: int = logging.INFO)

Attributes

Methods

LoggingCallback.on_run_start

async def on_run_start(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]]) -> None

Called before the run starts.

LoggingCallback.on_usage

async def on_usage(self, usage: Dict[str, Any]) -> None

Called when usage information is received.

LoggingCallback.on_run_end

async def on_run_end(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]], new_items: List[Dict[str, Any]]) -> None

Called after the run ends.

LoggingCallback.on_llm_start

async def on_llm_start(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]

Called before LLM processing starts.

LoggingCallback.on_llm_end

async def on_llm_end(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]

Called after LLM processing ends.

LoggingCallback.on_computer_call_start

async def on_computer_call_start(self, item: Dict[str, Any]) -> None

Called when a computer call starts.

LoggingCallback.on_computer_call_end

async def on_computer_call_end(self, item: Dict[str, Any], result: Any) -> None

Called when a computer call ends.

LoggingCallback.on_function_call_start

async def on_function_call_start(self, item: Dict[str, Any]) -> None

Called when a function call starts.

LoggingCallback.on_function_call_end

async def on_function_call_end(self, item: Dict[str, Any], result: Any) -> None

Called when a function call ends.

LoggingCallback.on_text

async def on_text(self, item: Dict[str, Any]) -> None

Called when a text message is encountered.

LoggingCallback.on_api_start

async def on_api_start(self, kwargs: Dict[str, Any]) -> None

Called when an API call is about to start.

LoggingCallback.on_api_end

async def on_api_end(self, kwargs: Dict[str, Any], result: Any) -> None

Called when an API call has completed.

LoggingCallback.on_screenshot

async def on_screenshot(self, item: Union[str, bytes], name: str = 'screenshot') -> None

Called when a screenshot is taken.

OperatorNormalizerCallback

Inherits from: AsyncCallbackHandler

Normalizes common computer call hallucinations / errors in computer call syntax.

Methods

OperatorNormalizerCallback.on_llm_end

async def on_llm_end(self, output: List[Dict[str, Any]]) -> List[Dict[str, Any]]

OtelCallback

Inherits from: AsyncCallbackHandler

OpenTelemetry callback handler for instrumentation.

Tracks:

• Agent session lifecycle (start/end)
• Agent run lifecycle (start/end with duration)
• Individual steps (with duration)
• Computer actions (with duration)
• Token usage
• Errors

Constructor

OtelCallback(self, agent: Any)

Attributes

Methods

OtelCallback.on_run_start

async def on_run_start(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]]) -> None

Called at the start of an agent run loop.

OtelCallback.on_run_end

async def on_run_end(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]], new_items: List[Dict[str, Any]]) -> None

Called at the end of an agent run loop.

OtelCallback.on_responses

async def on_responses(self, kwargs: Dict[str, Any], responses: Dict[str, Any]) -> None

Called when responses are received (each step).

OtelCallback.on_usage

async def on_usage(self, usage: Dict[str, Any]) -> None

Called when usage information is received.

OtelCallback.on_computer_call_start

async def on_computer_call_start(self, item: Dict[str, Any]) -> None

Called when a computer call is about to start.

OtelCallback.on_computer_call_end

async def on_computer_call_end(self, item: Dict[str, Any], result: List[Dict[str, Any]]) -> None

Called when a computer call has completed.

OtelCallback.on_api_start

async def on_api_start(self, kwargs: Dict[str, Any]) -> None

Called when an LLM API call is about to start.

OtelCallback.on_api_end

async def on_api_end(self, kwargs: Dict[str, Any], result: Any) -> None

Called when an LLM API call has completed.

OtelErrorCallback

Inherits from: AsyncCallbackHandler

Callback that captures errors and sends them to Sentry/OTEL.

Should be added early in the callback chain to catch all errors.

Constructor

OtelErrorCallback(self, agent: Any)

Attributes

Methods

OtelErrorCallback.on_error

async def on_error(self, error: Exception, context: Dict[str, Any]) -> None

Called when an error occurs during agent execution.

PromptInstructionsCallback

Inherits from: AsyncCallbackHandler

Prepend a user instructions message to the message list.

This is a minimal, non-invasive way to guide the agent's behavior without modifying agent loops or tools. It works with any provider/loop since it only alters the messages array before sending to the model.

Constructor

PromptInstructionsCallback(self, instructions: Optional[str]) -> None

Attributes

Methods

PromptInstructionsCallback.on_llm_start

async def on_llm_start(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]

TelemetryCallback

Inherits from: AsyncCallbackHandler

Telemetry callback handler for Computer-Use Agent (cua-agent)

Tracks agent usage, performance metrics, and optionally trajectory data.

Constructor

TelemetryCallback(self, agent, log_trajectory: bool = False)

Attributes

Methods

TelemetryCallback.on_run_start

async def on_run_start(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]]) -> None

Called at the start of an agent run loop.

TelemetryCallback.on_run_end

async def on_run_end(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]], new_items: List[Dict[str, Any]]) -> None

Called at the end of an agent run loop.

TelemetryCallback.on_usage

async def on_usage(self, usage: Dict[str, Any]) -> None

Called when usage information is received.

TelemetryCallback.on_responses

async def on_responses(self, kwargs: Dict[str, Any], responses: Dict[str, Any]) -> None

Called when responses are received.

TrajectorySaverCallback

Inherits from: AsyncCallbackHandler

Callback handler that saves agent trajectories to disk.

Saves each run as a separate trajectory with unique ID, and each turn within the trajectory gets its own folder with screenshots and responses.

Constructor

TrajectorySaverCallback(self, trajectory_dir: str, reset_on_run: bool = True, screenshot_dir: Optional[str] = None)

Attributes

Methods

TrajectorySaverCallback.on_run_start

async def on_run_start(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]]) -> None

Initialize trajectory tracking for a new run.

TrajectorySaverCallback.on_run_end

async def on_run_end(self, kwargs: Dict[str, Any], old_items: List[Dict[str, Any]], new_items: List[Dict[str, Any]]) -> None

Finalize run tracking by updating metadata with completion status, usage, and new items.

TrajectorySaverCallback.on_api_start

async def on_api_start(self, kwargs: Dict[str, Any]) -> None

TrajectorySaverCallback.on_api_end

async def on_api_end(self, kwargs: Dict[str, Any], result: Any) -> None

Save API call result.

TrajectorySaverCallback.on_screenshot

async def on_screenshot(self, screenshot: Union[str, bytes], name: str = 'screenshot') -> None

Save a screenshot.

TrajectorySaverCallback.on_usage

async def on_usage(self, usage: Dict[str, Any]) -> None

Called when usage information is received.

TrajectorySaverCallback.on_responses

async def on_responses(self, kwargs: Dict[str, Any], responses: Dict[str, Any]) -> None

Save responses to the current turn directory and update usage statistics.

TrajectorySaverCallback.on_computer_call_end

async def on_computer_call_end(self, item: Dict[str, Any], result: List[Dict[str, Any]]) -> None

Called when a computer call has completed. Saves screenshots and computer call output.