Callbacks

Callbacks let you hook into the agent's execution lifecycle. Use them to log events, track costs, save trajectories, limit context size, or add custom behavior at any point during agent execution.

Using Callbacks

Pass callbacks when creating your agent:

from agent import ComputerAgent
from agent.callbacks import LoggingCallback, BudgetManagerCallback
import logging

agent = ComputerAgent(
    model="anthropic/claude-sonnet-4-5-20250929",
    tools=[computer],
    callbacks=[
        LoggingCallback(level=logging.INFO),
        BudgetManagerCallback(max_budget=5.0)
    ]
)

Many callbacks also have shorthand parameters on ComputerAgent:

agent = ComputerAgent(
    model="anthropic/claude-sonnet-4-5-20250929",
    tools=[computer],
    verbosity=logging.INFO,        # Adds LoggingCallback
    max_trajectory_budget=5.0,     # Adds BudgetManagerCallback
    only_n_most_recent_images=3,   # Adds ImageRetentionCallback
    trajectory_dir="trajectories", # Adds TrajectorySaverCallback
    instructions="Be concise"      # Adds PromptInstructionsCallback
)

Built-in Callbacks

LoggingCallback

Logs agent lifecycle events with configurable verbosity:

from agent.callbacks import LoggingCallback
import logging

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

# Or use shorthand
agent = ComputerAgent(
    model="...",
    tools=[computer],
    verbosity=logging.INFO
)

Logs include API calls, computer actions, text responses, and usage statistics. Image URLs are automatically sanitized in logs.

BudgetManagerCallback

Tracks costs and stops execution when a budget limit is reached:

from agent.callbacks import BudgetManagerCallback

agent = ComputerAgent(
    model="...",
    tools=[computer],
    callbacks=[BudgetManagerCallback(
        max_budget=10.0,           # Dollar limit
        reset_after_each_run=True, # Reset per run (default)
        raise_error=False          # Graceful stop vs exception
    )]
)

# Or use shorthand
agent = ComputerAgent(
    model="...",
    tools=[computer],
    max_trajectory_budget=5.0
)

When raise_error=True, the agent raises BudgetExceededError instead of stopping gracefully.

ImageRetentionCallback

Limits how many screenshots are kept in the message history to prevent context overflow:

from agent.callbacks import ImageRetentionCallback

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

# Or use shorthand
agent = ComputerAgent(
    model="...",
    tools=[computer],
    only_n_most_recent_images=3
)

Older screenshots are removed along with their associated computer call items.

TrajectorySaverCallback

Saves complete agent conversations for debugging and analysis:

from agent.callbacks import TrajectorySaverCallback

agent = ComputerAgent(
    model="...",
    tools=[computer],
    callbacks=[TrajectorySaverCallback(
        trajectory_dir="trajectories",
        reset_on_run=True,              # New trajectory per run
        screenshot_dir="screenshots"    # Optional separate screenshot dir
    )]
)

# Or use shorthand
agent = ComputerAgent(
    model="...",
    tools=[computer],
    trajectory_dir="trajectories"
)

See Trajectories for details on the saved format.

PromptInstructionsCallback

Prepends custom instructions to every LLM call:

from agent.callbacks import PromptInstructionsCallback

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

# Or use shorthand
agent = ComputerAgent(
    model="...",
    tools=[computer],
    instructions="Always confirm before clicking buttons"
)

This is a simple way to guide agent behavior without modifying agent loops.

Lifecycle Hooks

Callbacks can implement these hooks (all are optional):

from agent.callbacks.base import AsyncCallbackHandler

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

    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 modified messages."""
        return messages

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

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

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

    async def on_computer_call_end(self, item, result):
        """Called after a computer action completes"""
        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"""
        pass

Creating Custom Callbacks

Build your own callback by extending AsyncCallbackHandler:

from agent.callbacks.base import AsyncCallbackHandler

class ActionCounterCallback(AsyncCallbackHandler):
    def __init__(self):
        self.action_count = 0

    async def on_computer_call_end(self, item, result):
        self.action_count += 1
        print(f"Actions taken: {self.action_count}")

    async def on_run_end(self, kwargs, old_items, new_items):
        print(f"Total actions in run: {self.action_count}")
        self.action_count = 0  # Reset for next run

# Use it
counter = ActionCounterCallback()
agent = ComputerAgent(
    model="...",
    tools=[computer],
    callbacks=[counter]
)

await agent.run("Open Firefox and search for Cua")
print(f"Final count: {counter.action_count}")

Callback Execution Order

Callbacks execute in this order during a run:

on_run_start - Once at the beginning
For each iteration:
- on_run_continue - Check if should continue
- on_llm_start - Before LLM call
- on_api_start - Before API request
- on_api_end - After API response
- on_usage - With usage stats
- on_llm_end - After LLM processing
- on_responses - With model responses
- on_text / on_computer_call_start / on_computer_call_end - Per response item
- on_screenshot - When screenshots are taken
on_run_end - Once at the end

Multiple callbacks are called in the order they appear in the callbacks list.

Was this page helpful?

Callbacks

On this page