Callback Handlers

Not supported in TypeScript

TypeScript does not support callback handlers. For real-time event handling in TypeScript, use the async iterator pattern with agent.stream() or see Hooks for lifecycle event handling.

Intercept and process agent events as they happen, synchronously, with a callback function in Python. Callback handlers drive real-time monitoring, custom output formatting, and integration with external systems.

For every event the stream can emit, including text, tool usage, lifecycle, and reasoning events, see the stream event types reference.

Note: For asynchronous applications, consider async iterators instead.

Basic Usage

The simplest way to use a callback handler is to pass a callback function to your agent:

from strands import Agent
from strands.vended_tools import notebook

def custom_callback_handler(**kwargs):
    # Process stream data
    if "data" in kwargs:
        print(f"MODEL OUTPUT: {kwargs['data']}")
    elif "current_tool_use" in kwargs and kwargs["current_tool_use"].get("name"):
        print(f"\nUSING TOOL: {kwargs['current_tool_use']['name']}")

# Create an agent with custom callback handler
agent = Agent(
    tools=[notebook],
    callback_handler=custom_callback_handler
)

agent('Create a notebook named "ideas" and add three project ideas.')

Default Callback Handler

Strands Agents provides a default callback handler that formats output to the console:

from strands import Agent
from strands.handlers.callback_handler import PrintingCallbackHandler

# The default callback handler prints text and shows tool usage
agent = Agent(callback_handler=PrintingCallbackHandler())

If you want to disable all output, specify None for the callback handler:

from strands import Agent

# No output will be displayed
agent = Agent(callback_handler=None)

Custom Callback Handlers

Custom callback handlers give you fine-grained control over what streams from your agents.

Example - Print all events in the stream sequence

Custom callback handlers can be useful to debug sequences of events in the agent loop:

from strands import Agent
from strands.vended_tools import notebook

def debugger_callback_handler(**kwargs):
    # Print the values in kwargs so that we can see everything
    print(kwargs)

agent = Agent(
    tools=[notebook],
    callback_handler=debugger_callback_handler
)

agent('Create a notebook named "ideas" and add three project ideas.')

This handler prints all calls to the callback handler including full event details.

Example - Buffering Output Per Message

This handler buffers text and shows it only when a complete message is generated, which suits chat interfaces that display polished, complete responses:

import json
from strands import Agent
from strands.vended_tools import notebook

def message_buffer_handler(**kwargs):
    # When a new message is created from the assistant, print its content
    if "message" in kwargs and kwargs["message"].get("role") == "assistant":
        print(json.dumps(kwargs["message"], indent=2))

# Usage with an agent
agent = Agent(
    tools=[notebook],
    callback_handler=message_buffer_handler
)

agent('Add two AWS Lambda project ideas to the "ideas" notebook.')

The message event fires when a complete message is created, so this handler buffers the incrementally streamed text and displays whole messages rather than partial fragments. Use it for conversational interfaces, or wherever responses read better as complete units.

Example - Event Loop Lifecycle Tracking

This handler prints each lifecycle event as it arrives, so you can watch the order the agent moves through its loop:

from strands import Agent
from strands.vended_tools import notebook

def event_loop_tracker(**kwargs):
    # Track event loop lifecycle
    if kwargs.get("init_event_loop", False):
        print("Event loop initialized")
    elif kwargs.get("start_event_loop", False):
        print("Event loop cycle starting")
    elif "message" in kwargs:
        print(f"New message created: {kwargs['message']['role']}")
    elif "result" in kwargs:
        print("Agent completed with result")
    elif kwargs.get("force_stop", False):
        print(f"Event loop force-stopped: {kwargs.get('force_stop_reason', 'unknown reason')}")

    # Track tool usage
    if "current_tool_use" in kwargs and kwargs["current_tool_use"].get("name"):
        tool_name = kwargs["current_tool_use"]["name"]
        print(f"Using tool: {tool_name}")

    # Show the first 20 characters of each text chunk to keep output readable
    if "data" in kwargs:
        data_snippet = kwargs["data"][:20] + ("..." if len(kwargs["data"]) > 20 else "")
        print(f"Text: {data_snippet}")

# Create agent with event loop tracker
agent = Agent(
    tools=[notebook],
    callback_handler=event_loop_tracker
)

# This will show the full event lifecycle in the console
agent('Read the "ideas" notebook and summarize it.')

The output will show the sequence of events:

1. First the event loop initializes (init_event_loop)
2. Then the cycle begins (start_event_loop)
3. New cycles may start multiple times during execution (start_event_loop)
4. Text generation and tool usage events occur during the cycle
5. Finally, the agent completes with a result event or may be force-stopped

Best Practices

When implementing callback handlers:

1. Keep Them Fast: Callback handlers run in the critical path of agent execution
2. Handle All Event Types: Be prepared for different event types
3. Graceful Errors: Handle exceptions within your handler
4. State Management: Store accumulated state in the request_state

Tags

streaming3