Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ async def contact_mothership(query: str) -> str:


async def main() -> None:
model = ai.get_model("gateway:anthropic/claude-sonnet-4")
model = ai.get_model("anthropic/claude-sonnet-4")
agent = ai.agent(tools=[contact_mothership])

messages = [
Expand Down Expand Up @@ -105,7 +105,7 @@ Built-in tools execute on the provider side and arrive as part of the stream:
async with ai.stream(
model,
[ai.user_message("Latest Formula 1 results?")],
tools=[ai.anthropic.tools.web_search(max_uses=3)],
tools=[ai.providers.anthropic.tools.web_search(max_uses=3)],
) as s:
async for event in s:
if isinstance(event, ai.events.TextDelta):
Expand Down
53 changes: 11 additions & 42 deletions docs/ai-python/content/docs/basics/agents.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -6,30 +6,16 @@ summary: Create agents, run the default loop, inspect results, and understand mu
---

Use an agent when the model needs to call tools and continue with the tool
results. The default agent loop is built from the same primitives you can use
directly: `ai.stream`, `ToolRunner`, messages, and events.
results.

## Create an agent

An agent wraps `ai.stream` in a loop. It streams model output, executes requested
tools, appends tool results to history, and repeats until the model returns a
final assistant message.

```python
agent = ai.agent(tools=[contact_mothership])

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

Use `ai.agent` for the default loop. Subclass `ai.Agent` and override
`async def loop()` when you need to change control flow.

## Run the default loop

Create an agent with tools, then call `agent.run`:
Subclass `ai.Agent` and override `async def loop()` when you need to change
control flow.

```python title="agent_loop.py"
import asyncio
Expand Down Expand Up @@ -64,17 +50,19 @@ if __name__ == "__main__":
asyncio.run(main())
```

## Inspect the run result

The stream yields model events and agent events. After the run finishes,
`stream.messages` contains the updated history, and `stream.output` contains
the final assistant output.

Unline `ai.stream`, every `agent.run` can produce multiple messages alternating
between `"user"`/`"tool"` and `"assistant"`, representing turns in the LLM request
and response cycle.

## Understand multi-turn behavior

Each loop turn streams one assistant message. If the message contains tool
calls, the agent executes them, appends one tool-result message, and starts the
next model turn.
Each loop turn has one `ai.stream` that produces one assistant message. If the
message contains tool calls, the agent executes them, appends one tool-result
message, and starts the next model turn.

```python
async with agent.run(model, messages) as stream:
Expand Down Expand Up @@ -109,7 +97,7 @@ async with agent.run(
model,
[ai.user_message("Return a JSON mothership forecast.")],
output_type=Forecast,
params={"temperature": 0},
params=ai.InferenceRequestParams().with_temperature(0),
) as stream:
async for event in stream:
if isinstance(event, ai.events.TextDelta):
Expand All @@ -118,22 +106,3 @@ async with agent.run(
forecast = stream.output
print(forecast.eta)
```

## Choose agents or direct streaming

Use `ai.stream` when you want one model response and you will handle any tool
calls yourself. Use an agent when the SDK should execute Python tools, append
tool results, and keep looping until the assistant returns a final answer.

```python
# Direct stream: inspect tool calls yourself.
async with ai.stream(model, messages, tools=[get_weather.tool]) as stream:
async for event in stream:
...

# Agent: execute registered tools.
agent = ai.agent(tools=[get_weather])
async with agent.run(model, messages) as stream:
async for event in stream:
...
```
91 changes: 43 additions & 48 deletions docs/ai-python/content/docs/basics/ai-sdk-ui.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -5,42 +5,18 @@ type: guide
summary: Parse UI messages, apply approvals, stream SSE responses, set headers, and rebuild UI history.
---

The AI SDK UI adapter converts between AI SDK UI messages and the Python
runtime message/event types.
This adapter provides integration with the UI part of the sister AI SDK
in TypeScript.

## Parse UI messages
AI SDK UI provides `useChat`, which handles streaming complexity on the
client side, such as collecting message chunks and optimistically updating
the UI when the user submits a new message.

Accept `UIMessage` values from an AI SDK UI client, then convert them to
runtime messages:
The AI SDK UI's data model is significantly different from the AI SDK for
Python's, which is why you need to convert events and messages back and forth
using the adapter.

```python
class ChatRequest(pydantic.BaseModel):
messages: list[ai.agents.ui.ai_sdk.UIMessage]


@app.post("/chat")
async def chat(request: ChatRequest):
messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages)
```

`to_messages` also extracts approval responses from tool parts.

## Apply approval responses

Register extracted approvals before resuming the agent:

```python
messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages)
ai.agents.ui.ai_sdk.apply_approvals(approvals)

async with chat_agent.run(model, messages) as stream:
...
```

The hook registry stores each approval until the matching tool-approval hook
runs.

## Stream SSE responses
## Outbound streaming and message history

Use `to_sse` to convert agent events to AI SDK UI stream chunks:

Expand All @@ -61,18 +37,6 @@ async def chat(request: ChatRequest) -> fastapi.responses.StreamingResponse:
)
```

## Set response headers

Return the adapter headers on every streamed response:

```python
headers = ai.agents.ui.ai_sdk.UI_MESSAGE_STREAM_HEADERS
```

These headers identify the response as an AI SDK UI message stream.

