---
title: Streaming Tools
description: Understand async-generator tools and aggregation.
type: conceptual
summary: Learn how streaming tools emit partial output while producing a final model-facing result.
---

# Streaming Tools



Async-generator tools yield values while they run. An aggregator turns those
values into the final tool result that the model sees on the next turn.

## Async-generator tools

Async-generator tools can stream partial output while they run. Use
`ai.StreamingTextTool` when yielded strings should be concatenated into the
tool result:

```python
@ai.tool
async def draft_mothership_reply(topic: str) -> ai.StreamingTextTool:
    """Draft a reply from the mothership."""
    yield "Consulting "
    yield "the "
    yield "mothership..."
```

## Partial tool results

Partial yields appear as `ai.events.PartialToolCallResult` events. The model
sees the aggregated result on the next turn.

## Aggregators

An aggregator receives each yielded value, keeps a snapshot for consumers, and
converts that snapshot to model-facing input:

```python
from collections.abc import AsyncGenerator
from typing import Annotated

import ai


type Lines = Annotated[
    AsyncGenerator[str],
    ai.agents.Aggregate(ai.agents.ConcatAggregator, delim="\n"),
]


@ai.tool
async def list_mothership_tasks() -> Lines:
    """List pending mothership tasks."""
    yield "Calibrate antenna"
    yield "Check orbit"
```

## Model-facing input

`ToolResultPart.result` stores the rich snapshot. `ToolResultPart.get_model_input`
returns the value sent back to the model:

```python
result_part = event.results[0]
print(result_part.result)
print(result_part.get_model_input())
```

For most tools, those values are the same. For subagents, the result can contain
messages while the model input is the final assistant text.

## Rich snapshots

Aggregators can preserve more than text. `MessageAggregator` stores nested
messages from a subagent:

```python
if isinstance(event, ai.events.ToolCallResult):
    result = event.results[0].result
    if isinstance(result, ai.agents.MessageBundle):
        print(result.messages[-1].text)
```

## Streaming text tools

Use `ai.StreamingTextTool` when every yielded string should be concatenated:

```python
@ai.tool
async def draft_mothership_reply(topic: str) -> ai.StreamingTextTool:
    """Draft a reply from the mothership."""
    yield "The "
    yield "mothership "
    yield f"reports on {topic}."
```

## Status tools

Use `ai.StreamingStatusTool[T]` when intermediate yields are progress updates
and the last yielded value is the final result:

```python
@ai.tool
async def check_alignment() -> ai.StreamingStatusTool[str]:
    """Check orbital alignment."""
    yield "opening channel"
    yield "checking telemetry"
    yield "alignment stable"
```

## Subagent tools

Use `ai.SubAgentTool` when a tool delegates work to another agent:

```python
@ai.tool
async def ask_mothership(topic: str) -> ai.SubAgentTool:
    """Ask the mothership subagent."""
    subagent = ai.agent()
    async with subagent.run(
        model,
        [ai.user_message(topic)],
    ) as stream:
        async for event in stream:
            yield event
```

The parent stream receives the nested events. The parent model receives the
subagent's final assistant text.


---

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

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