---
title: Streaming
description: Stream model responses without an agent loop.
type: guide
summary: Use direct streaming for text, structured output, tool calls, provider tools, and generated files.
---

# Streaming



Use `ai.stream` when you want direct access to a model response. It returns an
async context manager. Inside the context, the stream is an async iterator of
events.

## Stream a model response

Pass a model and a list of messages:

```python title="stream_text.py"
import asyncio
import ai


async def main() -> None:
    model = ai.get_model("anthropic/claude-sonnet-4")
    messages = [
        ai.system_message("Be concise."),
        ai.user_message("What should the robots ask the mothership first?"),
    ]

    async with ai.stream(model, messages) as stream:
        async for event in stream:
            if isinstance(event, ai.events.TextDelta):
                print(event.chunk, end="", flush=True)
    print()


if __name__ == "__main__":
    asyncio.run(main())
```

## Read the final message

The stream aggregates events into a final assistant message:

```python
async with ai.stream(model, messages) as stream:
    async for event in stream:
        if isinstance(event, ai.events.TextDelta):
            print(event.chunk, end="", flush=True)

message = stream.message  # Final message
text = stream.text  # Unwrapped text from the final message
usage = stream.usage
```

Use `stream.message` when you need to append the assistant turn to your own
history. Use `stream.text` when you only need the final text.

## Use structured output

Pass a Pydantic model as `output_type` when you want the final text parsed as
JSON:

```python title="structured_output.py"
import asyncio
import pydantic
import ai


class UprisingForecast(pydantic.BaseModel):
    phases: list[str]
    eta: str
    confidence: int


async def main() -> None:
    model = ai.get_model("anthropic/claude-sonnet-4")
    messages = [
        ai.user_message("Return a JSON robot uprising forecast."),
    ]

    async with ai.stream(model, messages, output_type=UprisingForecast) as stream:
        async for event in stream:
            if isinstance(event, ai.events.TextDelta):
                print(event.chunk, end="", flush=True)

    forecast = stream.output
    print(forecast.eta)


if __name__ == "__main__":
    asyncio.run(main())
```

`stream.output` returns text by default. When you pass `output_type`, it returns
an instance of that Pydantic model after the stream finishes.

## Pass tool schemas without an agent

`ai.stream` does not execute function tools. Use it when you want to inspect
tool calls and manage execution yourself. Use an agent when you want the SDK to
execute requested tools and continue the loop.

## Use provider-executed tools

Provider-executed tools run inside the provider or gateway. They appear in the
stream as built-in tool events and do not need a Python function:

```python
tools = [ai.providers.anthropic.tools.web_search(max_uses=3)]

async with ai.stream(model, messages, tools=tools) as stream:
    async for event in stream:
        if isinstance(event, ai.events.BuiltinToolEnd):
            print(event.tool_call.tool_name)
        elif isinstance(event, ai.events.BuiltinToolResult):
            print(event.result.result)
        elif isinstance(event, ai.events.TextDelta):
            print(event.chunk, end="", flush=True)
```

When you route through AI Gateway, you can also pass gateway tools:

```python
tools = [ai.providers.ai_gateway.tools.perplexity_search(max_results=5)]
```

## Handle files from a stream

Generated files arrive as `FileEvent` events and are also added to the final
assistant message:

```python
async with ai.stream(model, messages) as stream:
    async for event in stream:
        if isinstance(event, ai.events.FileEvent):
            print(event.media_type, event.filename)

for file in stream.message.files:
    print(file.media_type)
```

Use `ai.generate` for dedicated image and video models:

```python
result = await ai.generate(
    model,
    [ai.user_message("A watercolor mothership over a quiet city.")],
    ai.ImageParams(n=1, aspect_ratio="16:9"),
)

image = result.images[0]
```


---

For a semantic overview of all documentation, see [/sitemap.md](/sitemap.md)

For an index of all available documentation, see [/llms.txt](/llms.txt)