---
title: Subagents
description: Compose agents and stream nested agent output.
type: guide
summary: Run subagents as tools, stream nested output.
---

# Subagents



Subagents are regular agents used inside tools or custom loops. Their events
can stream through the parent run while their final text becomes model input.

## Run a subagent as a tool

In order for the SDK to correctly handle subagent's outputs, you need to
annotate the tool's output type with `ai.SubAgentTool`:

```python
mothership_model = ai.get_model("anthropic/claude-sonnet-4")


@ai.tool
async def ask_mothership(topic: str) -> ai.SubAgentTool:
    """Ask a specialist agent for mothership guidance."""
    sub_agent = ai.Agent()
    sub_messages = [
        ai.system_message("Answer as the mothership operations desk."),
        ai.user_message(topic),
    ]

    async with sub_agent.run(mothership_model, sub_messages) as stream:
        async for event in stream:
            yield event
```

This allows the framework to stream partial outputs, keep nested chat history
in `ai.MessageBundle` (which is a marker type for `list[ai.Message]`), and
pass the final output to the parent agent.

`ai.SubAgentTool` declares the aggregator for nested agent events. The parent
consumer receives each nested event as `PartialToolCallResult.value`:

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

## Run streams in parallel

Use `ai.yield_from` inside a custom loop to run branches concurrently and
forward their events:

```python
async with (
    mothership.run(model, mothership_messages) as mothership_stream,
    data_centers.run(model, data_center_messages) as data_center_stream,
):
    mothership_text, data_center_text = await asyncio.gather(
        ai.yield_from(
            mothership_stream,
            label="mothership",
            aggregator=ai.agents.MessageAggregator,
        ),
        ai.yield_from(
            data_center_stream,
            label="data_centers",
            aggregator=ai.agents.MessageAggregator,
        ),
    )
```

`yield_from` wraps events in `PartialToolCallResult` with the label
you pass, and forwards them outside via the internal runtime queue:

```python
if isinstance(event, ai.events.PartialToolCallResult):
    if event.label == "mothership":
        route_to_mothership_panel(event.value)
    elif event.label == "data_centers":
        route_to_data_center_panel(event.value)
```


---

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

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

For agent-facing discovery, including API and MCP surfaces, see [/agents.md](/agents.md)