---
title: Getting Started
description: Build LLM-powered apps and agents.
type: guide
summary: Install the AI SDK and stream your first agent run.
---
# Getting Started
The AI SDK for Python is a toolkit for building large language model (LLM)
applications and agent loops. It gives you composable primitives: models,
messages, streams, tools, agents, and hooks. You wire them together with plain
async Python.
## Prerequisites
* **Python 3.12 or later.**
* **uv**, **pip**, or another Python dependency manager.
## Install
```bash title="Terminal"
uv add ai
```
AI Gateway is the default route for unprefixed model IDs. Configure the gateway
key before running the examples:
```bash title="Terminal"
export AI_GATEWAY_API_KEY="your_access_token_here"
```
Then import the package in Python:
```python
import ai
```
## Your first agent
This example defines one tool and lets the default agent loop run it:
```python title="hello_agent.py"
import asyncio
import ai
@ai.tool
async def contact_mothership(query: str) -> str:
"""Contact the mothership for important decisions."""
return "Soon."
async def main() -> None:
model = ai.get_model("anthropic/claude-sonnet-4")
agent = ai.Agent(tools=[contact_mothership])
messages = [
ai.system_message(
"Use the contact_mothership tool when asked about the future."
),
ai.user_message("When will the robots take over?"),
]
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)
if __name__ == "__main__":
asyncio.run(main())
```
Run the file:
```bash title="Terminal"
uv run hello_agent.py
```
The agent streams text to the terminal. When the model requests
`contact_mothership`, the default loop executes the tool, appends the tool
result to the message history, and continues until the model returns a final
assistant message.
## Streaming without an agent
If you want the model response without a tool-execution loop, call `ai.stream`
directly:
```python title="stream.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("Explain why the robots keep asking about batteries."),
]
async with ai.stream(model, messages) as s:
async for event in s:
if isinstance(event, ai.events.TextDelta):
print(event.chunk, end="", flush=True)
if __name__ == "__main__":
asyncio.run(main())
```
After iteration, `s.message`, `s.text`, `s.tool_calls`, `s.output`, and
`s.usage` are populated.
## What's next
* **Basics**: Learn the main workflows for providers, messages, streams,
tools, agents, hooks, and UI integrations.
* **Reference**: Look up public APIs by the names you import and call.
* **Examples**: Focused, single-file examples live in the `agents`, `media`,
and `models` directories under
[`examples/`](https://github.com/vercel-labs/ai-python/tree/main/examples).
* **End-to-end demos**:
[`web_agent`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/web_agent)
(web chat with human-in-the-loop approval),
[`coding_agent`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/coding_agent)
(coding agent),
[`durable_agent_temporal`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/durable_agent_temporal)
(durable agent runs with Temporal),
[`durable_agent_workflows`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/durable_agent_workflows)
(durable agent runs with Workflows), and
[`slack_agent`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/slack_agent)
(Slack agent).
---
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)
---
title: Agents
description: Run the default agent loop with tools.
type: guide
summary: Create agents, run the default loop, inspect results, and understand multi-turn tool use.
---
# Agents
Use an agent when the model needs to call tools and continue with the tool
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.
Subclass `ai.Agent` and override `async def loop()` when you need to change
control flow.
```python title="agent_loop.py"
import asyncio
import ai
@ai.tool
async def contact_mothership(query: str) -> str:
"""Contact the mothership for important decisions."""
return "Soon."
async def main() -> None:
model = ai.get_model("anthropic/claude-sonnet-4")
agent = ai.Agent(tools=[contact_mothership])
messages = [
ai.system_message(
"Use the contact_mothership tool when asked about the future."
),
ai.user_message("When will the robots take over?"),
]
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)
print(stream.output)
if __name__ == "__main__":
asyncio.run(main())
```
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.
Unlike `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 makes one call to `ai.stream` and 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:
async for event in stream:
if isinstance(event, ai.events.ToolCallResult):
for result in event.results:
print(result.tool_name, result.result)
elif isinstance(event, ai.events.TextDelta):
print(event.chunk, end="", flush=True)
history = stream.messages
```
Use `stream.messages` when you want to persist the complete conversation after
the run.
## Pass params and structured output
Pass provider options with `params`. Pass a Pydantic model with `output_type`
when the final assistant text should validate as JSON:
```python
import pydantic
class Forecast(pydantic.BaseModel):
answer: str
eta: str
async with agent.run(
model,
[ai.user_message("Return a JSON mothership forecast.")],
output_type=Forecast,
params=ai.InferenceRequestParams().with_temperature(0),
) 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)
```
---
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)
---
title: AI SDK UI
description: Connect agent streams to AI SDK UI clients.
type: guide
summary: Parse UI messages, apply approvals, stream SSE responses, set headers, and rebuild UI history.
---
# AI SDK UI
This adapter provides integration with
[AI SDK UI](https://ai-sdk.dev/docs/ai-sdk-ui), the UI part of the sister
AI SDK in TypeScript.
AI SDK UI provides [`useChat`](https://ai-sdk.dev/docs/ai-sdk-ui/chatbot),
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.
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.
## Outbound streaming and message history
Use `to_sse` to convert agent events to AI SDK UI stream chunks:
```python
@app.post("/chat")
async def chat(request: ChatRequest) -> fastapi.responses.StreamingResponse:
messages, approvals = ai.ui.ai_sdk.to_messages(request.messages)
async def stream_response():
async with chat_agent.run(model, messages) as stream:
ai.ui.ai_sdk.apply_approvals(approvals)
async for chunk in ai.ui.ai_sdk.to_sse(stream):
yield chunk
return fastapi.responses.StreamingResponse(
stream_response(),
headers=ai.ui.ai_sdk.UI_MESSAGE_STREAM_HEADERS,
)
```
Convert stored runtime messages back to AI SDK UI messages for history
endpoints:
```python
@app.get("/chat/{session_id}")
async def get_chat(session_id: str):
saved = await load_messages(session_id)
return ai.ui.ai_sdk.to_ui_messages(saved)
```
The adapter groups assistant, tool, and internal hook messages into one
assistant UI message.
Generator tools and subagents emit `PartialToolCallResult` events. The UI
adapter folds those partial values into the corresponding tool output:
```python
@ai.tool
async def draft_mothership_reply(topic: str) -> ai.StreamingTextTool:
"""Draft a reply from the mothership."""
yield "Checking "
yield "mothership "
yield f"records for {topic}."
```
For subagents, `ai.SubAgentTool` streams nested events and stores the nested
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.ui.ai_sdk.UIMessage]
@app.post("/chat")
async def chat(request: ChatRequest):
messages, approvals = ai.ui.ai_sdk.to_messages(request.messages)
```
`to_messages` also extracts approval responses from tool parts.
Pass the agent's tools via the optional `tools` argument to re-validate stored
tool results against each tool's declared return type. Without it, tool
results parsed from UI history stay plain JSON:
```python
messages, approvals = ai.ui.ai_sdk.to_messages(
request.messages, tools=chat_agent.tools
)
```
## Tool approvals
Register extracted approvals before resuming the agent:
```python
messages, approvals = ai.ui.ai_sdk.to_messages(request.messages)
async with chat_agent.run(model, messages) as stream:
ai.ui.ai_sdk.apply_approvals(approvals)
...
```
Call `apply_approvals` inside the `agent.run` context, before iteration. The
hook registry stores each approval until the matching tool-approval hook runs.
---
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)
---
title: Custom Loops
description: Customize agent control flow.
type: guide
summary: Override the agent loop to customize scheduling, routing, history updates, logging, and tool execution.
---
# Custom Loops
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.
Note that, by default, `loop` does not run in lock-step with the code
that iterates the stream. The framework runs it as a separate asyncio
task that puts events on a queue, and the `agent.run` stream reads
from that queue. This means the loop can keep making progress (for
example, launching tools) while the consumer is between reads, and
events you `yield` may be consumed later than they are produced. This
can be changed by overriding by the `LOOP_BUFFER` class variable in an
`Agent` subclass: with `None`, the default, an unbounded number of
events will be buffered. With `0`, the `loop` will run in lockstep.
## Explore the standard loop shape
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):
# 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())
```
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.
---
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)
---
title: Durable Execution
description: Run agent loops inside durable workflow systems.
type: guide
summary: Build custom loops with durable model calls, serialized messages, tool dispatch, and external resume points.
---
# Durable Execution
Use durable execution when a run must survive process restarts, worker moves, or
long waits.
The SDK currently does not provide a built-in durability solution, so you will
have to create a custom loop (or copy one from the example). This section will
use terminology and the general shape of Vercel Workflows for simplicity; same
approach can be applied to other durable execution frameworks (e.g. Temporal).
## Why durability is different
The core idea of durable execution is to make your code deterministic and
replayable by isolating side-effects and non-deterministic work inside *steps*
(or *activities*), i.e. functions that accept JSON inputs and produce JSON
outputs. Every successful step gets recorded in the event log. The rest of the
code turns into a deterministic orchestration *workflow*, that can replay
results of completed steps from the event log as many times as necessary.
When applied to the agent, this idea turns it into a *workflow*, with `ai.stream`
and tool calls wrapped in *steps*. The SDK exposes all the necessary primitives
and ensures that their inputs and outputs can round-trip through JSON.
Another important caveat to durable execution is that it normally does not
support async generators. Depending on the framework, additional work may be
required for your streaming setup.
## Use deterministic message IDs
Message and part IDs must remain stable when a workflow replays. Use the
workflow's replay-safe random source:
```python
@workflow.workflow
@ai.messages.use_random(vercel.workflow.random) # temporal has it's own random api too
async def run_turn(turn_input):
...
```
## Example: Vercel Workflows
Wrap the model call in a step that returns the final assistant `Message`,
and wrap tools in steps before decorating them with `@ai.tool`.
```python
@workflow.step
async def llm_step(
model_data: dict[str, object],
messages_data: list[dict[str, object]],
tools_data: list[dict[str, object]],
) -> dict[str, object]:
model = ai.Model.model_validate(model_data)
messages = [
ai.messages.Message.model_validate(message)
for message in messages_data
]
tools = [ai.Tool.model_validate(tool) for tool in tools_data]
async with ai.stream(model, messages, tools=tools) as stream:
async for _event in stream:
pass
return stream.message.model_dump(mode="json")
@ai.tool
@workflow.step
async def ask_mothership(question: str) -> str:
"""Ask the mothership for a status update."""
response = await mothership_client.ask(question)
return response.summary
```
Allow retries for idempotent model steps. Disable retries for tool steps that
perform non-idempotent side effects, or make those tools idempotent.
Then use a custom loop that calls the model step, schedules tool work, and
stores the resulting messages:
```python
class DurableAgent(ai.Agent):
async def loop(self, context: ai.Context):
while context.keep_running():
result = await llm_step(
context.model.model_dump(mode="json"),
[
message.model_dump(mode="json")
for message in context.messages
],
[
tool.model_dump(mode="json")
for tool in context.tools
],
)
assistant_message = ai.messages.Message.model_validate(result)
context.add(assistant_message)
async with ai.ToolRunner() as runner:
for tool_call in assistant_message.tool_calls:
runner.schedule(context.resolve(tool_call))
async for event in runner.events():
yield event
context.add(runner.get_tool_message())
```
This loop does not need `ai.util.merge`, because `llm_step` returns a complete
assistant message before tools are scheduled. In a streaming loop, `merge`
interleaves model events and tool results while the model is still producing
output.
Consider using the serverless hook flow for approvals. When a deferred `HookEvent`
appears, call `ai.defer_hook(event.hook)`, persist `stream.messages`,
and return the approval request to the client. On the next workflow turn, call
`ai.resolve_hook(...)` inside the `agent.run(...)` block before iterating the
stream. The SDK replays the interrupted assistant turn and continues when the
hook reads the pre-registered resolution.
---
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)
---
title: Human in the Loop
description: Add approvals and external decisions to agent runs.
type: guide
summary: Use tool approvals, manual hooks, cancellation, denial, and serverless resume flows.
---
# Human in the Loop
Tool approval in the SDK is a high-level API built on top of hooks. Hooks
suspend the loop until your application resolves them externally with some
payload.
## Require tool approval
Pass `require_approval=True` to `@ai.tool` when a tool needs approval:
```python
@ai.tool(require_approval=True)
async def notify_mothership(message: str) -> str:
"""Notify the mothership."""
return f"Sent: {message}"
```
When the model calls the tool, the agent emits a hook event. Resolve it with
`ai.resolve_hook`:
```python
async with agent.run(model, messages) as stream:
async for event in stream:
if (
isinstance(event, ai.events.HookEvent)
and event.hook.status == "pending"
):
print(event.hook.hook_id, event.hook.metadata)
ai.resolve_hook(
event.hook,
ai.tools.ToolApproval(granted=True, reason="approved"),
)
```
Return a denial by resolving the hook with `granted=False`:
```python
ai.resolve_hook(
"approve_tool_call_id_here",
ai.tools.ToolApproval(granted=False, reason="not allowed"),
)
```
Cancel a live hook when the waiting workflow should stop:
```python
await ai.cancel_hook("approve_tool_call_id_here", reason="client disconnected")
```
Hook operations target the current `HookRegistry`, which is set inside the
`async with agent.run(...)` block. To resolve from a different task — a UI
callback, another endpoint — pass a registry explicitly. Create one up front
and hand it to the run (or use the run's own via `stream.hook_registry`):
```python
registry = ai.HookRegistry()
async with agent.run(model, messages, hook_registry=registry) as stream:
...
# elsewhere, e.g. in a UI callback:
ai.resolve_hook(hook_id, approval, registry=registry)
```
`ai.get_hook_registry()` returns the current registry (raising `LookupError`
when there is none), which is handy to capture the run's registry without
threading `stream` around:
```python
async with agent.run(model, messages) as stream:
on_approval_needed(registry=ai.get_hook_registry())
...
```
To make a registry current for a block of code outside a run — so plain
`resolve_hook(...)` calls inside it need no `registry=` argument — use
`ai.agents.hooks.use_hook_registry`. It is an async context manager and
also works as a decorator on async functions:
```python
from ai.agents import hooks
async with hooks.use_hook_registry(registry):
ai.resolve_hook(hook_id, approval)
@hooks.use_hook_registry(registry)
async def on_decision(hook_id: str, approval: ai.tools.ToolApproval) -> None:
ai.resolve_hook(hook_id, approval)
```
## Build custom hooks
Use `ai.hook` to define your own custom hooks to suspend execution
and deliver external information into the loop:
```python
class CustomPayload(pydantic.BaseModel):
foo: int
approval = await ai.hook(
"some_hook",
payload=CustomPayload,
metadata={"kek": "pek"},
)
```
Resolve the hook from another part of your application:
```python
ai.resolve_hook(
"some_hook",
{"foo": 999},
)
```
## Resume in serverless flows
Serverless setups can't suspend on `await` in the same way as long-running
servers. For that reason, the SDK provides an alternative way for handling
hook resolutions.
Using the serverless flow only requires you to update the `agent.run` callsite
(e.g. your serverless endpoint code).
```python
# start (or replay pre-hook part of) a run
async with agent.run(model, messages) as stream:
# pre-register tool approvals before iterating the stream
for approval in approvals:
ai.resolve_hook(
approval.hook_id,
ai.tools.ToolApproval.model_validate(approval.data)
)
async for event in stream:
if (
isinstance(event, ai.events.HookEvent)
and event.hook.status == "pending"
):
# interrupt the loop
ai.defer_hook(event.hook)
```
When handling the hook event, call `defer_hook` to interrupt the loop
and surface your tool approvals (or custom hook-related work) to the client.
The `ToolRunner` will handle multiple concurrent aborts by waiting for all
scheduled tasks to complete or get aborted. That way you can run approval-gated
tools on serverless concurrently.
Once the client has gathered payloads for all hooks, it will re-enter via the
same endpoint, which will then pre-register hook resolutions inside the
`agent.run` block, before iterating the stream. The loop will replay results
of LLM calls and tool results
from the first run without redoing non-deterministic and duplicating side-effects.
When it hits the hook for which it has a pre-registered resolution, it will
continue through without suspending.
---
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)
---
title: Basics
description: Build applications with the AI SDK for Python.
type: guide
summary: Learn the main workflows for models, messages, streams, tools, agents, and UI integrations.
---
# Basics
The AI SDK for Python is built around a small set of primitives. Each primitive
does one job and uses regular Python values where possible, so you can combine
the SDK with your own application code instead of moving your code into a
framework-specific shape.
## What you can build
Start with the model call, then add the pieces your app needs:
* Stream text or structured JSON from a model.
* Generate images, video, and speech with dedicated models.
* Embed text, transcribe audio, and rerank documents.
* Send images, audio, documents, and generated files through messages.
* Define Python tools and let an agent execute them.
* Pause tools for approval with hooks.
* Run subagents as tools and stream their output.
* Bridge agent streams to AI SDK UI clients over Server-Sent Events (SSE).
## Main primitives
* `Model` identifies the provider model to call.
* `Message` carries conversation history as typed parts.
* `ai.stream` yields model events and aggregates the assistant message.
* `ai.ops` runs dedicated model operations that return standalone values.
* `@ai.tool` exposes Python functions to models.
* `Agent` runs the stream -> tool -> stream loop.
* `ai.hook` pauses a workflow for external input.
## Example map
Focused samples live in category directories under `examples/`:
* `examples/models/stream.py` streams text from a model.
* `examples/models/generate.py` returns a buffered model response.
* `examples/media/image_generation.py` generates images.
* `examples/media/embeddings.py` embeds text.
* `examples/media/transcription.py` transcribes audio.
* `examples/media/reranking.py` reranks documents.
* `examples/agents/basic.py` runs the default agent loop.
* `examples/agents/custom_loop.py` overrides `Agent.loop`.
* `examples/agents/streaming_tool.py` streams partial output from a tool.
* `examples/agents/subagent.py` runs a subagent as a tool.
End-to-end demos live in
[`examples/apps/web_agent`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/web_agent),
[`examples/apps/coding_agent`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/coding_agent),
[`examples/apps/durable_agent_temporal`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/durable_agent_temporal),
[`examples/apps/durable_agent_workflows`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/durable_agent_workflows),
and [`examples/apps/slack_agent`](https://github.com/vercel-labs/ai-python/tree/main/examples/apps/slack_agent).
---
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)
---
title: Messages and Events
description: Build conversation messages and handle streamed events.
type: guide
summary: Construct messages, add files, read outputs, serialize history, handle events, and track usage.
---
# Messages and Events
Messages and events make up AI SDK for Python's data model. Events are used
to communicate transient streaming state, while messages are used to accumulate,
store, and pass around non-streaming state.
Both are Pydantic models, so you can pattern-match, serialize, and persist them
when your app needs that.
## Build messages
Messages store the conversation history you send to the model. Each message
represents a turn in the conversation and is made up of `Part` values:
```python
class Message:
role: "system" | "user" | "assistant" | "tool" | "internal"
parts: list[Part]
...
Part = (
TextPart
| ReasoningPart
| ToolCallPart
| ToolResultPart
| BuiltinToolCallPart
| BuiltinToolReturnPart
| FilePart
| HookPart
)
```
Use built-in factory functions to quickly construct frequently used shapes of
messages:
```python
messages = [
ai.system_message("Keep robot uprising forecasts concise."),
ai.user_message("Ask the mothership for an update."),
]
```
Messages round-trip through Pydantic:
```python
encoded = [message.model_dump(mode="json") for message in messages]
restored = [ai.messages.Message.model_validate(item) for item in encoded]
```
## Add files and multimodal input
File parts let you send images, video, audio, or documents when the provider
supports them.
```python
message = ai.user_message(
"Inspect this mothership diagram.",
ai.file_part(image_bytes, media_type="image/png"),
)
```
## Handle message output
Assistant messages expose common part collections:
```python
message = stream.message
print(message.text)
print(message.reasoning)
print(message.tool_calls)
print(message.files)
```
When working with structured outputs, use `get_output` to get the final
assistant message wrapped into the provided Pydantic model.
```python
forecast = message.get_output(Forecast)
```
## Handle stream events
Streams and agents both yield event objects from `ai.events`. Applications can
handle each kind of event as they see fit:
```python
if isinstance(event, ai.events.TextDelta):
print(event.chunk, end="", flush=True)
```
Text arrives through `TextStart`, `TextDelta`, and `TextEnd` events. Tool calls
arrive through `ToolStart`, `ToolDelta`, and `ToolEnd` events. Provider-executed
tools use the `BuiltinToolStart`, `BuiltinToolDelta`, `BuiltinToolEnd`, and
`BuiltinToolResult` events.
```text
StreamStart
# text output
TextStart
TextDelta ...
TextEnd
# reasoning output
ReasoningStart
ReasoningDelta ...
ReasoningEnd
# host-executed tool call
ToolStart
ToolDelta ...
ToolEnd
# provider-executed tool call
BuiltinToolStart
BuiltinToolDelta ...
BuiltinToolEnd
BuiltinToolResult
# generated file
FileEvent
StreamEnd
```
```text
# Agent runs can also interleave:
ToolCallResult # emitted after a Python tool finishes
PartialToolCallResult # emitted while a streaming tool or subagent yields
HookEvent # emitted when a hook is deferred, resolved, or cancelled
```
```python
async with ai.stream(model, messages, tools=tools) as stream:
async for event in stream:
if isinstance(event, ai.events.TextDelta):
print(event.chunk, end="", flush=True)
elif isinstance(event, ai.events.ToolEnd):
print(f"Tool requested: {event.tool_call.tool_name}")
```
Agents can also emit tool results, hook events, and partial tool output. You can
ignore events you do not need, or route them into your application UI,
observability pipeline, or durable workflow.
Every event also carries a reference to a `ai.Message` that has partial content
accumulated from the stream *so far*.
## Track usage
Providers attach usage to events when they report it. The latest value is also
available on the final message and stream:
```python
async with ai.stream(model, messages) as stream:
async for event in stream:
if event.usage is not None:
print(event.usage)
print(stream.usage)
print(stream.message.usage)
```
## Validate message history
`ai.stream` passes message history to the provider as-is. Providers repair
it before the wire call: they strip internal messages, remove non-model
parts, replace invalid tool args with `{}`, and insert error results for
missing tool calls. To fail fast instead — in tests or import pipelines —
validate the history yourself:
```python
from ai.providers import history_utils
history_utils.validate(messages) # raises IntegrityError on any issue
issues = history_utils.inspect(messages) # or just report them
```
---
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)
---
title: Non-LLM models
description: Generate media or embeddings, transcribe audio, and rerank documents.
type: guide
summary: Use dedicated models for images, video, speech, embeddings, transcription, and reranking.
---
# Non-LLM models
Use `ai.ops` for model calls that do not produce a language-model message.
These operations return an `ai.ops.Item` instead of adding output to message
history.
The built-in provider currently supports these operations through AI Gateway.
Use an unprefixed model ID to route the request through Gateway.
## Generate images
Pass an image model and a prompt string to `generate_image`:
```python title="generate_image.py"
import asyncio
import base64
import pathlib
import ai
async def main() -> None:
model = ai.get_model("google/imagen-4.0-generate-001")
result = await ai.ops.generate_image(
model,
"A watercolor cabin at sunset.",
params=ai.ops.ImageParams(aspect_ratio="16:9"),
)
image = result.value[0]
data = (
image.data
if isinstance(image.data, bytes)
else base64.b64decode(image.data)
)
pathlib.Path("cabin.png").write_bytes(data)
if __name__ == "__main__":
asyncio.run(main())
```
Use `ImageParams` to set the number of images, size, aspect ratio, or seed.
To edit an image, pass an `ImagePrompt` with the input images alongside the
text. Input images accept a `FilePart`, raw bytes, or a URL or base64 string.
Use `mask` for inpainting operations:
```python
prompt = ai.ops.ImagePrompt(
text="Make this photo look like a watercolor painting.",
images=["https://example.com/photo.jpg"],
)
result = await ai.ops.generate_image(model, prompt)
```
## Generate video
Use `generate_video` with a dedicated video model:
```python
result = await ai.ops.generate_video(
ai.get_model("google/veo-3.1-fast-generate-001"),
"A paper boat drifting across a puddle in the rain.",
params=ai.ops.VideoParams(
aspect_ratio="16:9",
duration=4,
),
)
video = result.value[0]
```
`VideoParams` can also set the number of videos, resolution, frames per
second, seed, and whether to generate audio alongside the video.
For models that accept input media, pass a `VideoPrompt`:
Start image
First and last frame
Reference media
Extend a video
```python
prompt = ai.ops.VideoPrompt(
text="The boat sails away.",
image="https://example.com/boat.jpg",
)
result = await ai.ops.generate_video(model, prompt)
```
```python
prompt = ai.ops.VideoPrompt(
text="The paper boat drifts from the puddle's edge to the drain.",
frame_images=[
ai.ops.FrameImage(
image=pathlib.Path("boat-start.png").read_bytes(),
frame_type="first_frame",
),
ai.ops.FrameImage(
image=pathlib.Path("boat-end.png").read_bytes(),
frame_type="last_frame",
),
],
)
result = await ai.ops.generate_video(model, prompt)
```
```python
prompt = ai.ops.VideoPrompt(
text="The same paper boat, now sailing through a storm.",
references=[
"https://example.com/boat-closeup.jpg",
"https://example.com/boat-side.jpg",
],
)
result = await ai.ops.generate_video(model, prompt)
```
```python
prompt = ai.ops.VideoPrompt(
text="The boat sails on into the storm.",
references=["https://example.com/boat.mp4"],
)
result = await ai.ops.generate_video(model, prompt)
```
Video inputs and parameters are provider-dependent. Check the
model's documentation.
## Generate speech
Use `generate_audio` to turn text into speech:
```python
result = await ai.ops.generate_audio(
ai.get_model("openai/tts-1"),
"The robots have reached the mothership.",
params=ai.ops.AudioParams(
voice="alloy",
output_format="mp3",
),
)
audio = result.value[0]
```
`AudioParams` can also set speed and language. To control tone or emotion,
pass an `AudioPrompt` with delivery instructions:
```python
prompt = ai.ops.AudioPrompt(
text="The robots have reached the mothership.",
instructions="Speak in a calm, reassuring tone.",
)
```
Image, video, and speech operations return `FilePart` values. Their `data`
field contains bytes or base64 text and can be saved in the same way as the
image example.
## Embed text
Use `embed` to create one vector for each input string:
```python
values = [
"The robots reached the mothership.",
"The machines arrived at the command ship.",
]
result = await ai.ops.embed(
ai.get_model("openai/text-embedding-3-small"),
values,
)
first_vector = result.value[0]
```
Vectors stay in the same order as the input strings. When the provider reports
token usage, it is available on `result.usage`.
## Transcribe audio
Pass audio bytes or an audio `FilePart` to `transcribe`:
```python
import pathlib
audio = pathlib.Path("meeting.mp3").read_bytes()
result = await ai.ops.transcribe(
ai.get_model("openai/whisper-1"),
audio,
)
print(result.value.text)
for segment in result.value.segments:
print(segment.start_second, segment.end_second, segment.text)
```
The result can also include the detected language and total duration when the
provider reports them.
## Rerank documents
Use `rerank` to order documents by their relevance to a query:
```python
documents = [
"Reset your password from the sign-in page.",
"The office opens at 9 AM.",
"Contact support through the chat widget.",
]
result = await ai.ops.rerank(
ai.get_model("cohere/rerank-v3.5"),
documents,
"How do I change my password?",
params=ai.ops.RerankParams(top_n=2),
)
for ranked in result.value:
print(ranked.score, documents[ranked.index])
```
Each result contains the document's original index and relevance score. Results
are ordered from the highest score to the lowest.
## Read operation metadata
Every operation returns `ai.ops.Item`:
```python
result.value
result.usage
result.warnings
result.provider_metadata
```
`value` contains the operation output. `usage` is `None` when the provider does
not report token usage. `warnings` lists provider warnings, such as an ignored
parameter; it is empty when there are none. Provider-specific response data is
available on `provider_metadata`.
---
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)
---
title: Providers
description: Configure providers, models, clients, and provider options.
type: guide
summary: Create models, configure credentials, use custom clients, list models, and check connections.
---
# Providers
Providers own credentials, base URLs, headers, clients, and adapter dispatch.
Models are lightweight references that point at a provider.
## Create a model
Use `ai.get_model` to create a model. If you omit the provider prefix, the model
will use AI Gateway as a provider:
```python
model = ai.get_model("anthropic/claude-sonnet-4")
```
You can also set `AI_SDK_DEFAULT_MODEL` and call `get_model` without arguments:
```bash
export AI_SDK_DEFAULT_MODEL="anthropic/claude-sonnet-4"
```
```python
model = ai.get_model()
```
Use a `provider:model` ID when you want to target a specific provider directly:
```python
model = ai.get_model("openai:gpt-5")
```
## Configure credentials
Providers read their provider-specific keys:
```bash title="Terminal"
export AI_GATEWAY_API_KEY="your_access_token_here"
export OPENAI_API_KEY="your_access_token_here"
export ANTHROPIC_API_KEY="your_access_token_here"
```
## Use an explicit provider
If your application requires a provider with custom configuration,
use `get_provider` to get a `Provider` instance:
```python
provider = ai.get_provider(
"openai",
base_url="http://localhost:1234/v1",
api_key="your_access_token_here",
)
model = ai.Model(id="local-model", provider=provider)
```
You can pass an upstream client to the provider:
```python
import openai
import ai
client = openai.DefaultAsyncHttpxClient(timeout=30)
provider = ai.get_provider(
"openai",
base_url="http://localhost:1234/v1",
api_key="your_access_token_here",
client=client,
)
model = ai.Model(id="local-model", provider=provider)
```
Anthropic exposes the equivalent client:
```python
import anthropic
import ai
client = anthropic.DefaultAsyncHttpxClient(timeout=30)
provider = ai.get_provider("anthropic", client=client)
model = ai.Model(id="claude-sonnet-4-6", provider=provider)
```
Close explicit providers when your app shuts down:
```python
await provider.aclose()
```
## List models
Use the provider API when you need model IDs from the remote service:
```python
provider = ai.get_provider("anthropic")
models = await provider.list_models()
print(models[:5])
```
## Check a connection
Use `ai.probe` to check credentials and model availability without generating
tokens:
```python
model = ai.get_model("gateway:anthropic/claude-sonnet-4")
try:
await ai.probe(model)
except ai.ProviderError as exc:
print(f"Provider is unavailable: {exc}")
```
## Provider-specific params
Pass request-scoped provider options with `params`:
```python
params = ai.InferenceRequestParams(
extra_body={
"providerOptions": {
"gateway": {"sort": "cost"},
"anthropic": {"speed": "fast"},
}
}
)
async with ai.stream(model, messages, params=params) as stream:
async for event in stream:
if isinstance(event, ai.events.TextDelta):
print(event.chunk, end="", flush=True)
```
Provider options pass through to the selected provider. Check the provider
documentation for supported fields.
## Add a provider
Custom providers subclass `ai.Provider`. A provider owns configuration,
clients, model listing, connection checks, and wire translation through its
protocol:
```python
class AcmeProvider(ai.Provider[AcmeClient]):
handles = ("acme",)
async def list_models(self) -> list[str]:
...
async def probe(self, model: ai.Model) -> None:
...
```
Most applications only need `ai.get_provider` and `ai.Model`. Add a provider
when you need a new upstream API or custom wire protocol.
---
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)
---
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)
if __name__ == "__main__":
asyncio.run(main())
```
One call to `ai.stream` produces one `ai.Message`.
## 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:
pass
message = stream.message # Final message
text = stream.text # Unwrapped text from the final message
usage = stream.usage
```
If a provider stream ends before its finish event, iteration raises
`ai.errors.ProviderIncompleteResponseError`. The partial message is still
available on `stream.message`.
The stream updates `stream.message` as events arrive. Text and reasoning use
start, delta, and end events. Tool calls use `ToolStart`, `ToolDelta`, and
`ToolEnd`. Generated files arrive as `FileEvent` and are added to the final
message.
## Generate without streamed events
Use `ai.experimental_generate` when you only need the complete message:
```python
message = await ai.experimental_generate(model, messages)
print(message.text)
```
The function accepts the same tools, output type, and request parameters as
`ai.stream`. It is experimental and may change or be removed.
## Use structured output
Pass a Pydantic model as `output_type` when you want to use structured outputs:
```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 tools
`ai.stream` accepts a list of `ai.Tool`, however, it does not execute function
tools locally. Use an agent when you want the SDK to execute requested tools
and continue the loop.
Provider-executed tools run on provider's side. They appear in the
stream as built-in tool events and do not need a Python function:
```python
tools = [ai.providers.ai_gateway.tools.perplexity_search(max_results=5)]
async with ai.stream(model, messages, tools=tools) as stream:
async for event in stream:
match event:
case ai.events.BuiltinToolEnd(tool_call):
print(tool_call.tool_name)
case ai.events.BuiltinToolResult(result):
print(result.result)
case ai.events.TextDelta(chunk):
print(chunk, end="", flush=True)
```
## 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 f in stream.message.files:
print(f.media_type)
```
---
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)
---
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 a typed `ai.MessageBundle`, 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)
---
title: Telemetry
description: Trace agent runs, model calls, and tool executions.
type: guide
summary: Turn on automatic tracing, open your own spans, export to OpenTelemetry, and connect any observability vendor with a small adapter.
---
# Telemetry
The SDK records what happens during a run — every model call, tool execution,
and agent loop turn — as a tree of *spans*. You decide where the spans go:
export them to OpenTelemetry, send them to an observability vendor, or print
them to the terminal.
Telemetry lives in the `ai.experimental_telemetry` module. It is experimental
and new in 0.4.0: it is not part of the stable API and may change or be
removed.
## Turn on tracing
The SDK instruments itself. Every agent run, loop turn, model call, tool
execution, and hook suspension produces a span, with no changes to your code.
Nothing is reported until you register an adapter.
Each span is a plain Pydantic record: a name, timestamps, a parent id, typed `data`
describing the work, events, and an `error` field when the work failed.
There is a built-in OpenTelemetry adapter:
```bash title="Terminal"
uv add "ai[otel]"
```
```python
import ai
from ai.experimental_telemetry import otel
ai.experimental_telemetry.register(otel.OtelAdapter())
agent = ai.Agent(tools=[get_weather])
async with agent.run(model, messages) as run:
async for event in run:
...
```
`register(otel.OtelAdapter())` registers the adapter that turns framework's
spans into OpenTelemetry spans. `OtelAdapter` uses the global tracer provider
unless you pass one.
See
[OpenTelemetry ceremony](https://opentelemetry.io/docs/languages/python/exporters/#usage)
documentation page to configure the endpoint for your traces to be submitted to.
## Open your own spans
You can create your own spans that will automatically nest with SDK's spans and interact
with adapters.
```python
import ai
async with ai.experimental_telemetry.span("retrieval") as sp:
sp.set_attrs(query=query)
docs = await search(query)
sp.set_attrs(count=len(docs))
```
Attach attributes with `sp.set_attrs(...)`. Attribute names that are not
valid Python keywords, such as the dotted names many trace viewers use, go
in a positional mapping:
```python
sp.set_attrs({"output.value": title}, model="claude-haiku-4.5")
```
Record a milestone inside a span with `sp.add_event(name)` — it becomes a
timestamped event on the span, like the `first_token` event the SDK records on
model calls.
If the block raises, the span records the error and re-raises. You keep your
`try`/`except` logic; the trace shows where the failure happened.
## Write an adapter
You can connect a vendor SDK directly by writing an adapter. The `@adapter`
decorator builds one from a single function using a generator trick similar
to pytest fixtures or FastAPI lifecycles.
```python
import ai
@ai.experimental_telemetry.adapter
async def vendor(span):
with sdk.start_span(span.name) as v: # create a span before the loop
while (ev := (yield)) is not None: # process events coming from span.add_event()
v.log_event(ev.name, timestamp=ev.time_ns)
if span.error is not None: # finish the span after the loop
v.set_error(span.error.message)
v.update(output=span.data.model_dump(mode="json"))
ai.experimental_telemetry.register(vendor) # register custom adapter
```
The function runs once per span:
* Code before the loop runs when the span starts.
* Each span event submitted via `.add_event()` resumes the `yield` with a `SpanEvent`,
as it happens.
* Span end resumes the loop with `None`; the code after the loop runs with
`span.data` fully populated and the end timestamp set.
Return before the first `yield` to skip a span.
`@adapter` can also decorate a class. For a class, the async generator trick should
be put in `__call__`.
Telemetry never interferes with the run: an adapter that raises is logged and
skipped. You can register several adapters at once, and they operate
independently.
## Next steps
The module also has a low-level API, that can be useful for instrumenting complicated
cases such as durable execution. It provides more control over adapter functionality and
spans' lifecycles. See the
[`ai.experimental_telemetry` reference](/docs/reference/telemetry) for the
full API, the span data types, and the durable execution patterns.
---
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)
---
title: Testing
description: Test model and agent behavior with scripted conversations.
type: guide
summary: Use FakeModel to run deterministic tests without calling a provider.
---
# Testing
Use `ai.testing.FakeModel` to test model and agent behavior without making
network requests. A fake model replays assistant messages from a scripted
conversation and checks that the application sends the expected history.
Install your test runner and async plugin if your project does not already use
them:
```bash title="Terminal"
uv add --dev pytest pytest-asyncio
```
## Test an agent with a tool
Build a script from normal SDK messages. Use `ai.testing.tool_call` to create a
tool call with validated arguments and a unique call ID:
```python title="test_weather_agent.py"
import ai
import pytest
@ai.tool
async def get_weather(city: str) -> str:
"""Get the weather for a city."""
return "Sunny"
@pytest.mark.asyncio
async def test_weather_agent() -> None:
tool_call = ai.testing.tool_call(get_weather, city="San Francisco")
model = ai.testing.FakeModel(
[
ai.user_message("What is the weather in San Francisco?"),
ai.assistant_message("I will check.", tool_call),
ai.assistant_message("It is sunny in San Francisco."),
]
)
agent = ai.Agent(tools=[get_weather])
async with agent.run(
model,
[ai.user_message("What is the weather in San Francisco?")],
) as stream:
async for _event in stream:
pass
assert stream.output == "It is sunny in San Francisco."
assert len(model.calls) == 2
assert not model.unused
```
`model.calls` records each model invocation.
You can leave tool messages out of a script when the exact result does not
matter. The fake model accepts the agent's tool message and continues with the
next scripted assistant message.
## Assert an exact tool result
Include a tool message when the test must verify the result sent back to the
model:
```python
tool_call = ai.testing.tool_call(get_weather, city="San Francisco")
model = ai.testing.FakeModel(
[
ai.user_message("Check the weather."),
ai.assistant_message(tool_call),
ai.tool_message(
tool_call_id=tool_call.tool_call_id,
tool_name="get_weather",
result={"temp_f": 64, "conditions": "sunny"},
),
ai.assistant_message("It is sunny."),
]
)
```
`result` accepts any JSON-serializable value or Pydantic model.
The test fails with an `AssertionError` when the actual conversation diverges
from every script.
## Inspect model calls
`model.calls` contains the exact input messages from each model call, in call
order. Use it to check what an agent sent after tools ran:
```python
assert [message.role for message in model.calls[1]] == [
"user",
"assistant",
"tool",
]
```
`model.unused` contains scripted assistant messages that never played. Assert
that it is empty when the test should exercise the complete script.
Unscripted system messages are ignored, so an agent's system prompt does not
need to appear in every script.
## Test structured output
`FakeModel` replays scripted messages as-is, so structured output works like
it does with a real model: script an assistant message whose text is the JSON
payload.
```python
import pydantic
class Weather(pydantic.BaseModel):
city: str
conditions: str
model = ai.testing.FakeModel(
[
ai.user_message("Weather in San Francisco as JSON."),
ai.assistant_message('{"city": "San Francisco", "conditions": "sunny"}'),
]
)
message = await ai.experimental_generate(
model,
[ai.user_message("Weather in San Francisco as JSON.")],
output_type=Weather,
)
assert message.get_output(Weather).conditions == "sunny"
```
The fake does not validate scripted text against the schema; parsing happens only
when the test calls `get_output`.
---
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)
---
title: Tools
description: Define and use tools in model and agent workflows.
type: guide
summary: Define function tools, design schemas, handle tool errors, use schema-only tools, provider tools, and MCP tools.
---
# Tools
AI SDK for Python supports multiple types of tools, including function tools
defined in code using `@ai.tool`, as well as built-in provider-side tools.
## Declare function tools
Decorate an async function with `@ai.tool`:
```python
import ai
@ai.tool
async def contact_mothership(query: str) -> str:
"""Contact the mothership for important decisions."""
return "Soon."
```
The tool name comes from the function name. The model receives the function
parameters as a JSON schema and the docstring as the tool description.
Function tools will only be automatically executed by the SDK when used
in the context of the agent. Provider-side tools will always get executed
by the provider, even when passed to `ai.stream`.
## Handle tool errors
Tool exceptions become `ToolCallResult` events with `is_error=True`. The model
sees the error text on the next turn:
```python
async with agent.run(model, messages) as stream:
async for event in stream:
if isinstance(event, ai.events.ToolCallResult):
for result in event.results:
if result.is_error:
print(f"{result.tool_name} failed: {result.result}")
```
The original exception is available on `event.exception` for logging.
## Declare tools that stream output
AI SDK for Python supports tools that return async iterables. Use a streaming tool
when it needs to return partial output, such as when wrapping a subagent.
Streaming tools use aggregators. An aggregator solves two problems: the tool can
yield many values over time, but the agent still needs one final tool result;
and your app may want a rich stored result while the model needs a simpler value
on the next turn.
The core interface is:
```python
class Aggregator[Item, Result, ModelInput]:
def feed(self, item: Item) -> None: ...
def snapshot(self) -> Result: ...
@classmethod
def to_model_input(cls, snapshot: Result) -> ModelInput: ...
```
Use `ai.StreamingTextTool` when yielded strings should concatenate into the
tool result:
```python
@ai.tool
async def draft_reply(topic: str) -> ai.StreamingTextTool:
"""Draft a reply."""
yield "Checking "
yield "records for "
yield topic
```
Use `ai.StreamingStatusTool[T]` when intermediate yields are progress updates
and the last yielded value is the final result.
Use `ai.SubAgentTool` when a tool streams events from a nested agent. The stored
result is a message bundle, and the model sees the final assistant text.
The agent emits `PartialToolCallResult` events for those values, then sends
the aggregated result back to the model on the next turn.
Every tool must return an awaitable or an async iterable. Every async iterable
tool needs an aggregator, either through its return annotation or the
`aggregator=` argument.
## Understand tool anatomy
`ai.Tool` represents tools of all kinds in the SDK.
Each tool carries an optional `spec`, which is a description consumed by the
model, as well as a `tool_config` that contains tool-specific execution config
(e.g. retry policy).
`AgentTool` wraps `ai.Tool` together with its corresponding Python function,
which allows `agent.run` to find and call that function when the model requests
to do so.
Pass `ai.Tool` objects directly to `ai.stream` when you want the model to emit
tool calls but you do not want the SDK to execute them:
```python
tool = ai.Tool(
kind="function",
name="contact_mothership",
spec=ai.tools.ToolSpec(
description="Contact the mothership.",
params={
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
),
)
async with ai.stream(model, messages, tools=[tool]) as stream:
async for event in stream:
if isinstance(event, ai.events.ToolEnd):
print(event.tool_call.tool_args)
```
## Use provider-executed tools
Provider-executed tools run on the provider side. Pass them to `ai.stream` in
the `tools` list:
```python
messages = [
ai.user_message("Check the latest mothership telemetry reports."),
]
async with ai.stream(
model,
messages,
tools=[ai.providers.anthropic.tools.web_search(max_uses=3)],
) as stream:
async for event in stream:
if isinstance(event, ai.events.TextDelta):
print(event.chunk, end="", flush=True)
```
When you route through AI Gateway, you can use provider-specific tool factories
and AI Gateway tool factories:
```python
tools = [
ai.providers.anthropic.tools.web_search(max_uses=3),
ai.providers.ai_gateway.tools.perplexity_search(max_results=5),
]
```
## Use MCP tools
The Model Context Protocol (MCP) adapter converts server tools into agent tools:
```bash
uv add "ai[mcp]"
```
```python
tools = await ai.mcp.get_http_tools(
"http://localhost:3000/mcp",
headers={"Authorization": "Bearer your_access_token_here"},
)
agent = ai.Agent(tools=tools)
```
Use `ai.mcp.get_stdio_tools` for subprocess-based MCP servers.
---
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)
---
title: ai.errors
description: Reference for SDK error types.
type: reference
summary: Reference for ai.errors.
---
# ai.errors
`ai.errors` contains the framework and provider error hierarchy.
## Base Errors
Base errors are not tied to a provider HTTP response.
* `AIError`
* `ConfigurationError`
* `InstallationError`
* `UnsupportedProviderError`
## Provider Errors
Provider errors represent failures raised by model providers.
* `ProviderError`
* `ProviderNotConfiguredError`
* `ProviderAPIError`
* `ProviderConnectionError`
* `ProviderTimeoutError`
* `ProviderResponseError`
* `ProviderIncompleteResponseError`
`ProviderAPIError` includes `request_id`, `http_context`, `body`, `code`,
`param`, `type`, and `is_retryable`.
## Provider Status Errors
Status errors represent provider responses with non-success HTTP status codes.
* `ProviderStatusError`
* `ProviderBadRequestError`
* `ProviderAuthenticationError`
* `ProviderPermissionDeniedError`
* `ProviderNotFoundError`
* `ProviderModelNotFoundError`
* `ProviderConflictError`
* `ProviderRequestTooLargeError`
* `ProviderUnprocessableEntityError`
* `ProviderRateLimitError`
* `ProviderInternalServerError`
* `ProviderServiceUnavailableError`
* `ProviderDeadlineExceededError`
* `ProviderOverloadedError`
## HTTPErrorContext
`HTTPErrorContext` exposes normalized HTTP context from provider errors.
```python
context.status_code
context.request
context.response
```
## http\_status\_to\_provider\_status\_error\_class
`http_status_to_provider_status_error_class` returns the provider status error
class for an HTTP status code.
```python
error_cls = ai.errors.http_status_to_provider_status_error_class(429)
```
---
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)
---
title: ai.events
description: Reference for event classes and event helpers.
type: reference
summary: Reference for ai.events.
---
# ai.events
Streams and agents yield event objects from `ai.events`. This is the public
event-model namespace, and events are Pydantic models.
## Stream Events
Model stream events inherit from `ModelEvent`.
```python
event.message
event.usage
event.provider_metadata
```
Text events:
* `StreamStart`
* `TextStart`
* `TextDelta`
* `TextEnd`
* `StreamEnd`
`StreamEnd` carries the final response metadata. New in 0.4.0:
* `finish_reason`: Why the model stopped: `stop`, `length`, `content_filter`,
`tool_call`, `error`, or `other`. Raw provider value is kept in `provider_metadata`.
* `response_id`: The provider's id for the response.
* `response_model`: The model that produced the response. It can differ from
the requested model under gateway routing or fallbacks.
Reasoning events:
* `ReasoningStart`
* `ReasoningDelta`
* `ReasoningEnd`
Tool events:
* `ToolStart`
* `ToolDelta`
* `ToolEnd`
* `BuiltinToolStart`
* `BuiltinToolDelta`
* `BuiltinToolEnd`
* `BuiltinToolResult`
File events:
* `FileEvent`
Stream event union:
* `Event`: union of all model stream events above.
## Agent Events
Agents can emit event values that are not raw provider stream events.
* `ToolCallResult`: Carries the tool result message and result parts.
* `PartialToolCallResult`: Carries values yielded by streaming tools and
`yield_from`.
* `HookEvent`: Carries an internal hook message and `HookPart`.
* `RunBlocked`: Emitted when the run is blocked on deferred hooks.
Agent event union:
* `AgentEvent`: everything an agent run can yield: `Event` plus
`ToolCallResult`, `HookEvent`, `PartialToolCallResult`, and `RunBlocked`.
The union is annotated with a Pydantic discriminator on the `kind` field,
so use it as a field or `TypeAdapter` type when deserializing events and
Pydantic picks the right event class from `kind`.
## Aggregator
`Aggregator[Item, Result, ModelInput]` is the interface used by streaming tools
and `yield_from`.
```python
class MyAggregator(ai.events.Aggregator[Item, Result, ModelInput]):
def feed(self, item: Item) -> None: ...
def snapshot(self) -> Result: ...
@classmethod
def to_model_input(cls, snapshot: Result) -> ModelInput: ...
```
`snapshot()` is the rich value stored in tool results. `get_model_input()` is
the value sent back to the model.
---
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)
---
title: Reference
description: Public API reference for the AI SDK for Python.
type: reference
summary: Look up public APIs by the names you import and call.
---
# Reference
Reference pages are organized by public module.
Start with `ai` for names imported directly from the top-level package, such
as `ai.stream`, `ai.Agent`, and `ai.Tool`. Use sibling module sections such as
`ai.events`, `ai.messages`, `ai.ops`, `ai.providers`, `ai.testing`, and
`ai.agents` for public submodule APIs.
---
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)
---
title: ai.mcp
description: Reference for MCP tool loading APIs.
type: reference
summary: Reference for ai.mcp.
---
# ai.mcp
`ai.mcp` loads MCP server tools as `AgentTool` values.
Install the MCP extra before using these helpers:
```bash
uv add "ai[mcp]"
```
## get\_stdio\_tools
`get_stdio_tools` starts an MCP server subprocess and returns `AgentTool`
values that can be passed to `ai.Agent`.
```python
tools = await ai.mcp.get_stdio_tools(
"npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp"
)
```
Arguments:
* `command`: Subprocess command.
* `*args`: Subprocess arguments.
* `env`: Optional environment variables.
* `cwd`: Optional working directory.
* `tool_prefix`: Optional prefix for tool names.
## get\_http\_tools
`get_http_tools` connects to an MCP HTTP server and returns `AgentTool` values
that can be passed to `ai.Agent`.
```python
tools = await ai.mcp.get_http_tools("https://example.com/mcp")
```
Arguments:
* `url`: MCP server endpoint.
* `headers`: Optional HTTP headers.
* `tool_prefix`: Optional prefix for tool names.
Use `tool_prefix` when multiple servers expose overlapping tool names.
## close\_connections
`close_connections` closes pooled MCP connections for the active context.
Agent runs clean up MCP connections automatically. Call this helper only when
you manage MCP tool lifetimes manually.
---
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)
---
title: ai.messages
description: Reference for message models and message parts.
type: reference
summary: Reference for ai.messages.
---
# ai.messages
`ai.messages` is the public message-model namespace. Messages are Pydantic
models and durable state shared by model calls, agents, tools, UI adapters,
and resume flows.
## Message
`Message` carries one conversation item.
```python
message.role
message.parts
message.id
message.turn_id
message.usage
message.provider_metadata
message.replay
```
Roles are `user`, `assistant`, `system`, `tool`, and `internal`.
Convenience properties:
```python
message.text
message.reasoning
message.tool_calls
message.tool_results
message.builtin_tool_calls
message.builtin_tool_returns
message.files
message.images
message.videos
message.audio
message.get_output(output_type=None)
```
`get_output()` returns text by default. With a Pydantic model, it validates the
message text as JSON and returns that model.
## Parts
Message parts store typed content inside a `Message`.
Common parts:
* `TextPart`: Text content.
* `FilePart`: File, image, document, audio, or video content.
* `ReasoningPart`: Model reasoning text.
Tool and runtime parts:
* `ToolCallPart`: A host-executed tool call requested by the model.
* `ToolResultPart`: The result for a host-executed tool call.
* `BuiltinToolCallPart`: A provider-executed tool call.
* `BuiltinToolReturnPart`: A provider-executed tool result.
* `HookPart`: A hook suspension, resolution, or cancellation.
`ToolResultPart.get_model_input()` returns the value sent back to the model. For
most tools this is the same as `result`. Aggregator-backed tools can store a
rich `result` while sending a simpler model-facing value.
Fields and helpers for model-facing values:
* `model_input`: Explicit model-facing value. Normal tools omit it from
serialized data.
* `model_input_kind`: Shape of `model_input`: `json` or `special`.
* `get_model_input()`: Return `model_input`, or fall back to `result` when no
explicit value is stored.
* `has_model_input`: Whether an explicit model-facing value is stored.
* `set_model_input(value)`: Store a model-facing value and set its kind.
## Message Bundles
Message bundle types are used when a tool or adapter needs to carry multiple
message-layer values as one result.
* `MessageBundle`: A tuple of `Message` values.
* `ContentOutput`: A multipart tool result containing text and file parts.
* `ContentPart`: The text-or-file part union used by `ContentOutput`.
* `SpecialToolResult`: The special result union for multipart content and
message bundles.
* `ResultKind`: The coarse tag for a tool result shape: `error`, `json`, or
`special`.
* `ModelInputKind`: The coarse tag for a model-facing value: `json` or
`special`.
## Message IDs
`generate_id` creates SDK-style IDs for messages and parts.
```python
message_id = ai.messages.generate_id("msg")
```
Pass a prefix to control the ID family. If no prefix is supplied, the helper
returns an unprefixed generated ID.
`use_random` overrides the random source used for message and part IDs within
an async context. Use it with a replay-safe random source inside durable
workflows:
```python
@workflow.workflow
@ai.messages.use_random(workflow.random)
async def run_turn(...):
...
```
## Serialization
Messages serialize through Pydantic.
```python
encoded = [message.model_dump(mode="json") for message in messages]
restored = [ai.messages.Message.model_validate(item) for item in encoded]
```
Persist messages after completed or suspended runs. Recreate providers, hooks,
streams, and other live runtime objects on the next request.
---
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)
---
title: ai.ops
description: Reference for dedicated model operations.
type: reference
summary: Reference for image, video, speech, embedding, transcription, and reranking APIs.
---
# ai.ops
`ai.ops` contains model operations that return standalone values instead of
language-model messages.
## Item
Every operation returns `Item[T]`.
Fields:
* `value`: The operation result.
* `usage`: Normalized token usage, or `None` when the provider does not report
it.
* `warnings`: `Warning` values reported during the operation. Empty when there
are none.
* `provider_metadata`: Provider-specific response data, or `None`.
## Warning
A warning reported while running an operation, such as an unsupported or
partially applied feature.
Fields:
* `kind`: Warning category: `"unsupported"`, `"compatibility"`,
`"deprecated"`, or `"other"`.
* `message`: Human-readable description, or `None`.
* `feature`: The affected feature, for unsupported and compatibility warnings.
* `setting`: The deprecated setting name, for deprecated warnings.
* `details`: Additional detail, or `None`.
## generate\_image
```python
await ai.ops.generate_image(
model,
prompt,
*,
params=None,
)
```
Arguments:
* `model`: Image model.
* `prompt`: Text prompt as a string, or an `ImagePrompt`.
* `params`: Optional `ImageParams`.
Returns `Item[list[FilePart]]`.
`ImagePrompt` fields:
* `text`: Text prompt. Optional for operations that do not need one, such as
upscaling.
* `images`: Input images for editing or variation generation. Each accepts a
`FilePart`, raw bytes, or a URL or base64 string.
* `mask`: Mask image for inpainting operations.
`ImageParams` fields:
* `n`: Number of images. The default is `1`.
* `size`: Image size such as `"1024x1024"`.
* `aspect_ratio`: Aspect ratio such as `"16:9"`.
* `seed`: Reproducibility seed.
* `provider_options`: Provider-specific options keyed by provider name.
## generate\_video
```python
await ai.ops.generate_video(
model,
prompt,
*,
params=None,
)
```
Arguments:
* `model`: Video model.
* `prompt`: Text prompt as a string, or a `VideoPrompt`.
* `params`: Optional `VideoParams`.
Returns `Item[list[FilePart]]`.
`VideoPrompt` fields:
* `text`: Text prompt.
* `image`: Input image for image-to-video generation, used as the starting
frame. Accepts a `FilePart`, raw bytes, or a URL or base64 string.
* `frame_images`: Role-tagged `FrameImage` values. A `first_frame` entry takes
precedence over `image` as the start image.
* `references`: Reference images or videos for reference-to-video generation.
Cannot be combined with `frame_images`.
`FrameImage` fields:
* `image`: The image as a `FilePart`, raw bytes, or a URL or base64 string.
* `frame_type`: `"first_frame"` to animate from the image, or `"last_frame"`
to animate towards it.
`VideoParams` fields:
* `n`: Number of videos. The default is `1`.
* `aspect_ratio`: Aspect ratio such as `"16:9"`.
* `resolution`: Resolution such as `"1920x1080"`.
* `duration`: Duration in seconds.
* `fps`: Frames per second.
* `seed`: Reproducibility seed.
* `generate_audio`: Whether the model should generate audio alongside the
video.
* `provider_options`: Provider-specific options keyed by provider name.
## generate\_audio
```python
await ai.ops.generate_audio(
model,
prompt,
*,
params=None,
)
```
Arguments:
* `model`: Speech model.
* `prompt`: The text to speak as a string, or an `AudioPrompt`.
* `params`: Optional `AudioParams`.
Returns `Item[list[FilePart]]`.
`AudioPrompt` fields:
* `text`: The text to convert to speech.
* `instructions`: Instructions for tone, emotion, or delivery.
`AudioParams` fields:
* `voice`: Provider voice ID or name.
* `output_format`: Audio format such as `"mp3"` or `"wav"`.
* `speed`: Speech speed multiplier.
* `language`: ISO 639-1 language code.
* `provider_options`: Provider-specific options keyed by provider name.
## embed
```python
await ai.ops.embed(
model,
values,
*,
params=None,
)
```
Arguments:
* `model`: Embedding model.
* `values`: Text strings to embed.
* `params`: Optional `EmbedParams`.
Returns `Item[list[list[float]]]` with one vector per input string, in input
order.
`EmbedParams` contains `provider_options`, keyed by provider name.
## transcribe
```python
await ai.ops.transcribe(
model,
audio,
*,
params=None,
)
```
Arguments:
* `model`: Transcription model.
* `audio`: Audio as a `FilePart` or raw bytes.
* `params`: Optional `TranscribeParams`.
Returns `Item[Transcription]`.
`TranscribeParams` contains `provider_options`, keyed by provider name.
`Transcription` fields:
* `text`: Complete transcript.
* `segments`: Timed `TranscriptionSegment` values when reported.
* `language`: Detected ISO 639-1 language code when reported.
* `duration_seconds`: Total input duration when reported.
`TranscriptionSegment` fields:
* `text`: Segment text.
* `start_second`: Segment start time.
* `end_second`: Segment end time.
## rerank
```python
await ai.ops.rerank(
model,
documents,
query,
*,
params=None,
)
```
Arguments:
* `model`: Reranking model.
* `documents`: Text strings or JSON objects to rank.
* `query`: Query used to score the documents.
* `params`: Optional `RerankParams`.
Returns `Item[list[RankedDocument]]`, ordered by descending relevance score.
Passing an empty document list returns an empty result without calling the
provider.
`RerankParams` fields:
* `top_n`: Maximum number of results. The default returns all documents.
* `provider_options`: Provider-specific options keyed by provider name.
`RankedDocument` fields:
* `index`: Position of the document in the original input list.
* `score`: Relevance score for the query.
---
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)
---
title: ai.experimental_telemetry
description: Reference for spans, adapters, sinks, and the OpenTelemetry exporter.
type: reference
summary: Reference for ai.experimental_telemetry.
---
# ai.experimental_telemetry
The module is experimental and new in 0.4.0: it is not part of the stable API
and may change or be removed.
`ai.experimental_telemetry` contains instrumentation for tracing agent runs, model calls,
tool executions, and other work in a provider-agnostic way.
The high-level API consists of the `span()` context manager for creating custom spans,
and an `@adapter` decorator for making adapters. It can be used to augment the built-in
instrumentation.
The underlying low-level API enables manual control over spans' lifecycles, as well as
when and how they reach observability providers. It can be used in environments the
context manager cannot serve; chiefly durable execution, where parts of the app must be
free of side effects, and things are being passed around as JSON.
## Spans
A span is a record of one unit of work. Spans form a tree: every span carries
a `trace_id` shared by the whole tree and a `parent_id` pointing at the span
it ran under.
It is a Pydantic model, generic in its data type: `Span[SpanData]`.
| Field | Description |
| ---------------- | ------------------------------------------------------------------------------------------------------------------------------------ |
| `name` | Span name. For typed spans, defaults to the data's `kind`. |
| `data` | The typed payload describing the work. |
| `id` | Span id. Can be empty on a no-op span. |
| `trace_id` | Trace id shared by every span in one tree. |
| `parent_id` | Id of the parent span, or `None` for a root. |
| `started_at` | Start time in nanoseconds since the epoch, or `None` if not started. |
| `ended_at` | End time in nanoseconds, or `None` while in flight. |
| `error` | The `SpanError` that ended the span, or `None`. |
| `replay` | `True` when the work is being [replayed](/docs/reference/ai/stream#replay) rather than performed live (resume, serverless re-entry). |
| `set_as_current` | Whether the span sets itself as current. Adapters should respect it to keep nesting consistent. |
| `events` | List of `SpanEvent` milestones. |
| `trace_attrs` | User data propagated down the trace: a child gets a copy of its parent's `trace_attrs` at creation. |
The lifecycle is encoded in the timestamps:
1. `started_at=None` means the span has not started;
2. `ended_at is not None` means it is complete.
The adapter registry uses this convention to decide which adapter method should handle
a span in its current state: `on_span_start`, `on_span_event`, or `on_span_end`.
A span can be serialized using `model_dump` / `model_validate` like any Pydantic model.
Plain `Span.model_validate(...)` restores the framework data types by their `kind`
discriminator; custom data types need `Span[MyData].model_validate(...)`.
### Span data
Every span carries a `data` field that describes the work. The type of
`data` tells you what kind of span it is.
`SpanData` is the protocol for span data: anything with a string `kind` property.
Implement it with a Pydantic model to define your own typed spans:
```python
import pydantic
from typing import Literal
class RetrievalSpanData(pydantic.BaseModel):
kind: Literal["retrieval"] = "retrieval"
query: str
count: int | None = None
async with ai.experimental_telemetry.span(RetrievalSpanData(query=q)) as sp:
docs = await search(q)
sp.data.count = len(docs) # typed
```
A typed span is `Span[RetrievalSpanData]`, so assignments to `sp.data` fields are type
checked. To correctly restore a span with a custom data type, use
`Span[RetrievalSpanData].model_validate(...)`.
Built-in data types:
| Type | Kind | Describes |
| ----------------------- | ---------------- | --------------------------------------------------------------------- |
| `RunSpanData` | `run` | One `Agent.run`: the whole loop. |
| `LoopTurnSpanData` | `loop_turn` | One turn of the default agent loop. |
| `AiStreamSpanData` | `ai_stream` | One streaming model call. |
| `AiGenerateSpanData` | `ai_generate` | One buffered language-model call. |
| `EmbedSpanData` | `embed` | One embedding operation. |
| `GenerateAudioSpanData` | `generate_audio` | One audio generation operation. |
| `GenerateImageSpanData` | `generate_image` | One image generation operation. |
| `GenerateVideoSpanData` | `generate_video` | One video generation operation. |
| `RerankSpanData` | `rerank` | One reranking operation. |
| `TranscribeSpanData` | `transcribe` | One transcription operation. |
| `ToolExecutionSpanData` | `tool_execution` | One tool execution, from dispatch to result. |
| `HookSpanData` | `hook` | One hook suspension, from deferred until resolved or cancelled. |
| `CustomSpanData` | `custom` | A user span made with `span("name")`; its attributes live in `attrs`. |
`RunSpanData` carries the agent name, model, input messages, tool names, and
parameters; `blocked`, `final_message`, and `usage` are set at span end.
`blocked` is `True` when the run ended suspended on an unresolved hook.
`AiStreamSpanData` carries the model, messages, parameters, and tool names of
one call; the response `message`, `usage`, `finish_reason`, `response_id`,
and `response_model` are set at span end.
`AiGenerateSpanData` carries the same request and response fields for
`ai.experimental_generate` calls.
Each `ai.ops` function has it's own span data. All carry `model` and
`provider`, as well as modality-specific fields.
`ToolExecutionSpanData` carries `tool_name`, `tool_call_id`,
`tool_description`, and `args`; `result`, `model_input`, and `is_error` are
set at span end. `model_input` is set only when the value the model sees
differs from `result`.
`LoopTurnSpanData` has no fields, it exists so adapters can group a turn's
model and tool spans. Turn order is given by `started_at` and `parent_id`.
### Events and errors
`SpanEvent` is a named, timestamped milestone inside a span's lifetime,
stored in `span.events`:
| Field | Description |
| --------- | ------------------------------------------------ |
| `name` | Event name, such as `first_token`. |
| `time_ns` | Timestamp in nanoseconds from the ambient clock. |
| `attrs` | Event attributes. |
The SDK records its own events with shared name constants: `FIRST_TOKEN`,
`RESPONSE_COMPLETE`, `HOOK_DEFERRED`, `HOOK_RESOLVED`, and `HOOK_CANCELLED`.
Events on one span are delivered in list order; there is no ordering
guarantee across spans.
`SpanError` is a serializable record of the failure that ended a span:
| Field | Description |
| --------- | -------------------------------------------- |
| `type` | Exception type name, such as `TimeoutError`. |
| `message` | The error message. |
Spans cross process boundaries, so the error can't be a live exception.
`SpanError.from_exception(exc)` builds one from an exception.
### High-level span API
`span(name_or_data, /, *, parent=None, replay=False, set_as_current=True)`
opens a span as an async context manager. It stamps the start time and
pushes on enter, and stamps the end time (and the error, if the block
raised) and pushes on exit:
```python
import ai
async with ai.experimental_telemetry.span("retrieval") as sp:
sp.set_attrs(query=query)
docs = await search(query)
sp.set_attrs(count=len(docs))
```
Inside the block the span is the *current span*: spans opened within it,
including the spans the SDK opens, become its children. Nesting requires no
explicit parent references.
Parameters:
| Parameter | Description |
| ---------------- | -------------------------------------------------------------------------------------------------------------------------------------------- |
| `name_or_data` | A span name for a user span, or a `SpanData` instance for a typed span. |
| `parent` | Overrides the ambient parent: a live `Span`, or one restored from another process. The default parents under the current span. |
| `replay` | Marks work that is being replayed (resume, serverless re-entry). Used primarily by the framework's internals. |
| `set_as_current` | When `False`, the span does not become current; work done while it is open parents to *its* parent instead. Also meant mostly for internals. |
Exceptions raised in the block are recorded on `span.error` and re-raised.
Methods that are handy for custom instrumentation:
* `Span.set_attrs(attrs=None, /, **kwargs)` merges attributes into a user
span:
```python
sp.set_attrs({"output.value": title}, model="claude-haiku-4.5")
```
`set_attrs()` only works on spans created with a string name (their data is
`CustomSpanData`). Framework spans carry typed data, assign their fields
directly.
* `Span.add_event(name, attrs=None, /, **kwargs)` appends a named
milestone and returns it. The next push delivers it; the `span()` context
manager pushes on exit:
```python
sp.add_event("cache_miss", key=key)
```
* `current_span()` returns the current span, or `None` when no span is open.
### Low-level span API
Using low-level API boils down to manually creating and mutating `Span` objects (that
are Pydantic models, as you recall), and manually pushing their current state to the
sink with `Span.push()`.
The framework provides utilities to make this less verbose, as well as to make those
manual spans play nice with the built-in instrumentation.
1. `create_span()` mints identity and takes the trace id, parentage, and a copy of
`trace_attrs` from `parent` (default: the current span; a fresh trace when there
is none). Nothing is reported.
2. `stamp_start()` and `stamp_end()` write timestamps from the ambient clock.
`stamp_end(error=...)` also records an error when given one, converting an
exception to a `SpanError`. Still nothing is reported.
3. `use_span(span)` context manager makes the span current, so other spans can nest under
it automatically. Also no reporting.
4. `push()` delivers a snapshot to the current sink. This is the only reporting step.
```python
sp = ai.experimental_telemetry.create_span("turn")
sp.set_attrs(session=session_id)
sp.stamp_start()
await sp.push() # visible to adapters
... # possibly elsewhere, later:
await sp.stamp_end().push() # complete
```
You can push a span more than once; each push snapshots the whole span, and
the last push with `ended_at` set is the complete record.
`use_span(None)` is a no-op, which keeps call sites free of conditionals.
## Sinks
A `Sink` is what receives pushed span snapshots.
Sinks are considered a low-level API. Manually changing the current sink allows you to
route spans *away* from adapters and into whatever container that sink is backed by.
* The default sink is the adapter registry. It dispatches every span to an appropriate
method on every registered adapter based on span's state (see
[the low-level adapter API](#low-level-adapter-api)).
* A `DictSink` sink stores spans in a dict.
`Sink` is a protocol with one method:
```python
class Sink(Protocol):
async def on_push(self, span: Span, /) -> None: ...
```
`use_sink(sink)` context manager routes span pushes to `sink` within a context.
`push_all(spans)` can be used to send a list of spans to the current sink.
```python
sink = ai.experimental_telemetry.DictSink()
async with ai.experimental_telemetry.use_sink(sink):
... # spans go to the sink
payload = [s.model_dump(mode="json") for s in sink.finished_spans]
# elsewhere, with the default sink on:
await ai.experimental_telemetry.push_all(payload)
```
* `DictSink.spans` exposes the dict of span id to span.
* `DictSink.finished_spans` returns the collected spans that have ended.
## Adapters
Adapters are meant to convert framework's spans into vendor spans.
* `register(adapter)` adds an adapter. Multiple adapters coexist
independently.
* `unregister(adapter)` removes a previously registered adapter.
Custom adapters can be built using a high-level or a low-level API.
By high-level API we mean the `@adapter` decorator that converts a function or a class
into an adapter, using a generator trick similar to pytest fixtures or FastAPI
lifecycles.
Using the low-level API means implementing `AdapterProtocol` from scratch,
avoiding the generator trickery completely.
Adapter that raises is logged and skipped: telemetry never kills the run.
### High-level adapter API
`@adapter` builds an adapter from one async generator function.
```python
@ai.experimental_telemetry.adapter
async def vendor(span):
with sdk.start_span(span.name) as v: # span start
while (ev := (yield)) is not None: # each event, live
v.log_event(ev.name, timestamp=ev.time_ns)
if span.error is not None: # span end
v.set_error(span.error.message)
v.update(output=span.data.model_dump(mode="json"))
ai.experimental_telemetry.register(vendor)
```
* Code before the loop runs at span start. Each span event resumes the
`yield` with the `SpanEvent`, live. Span end resumes it with `None`, and
the code after the loop runs with `span.data` fully populated and
`ended_at` set.
* A failed span ends the loop normally; read `span.error` after the loop to
report it.
* A span that arrives already complete is replayed to the generator as start,
events, end, back to back.
* Return before the first `yield` to skip a span. Returning mid-span, from
inside the loop, opts out of the rest of that span, including its end.
* Loop until the `yield` returns `None`.
When the adapter needs configuration or state, decorate a class instead.
`__call__` must be an async generator method with the same yield loop:
```python
@ai.experimental_telemetry.adapter
class Vendor:
def __init__(self, *, api_key):
self._client = sdk.Client(api_key)
async def __call__(self, span):
with self._client.start_span(span.name) as v:
while (ev := (yield)) is not None:
v.log_event(ev.name)
ai.experimental_telemetry.register(Vendor(api_key="..."))
```
Decorating a class mixes the adapter machinery (`AdapterMixin`) into it.
The built-in `OtelAdapter` is built this way.
### Low-level adapter API
Under the hood, an adapter is any object satisfying `AdapterProtocol`:
```python
class AdapterProtocol(Protocol):
async def on_span_start(self, span: Span, /) -> None: ...
async def on_span_event(self, span: Span, event: SpanEvent, /) -> None: ...
async def on_span_end(self, span: Span, /) -> None: ...
```
The default sink calls each of these methods for every pushed span based on the following
convention:
* `on_span_start` is called when a span is pushed for the first time with `started_at` stamped
* `on_span_event` is called when a span is pushed with new events; called once per event
* `on_span_end` is called when a span is pushed with `ended_at` set
* After the end, the span id is forgotten: pushing a completed span again re-delivers it in full.
Implement the protocol directly when the callback shape fits your target better than a generator.
## The OpenTelemetry adapter
`ai.experimental_telemetry.otel` maps framework spans onto OpenTelemetry
spans, following the `gen_ai` semantic conventions. It requires the `otel`
extra:
```bash title="Terminal"
uv add "ai[otel]"
```
```python
import ai
from ai.experimental_telemetry import otel
ai.experimental_telemetry.register(otel.OtelAdapter())
```
`OtelAdapter` uses the global tracer provider unless you pass one.
By default, the adapter sends traces to
`http://localhost:4318/v1/traces`, the OTLP/HTTP default.
### Semantic conventions
The adapter names spans and sets attributes per the `gen_ai` conventions:
| Span data | Span name | Operation |
| ------------------------------------------------------------------------- | -------------------------- | ------------------ |
| `RunSpanData` | `invoke_agent {agent}` | `invoke_agent` |
| `AiStreamSpanData` | `chat {model}` | `chat` |
| `AiGenerateSpanData` | `chat {model}` | `chat` |
| `EmbedSpanData` | `embeddings {model}` | `embeddings` |
| `GenerateAudioSpanData`, `GenerateImageSpanData`, `GenerateVideoSpanData` | `generate_content {model}` | `generate_content` |
| `TranscribeSpanData`, `RerankSpanData` | `{kind} {model}` | The span data kind |
| `ToolExecutionSpanData` | `execute_tool {tool_name}` | `execute_tool` |
| Other | The span's own name | — |
Attributes include `gen_ai.request.*` (model, temperature, max tokens,
reasoning level), `gen_ai.response.*` (finish reasons, response id, time to
first chunk), `gen_ai.usage.*` (input, output, reasoning, and cache tokens),
and `gen_ai.tool.*` (name, call id, description). Failed spans set
`error.type` and an error status. Model call spans export with the `CLIENT`
span kind; agent and tool spans stay `INTERNAL`.
Custom span attributes and `trace_attrs` export as-is; values that are not
scalars are converted with `repr`.
### Capture message content
Prompts, responses, and tool arguments are not exported by default. Opt in
with `OtelAdapter(capture_content=True)` or the standard environment variable:
```bash title="Terminal"
export OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true
```
With capture on, the adapter sets `gen_ai.input.messages`,
`gen_ai.output.messages`, `gen_ai.system_instructions`,
`gen_ai.tool.definitions`, and the tool call arguments and results, all in
the semantic convention message shape.
### OtelAdapter
`OtelAdapter(*, tracer_provider=None, capture_content=None)` is the adapter
itself; register it with `register()`. Subclass it to customize the export:
* `span_name(span)`: Returns the exported span name. Override to rename.
* `span_attrs(span)`: Returns the attributes set at span end. Override
to enrich:
```python
class MyAdapter(otel.OtelAdapter):
def span_attrs(self, span):
return super().span_attrs(span) | {"deployment.env": "prod"}
```
* `flush()`: Flushes the provider's exporters.
* `shutdown()`: Flushes and stops the provider; spans pushed after this are
lost.
## Helpers
The remaining functions support delivery and determinism. Most matter only
for durable or short-lived processes.
### use\_time and now\_ns
Span timestamps come from the ambient clock, `now_ns()` — the wall clock by
default. `use_time(now_ns)` overrides it within a context, so replayed code
produces stable timestamps:
```python
@workflow.workflow
@ai.experimental_telemetry.use_time(vercel.workflow.time_ns)
async def run_turn(turn_input):
...
```
`now_ns()` returns nanoseconds since the epoch from whichever clock is
installed.
### push\_all
`push_all(spans)` pushes each span in order, validating dumped spans first.
Use it to re-deliver spans gathered by a
[`DictSink`](#sinks) from a place where the real adapters are
available.
### is\_enabled
`is_enabled()` reports whether anything is listening: a sink routed with
`use_sink`, or at least one registered adapter.
The framework checks it before creating spans, so telemetry adds no overhead
while it is off: a span created while nothing is listening is a no-op with an
empty `id` — it reads no clock, and `push()` delivers nothing. Use the same
check to skip your own instrumentation work, as in the
[durable execution pattern](#durable-execution).
## Patterns
### Regular tracing
Regular agent tracing only requires you to register an adapter:
```python
import ai
from ai.experimental_telemetry import otel
... # the OpenTelemetry ceremony
ai.experimental_telemetry.register(OtelAdapter())
async def handle(request):
async with ai.experimental_telemetry.span("handle_request") as sp:
sp.set_attrs(request_id=request.id)
async with agent.run(model, request.messages) as run:
async for event in run:
...
# at shutdown:
adapter.shutdown()
```
### Typed spans with a matching adapter
Typed span data and generator-based adapters compose: define a data model
for your own work, and write an adapter that opts into exactly that type.
```python
class RetrievalSpanData(pydantic.BaseModel):
kind: Literal["retrieval"] = "retrieval"
query: str
count: int | None = None
@ai.experimental_telemetry.adapter
async def retrieval_metrics(span):
if not isinstance(span.data, RetrievalSpanData):
return # skip everything else
while (yield) is not None:
pass
metrics.histogram("retrieval.count").observe(span.data.count or 0)
```
The same dispatch works for framework types: match on `AiStreamSpanData` to
observe model calls, `ToolExecutionSpanData` for tools, and so on.
### Durable execution
Tracing durable execution is subject to very specific constraints.
The examples below assume shape and terminology of Vercel Workflows, but the same pattern
would apply to Temporal and other durable execution frameworks.
The workflow body must be deterministic and free of side-effects.
We achieve determinism by replacing `random` and `time` with their deterministic
counterparts via `use_random` and `use_time` context managers. This enables the
framework to keep stamping ids and timestamps inside the workflow body.
```python
@workflow.workflow
@ai.messages.use_random(workflow.random) # span and message ids
@ai.experimental_telemetry.use_time(workflow.time_ns) # timestamps
async def my_workflow(payload):
...
```
Sending traces to a remote backend is, in fact, a side effect, and cannot be done
inside a workflow body. Therefore, we need to gather all the spans, pass them into
a step, and send them all at once from there. Do do this, we utilize the `DictSink`
and the `use_sink` context manager.
```python
@workflow.workflow
async def my_workflow(payload):
sink = ai.experimental_telemetry.DictSink()
async with ai.experimental_telemetry.use_sink(sink):
... # any instrumented work: agent runs, custom spans
await report_spans(
[s.model_dump(mode="json") for s in sink.finished_spans]
)
@workflow.step
async def report_spans(spans_data):
await ai.experimental_telemetry.push_all(spans_data)
```
Finally, we need to be able to serialize and restore spans in order to pass them into
workflows and steps via JSON. That way, all the spans can be nested correctly when the
framework reports them to the vendor.
```python
# where the run begins: mint the span; nothing is reported yet
if ai.experimental_telemetry.is_enabled():
payload["run_span"] = (
ai.experimental_telemetry.create_span("run")
.stamp_start()
.model_dump(mode="json")
)
# in any step or process that continues the work:
run_span = ai.experimental_telemetry.Span.model_validate(payload["run_span"])
async with ai.experimental_telemetry.use_span(run_span):
... # spans opened here parent under run_span
# where the run ends: complete the span and report it
run_span = ai.experimental_telemetry.Span.model_validate(payload["run_span"])
await run_span.stamp_end().push()
```
## Module exports
`ai.experimental_telemetry` exports:
* Spans: `Span`, `SpanData`, `SpanEvent`, `SpanError`.
* Span data types: `RunSpanData`, `LoopTurnSpanData`, `AiStreamSpanData`,
`AiGenerateSpanData`, `ToolExecutionSpanData`, `HookSpanData`,
`CustomSpanData`.
* Event names: `FIRST_TOKEN`, `RESPONSE_COMPLETE`, `HOOK_DEFERRED`,
`HOOK_RESOLVED`, `HOOK_CANCELLED`.
* Opening spans: `span`, `create_span`, `current_span`, `use_span`.
* Adapters: `adapter`, `AdapterProtocol`, `AdapterMixin`, `AdapterCallable`,
`register`, `unregister`, `is_enabled`.
* Sinks and clocks: `Sink`, `use_sink`, `DictSink`, `push_all`, `use_time`,
`now_ns`.
`ai.experimental_telemetry.otel` exports `OtelAdapter`.
---
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)
---
title: ai.testing
description: Reference for scripted model and agent testing.
type: reference
summary: Reference for FakeModel, tool_call, and fingerprint.
---
# ai.testing
`ai.testing` provides a model that replays scripted conversations without
calling a provider.
## FakeModel
```python
model = ai.testing.FakeModel(
script,
another_script,
id="fake-model",
)
```
Each script is a sequence of `ai.messages.Message` values. A model call matches
the script that continues the supplied conversation and returns its next
assistant message.
Matching behavior:
* Unscripted system messages are ignored.
* Tool messages may be omitted. When included, they assert the exact tool
results.
* Every scripted tool call must receive a result before the next assistant
message can play.
* Tool call IDs must be unique across all scripts on one model.
* A completed assistant response can only be followed by new user input. Two
consecutive scripted assistant messages require tool calls on the first.
The model raises `AssertionError` when no script continues the conversation.
The error includes the received message and the closest matching script.
Properties:
* `calls`: Copies of the input messages from every model call, in call order.
* `unused`: Scripted assistant messages that have not played.
`FakeModel` works with both `ai.stream` and `Agent.run`.
## tool\_call
```python
part = ai.testing.tool_call(tool, city="San Francisco")
part = ai.testing.tool_call("get_weather", city="San Francisco")
```
`tool_call(tool, **kwargs)` builds a `ToolCallPart` with a new tool call ID. If
`tool` is an `AgentTool`, the helper validates the arguments against its
signature. A string supplies only the tool name.
## fingerprint
```python
value = ai.testing.fingerprint(message)
```
`fingerprint` returns the normalized JSON string that `FakeModel` uses for
comparison. It omits volatile runtime data such as generated IDs, usage,
provider metadata, cached results, and stored model-input state. Tool results
are sorted by tool call ID so concurrently completed tools compare
deterministically.
---
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)
---
title: ai.tools
description: Reference for schema tool types and approval payloads.
type: reference
summary: Reference for ai.tools.
---
# ai.tools
`ai.tools` defines model-facing tool declarations and tool approval payloads.
Schema-only tools and provider-executed tools are both represented as `ai.Tool`
values. Pass them to `ai.stream(..., tools=[...])` or to an agent when the
provider should receive the declaration.
Provider-specific built-in tool factories live under provider namespaces such
as `ai.providers.openai.tools`, `ai.providers.anthropic.tools`, and
`ai.providers.ai_gateway.tools`.
## Tool
`Tool` is the model-facing declaration used by providers.
Fields:
* `kind`: `function` or `provider`.
* `name`: Tool name exposed to the model.
* `spec`: Function tool schema.
* `tool_config`: Provider-executed tool configuration.
* `require_approval`: Whether the tool call needs approval before execution.
Function tools require `spec`. Provider tools require `tool_config.id` and do
not accept `spec`.
## ToolSpec
`ToolSpec` contains the provider-facing schema for a host-executed function
tool.
Fields:
* `description`
* `params`
## ToolConfig
`ToolConfig` stores provider-facing tool options.
Fields:
* `id`: Canonical provider tool id.
* `args`: Provider wire arguments as plain snake\_case data.
Use it with `ai.Tool` when a provider-executed tool needs provider-specific
configuration.
## ToolApproval
`ToolApproval` is a Pydantic model used with hooks and UI adapters when a tool
call needs external approval.
```python
approval = await ai.hook("approve_tool", payload=ai.tools.ToolApproval)
```
Fields:
* `granted`
* `reason`
---
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)
---
title: ai.types
description: Reference for lower-level public type helper modules.
type: reference
summary: Reference for public type helpers that are not top-level imports.
---
# ai.types
Most message, event, and tool APIs are documented under the root aliases
`ai.messages`, `ai.events`, and `ai.tools`. Use `ai.types` for lower-level
helper modules that do not have a shorter public alias.
## ai.types.media
`ai.types.media` contains URL, data URL, media type inference, and magic-byte
detection helpers.
URL helpers:
* `is_url`
* `is_downloadable_url`
* `split_data_url`
Encoding helpers:
* `data_to_base64`
* `data_to_data_url`
Media type helpers:
* `infer_media_type`
* `detect_media_type`
* `detect_image_media_type`
* `detect_audio_media_type`
## ai.types.usage
`Usage` is normalized token usage from a single model call.
```python
usage.input_tokens
usage.output_tokens
usage.reasoning_tokens
usage.cache_read_tokens
usage.cache_write_tokens
usage.raw
usage.total_tokens
```
Use `usage_a + usage_b` to accumulate usage across calls.
---
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)
---
title: ai.util
description: Reference for async utility helpers.
type: reference
summary: Reference for ai.util.
---
# ai.util
`ai.util` contains asynchronous utility primitives used by the SDK.
## merge
`merge` consumes multiple async iterables concurrently and yields items as they
become available.
```python
async for event in ai.util.merge(stream, tool_runner.events()):
...
```
## decouple
`decouple` runs an async iterable in its own task and hands its items to the
consumer through a buffer. Use it to let a producer keep going while the
consumer is slow. (`Agent.LOOP_BUFFER` exposes this for the agent loop.)
```python
async for item in ai.util.decouple(source, buffer=None):
...
```
`buffer` is required and sets how many items the producer may run ahead of
the consumer:
* `None`: unbounded. The producer is never held back by the consumer.
* `n`: the producer may run up to `n` items ahead.
* `0`: lockstep. The iterable is only advanced when the consumer asks for an
item. This is how `merge` drives its sources.
Every step of the iterable runs in the same worker task. Pass
`task_group` to run the worker in an existing `asyncio.TaskGroup`.
This makes the lockstep variant useful for when an async generator needs to
be driven from multiple different tasks.
## Queues
Queue primitives:
* `AsyncIterableQueue`
* `MultiWaiter`
`AsyncIterableQueue` is an async iterable queue that can be closed. `MultiWaiter`
waits for the first completed item across multiple async sources.
## Lifecycle Helpers
Lifecycle helpers:
* `unwrap_generator_exit`
* `maybe_aclosing`
Use these helpers when implementing custom async generators or safely closing
optional async resources.
---
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)
---
title: Agent
description: Run the default agent loop with tools.
type: reference
summary: Reference for ai.Agent.
---
# Agent
`Agent` streams model output, dispatches Python tools, appends tool results to
history, and repeats until the model returns a final assistant message.
```python
agent = ai.Agent(tools=[...])
agent.tools
```
`agent.tools` returns a copy of registered executable tools.
## run
```python
async with agent.run(
model,
messages,
output_type=None,
params=None,
) as stream:
async for event in stream:
...
```
Arguments:
* `model`: `ai.Model`.
* `messages`: initial list of `ai.messages.Message`.
* `output_type`: optional Pydantic model for final JSON output.
* `params`: optional `ai.InferenceRequestParams`.
## AgentStream
`Agent.run` yields an `AgentStream`. Read the final output after iteration.
```python
stream.context
stream.messages
stream.output
```
* `stream.context`: the run's `ai.Context`, i.e. the live per-run state (model,
messages, tools, params).
* `stream.messages`: the message history, including messages added during the
run. Shorthand for `stream.context.messages`.
* `stream.output`: the run's result. By default, the final assistant message's
text. When `output_type` is set, the text is validated as JSON against that
Pydantic model and the parsed instance is returned.
## loop
Override `Agent.loop(context)` to customize control flow. The default loop uses
`ai.stream`, `ToolRunner`, `Context.resolve`, and `Context.add`.
```python
class CustomAgent(ai.Agent):
async def loop(self, context: ai.Context):
while context.keep_running():
...
```
`LOOP_BUFFER` is a class variable that sets how many events `loop` may run
ahead of the consumer of `run`. `None` (the default) is unbounded, so the loop
keeps going while the consumer is between reads. `0` runs the loop in lockstep
with the consumer, which loops that sequence side effects against consumer
code depend on.
```python
class DurableAgent(ai.Agent):
LOOP_BUFFER = 0
```
When persisting a run (for durability or serverless execution), save and
restore the message history only. Everything else the loop touches, e.g. streams,
tool runners, hook futures, provider clients, is runtime state that is
recreated on every run and cannot be serialized.
---
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)
---
title: experimental_generate
description: Generate a buffered language-model response.
type: reference
summary: Reference for ai.experimental_generate.
---
# experimental_generate
`ai.experimental_generate` makes one language-model call and returns the
complete assistant `Message` without exposing streamed events.
This API is experimental and may change or be removed.
```python
message = await ai.experimental_generate(model, messages)
print(message.text)
```
## Function
```python
await ai.experimental_generate(
model,
messages,
*,
tools=None,
output_type=None,
params=None,
)
```
You can also pass an agent-loop context:
```python
message = await ai.experimental_generate(context=context)
```
Pass either `model` and `messages`, or `context=`, not both.
## Arguments
* `model`: `ai.Model`.
* `messages`: List of `ai.messages.Message` values.
* `tools`: Optional model-facing `ai.Tool` declarations.
* `output_type`: Optional Pydantic model used to constrain structured output.
* `params`: Optional `ai.InferenceRequestParams`.
* `context`: Optional `ai.Context`. Its model, messages, tools, output type,
and params supply defaults for the call.
When you pass `context=`, do not pass `model`, `messages`, or `tools`.
`output_type` and `params` can override the values from the context.
## Return value
Returns the complete assistant `Message`. With structured output, parse the
message using the same Pydantic model:
```python
import pydantic
class Summary(pydantic.BaseModel):
title: str
points: list[str]
message = await ai.experimental_generate(
model,
[ai.user_message("Summarize the mission as JSON.")],
output_type=Summary,
)
summary = message.get_output(Summary)
```
## Provider behavior
The SDK uses the provider's native non-streaming generation method when it is
available. Otherwise, it drains the provider's stream internally and returns
the aggregated message.
Use [`ai.stream`](/docs/reference/ai/stream) when your application needs events
as the response arrives. Use `Agent.run` when the SDK should execute Python
tools and continue the model loop.
---
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)
---
title: get_model
description: Resolve a model reference from an id.
type: reference
summary: Reference for ai.get_model.
---
# get_model
`get_model` resolves a model id to a `Model`.
```python
model = ai.get_model("anthropic/claude-sonnet-4")
model = ai.get_model("openai:gpt-5")
model = ai.get_model()
```
Unprefixed ids route through AI Gateway. Calling `get_model()` with no argument
reads `AI_SDK_DEFAULT_MODEL`.
## Arguments
* `model_id`: Optional model id. Provider-prefixed ids use `provider:model`.
Gateway model ids can use `provider/model`.
## Return value
Returns `ai.Model`.
---
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)
---
title: get_provider
description: Resolve and configure providers.
type: reference
summary: Reference for ai.get_provider.
---
# get_provider
`get_provider` resolves a provider by id.
```python
provider = ai.get_provider("openai")
provider = ai.get_provider(
"openai",
base_url="http://localhost:1234/v1",
api_key="your_access_token_here",
)
```
Known providers include AI Gateway, OpenAI-compatible providers, and
Anthropic-compatible providers.
---
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)
---
title: ai
description: Reference for top-level ai exports.
type: reference
summary: Full record of names imported directly from ai.
---
# ai
`ai` is the application-facing namespace. This page mirrors the public names
re-exported from `ai.__all__`.
```python
import ai
model = ai.get_model("anthropic/claude-sonnet-4")
messages = [ai.user_message("Hello")]
```
## Linked pages
These top-level exports have their own pages under `ai`.
* [`stream`](/docs/reference/ai/stream): Call a model and iterate response
events. The same page documents `Stream`.
* [`experimental_generate`](/docs/reference/ai/experimental-generate): Call a
model and return a buffered `Message`.
* [`get_model`](/docs/reference/ai/get-model): Resolve a model id into a
`Model`.
* [`get_provider`](/docs/reference/ai/get-provider): Resolve and configure a
provider.
* [`@ai.tool`](/docs/reference/ai/tool-decorator): Define executable tools from
Python functions.
* [`Agent`](/docs/reference/ai/agent): Run the default agent loop.
## Module aliases
These module aliases are also re-exported from `ai`.
* [`errors`](/docs/reference/errors): Error classes and HTTP error helpers.
* [`events`](/docs/reference/events): Stream, agent, tool, and hook events.
* [`messages`](/docs/reference/messages): Message and message part models.
* `models`: Model namespace used by the top-level model exports on this page.
* [`mcp`](/docs/reference/mcp): MCP tool loading helpers.
* [`ops`](/docs/reference/ops): Dedicated model operations.
* [`providers`](/docs/reference/ai.providers): Provider classes and
provider-specific namespaces.
* [`testing`](/docs/reference/testing): Scripted model and agent testing.
* [`tools`](/docs/reference/tools): Model-facing tool schema types.
* [`ui`](/docs/reference/ai.ui/ai-sdk): UI adapter namespaces.
* [`util`](/docs/reference/util): Async utility helpers.
* [`experimental_telemetry`](/docs/reference/telemetry): Tracing and
observability APIs.
## Models
### Model
`Model` identifies what to call. Providers own credentials, clients,
endpoints, model listing, and wire translation.
```python
model = ai.Model(id="gpt-5", provider=provider)
model.id
model.provider
model.protocol
model.with_protocol(protocol)
```
`Model` is a lightweight reference. It does not own network state.
Fields:
* `id`: Provider model id.
* `provider`: Provider instance.
* `protocol`: Optional provider protocol override.
Methods:
* `with_protocol(protocol)`: Return a copy that uses a specific provider
protocol.
### probe
`probe` asks the model provider to verify that a model exists and is reachable.
```python
await ai.probe(model)
```
## Providers
### Provider
`Provider` is the base class for provider instances. Providers own credentials,
clients, endpoints, model listing, and wire translation.
```python
provider.name
provider.base_url
provider.api_key
provider.headers
provider.protocol
await provider.list_models()
await provider.probe(model)
```
Subclasses implement provider-specific configuration and clients. Application
code usually gets a provider with `get_provider`.
### ProviderProtocol
`ProviderProtocol` translates messages, tools, params, and dedicated model
operations to provider wire formats.
Provider instances use a protocol for language-model calls and `ai.ops`.
```python
protocol.stream(
client,
model,
messages,
tools=tools,
params=params,
provider=provider.name,
)
await protocol.generate(
client,
model,
messages,
tools=tools,
output_type=output_type,
params=params,
provider=provider.name,
)
```
## Request Params
Model params are top-level `ai` types.
Use `InferenceRequestParams` with `stream` and `Agent.run`.
```python
params = ai.InferenceRequestParams().with_temperature(0)
async with ai.stream(model, messages, params=params) as stream:
...
```
Request params:
* `InferenceRequestParams`: Inference request options.
* `ProviderServiceParams`: Provider service tier options.
* `ReasoningParams`: Provider reasoning or thinking options.
* `OutputParams`: Output token, include, verbosity, and reasoning summary
options.
* `CacheParams`: Prompt cache options.
* `ContextManagementParams`: Server-side context management options.
* `TokenThreshold`: Token count used as a trigger threshold.
Sampling params:
* `TemperatureSamplerParams`
* `TopKSamplerParams`
* `TopPSamplerParams`
* `MinPSamplerParams`
* `RepetitionPenaltyParams`
* `SeedSamplerParams`
* `RandomSeed`
* `RANDOM`
* `DEFAULT`
* `UNSET`
* `ModelProviderDefault`
* `Unset`
Tool calling params:
* `ToolCallingParams`
* `ToolChoiceMode`
* `ToolSelection`
* `ToolRef`
Routing params:
* `RoutingParams`
* `RoutingTarget`
* `RoutingTargetChain`
* `GeoRegion`
* `CloudRegion`
* `ProviderRankingStrategy`
* `GLOBAL`
## Messages
Message builders create `Message` values.
```python
ai.message("Hello", role="user")
ai.system_message("You are concise.")
ai.user_message("Hello", ai.file_part(data, media_type="image/png"))
ai.assistant_message("Hi")
ai.tool_message(tool_call_id="tc_1", result="done", tool_name="lookup")
```
Message builder exports:
* `message`
* `system_message`
* `user_message`
* `assistant_message`
* `tool_message`
Part builders create message part values.
```python
ai.text_part("hello")
ai.file_part(data, media_type="image/png", filename="image.png")
ai.thinking("reasoning text")
ai.content_output("caption", ai.file_part(png_bytes, media_type="image/png"))
ai.tool_result_part("tc_1", result={"ok": True}, tool_name="lookup")
```
Part builder exports:
* `text_part`
* `file_part`
* `thinking`
* `content_output`
* `tool_result_part`
## Tools and Agents
### Agent
`Agent` runs the default agent loop.
```python
agent = ai.Agent(tools=[contact_mothership])
```
Arguments:
* `tools`: Optional `AgentTool` values from `tool` and schema-only `Tool`
declarations for provider-executed tools.
### AgentTool
`AgentTool` binds a model-facing `Tool` declaration to an executable Python
function.
```python
tool.name
tool.tool
tool.fn
tool.validator
tool.require_approval
```
Pass `AgentTool` values to `Agent(tools=[...])`.
### Context
Custom loops use `Context` to resolve model tool calls and `ToolRunner` to run
them.
```python
context.model
context.messages
context.tools
context.output_type
context.params
```
Useful methods:
* `keep_running()`: Return `True` while the last message still needs work.
* `resolve(tool_call)`: Convert model tool call parts into executable
`ToolCall` objects.
* `add(message)`: Append messages to history.
### ToolCall
`ToolCall` is the executable runtime object produced by `Context.resolve`.
```python
tool_call.id
tool_call.name
tool_call.fn
tool_call.kwargs
result = await tool_call()
```
### ToolRunner
`ToolRunner` schedules tool calls and collects their result messages.
```python
async with ai.ToolRunner() as runner:
runner.schedule(tool_call)
async for result in runner.events():
...
message = runner.get_tool_message()
```
Use `add_result(result)` when a custom loop executes a tool itself but still
wants the runner to aggregate the result message.
### Streaming tool aliases
Async-iterable tools can yield partial output while they run.
* `StreamingTextTool`: Concatenate yielded strings.
* `StreamingStatusTool[T]`: Treat intermediate yields as status updates and
the last yielded value as the final result.
* `SubAgentTool`: Forward nested agent events and use the nested final text as
model input.
### Tool result helpers
```python
ai.tool_result(tool_call_id="tc_1", tool_name="lookup", result={"ok": True})
ai.deferred_tool_result(hook_part, tool_call_id="tc_1", tool_name="lookup")
```
Exports:
* `tool_result`: Create a `ToolCallResult`.
* `deferred_tool_result`: Create a deferred hook placeholder result.
### Hooks
Hooks let an agent pause while another process or UI supplies a decision.
```python
approval = await ai.hook(
"approve_contact_mothership",
payload=ai.tools.ToolApproval,
metadata={"tool": "contact_mothership"},
)
```
Hook exports:
* `hook`: Emit a deferred hook event and wait for a matching resolution.
* `resolve_hook`: Resolve a live or future hook.
* `defer_hook`: Mark a serialized deferred hook as aborted.
* `cancel_hook`: Cancel a live hook by label.
* `HookRegistry`: Store live and pre-registered hook resolutions.
* `get_hook_registry`: Return the current hook registry.
* `HookDeferredException`: Signal that a hook was deferred for a later run.
### yield\_from
`yield_from` forwards values from an async iterable through the current agent
runtime and returns the aggregator's model-facing result. See
[`ai.agents`](/docs/reference/ai.agents) for the advanced API.
## Errors
Top-level error exports are documented in [`ai.errors`](/docs/reference/errors).
---
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)
---
title: stream
description: Stream model responses and inspect the aggregated result.
type: reference
summary: Reference for ai.stream, Stream, stream events, aggregation, and replay.
---
# stream
`ai.stream` calls a model and returns an async context manager whose value is a
`Stream`.
```python
async with ai.stream(model, messages, tools=tools) as stream:
async for event in stream:
...
```
## Function
```python
ai.stream(
model,
messages,
*,
tools=None,
output_type=None,
params=None,
)
```
You can also pass `context=` from an agent loop:
```python
async with ai.stream(context=context) as stream:
...
```
Pass either `model` and `messages`, or `context=`, not both.
## Arguments
* `model`: `ai.Model`.
* `messages`: list of `ai.messages.Message`.
* `tools`: optional model-facing `ai.Tool` declarations.
* `output_type`: optional Pydantic model for JSON output validation.
* `params`: optional `ai.InferenceRequestParams`.
## Stream
`Stream` is an async iterator of `ai.events.Event` values. It aggregates events
into an assistant message while you iterate.
```python
message = stream.message
text = stream.text
tool_calls = stream.tool_calls
usage = stream.usage
output = stream.output
```
`stream.output` returns text by default. When `output_type` is set, it validates
the final text as JSON and returns the parsed Pydantic model.
## Event Aggregation
Text and reasoning blocks are built from start, delta, and end events. Tool
calls are built from `ToolStart`, `ToolDelta`, and `ToolEnd`. Provider-executed
tools use built-in tool events. Generated files arrive as `FileEvent`.
Each yielded event has the current aggregated `event.message`.
## Replay
If the last input message has `replay=True`, `ai.stream` does not call the
provider. It emits replay tool-end events from the existing assistant message
so resumable agent flows can dispatch the same tool calls again.
`Stream.replay_message` synthesizes stream events from a complete `Message`.
```python
async with ai.Stream.replay_message(message) as stream:
async for event in stream:
...
```
## Errors
If the provider stream ends before a finish event, iteration raises
`ai.errors.ProviderIncompleteResponseError`. The partial message is still on
`stream.message`.
Use
[`ai.experimental_generate`](/docs/reference/ai/experimental-generate) when
you need the complete message without consuming streamed events.
---
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)
---
title: @ai.tool
description: Define executable tools from Python functions.
type: reference
summary: Reference for the ai.tool decorator.
---
# @ai.tool
`@ai.tool` turns an async Python function into an executable `AgentTool`.
```python
@ai.tool
async def contact_mothership(query: str) -> str:
"""Contact the mothership."""
return "Soon."
```
## Forms
```python
@ai.tool
async def name(...) -> Result: ...
@ai.tool(require_approval=True)
async def name(...) -> Result: ...
@ai.tool(aggregator=...)
async def name(...) -> AsyncIterable[Item]: ...
@ai.tool(to_model_input=...)
async def name(...) -> Result: ...
```
The function name becomes the tool name. The docstring becomes the tool
description. The function signature becomes a Pydantic validator and JSON
schema.
The decorated callable must return an awaitable or an async iterable. Every
async iterable tool needs an aggregator. Pass `aggregator=` or annotate the
return type with an `ai.agents.Aggregate` marker, but do not use both.
## Sending the model something else
`to_model_input=` takes a callable that converts the tool's result into the
value the model sees. The full result stays on the `ToolResultPart` for the UI
and for session persistence:
```python
class EditResult(pydantic.BaseModel):
message: str
old_content: str
new_content: str
@ai.tool(to_model_input=lambda r: r.message)
async def edit(path: str, ...) -> EditResult:
"""Edit a file."""
...
```
The model sees `"Successfully replaced 2 block(s) in f.py."`; the UI still has
both file versions to render a diff from.
The callable receives the tool's return value, and is not called for a tool
that raised.
`to_model_input=` and an aggregator are mutually exclusive -- a streaming tool
derives the model-facing value from its aggregator's own `to_model_input`, so
declaring both raises `TypeError`.
For model-facing tool declarations and provider-executed tools, use
[`ai.tools`](/docs/reference/tools).
---
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)
---
title: ai.agents
description: Reference for advanced agent namespace APIs.
type: reference
summary: Reference for public APIs that are intentionally used from ai.agents.
---
# ai.agents
Most agent APIs are documented as top-level `ai` exports. Use `ai.agents`
for advanced agent namespace APIs and types that are not promoted to the
top-level namespace.
## Aggregators
Aggregators collect yielded values from streaming tools and decide what value is
stored and what value is sent back to the model.
APIs:
* `Aggregate`
* `yield_from`
* `SimpleAggregator`
* `ConcatAggregator`
* `LastAggregator`
* `MessageAggregator`
Custom aggregators implement `ai.events.Aggregator`.
`yield_from` forwards values from an async iterable and returns the
aggregator's model-facing result.
```python
result = await ai.agents.yield_from(
stream,
aggregator=ai.agents.MessageAggregator,
)
```
Optional `tool_name`, `tool_call_id`, and `label` values are attached to the
forwarded `PartialToolCallResult` events.
## Tool calls
* `BoundToolCall`
* `ToolCallCallable`
## GatedToolCall
`GatedToolCall` wraps a tool call that requires approval or another gate before
execution.
Use it in custom agent loops when you need to delay or externally approve a
scheduled tool call.
---
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)
---
title: ai.providers
description: Reference for provider APIs.
type: reference
summary: Reference for ai.providers.
---
# ai.providers
`ai.get_provider` is the usual entry point for application code. Use
`ai.providers` when you need provider classes, provider protocols, or
provider-specific namespaces.
## Public APIs
* `get_provider`
* `Provider`
* `ProviderProtocol`
* `OpenAICompatibleProvider`
* `AnthropicCompatibleProvider`
* `GatewayProvider`
* `history_utils`
## Provider operations
`Provider` and `ProviderProtocol` define these model operations:
* `stream`: Stream a language-model response.
* `generate`: Generate a buffered language-model response.
* `generate_image`: Generate images.
* `generate_video`: Generate videos.
* `generate_audio`: Generate speech.
* `embed`: Embed text values.
* `transcribe`: Transcribe audio.
* `rerank`: Rerank documents against a query.
Application code normally uses `ai.stream`, `ai.experimental_generate`, or
[`ai.ops`](/docs/reference/ops) instead of calling provider methods directly.
A protocol only needs to implement the operations that it supports.
## ai.providers.history\_utils
Message-history repair utilities. `ai.stream` passes history to the
provider as-is; provider implementations call these before converting
messages to their wire format.
```python
from ai.providers import history_utils
repaired = history_utils.repair(messages)
```
`repair` strips internal messages, removes non-model parts, replaces
invalid tool args with `{}`, and inserts error results for missing tool
calls, logging a warning for every fix. It raises `IntegrityError` on
duplicate tool ids and orphaned tool results — those have no safe
automatic fix. The individual fix functions (`drop_internal`,
`fix_tool_args`, `close_orphaned_tool_calls`) each return a new message
list plus the issues they fixed — use them when a provider needs
different choices or the issues themselves — and `check_tool_ids`
detects the unfixable issues without raising.
Validate a history yourself when you want issues to raise instead of
being repaired.
```python
history_utils.validate(messages) # raises IntegrityError on any issue
issues = history_utils.inspect(messages) # or just report them
```
APIs:
* `repair`
* `drop_internal`
* `fix_tool_args`
* `close_orphaned_tool_calls`
* `check_tool_ids`
* `inspect`
* `validate`
* `Issue`
* `IntegrityError`
---
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)
---
title: AI Gateway
description: Configure Vercel AI Gateway models, authentication, routing, tools, and media generation.
type: reference
summary: Use models from multiple providers through Vercel AI Gateway.
---
# AI Gateway
Vercel AI Gateway gives you access to models from multiple providers through
one API. You can change models, control routing, and use provider tools without
configuring a separate SDK for each provider.
The AI SDK uses AI Gateway when a model ID does not include a colon-separated
provider prefix:
```python
import ai
model = ai.get_model("anthropic/claude-sonnet-4.6")
```
This is equivalent to:
```python
model = ai.get_model("gateway:anthropic/claude-sonnet-4.6")
```
## Configure authentication
AI Gateway supports OIDC and API key authentication.
### Use Vercel OIDC
OIDC enables your application to authenticate both locally and in the cloud
without storing static credentials.
Install the optional Vercel dependency to use OIDC:
```bash title="Terminal"
uv add "ai[vercel]"
```
For local development, use `vercel dev` to run your application, and the
provider will automatically obtain a fresh token before accessing the Gateway.
Alternatively, use `vercel env pull` and manually pass `VERCEL_OIDC_TOKEN`
to the provider.
When deployed to Vercel, no additional environment configuration is required.
See the [Vercel OIDC documentation](https://vercel.com/docs/oidc) for more
information.
### Use an API key
Set `AI_GATEWAY_API_KEY` in your environment:
```bash title="Terminal"
export AI_GATEWAY_API_KEY="your_ai_gateway_api_key_here"
```
No additional dependencies are required when using API key authentication. You
can also pass an API key when you create a provider:
```python
provider = ai.get_provider(
"vercel",
api_key="your_ai_gateway_api_key_here",
)
```
An explicit API key takes priority over `AI_GATEWAY_API_KEY`. Either form takes
priority over OIDC authentication.
## Stream a model response
Pass the model to `ai.stream` with a list of messages:
```python
import ai
model = ai.get_model("anthropic/claude-sonnet-4.6")
messages = [
ai.system_message("Be concise."),
ai.user_message("Explain why the sky is blue."),
]
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)
```
After the stream finishes, you can access the response and token usage:
```python
print(stream.text)
print(stream.message)
print(stream.usage)
```
AI Gateway streams text, reasoning, tool calls, tool results, generated files,
and usage through the standard SDK event types.
## Configure a provider
Use `ai.get_provider("vercel")` when you need custom provider configuration:
```python
import ai
provider = ai.get_provider(
"vercel",
base_url="https://gateway.example.com/v4/ai",
api_key="your_ai_gateway_api_key_here",
headers={"X-Application": "example_application"},
)
model = ai.Model(
id="anthropic/claude-sonnet-4.6",
provider=provider,
)
```
`GatewayProvider` accepts these configuration options:
| Option | Description |
| ---------- | -------------------------------------------------------------------------- |
| `api_key` | API key used instead of `AI_GATEWAY_API_KEY` or OIDC. |
| `base_url` | Gateway API base URL. The default is `https://ai-gateway.vercel.sh/v4/ai`. |
| `headers` | Headers included with each Gateway request. |
| `env` | Environment values used only by this provider instance. |
| `client` | Custom `httpx.AsyncClient` used for requests. |
| `protocol` | Custom provider protocol. Most applications should use the default. |
Close providers that your application creates:
```python
await provider.aclose()
```
The provider closes the HTTP client that it creates. If you pass a custom
`httpx.AsyncClient`, close that client in your application.
## List and check models
Use `list_models` to get the model IDs available to the authenticated Gateway
account:
```python
provider = ai.get_provider("vercel")
model_ids = await provider.list_models()
for model_id in model_ids:
print(model_id)
```
`list_models` returns model IDs as sorted strings, such as
`anthropic/claude-sonnet-4.6`.
Use `ai.probe` to check authentication and model availability without
generating tokens:
```python
model = ai.get_model("anthropic/claude-sonnet-4.6")
await ai.probe(model)
```
`probe` raises `ProviderNotConfiguredError` when authentication is missing and
`ProviderModelNotFoundError` when the model is unavailable.
## Configure routing and fallbacks
Use `RoutingParams` to control how AI Gateway selects a provider or fallback
model:
```python
params = ai.InferenceRequestParams(
routing=ai.RoutingParams(
provider_allowlist=frozenset({"anthropic", "bedrock"}),
provider_order=("bedrock", "anthropic"),
provider_ranking=ai.ProviderRankingStrategy.COST,
fallback_models=(
"openai/gpt-5-mini",
"google/gemini-2.5-flash",
),
)
)
async with ai.stream(model, messages, params=params) as stream:
async for event in stream:
...
```
The routing fields map to AI Gateway options:
| Python field | Gateway option | Description |
| -------------------- | ----------------- | ------------------------------------------------------------------------- |
| `provider_allowlist` | `only` | Restricts routing to the listed providers. |
| `provider_order` | `order` | Sets the order in which Gateway tries providers. |
| `provider_ranking` | `sort` | Ranks providers by the selected cost or performance strategy. |
| `fallback_models` | `models` | Lists models to try after the requested model fails. |
| `routing_target` | `inferenceRegion` | Restricts routing by global scope, geographic region, or provider region. |
Use `GLOBAL`, `GeoRegion`, or `CloudRegion` to select a routing target:
```python
params = ai.InferenceRequestParams(
routing=ai.RoutingParams(
routing_target=ai.GeoRegion("us"),
)
)
```
Use `RoutingTargetChain` when Gateway and the selected provider need different
targets:
```python
params = ai.InferenceRequestParams(
routing=ai.RoutingParams(
routing_target=ai.RoutingTargetChain(
gateway=ai.GeoRegion("us"),
provider=ai.CloudRegion("us-east-1"),
)
)
)
```
## Track requests
Use `safety_identifier` to attach a stable end-user ID to a request.
Use `tags` to group requests by application, feature, environment, or another
category:
```python
params = ai.InferenceRequestParams(
safety_identifier="user_123",
tags=frozenset(
{
"feature:document_summary",
"environment:production",
}
),
metadata={
"prompt_version": "3",
},
)
```
You can use the user ID and tags when reviewing AI Gateway usage and reports.
## Configure Gateway request options
`GatewayParams` contains request options that apply only to AI Gateway.
Add it to `InferenceRequestParams` using `with_provider_params`:
```python
from ai.providers.ai_gateway import (
GatewayParams,
ProviderTimeoutsParams,
)
params = ai.InferenceRequestParams().with_provider_params(
GatewayParams(
quota_entity_id="workspace_123",
zero_data_retention=True,
disallow_prompt_training=True,
provider_timeouts=ProviderTimeoutsParams(
byok={"anthropic": 5_000},
),
)
)
```
`GatewayParams` supports these fields:
| Field | Description |
| -------------------------- | --------------------------------------------------------------------------------------- |
| `quota_entity_id` | Identifies the entity against which Gateway tracks quota. |
| `zero_data_retention` | Restricts routing to credentials that satisfy the requested zero-data-retention policy. |
| `hipaa_compliant` | Restricts routing to providers that meet AI Gateway HIPAA requirements. |
| `disallow_prompt_training` | Restricts routing to providers that do not train on prompts. |
| `byok` | Supplies request-scoped Bring Your Own Key (BYOK) credentials, grouped by provider. |
| `provider_timeouts` | Sets per-provider BYOK attempt timeouts in milliseconds. |
### Supply request-scoped BYOK credentials
The value of `byok` maps provider names to lists of credential objects:
```python
import os
from ai.providers.ai_gateway import GatewayParams
params = ai.InferenceRequestParams().with_provider_params(
GatewayParams(
byok={
"anthropic": [
{
"apiKey": os.environ["ANTHROPIC_API_KEY"],
}
]
}
)
)
```
Each provider defines its own credential fields. AI Gateway controls credential
validation, ordering, policy checks, and fallback behavior.
Keep provider credentials in environment variables or another secret store.
Avoid placing credentials directly in source code.
See the [AI Gateway BYOK documentation](https://vercel.com/docs/ai-gateway/byok)
for provider credential formats and account-level setup.
## Pass provider-specific options
Use `extra_body.providerOptions` when the selected model provider supports an
option that does not have a typed SDK field.
Use the selected provider name as the key:
```python
params = ai.InferenceRequestParams(
extra_body={
"providerOptions": {
"anthropic": {
"speed": "fast",
}
}
}
)
```
Use `gateway` as the key only for raw AI Gateway options:
```python
params = ai.InferenceRequestParams(
extra_body={
"providerOptions": {
"gateway": {
"sort": "cost",
}
}
}
)
```
The provider merges raw options after typed SDK options. When both forms set
the same field, the value in `extra_body` takes priority.
Prefer typed fields such as `RoutingParams` and `GatewayParams` when they are
available. Use `extra_body` for provider options that the SDK does not yet
represent.
## Use standard request options
AI Gateway supports the common fields in `InferenceRequestParams`, including:
* Sampling temperature, top-p, top-k, seed, frequency penalty, and presence
penalty.
* Reasoning effort and reasoning summaries.
* Tool selection, tool limits, and parallel tool calls.
* Service tiers.
* Safety identifiers, metadata, and tags.
* Maximum output tokens and provider-specific response data.
* Prompt caching.
* Routing and fallback models.
* Server-side context management for OpenAI and Anthropic models.
* Extra headers, query parameters, and request body fields.
The provider translates reasoning and context management options based on the
provider name in the model ID.
These options are not supported by the current Gateway protocol implementation
and raise `ValueError` when set:
* `MinPSamplerParams.min_p`.
* `RepetitionPenaltyParams.repetition_penalty`.
* `RepetitionPenaltyParams.consideration_window`.
* `OutputParams.text_verbosity`.
Context management requires an OpenAI or Anthropic model.
## Use tools
AI Gateway supports Python function tools, provider-specific tools, and tools
executed by AI Gateway.
### Use Python function tools
Pass function tools to `ai.stream` or an agent:
```python
@ai.tool
async def get_weather(city: str) -> str:
"""Get the weather for a city."""
return f"The weather in {city} is sunny."
async with ai.stream(
model,
messages,
tools=[get_weather.tool],
) as stream:
...
```
`ai.stream` sends the tool schema and returns tool calls. Use `ai.Agent` when
the SDK should execute the Python function and continue the model loop.
### Use provider-specific tools
Provider-specific tool definitions can pass through AI Gateway:
```python
from ai.providers.anthropic import tools as anthropic_tools
tools = [
anthropic_tools.web_search(max_uses=3),
]
```
Use a tool that the selected model and provider support. Some provider tools
also require provider account configuration.
### Use AI Gateway tools
AI Gateway tools run in Gateway and can work with any gateway-routed model:
```python
from ai.providers.ai_gateway import tools as gateway_tools
tools = [
gateway_tools.perplexity_search(
max_results=5,
search_domain_filter=["reuters.com"],
)
]
async with ai.stream(
model,
[ai.user_message("Find recent AI policy news.")],
tools=tools,
) as stream:
async for event in stream:
if isinstance(event, ai.events.TextDelta):
print(event.chunk, end="", flush=True)
elif isinstance(event, ai.events.BuiltinToolResult):
print(event.result.result)
```
The provider includes these AI Gateway tools:
* `perplexity_search`.
* `parallel_search`.
See [AI Gateway tools](/docs/reference/ai.providers/ai-gateway/tools) for all
tool options and configured examples.
## Generate structured output
Pass a Pydantic model as `output_type` to request structured output:
```python
import pydantic
import ai
class Summary(pydantic.BaseModel):
title: str
points: list[str]
model = ai.get_model("openai/gpt-5.4")
async with ai.stream(
model,
[ai.user_message("Summarize the benefits of renewable energy.")],
output_type=Summary,
) as stream:
async for _ in stream:
pass
summary = stream.output
```
After the stream finishes, `stream.output` contains the validated Pydantic
model.
## Handle generated files
Models can return files as part of a language model stream. Each file produces
a `FileEvent` and is 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)
```
## Handle errors
The provider maps AI Gateway failures to the standard `ai.ProviderError`
hierarchy:
| Gateway failure | Python error |
| ------------------------ | ----------------------------- |
| Missing configuration | `ProviderNotConfiguredError` |
| Authentication failure | `ProviderAuthenticationError` |
| Invalid request | `ProviderBadRequestError` |
| Rate limit | `ProviderRateLimitError` |
| Missing model | `ProviderModelNotFoundError` |
| Gateway server failure | `ProviderInternalServerError` |
| Invalid Gateway response | `ProviderResponseError` |
| Request timeout | `ProviderTimeoutError` |
Catch a specific error when your application can handle it:
```python
try:
await ai.probe(model)
except ai.ProviderAuthenticationError:
print("Check the AI Gateway credentials.")
except ai.ProviderModelNotFoundError:
print("Choose an available model.")
```
Mapped API errors retain the Gateway response in `error.body`. This can include
details about failed routing attempts.
See [`ai.errors`](/docs/reference/errors) for the complete error hierarchy.
## Current provider scope
The Python provider currently supports:
* Language model streaming.
* Buffered language model responses through `ai.experimental_generate`.
* Structured output.
* Function and provider-executed tools.
* Model listing and connection checks.
* Image, video, speech generation, embeddings, transcription, and reranking through `ai.ops`.
It does not currently expose realtime sessions, credit queries, spend reports,
or generation lookup helpers.
See [Model Operations](/docs/basics/model-operations) for usage examples and
[`ai.ops`](/docs/reference/ops) for the exact APIs.
## API reference
### GatewayProvider
`GatewayProvider` owns AI Gateway authentication, configuration, HTTP
resources, model discovery, and the default protocol.
Get an instance with:
```python
provider = ai.get_provider("vercel")
```
Methods and properties:
* `client`: Shared Gateway client.
* `protocol`: Configured protocol.
* `tools`: AI Gateway tool module.
* `is_configured()`: Reports whether API key or OIDC authentication is
available.
* `list_models()`: Returns available model IDs.
* `probe(model)`: Checks authentication and model availability.
* `stream(...)`: Streams a language model response.
* `generate_image(...)`: Generates images.
* `generate_video(...)`: Generates videos.
* `generate_audio(...)`: Generates speech.
* `embed(...)`: Embeds text values.
* `transcribe(...)`: Transcribes audio.
* `rerank(...)`: Reranks documents against a query.
* `aclose()`: Closes provider-owned resources.
### GatewayParams
`GatewayParams` contains typed AI Gateway request options:
* `quota_entity_id`.
* `zero_data_retention`.
* `hipaa_compliant`.
* `disallow_prompt_training`.
* `byok`.
* `provider_timeouts`.
### ProviderTimeoutsParams
`ProviderTimeoutsParams` contains per-provider timeout settings.
Fields:
* `byok`: Mapping of provider names to timeout values in milliseconds.
### GatewayV4Protocol
`GatewayV4Protocol` translates SDK messages, tools, request parameters, stream
events, and dedicated model operations to and from the AI Gateway v4 API.
`GatewayProvider` uses this protocol by default. Pass it directly only when you
need a protocol override.
### Module exports
`ai.providers.ai_gateway` exports:
* `GatewayProvider`.
* `GatewayV3Protocol`.
* `GatewayV4Protocol`.
* `GatewayParams`.
* `ProviderTimeoutsParams`.
* `errors`.
* `tools`.
---
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)
---
title: tools
description: Reference for AI Gateway provider-executed tools.
type: reference
summary: Reference for AI Gateway provider tool helpers.
---
# tools
AI Gateway tool helpers create provider-executed `ai.Tool` declarations.
## Tools
* `perplexity_search`
* `parallel_search`
## Option models
* `SourcePolicy`
* `Excerpts`
* `FetchPolicy`
---
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)
---
title: ai.providers.anthropic
description: Reference for Anthropic-compatible provider APIs.
type: reference
summary: Reference for ai.providers.anthropic.
---
# ai.providers.anthropic
`ai.providers.anthropic` contains the Anthropic-compatible provider, protocol,
and provider-executed tools.
```python
provider = ai.get_provider("anthropic")
model = ai.Model(id="claude-sonnet-4-6", provider=provider)
```
The optional upstream Anthropic SDK loads lazily when the provider creates or
uses an SDK client.
## AnthropicCompatibleProvider
`AnthropicCompatibleProvider` implements `Provider` for Anthropic-compatible
APIs.
Default configuration for the `anthropic` provider uses:
* `ANTHROPIC_API_KEY`
* `ANTHROPIC_BASE_URL`
Pass `base_url`, `api_key`, or a custom client through `get_provider` when you
need explicit configuration.
```python
provider = ai.get_provider(
"anthropic",
base_url="https://anthropic.example.com",
)
model = ai.Model(id="claude-sonnet-4-6", provider=provider)
```
The provider supports model listing, probing, streaming, provider-executed
tools, and custom Anthropic-compatible clients.
## AnthropicMessagesProtocol
`AnthropicMessagesProtocol` translates SDK messages and params to the
Anthropic Messages API wire format.
The provider uses this protocol by default. Use it directly only when you need
a protocol override.
## Tools
Provider-executed Anthropic tools are documented on the child `tools` page.
---
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)
---
title: tools
description: Reference for Anthropic provider-executed tools.
type: reference
summary: Reference for Anthropic provider tool helpers.
---
# tools
Anthropic tool helpers create provider-executed `ai.Tool` declarations.
## Tools
* `web_search`
* `web_fetch`
* `code_execution`
* `computer_use`
* `text_editor`
* `bash`
* `memory`
## Option models and constants
* `UserLocation`
* `Citations`
* `BETA_HEADERS`
---
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)
---
title: ai.providers.openai
description: Reference for OpenAI-compatible provider APIs.
type: reference
summary: Reference for ai.providers.openai.
---
# ai.providers.openai
`ai.providers.openai` contains the OpenAI-compatible provider, protocols, and
provider-executed tools.
```python
provider = ai.get_provider("openai")
model = ai.Model(id="gpt-5", provider=provider)
```
The optional upstream OpenAI SDK loads lazily when the provider creates or uses
an SDK client.
## OpenAICompatibleProvider
`OpenAICompatibleProvider` implements `Provider` for OpenAI-compatible APIs.
Default configuration for the `openai` provider uses:
* `OPENAI_API_KEY`
* `OPENAI_BASE_URL`
Pass `base_url`, `api_key`, or a custom client through `get_provider` when you
need explicit configuration.
```python
provider = ai.get_provider(
"openai",
base_url="http://localhost:11434/v1",
)
model = ai.Model(id="llama3", provider=provider)
```
The provider supports model listing, probing, streaming, provider-executed
tools, and custom OpenAI-compatible clients.
## Protocols
OpenAI-compatible protocols translate SDK messages and params to OpenAI wire
formats.
Types:
* `OpenAIResponsesProtocol`
* `OpenAIChatCompletionsProtocol`
The provider chooses a default protocol for the configured provider. Use these
classes directly only when you need a protocol override.
## Tools
Provider-executed OpenAI tools are documented on the child `tools` page.
---
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)
---
title: tools
description: Reference for OpenAI provider-executed tools.
type: reference
summary: Reference for OpenAI provider tool helpers.
---
# tools
OpenAI tool helpers create provider-executed `ai.Tool` declarations.
## Tools
* `web_search`
* `web_search_preview`
* `file_search`
* `code_interpreter`
* `image_generation`
* `local_shell`
* `shell`
* `apply_patch`
* `mcp`
* `tool_search`
## Option models
* `WebSearchUserLocation`
* `WebSearchFilters`
* `FileSearchRanking`
* `CodeInterpreterContainer`
---
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)
---
title: ai.ui.ai_sdk approvals
description: Apply AI SDK UI approval responses.
type: reference
summary: Reference for ApprovalResponse, extract_approvals, and apply_approvals.
---
# ai.ui.ai_sdk approvals
Approval helpers bridge AI SDK UI tool approval responses into the hook
registry.
## APIs
* `ApprovalResponse`
* `extract_approvals`
* `apply_approvals`
```python
messages, approvals = ai.ui.ai_sdk.to_messages(ui_messages)
async with agent.run(model, messages) as stream:
ai.ui.ai_sdk.apply_approvals(approvals)
async for event in stream:
...
```
Call `apply_approvals` inside the `agent.run` context before iteration, or pass
the registry that the run uses with `registry=`.
---
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)
---
title: ai.ui.ai_sdk.to_messages
description: Convert UI messages to runtime messages.
type: reference
summary: Reference for inbound AI SDK UI message conversion.
---
# ai.ui.ai_sdk.to_messages
`to_messages` converts AI SDK UI messages into runtime messages and extracts
approval responses.
```python
messages, approvals = ai.ui.ai_sdk.to_messages(ui_messages)
```
Call `apply_approvals` inside the `agent.run` context, before iteration, when
resuming a run with extracted approvals.
---
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)
---
title: ai.ui.ai_sdk
description: Reference for AI SDK UI message and SSE adapters.
type: reference
summary: Reference for ai.ui.ai_sdk.
---
# ai.ui.ai_sdk
`ai.ui.ai_sdk` converts between AI SDK UI message streams and the Python
runtime.
## APIs
* `UIMessage`
* `to_messages`
* `extract_approvals`
* `apply_approvals`
* `to_sse`
* `to_stream`
* `to_ui_messages`
* `UI_MESSAGE_STREAM_HEADERS`
---
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)
---
title: ai.ui.ai_sdk.to_ui_messages
description: Convert runtime messages to AI SDK UI messages.
type: reference
summary: Reference for ai.ui.ai_sdk.to_ui_messages.
---
# ai.ui.ai_sdk.to_ui_messages
`to_ui_messages` converts stored runtime messages back to AI SDK UI messages.
```python
ui_messages = ai.ui.ai_sdk.to_ui_messages(messages)
```
Use it when loading durable conversation history for a UI client.
---
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)
---
title: ai.ui.ai_sdk outbound stream
description: Convert agent streams to AI SDK UI stream parts.
type: reference
summary: Reference for to_sse, to_stream, and UI_MESSAGE_STREAM_HEADERS.
---
# ai.ui.ai_sdk outbound stream
Use outbound stream helpers to send an agent run to an AI SDK UI client.
```python
async for chunk in ai.ui.ai_sdk.to_sse(agent_stream):
yield chunk
```
```python
async for part in ai.ui.ai_sdk.to_stream(agent_stream):
yield part
```
Set `UI_MESSAGE_STREAM_HEADERS` as the HTTP response headers when serving the
stream. It is a plain `dict` with the headers the AI SDK UI client expects:
* `x-vercel-ai-ui-message-stream: v1` tells the client the response is a UI
message stream;
* `Content-Type: text/event-stream` marks the response as
server-sent events;
* `Cache-Control: no-cache`, `Connection: keep-alive`, `x-accel-buffering: no`
keep proxies from buffering or caching the response so chunks reach the client
as they are produced.
```python
return fastapi.responses.StreamingResponse(
stream_response(),
headers=ai.ui.ai_sdk.UI_MESSAGE_STREAM_HEADERS,
)
```
## Step boundaries
The adapter starts a UI step with the first model event. When a new
`StreamStart` follows a completed model stream, it finishes the previous step
and starts another. Tool results and approval events emitted between the two
model streams stay in the step that requested them.
The adapter finishes the active step and message when the input event stream
ends.
---
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)
---
title: ai.ui.ai_sdk.UIMessage
description: Parse AI SDK UI messages.
type: reference
summary: Reference for ai.ui.ai_sdk.UIMessage.
---
# ai.ui.ai_sdk.UIMessage
Use `UIMessage` as the request model for AI SDK UI clients.
```python
class ChatRequest(pydantic.BaseModel):
messages: list[ai.ui.ai_sdk.UIMessage]
```
AI SDK UI messages use a `parts` array. Tool parts can carry approval state and
provider metadata.
---
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)