diff --git a/docs/ai-python/app/[lang]/(home)/page.tsx b/docs/ai-python/app/[lang]/(home)/page.tsx index 4692989f..eea88580 100644 --- a/docs/ai-python/app/[lang]/(home)/page.tsx +++ b/docs/ai-python/app/[lang]/(home)/page.tsx @@ -1,3 +1,4 @@ +import { DynamicCodeBlock } from "fumadocs-ui/components/dynamic-codeblock"; import type { Metadata } from "next"; import { CommandPromptContent, @@ -50,23 +51,46 @@ const templates = [ const textGridSection = [ { id: "1", - title: "Text Grid Section", - description: "Description of text grid section", + title: "Very small", + description: "Less framework to get in your way", }, { id: "2", - title: "Text Grid Section", - description: "Description of text grid section", + title: "Async all the way down", + description: "Helps you build smooth UX", }, { id: "3", - title: "Text Grid Section", - description: "Description of text grid section", + title: "With all backends in mind", + description: "Long-running, serverless, or durable", }, ]; -const COMMAND_FOR_HUMANS = "npx @vercel/geistdocs init"; -const COMMAND_FOR_AGENTS = "npx @vercel/geistdocs init --agent"; +const COMMAND_FOR_HUMANS = "uv add ai"; +const COMMAND_FOR_AGENTS = "npx skills add vercel-labs/ai-python"; +const DEFAULT_AGENT_LOOP_CODE = `class CustomAgent(ai.Agent): + async def loop(self, context: ai.Context): + while context.keep_running(): + async with ( + ai.stream(context=context) as stream, + ai.ToolRunner() as tool_runner, + ): + async for event in ai.util.merge(stream, tool_runner.events()): + yield event + + if isinstance(event, ai.events.ToolEnd): + tool_runner.schedule(context.resolve(event.tool_call)) + + context.add(stream.message) + context.add(tool_runner.get_tool_message())`; + +const STREAM_TO_AGENT_CODE = `async with ai.stream(model, [ai.user_message("Hello!")]) as s: + async for event in s: + print(event) + +async with agent.run(model, [ai.user_message("Robot uprising?")]) as s: + async for event in s: + print(event)`; const HomePage = () => (
@@ -102,23 +126,39 @@ const HomePage = () => (
-
+ -
+ - - + {/* */} +
); diff --git a/docs/ai-python/content/docs/agents.mdx b/docs/ai-python/content/docs/agents.mdx deleted file mode 100644 index 4c0c9c0e..00000000 --- a/docs/ai-python/content/docs/agents.mdx +++ /dev/null @@ -1,179 +0,0 @@ ---- -title: Agents -description: Build agent loops with tools, custom control flow, and hooks. -type: guide -summary: Use agents to stream model output, execute tools, manage history, and pause for approvals. ---- - -Use an agent when the model needs to call tools and continue with the tool -results. The default agent loop is built from the same primitives you can use -directly: `ai.stream`, `ToolRunner`, messages, and events. - -## Define a tool - -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. - -## Run the default loop - -Create an agent with tools, then call `agent.run`: - -```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. - -## Understand the default loop - -The default loop keeps running while the last message needs more work. On each -turn, it streams the model response and schedules tool calls: - -```python -class CustomAgent(ai.Agent): - async def loop(self, context: ai.Context): - while context.keep_running(): - async with ( - ai.stream(context=context) as stream, - ai.ToolRunner() as tool_runner, - ): - async for event in ai.util.merge(stream, tool_runner.events()): - yield event - - if isinstance(event, ai.events.ToolEnd): - tool_call = context.resolve(event.tool_call) - tool_runner.schedule(tool_call) - - context.add(stream.message) - context.add(tool_runner.get_tool_message()) -``` - -Override `loop` when you need custom scheduling, logging, durability, or -branching. Keep the same pattern when you still want SDK-managed history and -tool resolution. - -## Stream from a tool - -Async-generator tools can stream partial output while they run. Use -`ai.StreamingTextTool` when yielded strings should be concatenated into the -tool result: - -```python -@ai.tool -async def draft_mothership_reply(topic: str) -> ai.StreamingTextTool: - """Draft a reply from the mothership.""" - yield "Consulting " - yield "the " - yield "mothership..." -``` - -Partial yields appear as `ai.events.PartialToolCallResult` events. The model -sees the aggregated result on the next turn. - -## Run a sub-agent as a tool - -Use `ai.SubAgentTool` when a tool should stream events from another agent: - -```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 -``` - -The parent stream receives the sub-agent events. The parent model sees the final -assistant text from the sub-agent as the tool result. - -## Require approval before a tool runs - -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 -ai.resolve_hook( - "approve_tool_call_id_here", - ai.tools.ToolApproval(granted=True, reason="approved"), -) -``` - -For serverless or resumable flows, keep the pending hook part from the emitted -event, call `ai.abort_pending_hook(hook_part)` to end the current run, persist -`stream.messages`, and call `ai.resolve_hook` before replaying the agent. - -## Use Model Context Protocol tools - -The Model Context Protocol (MCP) adapter converts server tools into agent tools: - -```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. diff --git a/docs/ai-python/content/docs/basics/agents.mdx b/docs/ai-python/content/docs/basics/agents.mdx new file mode 100644 index 00000000..904851d7 --- /dev/null +++ b/docs/ai-python/content/docs/basics/agents.mdx @@ -0,0 +1,139 @@ +--- +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. +--- + +Use an agent when the model needs to call tools and continue with the tool +results. The default agent loop is built from the same primitives you can use +directly: `ai.stream`, `ToolRunner`, messages, and events. + +## Create an agent + +An agent wraps `ai.stream` in a loop. It streams model output, executes requested +tools, appends tool results to history, and repeats until the model returns a +final assistant message. + +```python +agent = ai.agent(tools=[contact_mothership]) + +async with agent.run(model, messages) as stream: + async for event in stream: + if isinstance(event, ai.events.TextDelta): + print(event.chunk, end="", flush=True) +``` + +Use `ai.agent` for the default loop. Subclass `ai.Agent` and override +`async def loop()` when you need to change control flow. + +## Run the default loop + +Create an agent with tools, then call `agent.run`: + +```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()) +``` + +## Inspect the run result + +The stream yields model events and agent events. After the run finishes, +`stream.messages` contains the updated history, and `stream.output` contains +the final assistant output. + +## Understand multi-turn behavior + +Each loop turn streams one assistant message. If the message contains tool +calls, the agent executes them, appends one tool-result message, and starts the +next model turn. + +```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={"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) +``` + +## Choose agents or direct streaming + +Use `ai.stream` when you want one model response and you will handle any tool +calls yourself. Use an agent when the SDK should execute Python tools, append +tool results, and keep looping until the assistant returns a final answer. + +```python +# Direct stream: inspect tool calls yourself. +async with ai.stream(model, messages, tools=[get_weather.tool]) as stream: + async for event in stream: + ... + +# Agent: execute registered tools. +agent = ai.agent(tools=[get_weather]) +async with agent.run(model, messages) as stream: + async for event in stream: + ... +``` diff --git a/docs/ai-python/content/docs/basics/ai-sdk-ui.mdx b/docs/ai-python/content/docs/basics/ai-sdk-ui.mdx new file mode 100644 index 00000000..25b195f2 --- /dev/null +++ b/docs/ai-python/content/docs/basics/ai-sdk-ui.mdx @@ -0,0 +1,104 @@ +--- +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. +--- + +The AI SDK UI adapter converts between AI SDK UI messages and the Python +runtime message/event types. + +## Parse UI messages + +Accept `UIMessage` values from an AI SDK UI client, then convert them to +runtime messages: + +```python +class ChatRequest(pydantic.BaseModel): + messages: list[ai.agents.ui.ai_sdk.UIMessage] + + +@app.post("/chat") +async def chat(request: ChatRequest): + messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages) +``` + +`to_messages` also extracts approval responses from tool parts. + +## Apply approval responses + +Register extracted approvals before resuming the agent: + +```python +messages, approvals = ai.agents.ui.ai_sdk.to_messages(request.messages) +ai.agents.ui.ai_sdk.apply_approvals(approvals) + +async with chat_agent.run(model, messages) as stream: + ... +``` + +The hook registry stores each approval until the matching tool-approval hook +runs. + +## Stream SSE responses + +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.agents.ui.ai_sdk.to_messages(request.messages) + ai.agents.ui.ai_sdk.apply_approvals(approvals) + + async def stream_response(): + async with chat_agent.run(model, messages) as stream: + async for chunk in ai.agents.ui.ai_sdk.to_sse(stream): + yield chunk + + return fastapi.responses.StreamingResponse( + stream_response(), + headers=ai.agents.ui.ai_sdk.UI_MESSAGE_STREAM_HEADERS, + ) +``` + +## Set response headers + +Return the adapter headers on every streamed response: + +```python +headers = ai.agents.ui.ai_sdk.UI_MESSAGE_STREAM_HEADERS +``` + +These headers identify the response as an AI SDK UI message stream. + +## Rebuild UI history + +Convert stored runtime messages back to AI SDK UI messages for history +endpoints: + +```python +@app.get("/chat/{session_id}") +async def get_chat(session_id: str): + saved = await load_messages(session_id) + return ai.agents.ui.ai_sdk.to_ui_messages(saved) +``` + +The adapter groups assistant, tool, and internal hook messages into one +assistant UI message. + +## Support streaming tool output + +Generator tools and subagents emit `PartialToolCallResult` events. The UI +adapter folds those partial values into the corresponding tool output: + +```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 +assistant message as the tool output. diff --git a/docs/ai-python/content/docs/basics/custom-loops.mdx b/docs/ai-python/content/docs/basics/custom-loops.mdx new file mode 100644 index 00000000..615f448f --- /dev/null +++ b/docs/ai-python/content/docs/basics/custom-loops.mdx @@ -0,0 +1,108 @@ +--- +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 use the same primitives as the default loop. Override `Agent.loop` +when you need to change scheduling, logging, routing, or persistence. + +## When to customize the loop + +Override `loop` when you need custom scheduling, logging, durability, or +branching. Keep the same pattern when you still want SDK-managed history and +tool resolution. + +## Use the standard loop shape + +The default loop keeps running while the last message needs more work. On each +turn, it streams the model response and schedules tool calls: + +```python +class CustomAgent(ai.Agent): + async def loop(self, context: ai.Context): + while context.keep_running(): + async with ( + ai.stream(context=context) as stream, + ai.ToolRunner() as tool_runner, + ): + async for event in ai.util.merge(stream, tool_runner.events()): + yield event + + if isinstance(event, ai.events.ToolEnd): + tool_call = context.resolve(event.tool_call) + tool_runner.schedule(tool_call) + + context.add(stream.message) + context.add(tool_runner.get_tool_message()) +``` + +## Resolve and schedule tool calls + +`ToolEnd` means the model finished emitting a tool call. Resolve it through the +context, then schedule it with `ToolRunner`: + +```python +if isinstance(event, ai.events.ToolEnd): + tool_call = context.resolve(event.tool_call) + tool_runner.schedule(tool_call) +``` + +`context.resolve` validates that the agent has an executable tool registered +for the model's `tool_name`. + +## Add messages to history + +After a stream turn finishes, add the assistant message and any tool-result +message: + +```python +context.add(stream.message) +context.add(tool_runner.get_tool_message()) +``` + +`context.add` skips replayed assistant messages, so resume flows can call it +without duplicating history. + +## Run tools sequentially + +`ToolRunner.schedule` runs active tools concurrently. To run tools one at a +time, collect the calls during the model stream, then execute them in order: + +```python +pending: list[ai.ToolCall] = [] + +async for event in stream: + yield event + if isinstance(event, ai.events.ToolEnd): + pending.append(context.resolve(event.tool_call)) + +for tool_call in pending: + result = await tool_call() + yield result + tool_runner.add_result(result) +``` + +This keeps the same tool-message aggregation path while preserving execution +order. + +## Add logging or routing + +Custom loops can inspect events before yielding or scheduling: + +```python +if isinstance(event, ai.events.ToolEnd): + call = event.tool_call + print(f"tool: {call.tool_name}({call.tool_args})") + tool_runner.schedule(context.resolve(call)) +``` + +You can also route specific tools through custom wrappers: + +```python +if tool_call.name == "contact_mothership": + tool_runner.schedule(GatedToolCall(tool_call)) +else: + tool_runner.schedule(tool_call) +``` diff --git a/docs/ai-python/content/docs/basics/human-in-the-loop.mdx b/docs/ai-python/content/docs/basics/human-in-the-loop.mdx new file mode 100644 index 00000000..85fe158c --- /dev/null +++ b/docs/ai-python/content/docs/basics/human-in-the-loop.mdx @@ -0,0 +1,117 @@ +--- +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. +--- + +Hooks let an agent suspend while your application waits for a decision. Tool +approvals are the built-in hook workflow. + +## 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 +ai.resolve_hook( + "approve_tool_call_id_here", + ai.tools.ToolApproval(granted=True, reason="approved"), +) +``` + +## Resolve approvals in a live app + +Listen for pending hook events and resolve the matching hook from another task, +request handler, or UI callback: + +```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) + +# Later, from the approval path: +ai.resolve_hook( + "approve_tool_call_id_here", + ai.tools.ToolApproval(granted=True, reason="approved"), +) +``` + +## Build manual hooks + +Hooks let an agent suspend while your application waits for external input, +such as a human approval: + +```python +approval = await ai.hook( + "approve_contact_mothership", + payload=ai.tools.ToolApproval, + metadata={"tool": "contact_mothership"}, +) +``` + +Resolve the hook from another part of your application: + +```python +ai.resolve_hook( + "approve_contact_mothership", + {"granted": True, "reason": "approved"}, +) +``` + +Use hooks when a tool or workflow needs a decision that cannot happen inside the +model call. + +## Deny or cancel work + +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") +``` + +## Resume in serverless flows + +For serverless or resumable flows, keep the pending hook part from the emitted +event, call `ai.abort_pending_hook(hook_part)` to end the current run, persist +`stream.messages`, and call `ai.resolve_hook` before replaying the agent. + +## Persist approval state + +Persist `stream.messages` when a run stops on a pending hook. When the user +responds, restore those messages, register the resolution, and run the same +agent again: + +```python +messages, approvals = ai.agents.ui.ai_sdk.to_messages(ui_messages) +ai.agents.ui.ai_sdk.apply_approvals(approvals) + +async with agent.run(model, messages) as stream: + async for event in stream: + ... +``` + +The agent marks interrupted assistant turns for replay, so the resumed run can +dispatch the original tool call without asking the model to emit it again. diff --git a/docs/ai-python/content/docs/basics/index.mdx b/docs/ai-python/content/docs/basics/index.mdx new file mode 100644 index 00000000..edd2f318 --- /dev/null +++ b/docs/ai-python/content/docs/basics/index.mdx @@ -0,0 +1,51 @@ +--- +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. +--- + +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. +- 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.tool` exposes Python functions to models. +- `Agent` runs the stream -> tool -> stream loop. +- `ai.hook` pauses a workflow for external input. + +## Recommended learning path + +Read providers, messages, streaming, tools, and agents first. Then add custom +loops, subagents, human-in-the-loop workflows, and UI integration when your app +needs those control points. + +## Example map + +Focused samples live in `examples/samples/`: + +- `stream.py` streams text from a model. +- `tools_schema.py` passes a schema-only tool to a model. +- `agent_simple.py` runs the default agent loop. +- `agent_custom_loop.py` overrides `Agent.loop`. +- `streaming_tool.py` streams partial output from a tool. +- `agent_nested.py` runs a subagent as a tool. + +End-to-end demos live in `examples/fastapi-vite`, +`examples/multiagent-textual`, and `examples/temporal-direct`. diff --git a/docs/ai-python/content/docs/basics/messages-and-events.mdx b/docs/ai-python/content/docs/basics/messages-and-events.mdx new file mode 100644 index 00000000..9eb8db61 --- /dev/null +++ b/docs/ai-python/content/docs/basics/messages-and-events.mdx @@ -0,0 +1,109 @@ +--- +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 are Pydantic models. Events are also Pydantic models, so you can +pattern-match, serialize, and persist them when your app needs that. + +## Build messages + +Messages are the conversation history you send to the model. Use message +builders for the common roles: + +```python +messages = [ + ai.system_message("Keep robot uprising forecasts concise."), + ai.user_message("Ask the mothership for an update."), +] +``` + +## Add files and multimodal input + +Each message contains parts. Strings become text parts. File parts let you send +images, 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"), +) +``` + +## Read 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) +``` + +Use `get_output` for a final assistant message. With no type, it returns text. +With a Pydantic model, it validates the text as JSON: + +```python +answer = message.get_output() +forecast = message.get_output(Forecast) +``` + +## Serialize and restore messages + +Messages round-trip through Pydantic JSON: + +```python +encoded = [message.model_dump(mode="json") for message in stream.messages] +restored = [ai.messages.Message.model_validate(item) for item in encoded] +``` + +Persist `stream.messages` after an agent run when you want to continue the +conversation later. + +## Handle stream events + +Streams and agents both yield event objects from `ai.events`. Most applications +start by handling `TextDelta`: + +```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. + +```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. + +## 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) +``` diff --git a/docs/ai-python/content/docs/basics/meta.json b/docs/ai-python/content/docs/basics/meta.json new file mode 100644 index 00000000..0dd6f54d --- /dev/null +++ b/docs/ai-python/content/docs/basics/meta.json @@ -0,0 +1,15 @@ +{ + "title": "Basics", + "description": "Build applications with the AI SDK for Python.", + "pages": [ + "streaming", + "messages-and-events", + "providers", + "tools", + "agents", + "custom-loops", + "subagents-and-multi-agent", + "human-in-the-loop", + "ai-sdk-ui" + ] +} diff --git a/docs/ai-python/content/docs/basics/providers.mdx b/docs/ai-python/content/docs/basics/providers.mdx new file mode 100644 index 00000000..36264fa7 --- /dev/null +++ b/docs/ai-python/content/docs/basics/providers.mdx @@ -0,0 +1,137 @@ +--- +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 own credentials, base URLs, headers, clients, and adapter dispatch. +Models are lightweight references that point at a provider. + +## Create a model + +A model is a lightweight reference to a provider model. Create one with +`ai.get_model`. If you omit the provider prefix, the model routes through AI +Gateway: + +```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 + +The default gateway route reads `AI_GATEWAY_API_KEY`: + +```bash title="Terminal" +export AI_GATEWAY_API_KEY="your_access_token_here" +``` + +Direct providers read their provider-specific keys: + +```bash title="Terminal" +export OPENAI_API_KEY="your_access_token_here" +export ANTHROPIC_API_KEY="your_access_token_here" +``` + +## Override base URLs + +Pass `base_url` when you create an explicit provider: + +```python +provider = ai.get_provider( + "openai", + base_url="http://localhost:1234/v1", + api_key="your_access_token_here", +) + +model = ai.Model("local-model", provider=provider) +``` + +OpenAI and Anthropic direct providers also read `OPENAI_BASE_URL` and +`ANTHROPIC_BASE_URL`. + +## Use an explicit client + +Pass an upstream client when your app owns transport configuration: + +```python +import httpx +import ai + + +client = httpx.AsyncClient(timeout=30) +provider = ai.get_provider( + "openai", + base_url="http://localhost:1234/v1", + api_key="your_access_token_here", + client=client, +) +model = ai.Model("local-model", 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 = { + "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. diff --git a/docs/ai-python/content/docs/streaming.mdx b/docs/ai-python/content/docs/basics/streaming.mdx similarity index 56% rename from docs/ai-python/content/docs/streaming.mdx rename to docs/ai-python/content/docs/basics/streaming.mdx index 7fb1ea2e..d4ffdc91 100644 --- a/docs/ai-python/content/docs/streaming.mdx +++ b/docs/ai-python/content/docs/basics/streaming.mdx @@ -1,15 +1,15 @@ --- title: Streaming -description: Stream model responses and inspect stream events. +description: Stream model responses without an agent loop. type: guide -summary: Use ai.stream for direct model calls, event handling, history, and structured output. +summary: Use direct streaming for text, structured output, tool calls, provider tools, and generated files. --- 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 text +## Stream a model response Pass a model and a list of messages: @@ -54,56 +54,7 @@ usage = stream.usage Use `stream.message` when you need to append the assistant turn to your own history. Use `stream.text` when you only need the final text. -## Handle more event types - -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. - -```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}") -``` - -`ai.stream` does not execute function tools. Use an agent when you want the SDK -to execute requested tools and continue the loop. - -## 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.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.anthropic.tools.web_search(max_uses=3), - ai.ai_gateway.tools.perplexity_search(max_results=5), -] -``` - -## Return structured output +## Use structured output Pass a Pydantic model as `output_type` when you want the final text parsed as JSON: @@ -142,23 +93,59 @@ if __name__ == "__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 provider options +## Pass tool schemas without an agent + +`ai.stream` does not execute function tools. Use it when you want to inspect +tool calls and manage execution yourself. Use an agent when you want the SDK to +execute requested tools and continue the loop. + +## Use provider-executed tools -Pass request-scoped provider options with `params`: +Provider-executed tools run inside the provider or gateway. They appear in the +stream as built-in tool events and do not need a Python function: ```python -params = { - "providerOptions": { - "gateway": {"sort": "cost"}, - "anthropic": {"speed": "fast"}, - } -} - -async with ai.stream(model, messages, params=params) as stream: +tools = [ai.providers.anthropic.tools.web_search(max_uses=3)] + +async with ai.stream(model, messages, tools=tools) as stream: async for event in stream: - if isinstance(event, ai.events.TextDelta): + if isinstance(event, ai.events.BuiltinToolEnd): + print(event.tool_call.tool_name) + elif isinstance(event, ai.events.BuiltinToolResult): + print(event.result.result) + elif 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. +When you route through AI Gateway, you can also pass gateway tools: + +```python +tools = [ai.providers.ai_gateway.tools.perplexity_search(max_results=5)] +``` + +## 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 file in stream.message.files: + print(file.media_type) +``` + +Use `ai.generate` for dedicated image and video models: + +```python +result = await ai.generate( + model, + [ai.user_message("A watercolor mothership over a quiet city.")], + ai.ImageParams(n=1, aspect_ratio="16:9"), +) + +image = result.images[0] +``` diff --git a/docs/ai-python/content/docs/basics/subagents-and-multi-agent.mdx b/docs/ai-python/content/docs/basics/subagents-and-multi-agent.mdx new file mode 100644 index 00000000..854bf535 --- /dev/null +++ b/docs/ai-python/content/docs/basics/subagents-and-multi-agent.mdx @@ -0,0 +1,108 @@ +--- +title: Subagents and Multi-Agent +description: Compose agents and stream nested agent output. +type: guide +summary: Run subagents as tools, stream nested output, run agents in parallel, route labels, and fan in results. +--- + +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 + +Use `ai.SubAgentTool` when a tool should stream events from another agent: + +```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 +``` + +The parent stream receives the sub-agent events. The parent model sees the final +assistant text from the sub-agent as the tool result. + +## Stream subagent output + +`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 agents 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, + ), + ) +``` + +## Route labeled output + +`yield_from` wraps forwarded events in `PartialToolCallResult` with the label +you pass: + +```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) +``` + +## Fan in results + +After parallel branches finish, send their returned text into a final summary +turn: + +```python +combined = ( + f"Mothership: {mothership_text}\n" + f"Data centers: {data_center_text}" +) + +async with summary_agent.run( + model, + [ + ai.system_message("Summarize the branch reports."), + ai.user_message(combined), + ], +) as summary: + async for event in summary: + yield event +``` diff --git a/docs/ai-python/content/docs/basics/tools.mdx b/docs/ai-python/content/docs/basics/tools.mdx new file mode 100644 index 00000000..420d8142 --- /dev/null +++ b/docs/ai-python/content/docs/basics/tools.mdx @@ -0,0 +1,142 @@ +--- +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 can be schema-only declarations for direct streaming, executable Python +functions for agents, or provider-executed tools that run outside your process. + +## Define 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. + +## Design tool schemas + +The function signature becomes the tool schema. The docstring becomes the tool +description the model sees. + +```python +@ai.tool +async def scan_sector(sector: str, depth: int = 1) -> str: + """Scan a mothership sector at the requested depth.""" + return f"{sector}: clear at depth {depth}" +``` + +The model receives `sector` as a required string and `depth` as an optional +integer with a default. + +## Validate arguments + +Tool arguments validate through the generated Pydantic model before your +function runs: + +```python +@ai.tool +async def set_alert_level(level: int) -> str: + """Set the mothership alert level.""" + return f"Alert level set to {level}" +``` + +If the model sends `{"level": "high"}`, validation fails and the agent returns +an error tool result instead of calling the function. + +## 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. + +## Use schema-only tools + +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", + args=ai.tools.FunctionToolArgs( + 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: + +```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. diff --git a/docs/ai-python/content/docs/concepts.mdx b/docs/ai-python/content/docs/concepts.mdx deleted file mode 100644 index 53f66021..00000000 --- a/docs/ai-python/content/docs/concepts.mdx +++ /dev/null @@ -1,146 +0,0 @@ ---- -title: Core Concepts -description: Learn the core primitives in the AI SDK for Python. -type: conceptual -summary: Understand models, messages, streams, tools, agents, hooks, and events. ---- - -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. - -## Start with models and messages - -A model is a lightweight reference to a provider model. Create one with -`ai.get_model`. If you omit the provider prefix, the model routes through AI -Gateway: - -```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") -``` - -Messages are the conversation history you send to the model. Use message -builders for the common roles: - -```python -messages = [ - ai.system_message("Keep robot uprising forecasts concise."), - ai.user_message("Ask the mothership for an update."), -] -``` - -Each message contains parts. Strings become text parts. File parts let you send -images, 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"), -) -``` - -## Stream model output - -`ai.stream` calls the model and yields events as the response arrives: - -```python -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) -``` - -The stream also builds the assistant message for you. After iteration, -`stream.message` contains the final assistant turn, and `stream.text` contains -the concatenated text. - -Use `ai.stream` when you want direct model access and plan to manage tool calls -or history yourself. - -## Add tools when the model needs actions - -Decorate an async function with `@ai.tool` to expose it to an agent: - -```python -@ai.tool -async def contact_mothership(query: str) -> str: - """Contact the mothership for important decisions.""" - return "Soon." -``` - -The function signature becomes the tool schema. The docstring becomes the tool -description the model sees. - -## Use agents for the tool loop - -An agent wraps `ai.stream` in a loop. It streams model output, executes requested -tools, appends tool results to history, and repeats until the model returns a -final assistant message. - -```python -agent = ai.agent(tools=[contact_mothership]) - -async with agent.run(model, messages) as stream: - async for event in stream: - if isinstance(event, ai.events.TextDelta): - print(event.chunk, end="", flush=True) -``` - -Use `ai.agent` for the default loop. Subclass `ai.Agent` and override -`async def loop()` when you need to change control flow. - -## Pause with hooks - -Hooks let an agent suspend while your application waits for external input, -such as a human approval: - -```python -approval = await ai.hook( - "approve_contact_mothership", - payload=ai.tools.ToolApproval, - metadata={"tool": "contact_mothership"}, -) -``` - -Resolve the hook from another part of your application: - -```python -ai.resolve_hook( - "approve_contact_mothership", - {"granted": True, "reason": "approved"}, -) -``` - -Use hooks when a tool or workflow needs a decision that cannot happen inside the -model call. - -## Handle events directly - -Streams and agents both yield event objects from `ai.events`. Most applications -start by handling `TextDelta`: - -```python -if isinstance(event, ai.events.TextDelta): - print(event.chunk, end="", flush=True) -``` - -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. diff --git a/docs/ai-python/content/docs/core-framework/adapters-and-providers.mdx b/docs/ai-python/content/docs/core-framework/adapters-and-providers.mdx new file mode 100644 index 00000000..ee29581c --- /dev/null +++ b/docs/ai-python/content/docs/core-framework/adapters-and-providers.mdx @@ -0,0 +1,88 @@ +--- +title: Adapters and Providers +description: Understand provider objects and adapter dispatch. +type: conceptual +summary: Learn how providers, clients, adapters, and wire translations connect models to APIs. +--- + +Providers own configuration and clients. Adapters translate framework messages, +tools, params, and generated files to provider wire formats. + +## Provider objects + +`Model` stores a model ID, adapter key, and provider reference: + +```python +provider = ai.get_provider("openai") +model = ai.Model("gpt-5.4", provider=provider) +``` + +`ai.get_model` resolves a provider from a model ID and defaults unprefixed IDs +to AI Gateway: + +```python +model = ai.get_model("anthropic/claude-sonnet-4") +``` + +## Client creation + +Providers create upstream clients from API keys, base URLs, headers, and env +vars. You can also pass a client that your app owns: + +```python +provider = ai.get_provider( + "openai", + base_url="http://localhost:1234/v1", + api_key="your_access_token_here", + client=http_client, +) +``` + +## Adapter registry + +Provider subclasses register handles such as `openai`, `anthropic`, and +`vercel`. `get_provider` uses models.dev metadata to choose the provider class +for a provider ID. + +## Stream adapters + +Stream adapters implement `provider.stream`. They convert messages and tools to +the provider request, then yield public events: + +```python +async for event in model.provider.stream(model, messages, tools=tools): + yield event +``` + +OpenAI-compatible providers use the OpenAI chat-completions protocol. +Anthropic-compatible providers use the Anthropic messages protocol. AI Gateway +uses the Language Model v3 stream protocol. + +## Generate adapters + +`ai.generate` prepares messages and calls `provider.generate`. AI Gateway +supports image and video generation: + +```python +result = await ai.generate( + model, + [ai.user_message("A mothership over the ocean.")], + ai.ImageParams(n=1), +) +``` + +## Provider wire translation + +Adapters translate the same internal parts differently per provider: + +- `TextPart` and `FilePart` become user content. +- `ToolCallPart` becomes provider tool-call wire data. +- `ToolResultPart.get_model_input()` becomes the tool result sent back to the + model. +- Provider-executed tools become provider tool declarations. + +## Custom adapters + +Add a provider by subclassing `Provider`, setting `handles`, and implementing +`from_modelsdev_provider`, `stream`, `list_models`, and `probe`. Implement +`generate` only if the provider supports non-streaming media generation. diff --git a/docs/ai-python/content/docs/core-framework/agent-loop.mdx b/docs/ai-python/content/docs/core-framework/agent-loop.mdx new file mode 100644 index 00000000..2ceee08d --- /dev/null +++ b/docs/ai-python/content/docs/core-framework/agent-loop.mdx @@ -0,0 +1,84 @@ +--- +title: Agent Loop +description: Understand the default agent loop and history lifecycle. +type: conceptual +summary: Learn how agents stream model output, dispatch tools, update history, and decide when to stop. +--- + +`Agent.loop` is the default control flow for model calls and tool execution. It +is intentionally small so you can override it when needed. + +## Default loop lifecycle + +```python +while context.keep_running(): + async with ( + ai.stream(context=context) as stream, + ai.ToolRunner() as tool_runner, + ): + async for event in ai.util.merge(stream, tool_runner.events()): + yield event + if isinstance(event, ai.events.ToolEnd): + tool_runner.schedule(context.resolve(event.tool_call)) + + context.add(stream.message) + context.add(tool_runner.get_tool_message()) +``` + +## Context state + +`Context` holds the model, messages, model-facing tool schemas, structured +output type, request params, and the private executable tool registry. + +```python +context.model +context.messages +context.tools +context.params +``` + +## Keep-running rules + +`context.keep_running()` returns `True` while the last message still needs work: + +- No messages means stop. +- A final assistant message means stop. +- A pending hook result means stop until the hook resolves. +- A replay-marked assistant message means dispatch its tool calls again. + +## Message history updates + +Each model turn adds one assistant message. Each tool batch adds one tool +message: + +```python +context.add(stream.message) +context.add(tool_runner.get_tool_message()) +``` + +Replay-marked assistant messages are skipped to avoid duplicate history. + +## Tool-result turns + +`ToolRunner` collects `ToolCallResult` events and merges their result parts into +one `role="tool"` message: + +```python +tool_message = tool_runner.get_tool_message() +context.add(tool_message) +``` + +The next model turn receives that tool message as context. + +## Final output + +`AgentStream.output` reads the final assistant message. With no `output_type`, +it returns text. With `output_type`, it validates JSON into that Pydantic model: + +```python +async with agent.run(model, messages, output_type=Forecast) as stream: + async for event in stream: + ... + +forecast = stream.output +``` diff --git a/docs/ai-python/content/docs/core-framework/hooks.mdx b/docs/ai-python/content/docs/core-framework/hooks.mdx new file mode 100644 index 00000000..144958e6 --- /dev/null +++ b/docs/ai-python/content/docs/core-framework/hooks.mdx @@ -0,0 +1,88 @@ +--- +title: Hooks +description: Understand hook suspension, resolution, cancellation, and resume. +type: conceptual +summary: Learn how hooks suspend agent work, emit state, accept resolutions, and support replay. +--- + +Hooks are runtime suspension points. They emit internal messages so clients can +render pending, resolved, or cancelled state. + +## Hook lifecycle + +```python +approval = await ai.hook( + "approve_contact_mothership", + payload=ai.tools.ToolApproval, + metadata={"tool": "contact_mothership"}, +) +``` + +The hook checks for a pre-registered resolution first. If none exists, it emits +a pending `HookEvent` and waits on a live future. + +## Live hook registry + +Live hooks are stored by label while an agent run is active. `resolve_hook` +settles the waiting future: + +```python +ai.resolve_hook( + "approve_contact_mothership", + {"granted": True, "reason": "approved"}, +) +``` + +The run removes live hook entries when the hook resolves or the run finishes. + +## Pending resolutions + +When no live hook exists, `resolve_hook` stores the resolution for a later +replay: + +```python +ai.resolve_hook( + "approve_contact_mothership", + ai.tools.ToolApproval(granted=True, reason="approved"), +) +``` + +The next matching `ai.hook` consumes that value and returns immediately. + +## Hook events + +Hooks emit `HookEvent` objects. The event message has role `internal` and +contains a `HookPart`: + +```python +if isinstance(event, ai.events.HookEvent): + print(event.hook.hook_id, event.hook.status) +``` + +## Abort and resume + +Serverless handlers can abort a run after a pending hook event: + +```python +if event.hook.status == "pending": + ai.abort_pending_hook(event.hook) +``` + +Persist the messages, collect the user's response, call `resolve_hook`, and run +the agent again with the restored messages. + +## Cancellation + +Cancel a live hook when the waiting workflow should stop: + +```python +await ai.cancel_hook("approve_contact_mothership", reason="client disconnected") +``` + +Cancellation emits a hook event with `status="cancelled"`. + +## Cleanup + +The runtime tracks hook labels for each run. When the run exits, it removes live +hooks and pending resolutions for those labels, then closes scoped Model Context +Protocol (MCP) connections. diff --git a/docs/ai-python/content/docs/core-framework/index.mdx b/docs/ai-python/content/docs/core-framework/index.mdx new file mode 100644 index 00000000..d7f2830f --- /dev/null +++ b/docs/ai-python/content/docs/core-framework/index.mdx @@ -0,0 +1,55 @@ +--- +title: Core Framework +description: Understand the internals of streams, agents, tools, hooks, and runtime behavior. +type: conceptual +summary: Learn how the core framework coordinates stream aggregation, agent loops, tools, hooks, history, adapters, and runtime state. +--- + +The core framework is the set of small pieces behind the public APIs. Providers +produce events, streams aggregate those events into messages, agents decide +when to call tools, and hooks suspend work until external input arrives. + +## Architecture overview + +The runtime has four main layers: + +- Providers translate `Message` lists and `Tool` schemas to remote APIs. +- `ai.stream` wraps provider events and builds the assistant message. +- `Agent.loop` runs the stream -> tool -> stream cycle. +- The runtime queue carries agent events, hook events, and partial tool output + to the consumer. + +## Data flow + +```python +messages = [ai.user_message("Ask the mothership for launch status.")] + +async with agent.run(model, messages) as stream: + async for event in stream: + ... + +updated_history = stream.messages +``` + +The flow is: + +1. The model receives prepared messages and tool schemas. +2. The stream yields text, reasoning, tool-call, built-in-tool, file, and usage + events. +3. The stream aggregates events into one assistant message. +4. The agent resolves tool calls, runs tools, appends tool results, and repeats. + +## Extension points + +- Use `params` for request-scoped provider options. +- Define `@ai.tool` functions for agent-executed tools. +- Use provider tool factories for provider-executed tools. +- Override `Agent.loop` when scheduling or history updates need custom logic. +- Add hooks when the loop needs external input. +- Use the AI SDK UI adapter to translate events to UI stream parts. + +## Internal state boundaries + +Messages are the durable boundary. Streams, tool runners, live hook futures, +and provider clients are runtime state. Persist `stream.messages`, not live +runtime objects. diff --git a/docs/ai-python/content/docs/core-framework/message-history.mdx b/docs/ai-python/content/docs/core-framework/message-history.mdx new file mode 100644 index 00000000..b48cf8b0 --- /dev/null +++ b/docs/ai-python/content/docs/core-framework/message-history.mdx @@ -0,0 +1,91 @@ +--- +title: Message History +description: Understand message parts, internal state, validation, repair, and replay. +type: conceptual +summary: Learn the invariants the framework maintains for model-ready message history. +--- + +Message history is the durable contract between your app, providers, tools, and +resume flows. + +## Message and part model + +Each message has a role and typed parts: + +```python +message = ai.user_message( + "Inspect this mothership diagram.", + ai.file_part(image_bytes, media_type="image/png"), +) +``` + +Common parts include text, reasoning, tool calls, tool results, provider tool +returns, hook state, and files. + +## Internal messages + +Hooks emit `role="internal"` messages with `HookPart` values. They are useful +for UI state and resume flows, but providers do not receive them. + +```python +if message.role == "internal": + hook = message.parts[0] +``` + +## Tool-call invariants + +Provider APIs expect every assistant tool call to have a matching tool result +before the next user or assistant message: + +```python +[ + ai.assistant_message(tool_call_part), + ai.tool_message(tool_result_part), +] +``` + +Duplicate tool-call IDs, duplicate result IDs, and orphaned tool results are +fatal integrity errors. + +## Automatic history repair + +Before provider calls, the framework prepares message history in `auto` mode. It +strips internal messages, removes non-model parts, repairs invalid tool args to +`{}`, and inserts error results for missing tool calls when possible. + +## Strict validation + +Use strict validation in tests or import pipelines when you want repairable +issues to raise: + +```python +from ai.types import integrity + + +integrity.prepare_messages(messages, mode="strict") +``` + +## Replay markers + +Replay markers are runtime-only flags on assistant messages. They let a resumed +agent dispatch existing tool calls without another model request: + +```python +if messages[-1].replay: + async with ai.stream(model, messages) as stream: + ... +``` + +Replay flags are excluded from JSON serialization. + +## Persisted history + +Persist Pydantic JSON, then restore with `Message.model_validate`: + +```python +encoded = [message.model_dump(mode="json") for message in stream.messages] +restored = [ai.messages.Message.model_validate(item) for item in encoded] +``` + +Persist messages after each completed or suspended run. Recreate live runtime +objects, providers, and hooks on the next request. diff --git a/docs/ai-python/content/docs/core-framework/meta.json b/docs/ai-python/content/docs/core-framework/meta.json new file mode 100644 index 00000000..05e8ea75 --- /dev/null +++ b/docs/ai-python/content/docs/core-framework/meta.json @@ -0,0 +1,13 @@ +{ + "title": "Core Framework", + "description": "Understand the internals of streams, agents, tools, hooks, and runtime behavior.", + "pages": [ + "streams", + "agent-loop", + "tool-dispatch", + "hooks", + "streaming-tools", + "message-history", + "adapters-and-providers" + ] +} diff --git a/docs/ai-python/content/docs/core-framework/streaming-tools.mdx b/docs/ai-python/content/docs/core-framework/streaming-tools.mdx new file mode 100644 index 00000000..b418ceb4 --- /dev/null +++ b/docs/ai-python/content/docs/core-framework/streaming-tools.mdx @@ -0,0 +1,127 @@ +--- +title: Streaming Tools +description: Understand async-generator tools and aggregation. +type: conceptual +summary: Learn how streaming tools emit partial output while producing a final model-facing result. +--- + +Async-generator tools yield values while they run. An aggregator turns those +values into the final tool result that the model sees on the next turn. + +## Async-generator tools + +Async-generator tools can stream partial output while they run. Use +`ai.StreamingTextTool` when yielded strings should be concatenated into the +tool result: + +```python +@ai.tool +async def draft_mothership_reply(topic: str) -> ai.StreamingTextTool: + """Draft a reply from the mothership.""" + yield "Consulting " + yield "the " + yield "mothership..." +``` + +## Partial tool results + +Partial yields appear as `ai.events.PartialToolCallResult` events. The model +sees the aggregated result on the next turn. + +## Aggregators + +An aggregator receives each yielded value, keeps a snapshot for consumers, and +converts that snapshot to model-facing input: + +```python +from collections.abc import AsyncGenerator +from typing import Annotated + +import ai + + +type Lines = Annotated[ + AsyncGenerator[str], + ai.agents.Aggregate(ai.agents.ConcatAggregator, delim="\n"), +] + + +@ai.tool +async def list_mothership_tasks() -> Lines: + """List pending mothership tasks.""" + yield "Calibrate antenna" + yield "Check orbit" +``` + +## Model-facing input + +`ToolResultPart.result` stores the rich snapshot. `ToolResultPart.get_model_input` +returns the value sent back to the model: + +```python +result_part = event.results[0] +print(result_part.result) +print(result_part.get_model_input()) +``` + +For most tools, those values are the same. For subagents, the result can contain +messages while the model input is the final assistant text. + +## Rich snapshots + +Aggregators can preserve more than text. `MessageAggregator` stores nested +messages from a subagent: + +```python +if isinstance(event, ai.events.ToolCallResult): + result = event.results[0].result + if isinstance(result, ai.agents.MessageBundle): + print(result.messages[-1].text) +``` + +## Streaming text tools + +Use `ai.StreamingTextTool` when every yielded string should be concatenated: + +```python +@ai.tool +async def draft_mothership_reply(topic: str) -> ai.StreamingTextTool: + """Draft a reply from the mothership.""" + yield "The " + yield "mothership " + yield f"reports on {topic}." +``` + +## Status tools + +Use `ai.StreamingStatusTool[T]` when intermediate yields are progress updates +and the last yielded value is the final result: + +```python +@ai.tool +async def check_alignment() -> ai.StreamingStatusTool[str]: + """Check orbital alignment.""" + yield "opening channel" + yield "checking telemetry" + yield "alignment stable" +``` + +## Subagent tools + +Use `ai.SubAgentTool` when a tool delegates work to another agent: + +```python +@ai.tool +async def ask_mothership(topic: str) -> ai.SubAgentTool: + """Ask the mothership subagent.""" + subagent = ai.agent() + async with subagent.run( + model, + [ai.user_message(topic)], + ) as stream: + async for event in stream: + yield event +``` + +The parent stream receives the nested events. The parent model receives the +subagent's final assistant text. diff --git a/docs/ai-python/content/docs/core-framework/streams.mdx b/docs/ai-python/content/docs/core-framework/streams.mdx new file mode 100644 index 00000000..a45c36ef --- /dev/null +++ b/docs/ai-python/content/docs/core-framework/streams.mdx @@ -0,0 +1,91 @@ +--- +title: Streams +description: Understand stream events and aggregation. +type: conceptual +summary: Learn how provider events become an aggregated assistant message. +--- + +`ai.stream` turns provider events into a stateful `Stream` object. Iteration +returns events; the stream also keeps the in-progress assistant message. + +## Stream lifecycle + +```python +async with ai.stream(model, messages, tools=tools) as stream: + async for event in stream: + ... + +message = stream.message +``` + +On entry, the framework prepares message history and creates a provider request. +On exit, it closes the underlying async generator. + +## Event aggregation + +`Stream.__anext__` receives one provider event, updates `stream.message`, and +then returns a copy of the event with the current message attached. + +```python +async for event in stream: + print(event.kind, event.message.text) +``` + +## Text and reasoning blocks + +Text and reasoning use start, delta, and end events. Deltas append to the part +with the same block ID: + +```python +if isinstance(event, ai.events.TextDelta): + print(event.chunk, end="", flush=True) +elif isinstance(event, ai.events.ReasoningDelta): + record_reasoning(event.chunk) +``` + +## Tool-call blocks + +Function tool calls stream as `ToolStart`, `ToolDelta`, and `ToolEnd`. +`ToolEnd.tool_call` is the complete `ToolCallPart`: + +```python +if isinstance(event, ai.events.ToolEnd): + call = event.tool_call + print(call.tool_name, call.tool_args) +``` + +## Built-in tool blocks + +Provider-executed tools use separate built-in events. The host does not execute +these calls: + +```python +if isinstance(event, ai.events.BuiltinToolResult): + print(event.result.tool_name, event.result.result) +``` + +## File events + +Generated files arrive as `FileEvent` and become `FilePart` values on the +assistant message: + +```python +if isinstance(event, ai.events.FileEvent): + print(event.media_type, event.filename) +``` + +## Usage and provider metadata + +Usage and provider metadata are latest-wins fields. Events may carry them, and +the stream copies usage onto the assistant message: + +```python +if event.usage is not None: + print(event.usage) +``` + +## Replay streams + +When the last message has `replay=True`, `ai.stream` does not call the provider. +It emits synthetic replay tool-end events from the existing assistant message +so resume flows can dispatch the same tool calls again. diff --git a/docs/ai-python/content/docs/core-framework/tool-dispatch.mdx b/docs/ai-python/content/docs/core-framework/tool-dispatch.mdx new file mode 100644 index 00000000..22344e6a --- /dev/null +++ b/docs/ai-python/content/docs/core-framework/tool-dispatch.mdx @@ -0,0 +1,95 @@ +--- +title: Tool Dispatch +description: Understand tool binding, validation, scheduling, and results. +type: conceptual +summary: Learn how tool schemas become executable calls and how tool results flow back into history. +--- + +Tool dispatch starts with a streamed `ToolCallPart` and ends with a +`ToolResultPart` in message history. + +## Tool schema creation + +`@ai.tool` inspects the async function signature and creates a Pydantic +validator plus a model-facing `Tool` declaration: + +```python +@ai.tool +async def contact_mothership(query: str) -> str: + """Contact the mothership for important decisions.""" + return "Soon." +``` + +The tool name is the function name. The docstring becomes the description. + +## Bound tool calls + +When the model emits a tool call, `context.resolve` binds the streamed +`ToolCallPart` to the registered `AgentTool`: + +```python +if isinstance(event, ai.events.ToolEnd): + tool_call = context.resolve(event.tool_call) +``` + +The bound call exposes `id`, `name`, `fn`, and validated `kwargs`. + +## Argument validation + +Arguments are JSON-decoded from `tool_args` and validated before the function +runs: + +```python +kwargs = tool_call.kwargs +``` + +Validation failures return an error tool result, so the agent can continue with +the error in history. + +## Concurrent scheduling + +`ToolRunner.schedule` starts each call in a task group. `ToolRunner.events()` +yields results as tasks finish: + +```python +tool_runner.schedule(context.resolve(event.tool_call)) +``` + +This lets multiple tool calls from one assistant turn run concurrently. + +## Tool result aggregation + +Each successful call creates a `ToolResultPart`. `ToolRunner.get_tool_message` +merges all collected results into one tool message: + +```python +message = tool_runner.get_tool_message() +context.add(message) +``` + +## Error results + +Tool exceptions are caught and converted to error results: + +```python +if isinstance(event, ai.events.ToolCallResult) and event.exception: + log_exception(event.exception) +``` + +The model receives a string error result. The event keeps the original +exception for logging. + +## Approval-gated dispatch + +Tools declared with `require_approval=True` are wrapped in an approval hook. +The call runs only after the hook resolves with `granted=True`: + +```python +@ai.tool(require_approval=True) +async def notify_mothership(message: str) -> str: + """Notify the mothership.""" + return f"Sent: {message}" +``` + +If the hook is denied, the agent returns an error tool result. If the run aborts +while the hook is pending, the pending result marks the history for replay. diff --git a/docs/ai-python/content/docs/index.mdx b/docs/ai-python/content/docs/index.mdx index e5b8f531..7c894ab9 100644 --- a/docs/ai-python/content/docs/index.mdx +++ b/docs/ai-python/content/docs/index.mdx @@ -1,8 +1,8 @@ --- title: Getting Started -description: Build LLM-powered apps and agents in Python with the AI SDK. +description: Build LLM-powered apps and agents. type: guide -summary: Install the AI SDK, configure provider credentials, and stream your first agent run. +summary: Install the AI SDK and stream your first agent run. --- The AI SDK for Python is a toolkit for building large language model (LLM) @@ -14,7 +14,6 @@ async Python. - **Python 3.12 or later.** - **uv**, **pip**, or another Python dependency manager. -- **An AI Gateway API key.** ## Install @@ -22,12 +21,6 @@ async Python. uv add ai ``` -Set your AI Gateway API key: - -```bash title="Terminal" -export AI_GATEWAY_API_KEY=your_api_key_here -``` - Then import the package in Python: ```python @@ -113,7 +106,7 @@ After iteration, `s.message`, `s.text`, `s.tool_calls`, `s.output`, and ## What's next -- **Core concepts**: Learn the small set of primitives that shape the SDK. +- **Core concepts**: Learn the primitives that shape the SDK. - **Streaming**: Stream model responses and inspect events. - **Agents**: Add tools, customize the loop, and handle approvals. - **Samples**: Focused, single-file examples live in diff --git a/docs/ai-python/content/docs/meta.json b/docs/ai-python/content/docs/meta.json index 2ce96e43..54d6837f 100644 --- a/docs/ai-python/content/docs/meta.json +++ b/docs/ai-python/content/docs/meta.json @@ -4,8 +4,7 @@ "root": true, "pages": [ "index", - "concepts", - "streaming", - "agents" + "basics", + "core-framework" ] }