Lesson 4: Adding Callbacks & Response Streaming

Code for this lesson: samples/04-callbacks-streaming

Callbacks Control What the User Sees

Callbacks control how agent output surfaces to users. The default handler streams text to stdout, which is fine for a terminal demo but not for a UI, an API, or a sub-agent running behind the scenes. You can replace the default with a custom handler, silence output entirely, or use stream_async() for async servers.

Custom Callback Handler

A callback handler is a function that accepts **kwargs. It fires for every agent event: text chunks, tool calls, and complete messages. This handler ignores streaming chunks and only prints complete assistant messages:

from strands import Agent, tool

@tool
def calculator(a: float, b: float, operation: str = "add") -> str:
    """Perform a math operation on two numbers.

    Args:
        a: First number
        b: Second number
        operation: One of "add", "subtract", "multiply", "divide", "power"
    """
    if operation == "add":
        result = a + b
    elif operation == "subtract":
        result = a - b
    elif operation == "multiply":
        result = a * b
    elif operation == "divide":
        result = a / b if b != 0 else "Error: division by zero"
    elif operation == "power":
        result = a ** b
    else:
        result = f"Unknown operation: {operation}"
    return str(result)

def buffered_handler(**kwargs):
    # Only show complete messages, not individual streaming chunks
    if "message" in kwargs and kwargs["message"].get("role") == "assistant":
        content = kwargs["message"].get("content", [])
        for block in content:
            if "text" in block:
                print(block["text"])

agent = Agent(tools=[calculator], callback_handler=buffered_handler)
agent("What is 2 to the power of 16, minus 1?")

📂 callbacks_streaming.py

Useful event keys you’ll see in kwargs:

data: a streamed text chunk
current_tool_use: the tool the model is invoking, including its input
message: a complete message once a turn finishes

Silent Mode

Set callback_handler=None to suppress all output. The agent runs and returns a result you can use programmatically:

agent = Agent(tools=[calculator], callback_handler=None)
result = agent("What is 42 * 42?")
print(f"Captured result: {result}")

This is essential for sub-agents in multi-agent systems that run behind the scenes. Only the orchestrator should be streaming to the user.

Async Streaming (FastAPI)

For async servers, use agent.stream_async(), an async generator that yields events as they happen. Here it powers a FastAPI streaming endpoint:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from strands import Agent, tool

@tool
def calculator(a: float, b: float, operation: str = "add") -> str:
    """Perform a math operation on two numbers.

    Args:
        a: First number
        b: Second number
        operation: One of "add", "subtract", "multiply", "divide", "power"
    """
    if operation == "add":
        result = a + b
    elif operation == "subtract":
        result = a - b
    elif operation == "multiply":
        result = a * b
    elif operation == "divide":
        result = a / b if b != 0 else "Error: division by zero"
    elif operation == "power":
        result = a ** b
    else:
        result = f"Unknown operation: {operation}"
    return str(result)

app = FastAPI()


class PromptRequest(BaseModel):
    prompt: str


@app.post("/stream")
async def stream_response(request: PromptRequest):
    async def generate():
        agent = Agent(tools=[calculator], callback_handler=None)
        async for event in agent.stream_async(request.prompt):
            if "data" in event:
                yield event["data"]

    return StreamingResponse(generate(), media_type="text/plain")

📂 fastapi_streaming.py · async_streaming.py

Run it with pip install fastapi uvicorn and uvicorn fastapi_streaming:app --reload, then POST to /stream.

Resources

📖 Callback Handlers
📖 Tools Overview