AI Python Streaming Tools Agent Skill

ai-python-streaming-tools

Use async-generator tools when a tool should show progress while it runs.

A streaming tool yields many values, but the agent still needs one final tool

result for the next model turn. The return type tells the SDK how to combine

those yields.

Text Chunks

Use ai.StreamingTextTool when yielded strings should concatenate into the

final tool result.

@ai.tool
async def draft_reply(topic: str) -> ai.StreamingTextTool:
    yield "Checking records for "
    yield topic

The caller sees each yield as ai.events.PartialToolCallResult. The model later

sees one string: "Checking records for {topic}".

Progress Then Result

Use ai.StreamingStatusTool[T] when the last yielded value is the tool result.

@ai.tool
async def ask_mothership(question: str) -> ai.StreamingStatusTool[str]:
    yield "connecting"
    yield "transmitting"
    yield f"The mothership says: {question} is under review."

The progress values stream to the caller. The model sees only the final yielded

value.

Subagents

Use ai.SubAgentTool when a tool runs another agent and streams its events.

@ai.tool
async def research(topic: str) -> ai.SubAgentTool:
    researcher = ai.Agent()

    messages = [
        ai.system_message("Research briefly."),
        ai.user_message(topic),
    ]

    async with researcher.run(model, messages) as stream:
        async for event in stream:
            yield event

The parent caller receives nested events as PartialToolCallResult.value.

async with agent.run(model, messages) as stream:
    async for event in stream:
        if isinstance(event, ai.events.PartialToolCallResult):
            if isinstance(event.value, ai.events.TextDelta):
                yield event.value.chunk

SubAgentTool stores a typed MessageBundle as the tool result. The parent

model sees the nested agent's final assistant text, not the raw bundle.

When saving history, keep the typed message data:

data = message.model_dump(mode="json")
message = ai.messages.Message.model_validate(data)

Do not stringify MessageBundle or drop result_kind.

Custom Aggregation

Prefer the aliases above. If you need custom aggregation, use either

@ai.tool(aggregator=...) or an Annotated return type. Do not use both.

from collections.abc import AsyncGenerator
from typing import Annotated

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


@ai.tool
async def outline(topic: str) -> JoinedLines:
    yield f"# {topic}"
    yield "- first point"

Custom aggregators implement ai.events.Aggregator.

Rules

• Streaming tools must be async generators.
• Every streaming tool needs an aggregator, usually from the return type alias.
• Consume live output from ai.events.PartialToolCallResult.
• The final aggregated value is sent back to the model as a normal tool result.
• In custom agent loops, keep ToolRunner events flowing; otherwise partial

tool output will not reach the caller.