*[Watch on YouTube](https://www.youtube.com/watch?v=PtdHZBbm1Ys&list=PLDzwjhH-4yhU&index=4)*

About this lesson

The videos in this course are a snapshot in time. Strands is under active development, so the code featured on this page reflects the most up-to-date patterns, but the concepts covered in the video still apply. When in doubt, trust the code.

*Code for this lesson: [`samples/04-callbacks-streaming`](https://github.com/aws-samples/sample-building-with-strands-course/tree/main/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:

```python
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](https://github.com/aws-samples/sample-building-with-strands-course/tree/main/samples/04-callbacks-streaming/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:

```python
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:

```python
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](https://github.com/aws-samples/sample-building-with-strands-course/tree/main/samples/04-callbacks-streaming/fastapi_streaming.py) · [async\_streaming.py](https://github.com/aws-samples/sample-building-with-strands-course/tree/main/samples/04-callbacks-streaming/async_streaming.py)

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

## Resources

-   📖 [Callback Handlers](/pr-cms-4519/docs/user-guide/sdk/streaming/callback-handlers/index.md)
-   📖 [Tools Overview](/pr-cms-4519/docs/user-guide/sdk/tools/index.md)