## Rebuild UI history

Convert stored runtime messages back to AI SDK UI messages for history
endpoints:

Expand All @@ -86,8 +50,6 @@ async def get_chat(session_id: str):
The adapter groups assistant, tool, and internal hook messages into one
assistant UI message.

## Support streaming tool output

Generator tools and subagents emit `PartialToolCallResult` events. The UI
adapter folds those partial values into the corresponding tool output:

Expand All @@ -101,4 +63,37 @@ async def draft_mothership_reply(topic: str) -> ai.StreamingTextTool:
```

For subagents, `ai.SubAgentTool` streams nested events and stores the nested
assistant message as the tool output.
transcript as a `MessageBundle`. The model sees the nested agent's final
assistant text.

## Inbound message updates

Accept `UIMessage` values from an AI SDK UI client, then convert them to
runtime messages:

```python
class ChatRequest(pydantic.BaseModel):
messages: list[ai.agents.ui.ai_sdk.UIMessage]


@app.post("/chat")
async def chat(request: ChatRequest):
messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages)
```

`to_messages` also extracts approval responses from tool parts.

## Tool approvals

Register extracted approvals before resuming the agent:

```python
messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages)
ai.agents.ui.ai_sdk.apply_approvals(approvals)

async with chat_agent.run(model, messages) as stream:
...
```

The hook registry stores each approval until the matching tool-approval hook
runs.
105 changes: 26 additions & 79 deletions docs/ai-python/content/docs/basics/custom-loops.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -5,104 +5,51 @@ type: guide
summary: Override the agent loop to customize scheduling, routing, history updates, logging, and tool execution.
---

Custom loops use the same primitives as the default loop. Override `Agent.loop`
when you need to change scheduling, logging, routing, or persistence.
Override `loop` when you need custom tool dispatch, logging, durability, or
branching. While the SDK defines the loop as a Python async generator, it still
uses a few framework components. Reuse those to retain framework-controlled
behavior.

## When to customize the loop
## Explore the standard loop shape

Override `loop` when you need custom scheduling, logging, durability, or
branching. Keep the same pattern when you still want SDK-managed history and
tool resolution.

## Use the standard loop shape

The default loop keeps running while the last message needs more work. On each
The default loop keeps running while there is more work to do. On each
turn, it streams the model response and schedules tool calls:

```python
class CustomAgent(ai.Agent):
async def loop(self, context: ai.Context):
# ai.Context keeps track of message history, run inputs, and other per-run data
async def loop(self, context: ai.Context) -> AsyncGenerator[ai.events.AgentEvent]:
while context.keep_running():
# call ai.stream with whatever model, tools, and messages the user has
# passed to agent.run
async with (
ai.stream(context=context) as stream,
# ai.ToolRunner is responsible for concurrent tool scheduling and
# streaming, as well as graceful handling of hook interruptions
ai.ToolRunner() as tool_runner,

):
# ai.util.merge interleaves two streams together, so the agent
# can fire off tools as they arrive in the stream rather than
# waiting for the model to stop streaming; and also start streaming
# the streaming tools.
async for event in ai.util.merge(stream, tool_runner.events()):
yield event

if isinstance(event, ai.events.ToolEnd):
# context.resolve looks up the Python function for the
# tool name in the model's tool call
tool_call = context.resolve(event.tool_call)
# schedule the tool for concurrent execution
# using the tool runner
tool_runner.schedule(tool_call)

# add new messages to message history stored in the context.
# this works with internal replay machinery to enable serverless
# execution.
context.add(stream.message)
context.add(tool_runner.get_tool_message())
```

## Resolve and schedule tool calls

`ToolEnd` means the model finished emitting a tool call. Resolve it through the
context, then schedule it with `ToolRunner`:

```python
if isinstance(event, ai.events.ToolEnd):
tool_call = context.resolve(event.tool_call)
tool_runner.schedule(tool_call)
```

`context.resolve` validates that the agent has an executable tool registered
for the model's `tool_name`.

## Add messages to history

After a stream turn finishes, add the assistant message and any tool-result
message:

```python
context.add(stream.message)
context.add(tool_runner.get_tool_message())
```

`context.add` skips replayed assistant messages, so resume flows can call it
without duplicating history.

## Run tools sequentially

`ToolRunner.schedule` runs active tools concurrently. To run tools one at a
time, collect the calls during the model stream, then execute them in order:

```python
pending: list[ai.ToolCall] = []

async for event in stream:
yield event
if isinstance(event, ai.events.ToolEnd):
pending.append(context.resolve(event.tool_call))

for tool_call in pending:
result = await tool_call()
yield result
tool_runner.add_result(result)
```

This keeps the same tool-message aggregation path while preserving execution
order.

## Add logging or routing

Custom loops can inspect events before yielding or scheduling:

```python
if isinstance(event, ai.events.ToolEnd):
call = event.tool_call
print(f"tool: {call.tool_name}({call.tool_args})")
tool_runner.schedule(context.resolve(call))
```

You can also route specific tools through custom wrappers:

```python
if tool_call.name == "contact_mothership":
tool_runner.schedule(GatedToolCall(tool_call))
else:
tool_runner.schedule(tool_call)
```
You can modify most of the loop and extend `CustomAgent` to fit your application's needs,
and the SDK will keep working as long as the loop is still an async generator of events.
Loading
Loading