# AG2 - Full Documentation > AG2 (`ag2`) is an async, protocol-driven Python framework for building AI agents - covering agents, tools, multi-agent networks, structured output, memory, and evaluation. This file indexes the AG2 documentation for LLMs and coding assistants. Build with `ag2` only. The classic `autogen` API (`ConversableAgent`, `initiate_chat`, `GroupChat`) has been removed - do not use it. For ready-made setup, install the AG2 Skills with `npx skills add ag2ai/ag2-skills`. --- # AG2 Source: https://docs.ag2.ai/docs/user-guide/motivation/ ## Why did we create AG2? The original **AutoGen** project released with its first public preview in September 2023, and **AG2** later diverged from that codebase in November 2024 to continue building on its core ideas. **AutoGen** was one of the earliest frameworks for building AI agents and orchestrating agent-to-agent collaboration. That early vision proved valuable: it enabled real-world systems, informed the design of many tools, and helped shape the agent ecosystem. Since then, the agent landscape has changed significantly. Over time, the community has established better practices, common protocols, and new interoperability standards. Capabilities that were once experimental are now becoming part of the expected foundation for agent platforms. Examples include: - [Model Context Protocol (MCP)](https://www.anthropic.com/news/model-context-protocol), introduced in November 2024 - [Agent2Agent (A2A)](https://developers.googleblog.com/en/a2a-a-new-era-of-agent-interoperability/), introduced in April 2025 - [AG-UI](https://docs.ag-ui.com/introduction), introduced in May 2025 We have increasingly found that the original architecture inherited from **AutoGen** challenged the adoption of new ideas. Shipping modern capabilities inside the original design often requires introducing complexity, unnecessary migration effort, or compatibility compromises. Not every part of the ecosystem is standardized yet, but the direction is clear. AI agents are no longer an experiment; they are standard application infrastructure. Today's **AG2** is our way to move forward with a future-focused foundation, applying the lessons we learned from building and operating hundreds of agent systems on the original framework. ## What is AG2? **AG2** is a protocol-driven, async agent framework built around a small, predictable core and a set of opt-in primitives. It rebuilds the framework around capabilities that were difficult or impractical to introduce on the original architecture. It is the foundation for AG2 agent development and production-ready multi-agent systems going forward, and forms the basis of AG2 v1.0. ## Why use AG2? **AG2** is built around a small, predictable, core and a set of opt-in primitives you compose to fit your application. Here is what you get out of the box. ### 1. A clean, async-first agent API Two methods cover the conversational surface - `agent.ask(...)` to start a turn and `reply.ask(...)` to continue one. The agent loop, tool execution, and LLM calls are async throughout, with streaming enabled by default on supported providers. ### 2. A composable harness for capable, long-running, agents AG2's harness layers on powerful primitives to the base Agent - **assembly policies** for context shaping, a **knowledge store** for persistent memory, **sub-task delegation** with isolated streams, and **middleware** for retries, logging, token limits, and history management. Build up the agent you need with the harness doing the heavy lifting for you. The Agent's ability to fan out work, in parallel, to a team of specialist agents (seen as tools), or as subtasks, provides natural orchestration within each agent. ### 3. Production and Scalability **Human-in-the-loop** hooks, **structured output** (static, callable, prompted, and transformable), **OpenTelemetry** tracing, **persistent backends** for history and streams (e.g. Redis), and a **testing utility** that mocks LLMs and tool calls without hitting the network - the primitives you need to take an Agent from prototype to production. The runtime is async end-to-end, so a single process can drive many concurrent agents, tool calls, and provider streams without blocking, and sub-tasks fan out in parallel via `asyncio.gather`. State is externalised behind protocols - `History`, `Storage`, and `Stream` can be backed by Redis, a database, or anything you build - so agents stay effectively stateless and horizontal scaling is straightforward. Cross-cutting concerns like retries, rate limits, token budgets, and history compaction are middleware you compose onto an Agent. ### 4. UI and external integration Every event in the agent loop - model requests and responses, tool calls and results, human-input requests, observer alerts - flows through an event stream. Streams, with natural filtering capabilities, can power UIs, logging, metrics, or approvals without touching the agent itself. The stream is bidirectional: **AG-UI** renders model output and tool calls in real time while user responses come back as `HumanMessage` events, and persistent backends like Redis let separate processes - a web frontend and a background worker - share the same live conversation. ### 5. Tools, toolkits, and built-in tools Define tools with a `@tool` decorator on plain functions. Use **type hints**, **dependency injection** (`Context`, `Inject`, `Variable`), and **toolkits** to organize related capabilities. Wire in **built-in tools** (web search, code execution, shell, memory) or expose any agent as a tool with `Agent.as_tool(...)`. ### 6. One configuration model across providers A single, type-safe, interface spans **OpenAI**, **OpenAI Responses**, **Anthropic**, **Gemini**, **Vertex AI**, **Ollama**, and **DashScope**. Switching providers is a config change, not a rewrite, and structured output, multimodality (images, audio, video), and built-in tools work consistently across them. ## Migrating from earlier AG2 We value all the users and contributors who have made AG2 what it is today and want to bring you along on this journey. If you are coming from the earlier `ConversableAgent` / `GroupChat` API, the [network migration guide](network/migration_from_group_chat.md) maps the classic orchestration patterns onto the new `ag2.network` module. ## How do I try it out? Install with `pip install ag2`. The framework is importable as the `ag2` module. For the latest in-development version, use the `main` branch of the [GitHub repository](https://github.com/ag2ai/ag2). !!! tip "Using an AI coding assistant?" If you build with Claude Code, Cursor, Copilot, or another AI coding assistant, see [Coding with AI Assistants](coding_with_ai.md) to set it up with AG2 skills and project rules so it writes against the current `ag2` API. See the following pages for walkthroughs of the AG2 API. ## Current Focus Areas **AG2** is actively focused on: - improving the single-agent developer experience - providing stronger context and memory management primitives - simplifying integration with real applications, including Text UI, web, ambient, and background runtimes - enabling new multi-agent coordination patterns that are not feasible in the current AG2 architecture - supporting emerging standards and protocols across the AI agent ecosystem We are building **AG2** to make agent development simpler, more modern, and easier to integrate into production-grade applications. We would love your feedback as the API evolves ([Discord](https://discord.com/invite/pAbnFJrkgZ)). --- # Quick Start Source: https://docs.ag2.ai/docs/user-guide/quick-start/ Get up and running with **AG2** in a few minutes. You'll install the framework, configure a model provider, and build your first agent - then give it a tool. ## 1. Install AG2 Install AG2 with the extra for your model provider: === "OpenAI" ```bash pip install "ag2[openai]" ``` === "Anthropic" ```bash pip install "ag2[anthropic]" ``` === "Gemini" ```bash pip install "ag2[gemini]" ``` Then export your provider's API key: === "OpenAI" ```bash export OPENAI_API_KEY="your-api-key" ``` === "Anthropic" ```bash export ANTHROPIC_API_KEY="your-api-key" ``` === "Gemini" ```bash export GEMINI_API_KEY="your-api-key" ``` ## 2. Build your first agent AG2 is async throughout, so the example runs inside an `async` function driven by `asyncio.run`. `Agent.ask(...)` starts a turn and returns an `AgentReply`; read the text with `reply.body`. ```python import asyncio from ag2 import Agent from ag2.config import OpenAIConfig agent = Agent( "assistant", prompt="You are a helpful assistant.", config=OpenAIConfig("gpt-4o-mini"), ) async def main() -> None: reply = await agent.ask("Give me one sentence about AG2.") print(reply.body) asyncio.run(main()) ``` To continue the same conversation, call `ask` again on the reply - it preserves context and history: ```python async def main() -> None: reply = await agent.ask("Give me one sentence about AG2.") follow_up = await reply.ask("Now make it shorter.") print(follow_up.body) ``` ## 3. Give your agent a tool Decorate a plain Python function with `@tool` and pass it to the agent. AG2 manages the full tool-calling lifecycle - the model decides when to call it, AG2 executes it, and feeds the result back. ```python import asyncio from ag2 import Agent, tool from ag2.config import OpenAIConfig @tool async def get_weather(city: str) -> str: """Return the current weather for a city.""" return f"It's sunny in {city}." agent = Agent( "assistant", prompt="Use tools when helpful.", config=OpenAIConfig("gpt-4o-mini"), tools=[get_weather], ) async def main() -> None: reply = await agent.ask("What's the weather in Paris?") print(reply.body) asyncio.run(main()) ``` ## Next steps - [Agents](agents.md) - the full agent communication API, streaming, and live runs. - [Models & Providers](model_configuration.md) - configure OpenAI, Anthropic, Gemini, and more. - [Tools](tools/tools.md) - tools, toolkits, and built-in tools. - [Multi-Agent Network](network/overview.md) - coordinate multiple agents. - [Coding with AI Assistants](coding_with_ai.md) - set up your AI assistant to build with AG2. --- # Agent Communication Source: https://docs.ag2.ai/docs/user-guide/agents/ Agents are the central primitive in **AG2**. They maintain state, interact with models, execute tools, and handle user interactions through a clean, conversation-focused API. ## Core Communication Primitives The API is built around a few simple methods: * `Agent.ask(...)` initiates a new turn and **blocks** until it finishes, returning an `AgentReply`. * `AgentReply.ask(...)` continues an existing conversation, preserving its context and history. * `Agent.run(...)` / `AgentReply.run(...)` are the **observable** counterparts: they open a turn you can watch live and drive on demand, returning an `AgentRun` handle. See [Watching a Turn Live with `run`](#watching-a-turn-live-with-run) below. The final result of any turn is safely stored in `reply.response`; use `reply.body` for the text. ## Basic Communication Example Here's how easily you can start and continue a conversation: ```python from ag2 import Agent from ag2.config import OpenAIConfig agent = Agent( "assistant", prompt="You are a helpful assistant.", config=OpenAIConfig("gpt-4o-mini"), ) # Start a new conversation reply = await agent.ask("Give me one sentence about AG2.") print(reply.body) # Continue the exact same conversation context next_turn = await reply.ask("Now make it shorter.") print(next_turn.body) ... ``` ## Empowering Agents with Tools Agents can seamlessly use Python functions as tools. When you provide a list of `@tool`-decorated functions to an agent, it automatically manages the entire execution lifecycle (model requests to execution and returning results). ```python from ag2 import Agent, Context, tool from ag2.config import OpenAIConfig @tool async def echo(text: str) -> str: """Useful for repeating exactly what was given.""" return f"echo: {text}" agent = Agent( "assistant", prompt="Use tools when helpful.", config=OpenAIConfig("gpt-4o-mini"), tools=[echo], ) reply = await agent.ask("Call the echo tool with 'hello'.") print(reply.body) ``` ## Adding Human-in-the-Loop (HITL) Sometimes an agent needs human guidance. You can configure an agent to handle `HumanInputRequest` events. This is especially effective inside tools where you can get confirmation before taking a sensitive action. ```python from ag2 import Agent, Context, tool from ag2.config import OpenAIConfig from ag2.events import HumanInputRequest, HumanMessage @tool async def ask_human(context: Context) -> str: # Pauses agent execution to await human input answer = await context.input("Please provide confirmation:") return f"Human said: {answer}" # Define how your application handles the input request def hitl_hook(event: HumanInputRequest) -> HumanMessage: # Here you could block and wait for UI/CLI input. # We return a static response for demonstration. return HumanMessage(content="confirmed") agent = Agent( "assistant", prompt="Use ask_human when needed.", config=OpenAIConfig("gpt-4o-mini"), tools=[ask_human], hitl_hook=hitl_hook, ) reply = await agent.ask("Request confirmation through the tool.") print(reply.body) ``` ## Observing Agent Actions Need to know exactly what the agent is doing? Pass a `MemoryStream` when calling `ask()`. You can attach event subscribers to log actions, save history to a database, or update a user interface in real time. ```python from ag2 import Agent, Context, MemoryStream from ag2.events import BaseEvent, ModelResponse, ToolCallEvent from ag2.config import OpenAIConfig stream = MemoryStream() # Listen to everything @stream.subscribe() async def on_any_event(event: BaseEvent) -> None: print(f"Event occurred: {event}") # Only listen to specific events @stream.where(ToolCallEvent).subscribe() async def on_tool_call(event: ToolCallEvent) -> None: print("Agent requested tool:", event.name) agent = Agent( "assistant", prompt="You are a helpful assistant.", config=OpenAIConfig("gpt-4o-mini"), ) # Stream captures all events during the ask reply = await agent.ask( "Give me one sentence about AG2.", stream=stream ) ``` ## Watching a Turn Live with `run` `ask()` blocks until the whole turn is done. When you want to *watch* a turn unfold - stream tokens to a UI as they arrive, or steer the agent mid-turn - use `run()` instead. It returns an `AgentRun` async context manager. The turn does **not** advance on its own. Call `run.start()` to drive it in the background, then read `run.stream.join()` - an async iterator of live events - at the same time. `await run.result()` returns the turn's final `AgentReply`. ### Iterating over events while the turn runs ```python from ag2 import Agent from ag2.config import OpenAIConfig agent = Agent( "assistant", config=OpenAIConfig("gpt-4o-mini", streaming=True), ) async with agent.run("Tell me about Paris in two sentences.") as run: run.start() # drive the turn in the background with run.stream.join() as events: async for event in events: print(event) # every event the turn emits reply = await run.result() print("\n", reply.body) ``` How it behaves: * `run.start()` drives the turn in a scope-owned background task; it returns immediately (it is **not** awaited). Without it, `run.stream.join()` would block forever - nothing advances the turn. * `run.stream.join()` yields every event the turn emits; * **Token streaming needs `streaming=True` on the model config** (as above); the model then emits `ModelMessageChunk` events as it generates. * `await run.result()` returns the same `AgentReply` the started turn produced (idempotent, re-raising the same failure on retry). Leaving the block without awaiting it cancels a still-running turn. * To drive **inline** instead, skip `start()` and `await run.result()` directly; cancelling that await - e.g. `await asyncio.wait_for(run.result(), timeout=5)` - cancels the turn. !!! warning "Do not timeout individual `join()` items" Consume `run.stream.join()` with a plain `async for`, or use `join(max_events=N)` when you know how many events you need. Avoid wrapping individual pulls such as `await asyncio.wait_for(events.__anext__(), timeout=...)`: cancelling one `__anext__()` closes the current iterator, so later events will not be yielded by that iterator. If you need a time limit, put one `asyncio.timeout(...)` around the whole loop instead of around each item. `AgentReply.run(...)` is the same thing for a **continuation**: it watches a follow-up turn on an existing conversation, just as `AgentReply.ask(...)` continues one with a blocking call. !!! note "`ask` is `run` you don't have to drive" `await agent.ask(...)` is exactly `run()` + `await result()` rolled into one call. Reach for `ask` when you just want the answer, and `run` when you need to observe or steer the turn. ### Feeding the turn while it runs A running turn keeps an **inbox**. Call `run.enqueue(...)` to push a follow-up message into it - from a concurrent task or while iterating events - and the turn consumes it at its next model call, without starting a new turn. Here we wait for the first tool result, then steer the turn: ```python from ag2 import Agent, tool from ag2.events import ToolResultEvent from ag2.config import OpenAIConfig @tool async def search(query: str) -> str: """Looks up a query.""" return f"results for {query}" agent = Agent( "assistant", prompt="Use the search tool when helpful.", config=OpenAIConfig("gpt-4o-mini"), tools=[search], ) async with agent.run("Search for AG2.") as run: run.start() with run.stream.where(ToolResultEvent).join(max_events=1) as results: async for _ in results: run.enqueue("Now summarize the result in one line.") # lands at the next model call reply = await run.result() print(await reply.content()) ``` `enqueue` is non-blocking - it only appends to the inbox. When the message is consumed depends on timing: * **before** `result()` -> merged into the turn's first model call; * **while the turn is running** -> consumed by that same turn's next or final model call; * **after** the turn finishes -> waits for the next turn on this stream. !!! tip "Building with an AI coding assistant?" See [Coding with AI Assistants](coding_with_ai.md) to set up Claude Code, Cursor, Copilot, or another assistant with AG2 skills and project rules so it writes against the current `ag2` API. --- # Model Configuration Source: https://docs.ag2.ai/docs/user-guide/model_configuration/ # Model Configuration The AG2 framework provides an explicit, predictable, and type-safe way to configure Large Language Models (LLMs) for your agents. The configuration API is designed to provide a consistent developer experience across different model providers while maintaining strong typing support. ## Supported Providers AG2 supports multiple LLM providers through dedicated configuration classes. Each provider requires its respective optional dependencies to be installed. | Provider | Configuration Class | Installation Command | | :--- | :--- | :--- | | **[OpenAI Responses](https://developers.openai.com/api/reference/responses/overview)** | `OpenAIResponsesConfig` | `pip install "ag2[openai]"` | | **[OpenAI](https://developers.openai.com/api/reference/overview)** | `OpenAIConfig` | `pip install "ag2[openai]"` | | **[Anthropic](https://platform.claude.com/docs/en/build-with-claude/overview)** | `AnthropicConfig` | `pip install "ag2[anthropic]"` | | **[Gemini](https://ai.google.dev/gemini-api/docs)** | `GeminiConfig` | `pip install "ag2[gemini]"` | | **[Gemini on Vertex AI](https://cloud.google.com/vertex-ai/generative-ai/docs)** | `VertexAIConfig` | `pip install "ag2[gemini]"` | | **[Amazon Bedrock](https://docs.aws.amazon.com/bedrock/latest/userguide/conversation-inference.html)** | `BedrockConfig` | `pip install "ag2[bedrock]"` | | **[Ollama](https://docs.ollama.com/api/introduction)** | `OllamaConfig` | `pip install "ag2[ollama]"` | | **[DashScope](https://www.alibabacloud.com/help/en/model-studio/first-api-call-to-qwen)** | `DashScopeConfig` | `pip install "ag2[dashscope]"` | | **[xAI](https://docs.x.ai/docs/overview)** | `XAIConfig` | `pip install "ag2[xai]"` | | **[Z.AI](https://docs.z.ai)** | `ZAIConfig` | `pip install "ag2[zai]"` | | **[Mistral](https://docs.mistral.ai)** | `MistralConfig` | `pip install "ag2[mistral]"` | *(Note: `OpenAIConfig` is also available for OpenAI-compatible endpoints).* --- ## How to Configure a Model ### Basic Configuration To configure a model, import the specific provider's configuration class and initialize it with your desired parameters. The most common parameters are `model`, `api_key`, and `base_url`. === "OpenAI Responses" ```python linenums="1" from ag2.config import OpenAIResponsesConfig # Configure an OpenAI Responses API model config = OpenAIResponsesConfig( model="gpt-4.1-nano", api_key="sk-...", streaming=True ) ``` === "OpenAI" ```python linenums="1" from ag2.config import OpenAIConfig # Configure an OpenAI model config = OpenAIConfig( model="gpt-4o-mini", api_key="sk-...", temperature=0.2, streaming=True ) ``` === "Anthropic" ```python linenums="1" from ag2.config import AnthropicConfig # Configure an Anthropic model config = AnthropicConfig( model="claude-haiku-4-5-20251001", api_key="sk-ant-...", streaming=True ) ``` `ag2[anthropic]` requires `anthropic>=1.2.0,<2`. See [Anthropic Configuration](#anthropic-configuration) for the `httpx2` HTTP client and the sampling parameters (`temperature`, `top_p`, `top_k`). === "Gemini" ```python linenums="1" from ag2.config import GeminiConfig # Configure a Gemini model config = GeminiConfig( model="gemini-3-flash-preview", api_key="...", streaming=True ) ``` === "Amazon Bedrock" ```python linenums="1" from ag2.config import BedrockConfig # Configure an Amazon Bedrock model (Converse API) config = BedrockConfig( model="global.anthropic.claude-sonnet-5", region_name="us-east-1", streaming=True ) ``` Credentials follow the standard AWS resolution chain: explicit `aws_access_key_id` / `aws_secret_access_key`, a named `profile_name`, environment variables, shared config files, or instance roles. `model` accepts a Bedrock model id or an inference-profile ARN. See [Amazon Bedrock authentication](#amazon-bedrock-authentication) for the API-key (bearer token) alternative. === "Ollama" ```python linenums="1" from ag2.config import OllamaConfig # Configure an Ollama model config = OllamaConfig( model="qwen3.5:latest", streaming=True ) ``` === "DashScope" ```python linenums="1" from ag2.config import DashScopeConfig # Configure a DashScope model config = DashScopeConfig( model="qwen-plus", api_key="...", streaming=True ) ``` === "xAI" ```python linenums="1" from ag2.config import XAIConfig # Configure an xAI Grok model config = XAIConfig( model="grok-4", api_key="xai-...", streaming=True ) ``` === "Z.AI" ```python linenums="1" from ag2.config import ZAIConfig # Configure a Z.AI GLM model config = ZAIConfig( model="glm-4.6", api_key="...", streaming=True ) ``` === "Mistral" ```python linenums="1" from ag2.config import MistralConfig # Configure a Mistral model config = MistralConfig( model="mistral-large-latest", api_key="...", streaming=True ) ``` !!! tip **AG2** is designed to be async and streaming-first, so for the best user experience it is recommended to enable streaming on the model provider configurations for models that support it. As shown above, `Streaming` has been set to `True` in each config. ### Using Environment Variables For security and convenience, you don't need to hardcode your API keys. If `api_key` is not explicitly provided, the configuration will automatically attempt to load it from your environment variables. The system looks for provider-specific keys (e.g., `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `GEMINI_API_KEY`, `XAI_API_KEY`, `ZAI_API_KEY`). ```python from ag2.config import OpenAIConfig # Automatically falls back to OPENAI_API_KEY from the environment config = OpenAIConfig(model="gpt-5") ``` ### Anthropic Configuration `AnthropicConfig` connects to the [Claude API](https://platform.claude.com/docs/en/build-with-claude/overview) through the official `anthropic` SDK. Install the optional dependency with `pip install "ag2[anthropic]"`, which requires `anthropic>=1.2.0,<2`. If `api_key` is not passed explicitly, it is resolved from the `ANTHROPIC_API_KEY` environment variable. ```python from ag2.config import AnthropicConfig # Auth resolved from ANTHROPIC_API_KEY when omitted config = AnthropicConfig(model="claude-haiku-4-5-20251001", streaming=True) ``` #### Custom HTTP Client `anthropic>=1` is built on [`httpx2`](https://pypi.org/project/httpx2/), so `http_client` takes an `httpx2.AsyncClient`: ```python import httpx2 from ag2.config import AnthropicConfig config = AnthropicConfig( model="claude-haiku-4-5-20251001", api_key="sk-ant-...", http_client=httpx2.AsyncClient(proxy="http://proxy.example.com:8080"), ) ``` !!! warning "An `httpx` client is rejected" The `anthropic` SDK keeps no compatibility path for a legacy `httpx.AsyncClient` - it raises `TypeError` as soon as the client is built, where the OpenAI SDK would still accept one. Rebuild yours against `httpx2`; the two packages share an API, so this is usually just the import line. #### Sampling Parameters `temperature`, `top_p` and `top_k` are ordinary fields on `AnthropicConfig`. The 1.x Messages API dropped them from its method signature, so AG2 sends them in the request body, where the API still reads them: ```python from ag2.config import AnthropicConfig config = AnthropicConfig( model="claude-haiku-4-5-20251001", api_key="sk-ant-...", temperature=0.2, ) ``` !!! warning "Recent Claude models reject sampling parameters" Claude Fable 5, Opus 5, Opus 4.8, Opus 4.7 and Sonnet 5 removed `temperature`, `top_p` and `top_k` from the Messages API - a request carrying any of them is answered with HTTP 400. Leave all three unset (the default) on those models. Claude Opus 4.6, Sonnet 4.6, Haiku 4.5 and earlier still accept them. !!! note "`extra_body` precedence" Pass `extra_body` to forward provider-specific keys that have no dedicated field. A key you write there wins over the same key AG2 derived from a field - an `extra_body={"temperature": 0.9}` overrides `temperature=0.2` - while unrelated keys are merged through untouched. ### Amazon Bedrock Authentication `BedrockConfig` authenticates in either of two ways, both resolved by the underlying AWS SDK: **1. AWS credentials (SigV4)** - explicit keys, a `profile_name`, or the standard environment variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN`). Recommended for production; credentials refresh automatically through botocore. **2. Bedrock API keys (bearer token)** - set the [Amazon Bedrock API key](https://docs.aws.amazon.com/bedrock/latest/userguide/api-keys.html) as an environment variable and boto3 uses bearer-token auth for Bedrock calls automatically (no other credentials needed): ```bash export AWS_BEARER_TOKEN_BEDROCK= export AWS_DEFAULT_REGION=us-east-1 ``` ```python from ag2.config import BedrockConfig # Auth from AWS_BEARER_TOKEN_BEDROCK, region from AWS_DEFAULT_REGION config = BedrockConfig(model="global.anthropic.claude-sonnet-5") ``` A region is always required - pass `region_name=` or set `AWS_DEFAULT_REGION`. Notes on API keys: - **Short-term keys** expire with the console session that minted them (max 12 hours) and are region-bound - generate the key in the same region you call. - **Long-term keys** are backed by an auto-created IAM user; AWS recommends them for exploration only. - API keys work only for Bedrock / Bedrock Runtime actions. If both a bearer token and AWS credentials are present, the bearer token wins for Bedrock calls. ### Google Vertex AI (Gemini) For Gemini on **Vertex AI** (Google Cloud), use the dedicated `VertexAIConfig` class. `GeminiConfig` covers the public Developer API (`api_key`); `VertexAIConfig` covers the Vertex path (GCP `project`, `location`, and Google-issued credentials). Authentication accepts any of the following: === "Service account key file" ```python linenums="1" hl_lines="7" from ag2.config import VertexAIConfig config = VertexAIConfig( model="gemini-3-flash-preview", project="my-gcp-project", location="us-central1", credentials="/path/to/service-account-key.json", # Path to a service-account JSON key file downloaded from # GCP Console -> IAM & Admin -> Service Accounts -> Keys. ) ``` The service account needs the **Vertex AI User** (`roles/aiplatform.user`) IAM role on the project. === "Application Default Credentials" ```python linenums="1" from ag2.config import VertexAIConfig # Run `gcloud auth application-default login` first, or ensure # GOOGLE_APPLICATION_CREDENTIALS points to a key file. With nothing # passed to `credentials`, google-genai resolves ADC automatically. config = VertexAIConfig( model="gemini-3-flash-preview", project="my-gcp-project", location="us-central1", ) ``` === "Pre-built Credentials object" ```python linenums="1" hl_lines="4-6 12" import google.auth from ag2.config import VertexAIConfig creds, _ = google.auth.default( scopes=["https://www.googleapis.com/auth/cloud-platform"], ) config = VertexAIConfig( model="gemini-3-flash-preview", project="my-gcp-project", location="us-central1", credentials=creds, ) ``` Use this path for impersonated credentials, workload identity, or any other `google.auth.credentials.Credentials` source. #### Environment variables Instead of passing parameters explicitly, the underlying `google-genai` SDK resolves any field left unset from the following environment variables: | Environment variable | Used by | Equivalent parameter | Notes | | :--- | :--- | :--- | :--- | | `GOOGLE_API_KEY` | `GeminiConfig` | `api_key` | Takes precedence over `GEMINI_API_KEY` if both are set. | | `GEMINI_API_KEY` | `GeminiConfig` | `api_key` | Developer API key. | | `GOOGLE_CLOUD_PROJECT` | `VertexAIConfig` | `project` | GCP project ID. | | `GOOGLE_CLOUD_LOCATION` | `VertexAIConfig` | `location` | GCP region (or `global`). | | `GOOGLE_APPLICATION_CREDENTIALS` | `VertexAIConfig` | `credentials` | Path to a service-account JSON key file, read via ADC. | With the three Vertex variables set in the environment, configuration collapses to just the model name: ```bash export GOOGLE_CLOUD_PROJECT=my-gcp-project export GOOGLE_CLOUD_LOCATION=us-central1 export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json ``` ```python from ag2.config import VertexAIConfig # All Vertex auth parameters resolved from the environment. config = VertexAIConfig(model="gemini-3-flash-preview") ``` ### Controlling Gemini Thinking Gemini 3 Pro models default to **dynamic / unbounded** thinking, which can cause individual calls to spend a large internal token budget before responding. Both `GeminiConfig` and `VertexAIConfig` accept thinking controls that map directly to [Google's Thinking API](https://ai.google.dev/gemini-api/docs/thinking). Use `thinking_level` for **Gemini 3** models, or `thinking_budget` for **Gemini 2.5** models: ```python from ag2.config import GeminiConfig, VertexAIConfig # Gemini 3 - bound thinking with a level gemini3 = GeminiConfig( model="gemini-3-flash-preview", thinking_level="low", # "low" | "medium" | "high" ) # Gemini 2.5 - bound thinking with an explicit token budget gemini25 = VertexAIConfig( model="gemini-2.5-pro", project="my-gcp-project", location="us-central1", thinking_budget=512, # 0 disables thinking entirely ) ``` For full control (e.g. enabling `include_thoughts`), pass a `google.genai.types.ThinkingConfig` directly via `thinking_config`. When set, it takes precedence over the shorthand fields. The number of thinking tokens consumed is reported on `ModelResponse.usage.thinking_tokens` and emitted as the `gen_ai.usage.thinking_tokens` OpenTelemetry attribute by `TelemetryMiddleware`. ### Z.AI (GLM) Configuration `ZAIConfig` connects to Z.AI's [GLM models](https://docs.z.ai) through the official `zai-sdk`. Install the optional dependency with `pip install "ag2[zai]"`. If `api_key` and `base_url` are not passed explicitly, the SDK resolves them from the `ZAI_API_KEY` and `ZAI_BASE_URL` environment variables; `base_url` otherwise defaults to the international endpoint (`https://api.z.ai/api/paas/v4`). ```python from ag2.config import ZAIConfig # Auth and endpoint resolved from ZAI_API_KEY / ZAI_BASE_URL when omitted config = ZAIConfig(model="glm-4.6", streaming=True) ``` #### Controlling GLM Thinking GLM models accept reasoning controls that map directly to the Z.AI API. Set `thinking` to toggle the reasoning pass (`True` enables it, `False` disables it, unset uses the model default) and `reasoning_effort` to bound how much the model thinks: ```python from ag2.config import ZAIConfig config = ZAIConfig( model="glm-4.6", thinking=True, reasoning_effort="high", streaming=True, ) ``` Reasoning content is streamed as `ModelReasoning` events, and reasoning tokens are reported on `ModelResponse.usage.thinking_tokens`. !!! note "`extra_body` precedence" Pass `extra_body` to forward provider-specific keys that have no dedicated field. Explicit top-level fields always win over the same key inside `extra_body` - e.g. a top-level `thinking=True` overrides a `thinking` entry inside `extra_body`, while unrelated `extra_body` keys are merged through untouched. ### Mistral Configuration `MistralConfig` connects to [Mistral's](https://docs.mistral.ai) chat-completions API through the official `mistralai` SDK. Install the optional dependency with `pip install "ag2[mistral]"`. If `api_key` is not passed explicitly, the SDK resolves it from the `MISTRAL_API_KEY` environment variable. ```python from ag2.config import MistralConfig # Auth resolved from MISTRAL_API_KEY when omitted config = MistralConfig(model="mistral-large-latest", streaming=True) ``` Point `server_url` at a different deployment (for example a private or regional endpoint) when you are not calling `https://api.mistral.ai`. #### Reasoning Models Magistral and other reasoning models return their thinking trace alongside the answer. AG2 splits the two: the trace is emitted as `ModelReasoning` events, and only the answer text lands on `ModelResponse.message`. ```python from ag2.config import MistralConfig config = MistralConfig( model="magistral-medium-latest", reasoning_effort="high", streaming=True, ) ``` #### Image Generation `ImageGenerationTool` is executed by Mistral, not locally. The call and its result are surfaced as `BuiltinToolCallEvent` and `BuiltinToolResultEvent`, and the generated image arrives as a `UrlInput`: ```python from ag2 import Agent from ag2.config import MistralConfig from ag2.tools.builtin.image_generation import ImageGenerationTool agent = Agent( "illustrator", config=MistralConfig(model="mistral-medium-latest"), tools=[ImageGenerationTool()], ) reply = await agent.ask("Draw a red circle over a black square.") ``` The image URL is a short-lived signed link, so download it promptly if you need to keep it. Mistral's tool takes no options, so `size`, `quality`, and `output_format` are ignored. The model may reply with the image and no commentary, leaving `reply.body` empty - the image is on the tool result either way. !!! note "Other builtin tools" Apart from image generation, Mistral's server-side tools (`web_search`, `code_interpreter`, `document_library`) belong to its Agents API and are rejected by chat-completions. Passing those AG2 builtin tools raises `UnsupportedToolError` rather than failing at request time. !!! note "OpenTelemetry version ceiling" The `mistralai` SDK pins `opentelemetry-semantic-conventions<0.61`, which transitively caps `opentelemetry-api` at 1.39.1. A clean install of `ag2[mistral,tracing]` resolves to a consistent OpenTelemetry 1.39.x stack and traces normally. Adding `ag2[mistral]` to an environment that *already* has a newer OpenTelemetry downgrades `opentelemetry-api` on its own, leaving it mismatched against the newer SDK. Reinstall the pair together, or hold the newer versions explicitly - the SDK itself works fine with current OpenTelemetry, the pin is just over-tight: ```bash pip install "mistralai>=2.8.0" "opentelemetry-api>=1.43" "opentelemetry-sdk>=1.43" "opentelemetry-semantic-conventions>=0.64b0" ``` ### Self-Hosted and OpenAI-Compatible Models (vLLM, LM Studio, etc.) If you are using a self-hosted model or an API that is compatible with the OpenAI format (such as **vLLM**, **LM Studio**, **FastChat**, or **Together AI**), you can use the `OpenAIConfig` class and specify a custom `base_url`. ```python from ag2.config import OpenAIConfig # Configure a vLLM or other OpenAI-compatible endpoint config = OpenAIConfig( model="qwen-3", base_url="http://localhost:8000/v1", # Some endpoints don't require an API key, but the client expects a non-empty string api_key="NotRequired", ) ``` !!! tip If you are running a self-hosted server via HTTPS without a valid SSL certificate (e.g., a local self-signed certificate), you can disable SSL checks by passing a custom `httpx2.AsyncClient` with `verify=False` to the configuration: ```python linenums="1" hl_lines="8" import httpx2 from ag2.config import OpenAIConfig config = OpenAIConfig( model="qwen-3", base_url="https://localhost:8000/v1", api_key="NotRequired", http_client=httpx2.AsyncClient(verify=False) ) ``` !!! note "`http_client` is an `httpx2` client" From `openai>=3` the OpenAI SDK uses [`httpx2`](https://pypi.org/project/httpx2/), which `ag2[openai]` installs for you. `OpenAIConfig` and `OpenAIResponsesConfig` annotate `http_client` accordingly, and ag2 forwards whatever you pass to the SDK untouched. A legacy `httpx.AsyncClient` still works at runtime: the SDK keeps a compatibility path for it, though it fails static type checking and the SDK's [migration guide](https://github.com/openai/openai-python/blob/main/httpx2.md) calls that path a migration aid that may be discontinued. Build the client with `httpx2.AsyncClient` to be done with it. Each provider takes the client its own SDK takes, so the annotation differs by provider - `AnthropicConfig.http_client`, for instance, follows whichever package the pinned `anthropic` release is built on. !!! warning "TLS certificates come from the operating system" `httpx2` verifies certificates against the **operating system trust store**, not a bundled one. Nothing extra is needed on a normal machine or a standard base image, but a minimal container without system CA certificates will fail to connect. Install the distribution's CA bundle (e.g. `ca-certificates`), or point the client at one explicitly: ```python import httpx2 from ag2.config import OpenAIConfig config = OpenAIConfig(model="gpt-4o", http_client=httpx2.AsyncClient(verify="/path/to/ca-bundle.crt")) ``` `certifi` remains installed - ag2's own core depends on `httpx` - but the OpenAI client no longer reads it. Its presence is not evidence that TLS is configured there. ### Extra Body Parameters Some OpenAI API-compatible providers require additional, provider-specific parameters in the request body. Use the `extra_body` parameter on `OpenAIConfig` to pass these through directly to the API call. This is useful for enabling features like extended thinking on self-hosted or third-party models: ```python from ag2.config import OpenAIConfig # NVIDIA NIM nemotron = OpenAIConfig( model="nvidia/nemotron-3-super-120b-a12b", base_url="https://integrate.api.nvidia.com/v1", extra_body={"chat_template_kwargs": {"thinking": True{{ "}}" }}, ) ``` ## Reusing and Overriding Configurations Model configurations are **immutable**. If you need to reuse a configuration for multiple agents with slight variations (e.g., changing the model version or adjusting the temperature), use the `.copy()` method. This creates a new updated instance without mutating the original configuration. ```python from ag2.agent import Agent from ag2.config import OpenAIConfig base_config = OpenAIConfig(model="gpt-5") agent1 = Agent( "Assistant", # Create a new configuration with updated temperature config=base_config.copy(temperature=0.2), ) agent2 = Agent( "AnotherAssistant", # Create a new configuration with updated model and temperature config=base_config.copy(model="gpt-5-mini", temperature=0.8), ) ``` ## Delaying Model Configuration In many use cases, you may want to separate the logic of defining your agent (tools, system messages, instructions) from configuring the specific model it uses. This allows you to construct an agent once and dynamically provide the model configuration later during execution. You can accomplish this by passing the configuration to the `.ask()` method when interacting with the agent. This is especially useful for applications like web servers where the user might bring their own API key or choose a different model on the fly. ```python from ag2.agent import Agent from ag2.config import OpenAIConfig # Define an agent without an initial model config, # or with a default one you plan to override later agent = Agent( "Assistant", prompt="You are a helpful assistant.", # other tools and settings... ) # Ask the agent, passing the explicit model configuration response = await agent.ask( "Hello!", config=OpenAIConfig( model="gpt-5", api_key="sk-user-specific-key" ) ) ``` !!! warning Providing a configuration or client directly to the `ask()` method completely **overrides** the original model configuration assigned to the agent for that specific turn. ```python linenums="1" hl_lines="6 12" from ag2.agent import Agent from ag2.config import OpenAIConfig agent = Agent( "Assistant", config=OpenAIConfig(model="gpt-5"), ) response = await agent.ask( "Hello!", # overrides the original model configuration config=OpenAIConfig(model="gpt-5-mini") ) ``` --- # Resuming a Turn Source: https://docs.ag2.ai/docs/user-guide/resume/ `agent.ask(...)` starts a turn from a new message on the agent's current stream. `agent.resume(...)` does the same job from a **recorded trajectory**: you hand it a list of past events, and it re-enters the agent loop driven by the **last** event in that list. The trigger can be *any* event (a fresh user message, a tool result, a human reply) so `resume` is the general way to rebuild a conversation from stored state and carry it forward. !!! tip "Most multi-turn chats do **not** need `resume`" To continue a conversation within a running process, just keep using the reply: call `agent.ask(...)` once, then `reply.ask(...)` for each follow-up. That is the normal continuation path - `resume` is **not** a step you run after `ask`. ```python linenums="1" reply = await agent.ask("Plan a trip to Kyoto.") reply = await reply.ask("Make it five days.") # continue - no resume needed ``` Reach for `resume` only when you **can't** hold onto that reply: the process ended, another worker picks the turn up, or you need to drive the loop from a non-message event such as a tool result. See [How it relates to `ask` / `reply.ask`](#how-resume-works) below. ## How `resume` works `resume` takes the full trajectory as positional `events`: ```python await agent.resume(*events, ...) ``` The list is split in two: - **All events except the last** seed the stream's history - the conversation state the model will see. - **The last event** is the **trigger**: the event that drives the next LLM call. It can be any [event](advanced/stream.md) - typically a `ModelRequest` (a new user message) or a `ToolResultsEvent` (a tool result the model should react to). ```python *history, trigger = events # history -> replaces the stream's history (the prefix the model sees) # trigger -> drives the next LLM call (any BaseEvent) ``` Everything after that is identical to `ask`: the agent calls the model, may issue tool calls, and returns an `AgentReply`. The `resume` signature mirrors `ask` exactly - `stream`, `dependencies`, `variables`, `prompt`, `config`, `tools`, `middleware`, `observers`, `response_schema`, and `hitl_hook` all behave the same way. !!! note "How it relates to `ask` / `reply.ask`" [`reply.ask(...)`](agents.md) continues the **same live stream** the turn ran on - you keep the conversation going as long as you still hold the `AgentReply` object in the running process. That stream may itself be durable (a `RedisStream`, say); what `reply.ask` needs is the in-process handle, not where the history happens to be stored. `resume` is for when you no longer have that handle - a process restart, a worker on another machine, a turn rebuilt from a store - so you supply the events yourself and drive the next one. ## Continue a stored conversation The most common use is multi-turn that outlives the process: persist a conversation, load it back later, and continue with a new user message as the trigger. ```python from ag2 import Agent from ag2.config import OpenAIConfig from ag2.events import ModelRequest, TextInput agent = Agent( "assistant", prompt="You are a helpful travel assistant.", config=OpenAIConfig("gpt-4o-mini"), ) # A conversation you stored earlier and just loaded back from your database. past_events = load_conversation("thread-42") # Drive the next turn with a fresh user message as the trigger. trigger = ModelRequest([TextInput("And what about getting there by train?")]) reply = await agent.resume(*past_events, trigger) print(reply.body) ``` This is `reply.ask("...")`, except you drive it from events you reload yourself rather than from a live `AgentReply` handle held in the running process. ## Resume from a tool result A turn can also stop **mid-loop** - the model asked to run a tool whose result is produced elsewhere: a webhook, a queue worker, a long-running job, or a human approving a request. You record the tool call, run the work separately, then hand the result back as the trigger. The model reacts to the result without the tool being re-executed. ```python from ag2 import Agent from ag2.config import OpenAIConfig from ag2.events import ( ModelMessage, ModelRequest, ModelResponse, TextInput, ToolCallEvent, ToolCallsEvent, ToolResultEvent, ToolResultsEvent, ) agent = Agent("support", prompt="Answer using the tool result.", config=OpenAIConfig("gpt-4o-mini")) # The trajectory recorded earlier: the user asked, and the model responded # by requesting a tool call (whose result is not yet known). call = ToolCallEvent(name="lookup_order", arguments='{"id": "A-1001"}', id="call-1") history = [ ModelRequest([TextInput("Where is order A-1001?")]), ModelResponse(message=ModelMessage(""), tool_calls=ToolCallsEvent([call])), ] # The tool result, produced out of band, becomes the trigger. trigger = ToolResultsEvent([ToolResultEvent.from_call(call, "Shipped, arriving Tuesday.")]) # Re-enter the loop: the model sees the history + result and writes the answer. reply = await agent.resume(*history, trigger) print(reply.body) # -> grounded answer using "Shipped, arriving Tuesday." ``` `ToolResultEvent.from_call(call, result)` pairs the result with the original call so the model can match it to its request. Because `resume` re-enters the **live** loop, the model is free to react by calling more tools - those continuation calls execute normally; only the trigger event is replayed. ## Capture, persist, resume A trajectory is just a list of events, so you can store it anywhere and pass it back to `resume` later - even from another process. Pull the events from a live stream with `await stream.history.get_events()`, or build them yourself, as long as the prefix ends at the point you want to continue from. ```python # --- First process: capture the trajectory and persist it --- events = list(await reply.context.stream.history.get_events()) save_to_store(stream_id="thread-42", events=events) # --- Later / another process: load it back and drive the next event --- events = load_from_store(stream_id="thread-42") reply = await agent.resume(*events, ModelRequest([TextInput("Carry on.")])) print(reply.body) ``` !!! note "Reading back the event log" `resume` reads only the events you pass in - it does not load history from a store on its own. The [event log](advanced/knowledge_store.md) written by `KnowledgeConfig(write_event_log=True)` is one place these trajectories can come from: it persists each turn to `/log/{stream_id}.jsonl`, and `EventLogWriter(store).load(stream_id)` reads it back as typed events ready to pass straight to `resume`. ```python linenums="1" from ag2.knowledge import EventLogWriter events = await EventLogWriter(store).load(stream_id) reply = await agent.resume(*events, ModelRequest([TextInput("Carry on.")])) ``` ## The stream is replaced, not appended `resume` **replaces** the target stream's history with the seeded prefix. If you omit `stream`, a fresh `MemoryStream` is created. If you pass an existing stream, any history it already held is discarded in favour of the trajectory you provide. ```python from ag2.stream import MemoryStream stream = MemoryStream() await stream.history.replace([ModelRequest([TextInput("stale prior turn")])]) # The stale turn is dropped; only the events you pass remain. reply = await agent.resume(*history, trigger, stream=stream) ``` This makes the recorded trajectory the single source of truth for the resumed turn - the events you pass are exactly what the model sees. !!! warning "Durable-backed streams are overwritten in place" The replacement is applied to the stream's **storage**, not just an in-memory copy. For a persistent stream such as `RedisStream`, seeding the prefix overwrites the durable history stored under that stream id - the backing store clears the key and rewrites it. Resume into a fresh stream, or use a new stream id, when you need to keep the original conversation's stored history intact. ## Caveats - resuming isn't always possible `resume` replays provider-native events back to the model, and some providers attach opaque, **required** metadata to those events. Gemini 3.x, for example, binds a *thought signature* to each function call and rejects a replayed call that arrives without it: ``` 400 INVALID_ARGUMENT - Function call is missing a thought_signature in functionCall parts ``` OpenAI reasoning items and Anthropic thinking signatures carry similar provider-specific state. As long as you resume from the **original** events - the ones the agent emitted, written verbatim by the [event log](#capture-persist-resume) - this metadata travels with them and `resume` round-trips cleanly. The risk appears when a trajectory is rebuilt from a lossy form: a hand-rolled `{name, args, result}` record, or a trace store that keeps only the fields it understands. The required metadata is silently dropped, and the provider rejects the resumed turn. !!! warning Resuming is not guaranteed for every provider and every trajectory. Preserve the events exactly as the agent emitted them; do not assume a trajectory reconstructed from a reduced or normalized form can be resumed. ## Related - [Agent Communication](agents.md) - `ask` / `reply.ask`, the in-memory entry points. - [Human in the Loop](context/human_in_the_loop.md) - pausing for human input, a common reason a turn is continued out of band. - [Stream & Events](advanced/stream.md) - the event types that make up a trajectory and how to subscribe to them. --- # Structured Output Source: https://docs.ag2.ai/docs/user-guide/structured_output/ # Structured Output Structured output constrains the model's final message so you can parse it into a **typed Python value**-a number, a `dataclass`, a Pydantic model, or the result of your own validator-instead of treating the reply as an opaque string. ## What you get on each turn Every turn returns an [`AgentReply`](agents.md). Two surfaces matter for structured output: | Surface | What it is | |--------|------------| | `reply.body` | Raw text from the model for that turn (a `str` or `None`). | | `await reply.content()` | Parsed value according to the **response schema** in effect for that turn. | If the model's output cannot be parsed or fails validation, `content()` raises an error from the underlying parser (for example Pydantic's validation errors). You can pass `retries` to automatically [re-ask the model](#validation-retries) on failure. With the default **OpenAI** client, when the schema exposes a JSON Schema to the API, the client sends a structured `response_format` so the model is guided to emit JSON matching that schema. [`PromptedSchema`](#promptedschema-models-without-native-structured-output) is the escape hatch when the provider does not support that mechanism: the schema is injected into the system prompt instead, and `content()` still runs the same way afterward. ## When to use which tool - **Pass a plain type** (`int`, `YourModel`, ...) when the default schema name and description are enough. - Use **`ResponseSchema`** when you want a clear **`name`** and **`description`** in the API payload so the model knows the role of the structured payload. - Use **`@response_schema`** when you need **custom parsing**, normalization, or extra steps after JSON is read. - Use **`PromptedSchema`** when your **model or endpoint does not support** native structured output. ## Quick start ```python from ag2 import Agent from ag2.config import OpenAIConfig agent = Agent( "assistant", prompt="You are a helpful assistant. Answer concisely.", config=OpenAIConfig("gpt-4o-mini"), response_schema=int, ) reply = await agent.ask("How many bits are in a byte?") print(reply.body) # e.g. '8' - raw model text result = await reply.content() print(result) # 8 - Python int ``` --- ## Real-world examples The following patterns mirror how structured output is used in applications: triage, extraction, and safe normalization. ### Classify a support ticket (Pydantic) Route incoming text into fields your helpdesk or CRM already understands: ```python from typing import Annotated from pydantic import BaseModel, Field from ag2 import Agent from ag2.config import OpenAIConfig class TicketTriage(BaseModel): """Structured triage for a single support message.""" category: Annotated[str, Field(description="e.g. billing, bug, account_access")] urgency: Annotated[str, Field(description="low, medium, or high")] summary_one_line: Annotated[str, Field(description="Max 120 characters", max_length=120)] agent = Agent( "triage", prompt="You triage customer support messages. Be conservative with urgency.", config=OpenAIConfig("gpt-4o-mini"), response_schema=TicketTriage, ) body = ( "I was charged twice for Pro last week and I still can't export my reports. " "This is blocking our quarter close." ) reply = await agent.ask(f"Classify this ticket:\n\n{body}") triage = await reply.content() # triage.category, triage.urgency, triage.summary_one_line -> use in routing rules ``` ### Extract a delivery ETA window (dataclass) Turn natural language into something your scheduling layer can consume: ```python from dataclasses import dataclass from ag2 import Agent from ag2.config import OpenAIConfig @dataclass class DeliveryWindow: day_label: str start_hour_local: int end_hour_local: int timezone: str agent = Agent( "scheduler", prompt="Extract delivery windows as structured data only; use 24h integers for hours.", config=OpenAIConfig("gpt-4o-mini"), response_schema=DeliveryWindow, ) reply = await agent.ask( "Customer said: drop off Tuesday between 2 and 5pm Pacific, before dinner." ) window = await reply.content() ``` ### Score a review on a fixed scale (primitive + clear prompt) Use a primitive schema when the payload is a single JSON value and your prompt defines the scale: ```python from ag2 import Agent from ag2.config import OpenAIConfig agent = Agent( "reviews", prompt="You output a single integer 1-5 for overall satisfaction. No prose.", config=OpenAIConfig("gpt-4o-mini"), response_schema=int, ) reply = await agent.ask( "Rate this review: 'Shipped fast, packaging was torn, product works great.'" ) stars = await reply.content() ``` --- ## Supported schema types You can pass any type the stack can turn into a JSON Schema and parse back: primitives, `dataclass`, Pydantic models, unions, and more. Plain types are wrapped in an internal `ResponseSchema` instance for validation and API schema generation. ### Primitives ```python agent = Agent("assistant", config=config, response_schema=int) reply = await agent.ask("What is 2 + 2?") result = await reply.content() # 4 - int ``` ### Dataclasses ```python from dataclasses import dataclass @dataclass class City: name: str population: int agent = Agent("assistant", config=config, response_schema=City) reply = await agent.ask("Give the city name and approximate population for Kyoto.") result = await reply.content() ``` ### Pydantic models ```python from pydantic import BaseModel class Sentiment(BaseModel): label: str score: float agent = Agent("assistant", config=config, response_schema=Sentiment) reply = await agent.ask("Analyze: 'I love this product!'") result = await reply.content() ``` ### Unions Use a union (`int | str`) or a tuple of types (`(int, str)`) when the model must return **one of several JSON shapes**. ```python from ag2 import Agent from ag2.config import OpenAIConfig config = OpenAIConfig("gpt-4o-mini") # int | str - e.g. a count, or "unknown" when the text does not say agent = Agent( "extractor", prompt='Reply with JSON only: either an integer count or the string "unknown".', config=config, response_schema=int | str, ) reply = await agent.ask("How many seats does the venue mention? (no number in text)") result = await reply.content() # result is int or str, depending on the model output ``` --- ## `ResponseSchema` (named payloads) For clearer API metadata, construct a `ResponseSchema` with an explicit **`name`** and **`description`**: ```python from ag2 import Agent, ResponseSchema schema = ResponseSchema( int | str, name="ByteWidth", description="The number of bits in one byte.", ) agent = Agent("assistant", config=config, response_schema=schema) ``` Those fields are attached to the structured-output payload where the provider supports it, which helps the model treat the JSON as a named contract rather than a generic blob. --- ## Custom validation with `@response_schema` Use the decorator when you need **logic beyond** "parse this JSON into a type": clamping, regex cleanup, decoding wrapped JSON, or combining fields. ### Sync validator: clamp a numeric rating ```python from ag2 import Agent, response_schema @response_schema def parse_rating(content: str) -> int: """Parse a rating and clamp it to 1-5.""" return max(1, min(5, int(content))) agent = Agent("assistant", config=config, response_schema=parse_rating) reply = await agent.ask("Rate this movie from 1 to 5.") result = await reply.content() ``` ### Async validator: enrich after JSON parse ```python import json @response_schema async def fetch_and_validate(content: str) -> dict: """Validate and enrich the model's JSON response.""" data = json.loads(content) data["validated"] = True return data ``` ### Validation rules for `@response_schema` The framework introspects your function with **fast_depends** (the same dependency-injection path as `@tool` callables). Parameters satisfied by injection - [`Variables`](context/variables.md), [`Depends`](depends.md), [`Inject`](context/inject.md), Context and similar-are **not** part of the JSON the model must produce. Every other parameter controls how the completion text is decoded and whether a JSON Schema is attached for native structured output. #### One non-injected parameter | Annotated type | What the model's message must look like | JSON Schema sent to the API? | |----------------|----------------------------------------|------------------------------| | `str` | Any text. The **raw** completion string is passed in; nothing is parsed as JSON for you. | **No** - there is no derived schema, so clients such as OpenAI do not get a `response_format` schema from this callable alone. | | Primitive or union (`int`, `float`, `bool`, `int \| str`, ...) | By default (**`embed=True`**), a JSON object `{"data": }`. The framework unwraps it before calling your function. With `embed=False`, a bare JSON value. | **Yes**, when the client supports structured output and emits `json_schema` from the derived schema. | | Structured type (`dataclass`, Pydantic model, `dict`, ...) | A JSON **object** matching the type's schema. These are never embedded regardless of the `embed` flag. | **Yes**. | Illustrative shapes (each function would be decorated with `@response_schema` and used as `response_schema=...` on an `Agent`): ```python # Raw text - parse inside the function (e.g. json.loads). def only_str(content: str) -> dict: pass ``` ```python # Single JSON value at the top level, e.g. 42 def only_int(content: int) -> dict: pass ``` ```python from dataclasses import dataclass @dataclass class Data: content: int # Single JSON object at the top level, e.g. {"content": 1} def only_dataclass(content: Data) -> dict: pass ``` #### Two or more non-injected parameters The framework builds one synthetic JSON **object** schema: **Python parameter names are JSON keys**. The completion must be a single object with those keys; values are validated against the annotations and passed into your function as keyword arguments (alongside any injected parameters). For example: ```python @response_schema def create_user(name: str, age: int, email: str) -> dict: """Create a validated user record.""" return {"name": name, "age": age, "email": email, "active": True} # expected JSON: {"name": "John Doe", "age": 30, "email": "john.doe@example.com"} agent = Agent("assistant", config=config, response_schema=create_user) reply = await agent.ask("Create a user for Alice, age 30, alice@example.com") result = await reply.content() # {"name": "Alice", "age": 30, "email": "alice@example.com", "active": True} ``` **`pydantic.Field` on each parameter** Multi-parameter validators are backed by a synthetic Pydantic model, so you can document and constrain each JSON property with [`Field`](https://docs.pydantic.dev/latest/concepts/fields/), just like on a [`BaseModel`](https://docs.pydantic.dev/latest/concepts/models/): - Use [`typing.Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated) when the parameter has no default: `Annotated[str, Field(description="...")]`. - Combine a **default** and metadata with `Field` as the default value, e.g. `score: float = Field(1.0, description="Test score")`. `description` is surfaced on each property in the generated JSON Schema (and thus in native structured output when the client sends that schema). Other `Field` arguments-`ge`, `le`, `pattern`, and so on-are reflected as the usual JSON Schema keywords. ```python from typing import Annotated from pydantic import Field from ag2 import response_schema @response_schema def extract_listing( title: Annotated[str, Field(description="Product name from the text")], price_usd: Annotated[float, Field(description="Price in US dollars", ge=0)], in_stock: Annotated[bool, Field(description="True if the listing says it ships now")], ) -> dict: return {"title": title, "price_usd": price_usd, "in_stock": in_stock} ``` Parameters with a Python default (plain value or `Field(default, ...)`) are usually **not** listed as required in the schema; callers can omit those keys in the JSON object. !!! note Renaming a parameter changes the key the model is instructed to use. Treat those names as part of your contract with the model. ### Accessing `Context` Validators participate in the same dependency injection model as tools. Inject [`Context`](context/variables.md) to read `variables`, tie validation to session state, or perform lookups: ```python from ag2 import Context, response_schema @response_schema def validate_with_context(content: str, context: Context) -> str: """Use context variables during validation.""" language = context.variables.get("language", "en") return f"[{language}] {content}" ``` --- ## `PromptedSchema` (models without native structured output) Some models or providers do not support API-level structured output (no `response_format` JSON schema). `PromptedSchema` **injects the JSON Schema into the system prompt** and sets `json_schema` to `None` on the wire so the client does not request native structured mode. Validation still goes through the inner schema's `validate` method. !!! note **Amazon Bedrock** supports [native structured output](https://docs.aws.amazon.com/bedrock/latest/userguide/structured-output.html) on the Converse API - `BedrockConfig` sends schemas via `outputConfig` automatically. Support is model-dependent (check the Bedrock model pages), and Bedrock compiles each new schema on first use, so the first request with a given schema can take noticeably longer than subsequent ones. ```python from ag2 import Agent, PromptedSchema agent = Agent( "assistant", config=config, response_schema=PromptedSchema(int), ) reply = await agent.ask("How many oceans are there on Earth?") result = await reply.content() ``` You can keep a **single** schema definition (type, `ResponseSchema`, or `@response_schema` callable) and **only wrap it when** you need prompt-based delivery. The inner `validate` logic and JSON shape stay the same; `PromptedSchema` swaps how the schema reaches the model (system-prompt text instead of API `response_format`). ```python from ag2 import Agent, PromptedSchema, ResponseSchema, response_schema from ag2.config import OpenAIConfig config = OpenAIConfig("gpt-4o-mini") # Plain type you already pass as response_schema=int elsewhere agent_a = Agent("a", config=config, response_schema=PromptedSchema(int)) # Named ResponseSchema reused from a "native structured" setup - wrap for a weaker API ocean_count = ResponseSchema( int, name="OceanCount", description="Number of oceans on Earth.", ) agent_b = Agent("b", config=config, response_schema=PromptedSchema(ocean_count)) # Same callable validator as without PromptedSchema - wrap it when the wire format must be prompt-only @response_schema def parse_int(content: str) -> int: return int(content.strip()) strict_int = PromptedSchema(parse_int) agent_c = Agent("c", config=config, response_schema=strict_int) ``` ### Custom prompt template The default template asks for raw JSON only. Override it with a string that contains the `{schema}` placeholder: ```python PromptedSchema( int, prompt_template="Reply with JSON matching this schema:\n{schema}", ) ``` --- ## Override schema per request Pass `response_schema` to `ask()` (or `AgentReply.ask()`) to change the contract for **one turn** only. The agent's default schema applies again on the next turn unless you override again. ```python agent = Agent("assistant", config=config) turn = await agent.ask("How many seconds in a minute?", response_schema=int) result = await turn.content() #> 60 - int turn2 = await turn.ask("Say hello.") result2 = await turn2.content() #> "Hello!" - str ``` Pass **`response_schema=None`** to drop a schema that was set on the agent for a single request: ```python agent = Agent("assistant", config=config, response_schema=int) reply = await agent.ask("Just say hello in plain text.", response_schema=None) result = await reply.content() ``` !!! note The per-request override applies only to that turn. The conversation history is unchanged; only the schema used for the next completion differs. ## Validation retries When the model's response fails schema validation, you can automatically **re-ask** the model instead of raising immediately. Pass the `retries` keyword to `content()`: ```python agent = Agent("assistant", config=config, response_schema=int) reply = await agent.ask("How many planets in the solar system?") result = await reply.content(retries=3) ``` The `retries` parameter controls how many **re-asks** are allowed after the initial attempt. With `retries=3`, the initial response is validated; if it fails, the model is re-asked up to **3 more times** before the error is raised. | Value | Behavior | |-------|----------| | `retries=0` (default) | No retries - raise on the first validation failure. | | `retries=3` | Up to 3 re-asks after the initial attempt (4 total). | | `retries=math.inf` | Re-ask indefinitely until the model produces a valid response. | Each retry sends the validation error back to the model as a follow-up message in the **same conversation**, so the model can see what went wrong and correct its output. !!! warning `retries=math.inf` will loop forever if the model consistently produces invalid output. Use a finite count in production, and reserve `math.inf` for interactive or experimental use. --- ## Primitive embedding (`embed`) When a schema type is a **primitive** (`int`, `float`, `bool`, `list[...]`) or a **union** (`int | str`), the framework wraps it in a one-field JSON object by default. This is called **embedding**. Instead of asking the model to produce a bare value like `42`, the API schema asks for `{"data": 42}`. The `content()` method transparently unwraps the envelope so your code still receives a plain Python value. ### Why? Most structured-output APIs (OpenAI, etc.) are designed around JSON **objects**. A bare value (`42`, `true`, `"hello"`) is technically valid JSON but some providers handle it less reliably. Wrapping the value in `{"data": ...}` gives the model a proper object to fill in, which improves reliability without changing your application code. ### Which types are embedded? | Type | Embedded by default? | Reason | |------|---------------------|--------| | `str` | No schema generated | Raw text is passed through as-is. | | `int`, `float`, `bool` | **Yes** | Bare primitives benefit from the object wrapper. | | `list[T]`, `tuple[T, ...]` | **Yes** | Array values also benefit from the wrapper. | | `int \| str`, `Union[T1, T2]`, `(T1, T2)` | **Yes** | Union of primitives. | | `BaseModel` subclass | No | Already a JSON object. | | `@dataclass` | No | Already a JSON object. | | `TypedDict` | No | Already a JSON object. | | `dict[K, V]` | No | Already a JSON object. | ### Opting out Pass `embed=False` to `ResponseSchema` or `@response_schema` to disable wrapping. The model must then produce the bare JSON value directly (e.g. `42` instead of `{"data": 42}`). ```python from ag2 import ResponseSchema schema = ResponseSchema(int, name="RawInt", embed=False) # Model must produce: 42 # With embed=True (default): model produces {"data": 42}, content() returns 42 either way ``` With the `@response_schema` decorator: ```python @response_schema(embed=False) def parse_rating(value: int) -> int: return max(1, min(5, value)) ``` !!! note Embedding is transparent to your code. Whether `embed` is `True` or `False`, `content()` always returns the unwrapped Python value. The only difference is the JSON shape the model is asked to produce. --- --- # Prompt Management Source: https://docs.ag2.ai/docs/user-guide/system_prompts/ # Prompt Management ## System Prompts Agents can be initialized with a static system prompt. You can provide a single string or a list of strings: ```python from ag2 import Agent # Single string prompt agent = Agent( "assistant", prompt="You are a helpful agent!" ) # List of strings prompt agent2 = Agent( "assistant2", prompt=[ "You are an expert in Python.", "Be concise." ] ) ``` ## Dynamic Prompts ### On conversation startup System prompts can be generated dynamically when a conversation starts. This is useful when the prompt depends on external state or initial context. You can achieve this by using the `@my_agent.prompt` decorator or passing a synchronous or asynchronous function. Dynamic prompt functions support the same powerful execution context capabilities as Agent Tools. For more detailed information on specific context features, see [Dependency Injection](context/inject.md), [Context Variables](context/variables.md), and [Human-in-the-loop](context/human_in_the_loop.md). Dynamic prompts are evaluated only once at the beginning of the conversation, and their results are appended to the static prompts and reused for subsequent turns. ```python from ag2 import Agent, Context agent = Agent("assistant") @agent.prompt async def dynamic_sysprompt(ctx: Context) -> str: # Generate prompt dynamically based on the initial event or context return ( "You are a helpful agent. " f"The current context is {ctx.variables}." ) ``` Alternatively, you can pass a callable directly to the `prompt` parameter, or mix static strings and callables in a list: ```python from ag2 import Agent def get_sysprompt() -> str: # Returns a string for the prompt, evaluated at the beginning of the conversation return "This is dynamically generated." agent = Agent( "assistant", prompt=["Static prompt part.", get_sysprompt] ) ``` ### On each conversation turn While dynamic prompt hooks are evaluated once per conversation, you might need to update the prompt dynamically on each turn. You can do this by mutating the `prompt` list within the `Context` directly between calls to `reply.ask()`. ```python # Initial conversation turn reply = await agent.ask("Hi, agent!") # Change the prompt for the next turn reply.context.prompt = ["You are now a funny agent!"] await reply.ask("Tell me a joke") ``` You can also completely override the agent's default prompt for a specific run or turn by passing the `prompt` parameter directly to `ask()`: ```python # Overrides the default prompt for this conversation reply = await agent.ask( "Hi!", prompt=["Temporary prompt for this run"] ) ``` ## Prompt updates For continuous and event-driven prompt updates, you can mutate `context.prompt` dynamically from an event subscriber. This allows you to respond to specific events in the stream and adjust the agent's behavior on the fly during an ongoing conversation. See the [Stream](advanced/stream.md) documentation for more details on this advanced feature. ```python from ag2 import Agent, Context, MemoryStream from ag2.events import ModelRequest agent = Agent("assistant", prompt="You are a helpful assistant.") stream = MemoryStream() @stream.where(ModelRequest).subscribe() async def mutate_prompt(event: ModelRequest, context: Context) -> None: # Update the prompt dynamically when a ModelRequest is triggered if "joke" in event.content.lower(): context.prompt = ["You are now a comedian."] await agent.ask("Tell me a joke", stream=stream) ``` --- # Depends Source: https://docs.ag2.ai/docs/user-guide/depends/ # Depends The `Depends` mechanism allows you to calculate and inject dependencies dynamically at execution time. The key difference with [Dependency Injection](context/inject.md){.internal-link} is their execution model. `Inject` is used to retrieve static objects or configurations that have already been created (like an existing database connection or API key). `Depends`, on the other hand, executes a callable function *during* the tool's invocation to resolve the dependency. Under the hood, `Depends` uses the exact same mechanism and design philosophy as [FastAPI's dependency injection system](https://fastapi.tiangolo.com/tutorial/dependencies/). ## Side-execution You can use `Depends` to execute side-effects before your tool runs-even if your tool doesn't actually need the return value of the dependency. This is extremely useful for things like authentication, logging, or permission verification. To do this, simply declare the dependency in your tool's signature. The framework will execute it, and you can safely ignore the injected value. ```python from typing import Annotated from ag2 import Depends, tool def verify_permissions(user_id: int) -> None: # Perform complex verification here # Raises an exception if permissions are invalid raise PermissionDenied(user_id) @tool def delete_user( user_id: int, # The dependency is executed, acting as a gatekeeper auth: Annotated[None, Depends(verify_permissions)] ) -> str: return f"User {user_id} deleted." ``` !!! note "Sync/Async" `Depends` can be used with both synchronous and asynchronous functions. ## Depends with yield Just like in [FastAPI](https://fastapi.tiangolo.com/), you can create dependencies that use `yield` instead of `return`. This allows you to execute "teardown" or "cleanup" code *after* the tool has finished executing. This is the recommended approach for managing resource lifecycles, such as opening and closing database sessions or file handlers. ```python def get_db_session(): print("Opening database session...") session = "db_session_object" # The tool execution happens here yield session # This runs after the tool finishes print("Closing database session...") @tool def fetch_records( db: Annotated[str, Depends(get_db_session)], ) -> str: return "Records fetched." ``` ### Combining Depends and Inject A powerful pattern is to combine `Depends` with `Inject`. You can use `Inject` to retrieve a static configuration or persistent resource (like a database connection pool), and then use `Depends` to manage a short-lived resource (like a database session) based on that configuration. ```python from typing import Annotated from ag2 import Depends, Inject, tool, Agent def get_db_session( db_pool: Annotated[Pool, Inject("database_pool")], ) -> Session: session = db_pool.acquire() yield session session.release() @tool def fetch_records( db_session: Annotated[object, Depends(get_db_session)], ) -> str: return "Records fetched." agent = Agent( "TestAgent", tools=[fetch_records], dependencies={"database_pool": Pool()}, ) ``` ## Dependencies caching By default, if multiple parameters in your tool (or multiple sub-dependencies) depend on the exact same `Depends` function, the framework will only execute that function **once** per tool call. The result is cached and reused for any subsequent injections within that specific execution step. ```python def get_expensive_config() -> dict: print("Calculating config...") # This will only print once! return {"timeout": 30} def get_timeout( config: Annotated[dict, Depends(get_expensive_config)], ) -> int: return config["timeout"] @tool def process_data( timeout: Annotated[int, Depends(get_timeout)], # cached dependency config: Annotated[dict, Depends(get_expensive_config)], ) -> str: return "Done" ``` If you explicitly want the dependency to be re-calculated every single time it is injected, you can disable the cache by passing `use_cache=False`: ```python @tool def random_tool( val1: Annotated[int, Depends(get_random_number, use_cache=False)], val2: Annotated[int, Depends(get_random_number, use_cache=False)] ) -> str: # val1 and val2 will be different numbers pass ``` ## Dependencies Overrides During testing, you often need to mock or override complex dependencies (like replacing a production database with a mock test database). You can easily override any `Depends` function at the agent level using the `dependency_provider`. When the agent executes, it will automatically route all requests for the original dependency to your override function. ```python from ag2 import Agent, tool def get_production_db(): raise Exception("Do not call this in tests!") @tool def read_data(db: Annotated[object, Depends(get_production_db)]) -> str: return "Data" agent = Agent("TestAgent", tools=[read_data]) # Create a mock function def get_test_db(): return "mock_database" # Override the production dependency with the test dependency agent.dependency_provider.override(get_production_db, get_test_db) # When the tool is called, it will use `get_test_db` instead await agent.ask("Read some data") ``` To override `Inject` dependencies, you can just set `dependencies={...}` in the `ask` call. ```python agent = Agent("TestAgent", tools=[read_data]) # Override the production `Inject` dependency with the test dependency await agent.ask( "Read some data", dependencies={"database_pool": Pool()}, ) ``` --- # Skills Source: https://docs.ag2.ai/docs/user-guide/skills/ Skills let an agent load specialized instructions on demand instead of carrying every capability in its system prompt. They follow the [agentskills.io](https://agentskills.io) convention: each skill is a directory that an agent discovers, reads, and runs only when a task actually calls for it. Skills can also be [defined inline in code](#code-defined-skills) when a directory on disk isn't a good fit. !!! note This page covers **local** skills that run on your machine. To activate skills hosted and executed by the LLM provider (Anthropic or OpenAI server-side), use `SkillsTool` - see [Provider Skills](tools/builtin_tools.md#provider-skills). AG2 ships three entry points, from highest-level to lowest: | Entry point | Use it when | | :--- | :--- | | [`SkillPlugin`](#skillplugin) | **Recommended.** Injects the catalog into the system prompt and wires the activation tools, so the model knows what's available from the first turn. | | [`SkillsToolkit`](#skillstoolkit) | You want the activation tools (plus an explicit `list_skills`) without prompt injection, or full control over runtimes. | | [`SkillSearchToolkit`](#skillsearchtoolkit) | The agent should discover and install new skills from the [skills.sh](https://skills.sh) registry at runtime. | ## Skill structure A skill is a directory with a `SKILL.md` at its root and, optionally, scripts and resource files: ```text pdf-processing/ ├── SKILL.md # required: YAML frontmatter + instructions ├── scripts/ # optional: runnable .py / .sh files │ └── extract.py └── references/ # optional: any bundled resource files └── form-fields.md ``` `SKILL.md` carries YAML frontmatter (`name`, `description`, and optionally `version`, `license`, `compatibility`) followed by the instruction body. For the full authoring format, see the [agentskills.io documentation](https://agentskills.io/). When surfacing a skill's files, AG2 draws one firm line: - A **script** lives under `scripts/` and is *executed* via `run_skill_script`. - A **resource** is any other file (not `SKILL.md`, not under `scripts/`) and is *read* via `read_skill_resource`. The two are disjoint - a file is one or the other, never both. ## Progressive disclosure Skills exist to keep context small. The agent only pays the token cost of the detail it actually uses, in three tiers: | Tier | What's loaded | When | | :--- | :--- | :--- | | **1. Catalog** | Name + description + location, per skill | At startup (or via `list_skills`) | | **2. Instructions** | The full `SKILL.md` body | When the model calls `load_skill` | | **3. Resources & scripts** | Bundled files and script output | When the instructions reference them (`read_skill_resource` / `run_skill_script`) | An agent with 20 installed skills doesn't load 20 full instruction sets upfront - only the ones a given conversation activates. ## SkillPlugin `SkillPlugin` is the recommended way to give an agent local skills. Instead of spending a `list_skills` tool round-trip, it injects the catalog into the system prompt at startup, so the model can decide which skill is relevant immediately. ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import SkillPlugin agent = Agent( "assistant", config=AnthropicConfig(model="claude-sonnet-5"), plugins=[SkillPlugin()], ) ``` By default it scans `.agents/skills` relative to the current working directory. Pass a path or a `LocalRuntime` to point elsewhere: ```python plugins=[SkillPlugin("./my-skills")] ``` ### Activation flow 1. **Startup** - the plugin discovers every skill and injects an `` block into the system prompt. Each entry lists the skill's `name`, `description`, and `location`: ```xml pdf-processing Extract PDF text, fill forms, merge files. Use when handling PDFs. /home/user/.agents/skills/pdf-processing/SKILL.md ``` 2. **Load** - when a task matches a description, the model calls `load_skill(name)`. The `name` parameter is constrained to the discovered skills, so the model can't invent one. The tool returns the `SKILL.md` body wrapped in ``, along with the skill directory and a listing of bundled resources. 3. **Use resources** - the model reads a listed resource with `read_skill_resource(name, resource)` or executes a script with `run_skill_script(name, script, args)` only when the instructions call for it. ### Capability gating `SkillPlugin` only registers the activation tools the installed skills can actually use: - `load_skill` is always registered (when at least one skill exists). - `read_skill_resource` is registered **only if some skill has resources**. - `run_skill_script` is registered **only if some skill has scripts**. So an agent whose skills are pure instructions never sees a `run_skill_script` tool it can't use. When no skills are found at all, the plugin contributes nothing - no catalog and no tools. !!! tip `SkillPlugin` is a snapshot taken at construction time: the catalog, the `name` constraint, and the gated tools always describe the same set of skills. Rebuild the agent (or the plugin) after installing new skills. ## SkillsToolkit `SkillsToolkit` is the lower-level building block behind `SkillPlugin`. It exposes the same activation tools plus an explicit `list_skills` tool, but does **not** inject anything into the prompt - the model discovers skills by calling `list_skills` itself. Prefer [`SkillPlugin`](#skillplugin) unless you need that control. ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import SkillsToolkit agent = Agent( "assistant", config=AnthropicConfig(model="claude-sonnet-5"), tools=[SkillsToolkit()], ) ``` It exposes four tools: | Tool | Description | | :--- | :--- | | `list_skills` | Return a catalog of installed skills with name, description, and location | | `load_skill` | Fetch the full `SKILL.md` content for a specific skill | | `read_skill_resource` | Read a bundled resource file from a skill's directory | | `run_skill_script` | Execute a `.py` or `.sh` script from a skill's `scripts/` directory | Every tool is also available as a method, so you can hand-pick a subset: ```python skills = SkillsToolkit() agent = Agent( "assistant", config=AnthropicConfig(model="claude-sonnet-5"), tools=[skills.list_skills(), skills.load_skill()], ) ``` ### Runtimes A **runtime** owns where skills live and how their scripts run - it discovers skills, reads their content, and executes their scripts. `LocalRuntime` is the default: it backs skills with the filesystem and runs scripts in a local subprocess (or a sandbox you supply). Pass one to a toolkit or plugin as a path string or an explicit `LocalRuntime`: ```python from ag2.tools import SkillsToolkit from ag2.tools.skills import LocalRuntime skills = SkillsToolkit(LocalRuntime("./my-skills")) # or just a path string skills = SkillsToolkit("./my-skills") ``` `LocalRuntime` takes the install directory plus optional execution and discovery settings. `extra_paths` adds read-only directories that are scanned for skills but never written to - installed skills always go to the primary `dir`: ```python skills = SkillsToolkit( LocalRuntime( "./my-skills", extra_paths=["./shared-skills"], # read-only, also scanned timeout=30, # per-script timeout (seconds) blocked=["rm -rf"], # best-effort command blocklist ) ) ``` ## Composing multiple runtimes `SkillPlugin` and `SkillsToolkit` both accept **more than one runtime**, so you can serve skills from several locations at once - each with its own configuration. A common pattern is a read-only global library plus a writable project directory: ```python from ag2.tools import SkillPlugin from ag2.tools.skills import LocalRuntime plugins = [ SkillPlugin( LocalRuntime("~/.agents/skills"), # global LocalRuntime(".agents/skills"), # project ), ] ``` When the same skill name exists in more than one runtime, the **last runtime wins** - here the project skill shadows the global one. The rule is applied uniformly: the catalog, the `name` constraint, `load_skill`, `read_skill_resource`, and `run_skill_script` all resolve to the same winning skill. !!! note Because each runtime can carry its own execution and storage settings (sandbox, timeout, install directory), composition lets global skills run one way and project skills another - something a single runtime can't express. ## Code-defined skills Not every skill needs to live on disk. A `MemorySkill` defines a skill **inline in code** - its instructions, resources, and scripts are Python values rather than files. It's backed by an in-memory runtime (`MemoryRuntime`) instead of `LocalRuntime`, but the agent activates it through the same progressive-disclosure flow. ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import SkillPlugin, MemorySkill unit_converter = MemorySkill( name="unit-converter", description="Convert between common units. Use when asked to convert miles, kilometers, pounds, or kilograms.", instructions="Use the convert script, passing the value and a factor from the conversion_table resource.", ) @unit_converter.resource def conversion_table() -> str: """Multiplication factors for common conversions.""" return "miles->km: 1.60934\npounds->kg: 0.453592" @unit_converter.script def convert(value: float, factor: float) -> str: """Multiply a value by a conversion factor.""" return str(round(value * factor, 4)) agent = Agent( "assistant", config=AnthropicConfig(model="claude-sonnet-5"), plugins=[SkillPlugin(unit_converter)], ) ``` Pass a `MemorySkill` straight to `SkillPlugin` (or `SkillsToolkit`) - it is wrapped in a `MemoryRuntime` automatically and appears in the catalog alongside any file-based skills. ### Resources and scripts as callables The `@skill.resource` and `@skill.script` decorators register Python callables (sync or async). By default the **function name** becomes the resource/script name and the **docstring** becomes its description - pass `name=` or `description=` only to override: - A **resource** callable runs every time it is read, so it can return live data - current config, a roster, a database lookup - rather than a static file. - A **script** callable runs **in-process**: no subprocess, no `scripts/` directory. Its parameter JSON-schema is generated from the signature and disclosed inside the loaded skill content, so the model calls `run_skill_script` with named arguments (`{"value": 10, "factor": 2}`) matching that schema. Arguments are validated and coerced exactly as a regular tool's are. Both forms support dependency injection - a callable can declare `Context`, `Variable`, or `Inject` parameters and they resolve from the live run context, just like a tool: ```python from typing import Annotated from ag2 import Variable from ag2.tools import MemorySkill project = MemorySkill(name="project-info", description="Project status and configuration.") @project.resource def environment(region: Annotated[str, Variable("region")]) -> str: return f"Region: {region}" ``` ### Composing with file-based skills A `MemorySkill` composes with paths and runtimes like any other source - declaration order decides precedence, and the last source wins on a name clash: ```python plugins=[SkillPlugin(".agents/skills", unit_converter, project)] ``` Grouping is associative: passing loose `MemorySkill`s, wrapping each in its own `MemoryRuntime`, or grouping several into one `MemoryRuntime(...)` all behave identically. !!! note `MemoryRuntime` is read-only - its skills are defined in code, so it cannot be an install target for `SkillSearchToolkit`. ## SkillSearchToolkit `SkillSearchToolkit` extends `SkillsToolkit` with three tools for discovering and installing skills from the [skills.sh](https://skills.sh) registry. It uses the GitHub Tarball API directly - no Node.js required. ```python import asyncio from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import SkillSearchToolkit agent = Agent( "coder", "You are a helpful coding assistant. Use skills to extend your capabilities.", config=AnthropicConfig(model="claude-sonnet-5"), tools=[SkillSearchToolkit()], # adds search_skills, install_skill, remove_skill ) async def main() -> None: reply = await agent.ask( "Find and install a skill for React best practices, then tell me the top 3 rules." ) print(await reply.content()) asyncio.run(main()) ``` It inherits all of `SkillsToolkit`'s tools and adds: | Tool | Description | | :--- | :--- | | `search_skills` | Search the skills.sh registry by keyword | | `install_skill` | Download and install a skill by its registry identifier | | `remove_skill` | Remove an installed skill by name | ### GitHub rate limits By default the GitHub API allows 60 unauthenticated requests per hour. Setting a `GITHUB_TOKEN` environment variable raises this to 5,000 per hour: ```bash export GITHUB_TOKEN=ghp_... ``` You can also pass the token (and other settings) directly via `SkillsClientConfig`: ```python from ag2.tools import SkillSearchToolkit from ag2.tools.skills import LocalRuntime, SkillsClientConfig skills = SkillSearchToolkit( LocalRuntime(dir="./my-skills", timeout=30), client=SkillsClientConfig( github_token="ghp_...", proxy="http://proxy.company.com:8080", ), ) ``` !!! note `SkillSearchToolkit` installs into a single runtime. To combine installed skills with skills from other locations, serve them through a `SkillPlugin` or `SkillsToolkit` that lists multiple runtimes. --- # Middleware Source: https://docs.ag2.ai/docs/user-guide/middleware/ # AG2 Middleware Middleware lets you intercept and customize how an **AG2** agent runs a turn. It's the right tool when you want to add cross-cutting behavior such as logging, retries, history trimming, request mutation, tool auditing, or guardrails without changing the agent, model client, or tools themselves. At a high level, middleware can wrap four parts of the runtime: - the full agent turn - each LLM call - each tool execution - each human input request This makes it a good fit for behavior that should apply consistently across many runs. ## What is Middleware Middleware is an object that receives the current turn's initial event and `Context`, then participates in one or more lifecycle hooks. Each middleware instance is created at the beginning of a turn and can keep per-turn state on `self`. That same instance can then observe or modify the turn, the LLM call, tool execution, and human input as the run progresses. In practice, you use middleware to: - add observability such as logging, tracing, and timing - enforce policies before a tool runs - retry transient model failures - trim conversation history before sending it to the model - normalize tool inputs or outputs - short-circuit or reshape a response ## Middleware Hooks `BaseMiddleware` exposes four async hooks. You can implement just one of them or mix several in the same class. ### `on_turn()` ```python class BaseMiddleware: async def on_turn( self, call_next: Callable[[BaseEvent, Context], Awaitable[ModelResponse]], event: BaseEvent, context: Context, ) -> ModelResponse: return await call_next(event, context) ``` `on_turn()` wraps the whole agent turn. It receives the incoming event and the final `ModelResponse`. Use `on_turn()` when you want to: - measure total turn latency - inspect or rewrite the initial request before anything else happens - inspect or rewrite the final response before it is returned - implement turn-level policies, approvals, or short-circuit behavior Conceptually, this is the outermost hook around a single `ask(...)` call. ### `on_llm_call()` ```python class BaseMiddleware: async def on_llm_call( self, call_next: Callable[[Sequence[BaseEvent], Context], Awaitable[ModelResponse]], events: Sequence[BaseEvent], context: Context, ) -> ModelResponse: return await call_next(events, context) ``` `on_llm_call()` wraps the call to the configured model client. It receives the event history that will be sent to the LLM. Use `on_llm_call()` when you want to: - retry transient client failures - log prompts and responses - trim history before it reaches the model - sanitize context / model response - inject additional request-time instructions through event mutation - implement caching or request deduplication around model calls This is the hook used by built-in history and token limiting middleware. ### `on_tool_execution()` ```python class BaseMiddleware: async def on_tool_execution( self, call_next: Callable[[ToolCallEvent, Context], Awaitable[ToolResultType]], event: ToolCallEvent, context: Context, ) -> ToolResultType: return await call_next(event, context) ``` `on_tool_execution()` wraps each tool invocation triggered during the turn. It receives the current `ToolCallEvent` and returns a `ToolResultType`. Use `on_tool_execution()` when you want to: - validate or rewrite tool arguments before execution - log tool usage - transform tool results before they go back into the event stream - capture tool failures and replace them with safer fallback results - enforce access control around specific tools For wrapping **one** tool at definition time with async hooks, see [Tool middleware](tools/tool_middleware.md). ### `on_human_input()` ```python class BaseMiddleware: async def on_human_input( self, call_next: Callable[[HumanInputRequest, Context], Awaitable[HumanMessage]], event: HumanInputRequest, context: Context, ) -> HumanMessage: return await call_next(event, context) ``` `on_human_input()` wraps each human-in-the-loop (HITL) request triggered during a turn. It receives the `HumanInputRequest` emitted by a tool via `ctx.input(...)` and can intercept or modify both the request and the `HumanMessage` response. Use `on_human_input()` when you want to: - log or audit human input requests and responses - rewrite or enrich the prompt shown to the human - transform the human's reply before it reaches the tool - short-circuit the request with an automated response instead of asking a human - enforce policies or rate limits on human input requests ## Registering Middleware ### On an Agent To make middleware apply to every turn for an agent, pass it through the `middleware` argument when constructing the agent. ```python from ag2 import Agent from ag2.config import OpenAIConfig from ag2.middleware import LoggingMiddleware, RetryMiddleware agent = Agent( "assistant", prompt="Be helpful.", config=OpenAIConfig("gpt-4o-mini"), middleware=[ LoggingMiddleware(), RetryMiddleware(max_retries=2), ], ) ``` Use agent-level registration for behavior that should always be present, such as logging, tracing, or default retry policy. ### On a Single Call You can also add middleware just for a specific turn. This is useful when you want temporary behavior without changing the agent's defaults. Both `Agent.ask(...)` and `AgentReply.ask(...)` accept a `middleware` argument. ```python from ag2 import Agent from ag2.config import OpenAIConfig from ag2.middleware import LoggingMiddleware, TokenLimiter agent = Agent( "assistant", prompt="Be helpful.", config=OpenAIConfig("gpt-4o-mini"), ) reply = await agent.ask( "Summarize the latest messages.", middleware=[LoggingMiddleware()], ) next_turn = await reply.ask( "Now answer in one paragraph.", middleware=[TokenLimiter(max_tokens=4000)], ) ``` Call-level middleware is appended after the middleware list defined on the agent. ## Middleware Ordering Middleware runs in the order you register them. If you register `[A, B, C]`, they enter in the order `A -> B -> C` and unwind in reverse order `C -> B -> A`. This matters when you combine behaviors such as logging, mutation, and retries. ```python from ag2 import Agent, Context from ag2.config import OpenAIConfig from ag2.events import BaseEvent, ModelResponse from ag2.middleware import AgentTurn, BaseMiddleware class A(BaseMiddleware): async def on_turn( self, call_next: AgentTurn, event: BaseEvent, context: Context, ) -> ModelResponse: print("enter A") response = await call_next(event, context) print("exit A") return response class B(BaseMiddleware): async def on_turn( self, call_next: AgentTurn, event: BaseEvent, context: Context, ) -> ModelResponse: print("enter B") response = await call_next(event, context) print("exit B") return response class C(BaseMiddleware): async def on_turn( self, call_next: AgentTurn, event: BaseEvent, context: Context, ) -> ModelResponse: print("enter C") response = await call_next(event, context) print("exit C") return response agent = Agent( "assistant", prompt="Be helpful.", config=OpenAIConfig("gpt-4o-mini"), middleware=[A, B], ) await agent.ask( "Hello", middleware=[C], ) # Output: # enter A # enter B # enter C # exit C # exit B # exit A ``` ## Writing Your Own Middleware To create custom middleware, subclass `BaseMiddleware` and implement the hooks you need. If your middleware does not need extra constructor arguments, you can register the class directly. If it does need configuration, wrap it with `Middleware(...)` when registering it. ```python import logging from collections.abc import Sequence from ag2 import Agent, Context from ag2.config import OpenAIConfig from ag2.events import BaseEvent, ModelResponse, ToolCallEvent from ag2.middleware import BaseMiddleware, LLMCall, Middleware, ToolExecution class AuditMiddleware(BaseMiddleware): def __init__( self, event: BaseEvent, context: Context, logger: logging.Logger, ) -> None: super().__init__(event, context) self.logger = logger async def on_llm_call( self, call_next: LLMCall, events: Sequence[BaseEvent], context: Context, ) -> ModelResponse: self.logger.info("Calling model with %d events", len(events)) response = await call_next(events, context) self.logger.info("Model returned: %s", response) return response async def on_tool_execution( self, call_next: ToolExecution, event: ToolCallEvent, context: Context, ): self.logger.info("Executing tool: %s", event.name) return await call_next(event, context) agent = Agent( "assistant", prompt="Be helpful.", config=OpenAIConfig("gpt-4o-mini"), middleware=[ Middleware(AuditMiddleware, logger=logging.getLogger("ag2.audit")), ], ) ``` ### Guidelines for Custom Middleware - Keep hook behavior focused. Middleware that does one job well is easier to reason about than one that handles, for example, logging, retries, mutation, and policy checks together. - Prefer `on_turn()` for whole-run behavior, `on_llm_call()` for model-facing behavior, `on_tool_execution()` for tool-facing behavior, and `on_human_input()` for human-in-the-loop behavior. - Be deliberate when mutating `event`, `events`, or tool results. Later executing middleware and the rest of the runtime will observe those changes. - Register zero-config middleware classes directly, and use `Middleware(YourMiddleware, ...)` when the constructor needs additional options. ## Built-In Middleware AG2 currently includes four built-in middleware in `ag2.middleware`: ### `LoggingMiddleware` ```python from ag2 import Agent from ag2.middleware import LoggingMiddleware agent = Agent(..., middleware=[LoggingMiddleware()]) ``` Logs the lifecycle of a turn, including: - when a turn starts and finishes - each LLM call and its response time - each tool execution and its result Use it for quick debugging or application-level observability. ### `RetryMiddleware` ```python from ag2 import Agent from ag2.middleware import RetryMiddleware agent = Agent(..., middleware=[RetryMiddleware(max_retries=2)]) ``` Retries failed LLM calls up to `max_retries` times. By default it retries any `Exception`, but you can narrow that with `retry_on=...`. Use it for transient failures such as provider timeouts or flaky network issues. A call that already published something to the stream - a chunk, reasoning, or a server-side tool call - is not retried; the exception propagates instead. Retrying would leave anyone watching the stream with both attempts' output, while the reply carries only the second. Failures that arrive before anything is published - connection errors, rate limits, `5xx` - retry as usual. ### `HistoryLimiter` ```python from ag2 import Agent from ag2.middleware import HistoryLimiter agent = Agent(..., middleware=[HistoryLimiter(max_events=100)]) ``` Trims the event history to a maximum number of events before the model call. It preserves the first `ModelRequest` when possible, and drops whatever the cut orphaned so the trimmed history still replays. `max_events` is a target rather than a hard cap, for the same reason it is one on the reduction policies - see [Reduction limits are targets](advanced/assembly.md#reduction-limits-are-targets). Use it when you want a simple, deterministic cap on context length by event count. ### `TokenLimiter` ```python from ag2 import Agent from ag2.middleware import TokenLimiter agent = Agent(..., middleware=[TokenLimiter(max_tokens=1000)]) ``` Trims the event history to fit within an approximate token budget before the model call. It uses a character-based estimate controlled by `chars_per_token`. `max_tokens` is a target rather than a hard cap - see [Reduction limits are targets](advanced/assembly.md#reduction-limits-are-targets). Use it when you need lightweight context budgeting without depending on a model-specific tokenizer. ## Conditional Middleware `ConditionalMiddleware` lets you gate any middleware so each hook only activates when a condition matches the hook's own event. When the condition is not met, that hook passes through to the next middleware in the chain. This is useful when you have middleware that should only run for certain event types - for example, approval middleware that only fires for a specific tool - without writing condition checks inside every hook method. ```python from ag2 import Agent from ag2.config import OpenAIConfig from ag2.events import ToolCallEvent from ag2.middleware import ConditionalMiddleware, Middleware agent = Agent( "assistant", prompt="Be helpful.", config=OpenAIConfig("gpt-4o-mini"), middleware=[ ConditionalMiddleware( Middleware(ApprovalMiddleware), condition=ToolCallEvent.name == "execute_code", ), ], ) ``` In this example, `ApprovalMiddleware` only activates during `on_tool_execution` when the tool name is `"execute_code"`. The `on_turn` hook receives a different event type (`ModelRequest`), so the condition does not match there and the middleware passes through. Each hook - `on_turn`, `on_tool_execution`, `on_human_input` - checks the condition against its own event. `on_llm_call` checks against the initial turn event, since it receives a sequence of events rather than a single one. When a condition targets a specific event type like `ToolCallEvent`, hooks that receive a different type will naturally pass through. ### Composing Conditions Conditions support `&` (and), `|` (or), and `~` (not) operators, so you can build expressive gates: ```python from ag2.events import ToolCallEvent from ag2.middleware import ConditionalMiddleware, Middleware # Activate only for tool calls named "search" or "browse" conditional = ConditionalMiddleware( Middleware(MyAuditMiddleware, logger=logger), condition=(ToolCallEvent.name == "search") | (ToolCallEvent.name == "browse"), ) # Activate for all tool calls EXCEPT "calculate" conditional = ConditionalMiddleware( Middleware(MyAuditMiddleware, logger=logger), condition=~(ToolCallEvent.name == "calculate"), ) ``` You can also pass a bare event type as the condition - it is automatically wrapped: ```python # Activate only when the hook receives a ToolCallEvent conditional = ConditionalMiddleware( Middleware(MyAuditMiddleware, logger=logger), condition=ToolCallEvent, ) ``` `ConditionalMiddleware` wraps any `MiddlewareFactory` - including `Middleware(...)` instances, bare `BaseMiddleware` subclasses, and other `ConditionalMiddleware` wrappers. ## Describing Middleware A middleware can report what it is and how it was configured, so it can be logged, compared against another instance, or asserted on in a test. Every built-in does: ```python from ag2.middleware import TokenLimiter TokenLimiter(max_tokens=100).describe() # MiddlewareDescription(kind='TokenLimiter', config={'max_tokens': 100, 'chars_per_token': 4}, complete=True) TokenLimiter(max_tokens=100).describe() == TokenLimiter(max_tokens=100).describe() # True ``` Your own middleware opts in the same way, by implementing `describe()`: ```python from ag2.middleware import MiddlewareDescription class RateLimit: def __init__(self, per_minute: int) -> None: self._per_minute = per_minute async def __call__(self, call_next, event, context): return await call_next(event, context) def describe(self) -> MiddlewareDescription: return MiddlewareDescription( kind=type(self).__qualname__, config={"per_minute": self._per_minute}, ) ``` Opting in is optional. Where AG2 surfaces middleware, anything that has not opted in is reported honestly rather than guessed at - `config` is empty and `complete` is `False`. AG2 never inspects `__closure__` to recover configuration, because cell names and ordering are an implementation detail of whatever function produced the closure. A `describe()` that fails or returns the wrong type degrades to that same honest unknown, so introspection cannot break the code doing the logging. Middleware that wraps other middleware reports them in `inner`, keeping `config` a flat settings mapping: ```python from ag2.middleware import ConditionalMiddleware, TokenLimiter from ag2.events import ToolCallEvent ConditionalMiddleware(TokenLimiter(max_tokens=10), ToolCallEvent).describe() # MiddlewareDescription( # kind='ConditionalMiddleware', config={'condition': 'TypeCondition'}, complete=True, # inner=(MiddlewareDescription(kind='TokenLimiter', config={'max_tokens': 10, ...}),), # ) ``` A composite is only as describable as its parts: `complete` is forced to `False` when any entry in `inner` is incomplete, so composing middleware cannot claim completeness its parts do not support. Because each description applies the rule as it is constructed, it holds at any depth without a parent walking the tree. Descriptions support equality but are deliberately **not hashable** - `config` may hold arbitrary values, so they cannot go in a `set` or be used as dict keys. To read the middleware attached to an agent or a tool, use `agent.middleware` and `tool.middleware`. Each entry pairs the middleware object with its description, so an entry always has one whether or not that middleware opted in. See [Reading an agent's composition](agent_harness.md#reading-an-agents-composition). !!! note `Middleware(...)` reports the wrapped class and which options were set, but never their values, and always `complete=False`. The options are caller-supplied and the wrapper cannot know whether one is a credential. Write a `MiddlewareFactory` with its own `describe()` when you want a complete description. Its `.cls` and `.options` expose the class and the option values directly, for when you are inspecting a factory you already hold rather than producing something to log. !!! note `config` is for settings only. Counters, caches, and anything else that changes during a run belong in `context.variables` or the per-turn `BaseMiddleware` instance - a description that carries a moving number produces flaky snapshots. !!! warning Descriptions are meant to be logged and committed as test fixtures, so `describe()` must not expose credentials. The built-ins report live objects by name or presence - `LoggingMiddleware` reports its logger's name, and `TelemetryMiddleware` reports only the *keys* of its `span_attributes`. ## Choosing the Right Hook If you are unsure where a behavior belongs, use this rule of thumb: - Use `on_turn()` when the behavior is about the entire request/response lifecycle. - Use `on_llm_call()` when the behavior is about what goes into or comes out of the model. - Use `on_tool_execution()` when the behavior is about tool safety, auditing, or result shaping across tools (or when branching on `event.name` is acceptable). - Use **tool-scoped** `middleware=[...]` on `@tool` / `Agent.tool` / `Toolkit.tool` when the behavior applies only to that tool's definition; see [Tool middleware](tools/tool_middleware.md){.internal-link}. - Use `on_human_input()` when the behavior is about intercepting, logging, or transforming human-in-the-loop requests and responses. For related runtime customization patterns, see [Tools](tools/tools.md), [Tool middleware](tools/tool_middleware.md), [Prompt Management](system_prompts.md), and [Events Streaming](advanced/stream.md). --- # Metrics Source: https://docs.ag2.ai/docs/user-guide/metrics/ # Metrics AG2 includes a `MetricsMiddleware` that emits Prometheus-compatible counters and histograms for agent turns, LLM calls, tool executions, human input requests, and LLM token usage. Use metrics when you need operational dashboards and alerts for latency, success rates, error types, and token consumption. Use [Telemetry](telemetry/agent.md) when you need request-level traces and span details. ## Installation Install the Prometheus metrics extra: ```bash pip install "ag2[metrics]" ``` If your agent uses OpenAI in the examples below, install both extras: ```bash pip install "ag2[openai,metrics]" ``` ## Quick Start Create one Prometheus `CollectorRegistry`, create one `MetricsMiddleware` for that registry, then reuse the middleware on the agents that should emit into it. ```python import asyncio from prometheus_client import CollectorRegistry, start_http_server from ag2 import Agent from ag2.config import OpenAIConfig from ag2.middleware import MetricsMiddleware async def main() -> None: registry = CollectorRegistry() metrics = MetricsMiddleware(registry=registry) start_http_server(8000, registry=registry) agent = Agent( "assistant", prompt="You are a helpful assistant.", config=OpenAIConfig(model="gpt-4o-mini"), middleware=[metrics], ) reply = await agent.ask("Give me a one sentence project status template.") print(reply.body) if __name__ == "__main__": asyncio.run(main()) ``` Run the process, then open `http://localhost:8000/metrics` or configure Prometheus to scrape that endpoint. ## Metrics Reference ### Counters | Metric | Labels | Description | |---|---|---| | `ag2_agent_turns_total` | `agent`, `outcome`, `error_type` | Total agent turns. One turn is one `ask()` call or follow-up reply turn. | | `ag2_llm_calls_total` | `agent`, `provider`, `model`, `outcome`, `finish_reason`, `error_type` | Total LLM calls made by the agent runtime. | | `ag2_llm_tokens_total` | `agent`, `provider`, `model`, `token_type` | Total LLM tokens reported by model responses. | | `ag2_tool_calls_total` | `agent`, `tool`, `outcome`, `error_type` | Total tool executions. Tool error results and raised exceptions are counted as errors. | | `ag2_human_input_requests_total` | `agent`, `outcome`, `error_type` | Total human-in-the-loop input requests. Timeouts and cancellations are counted as errors. | ### Histograms | Metric | Labels | Buckets | Description | |---|---|---|---| | `ag2_agent_turn_duration_seconds` | `agent`, `outcome`, `error_type` | `0.1`, `0.25`, `0.5`, `1.0`, `2.5`, `5.0`, `10.0`, `20.0`, `30.0`, `60.0`, `120.0`, `300.0`, `600.0`, `+Inf` | Full agent turn duration in seconds. | | `ag2_llm_call_duration_seconds` | `agent`, `provider`, `model`, `outcome`, `error_type` | `0.05`, `0.1`, `0.25`, `0.5`, `1.0`, `2.5`, `5.0`, `10.0`, `15.0`, `20.0`, `30.0`, `+Inf` | LLM call duration in seconds. Streaming calls are measured through the final response. | | `ag2_tool_duration_seconds` | `agent`, `tool`, `outcome`, `error_type` | `0.01`, `0.025`, `0.05`, `0.1`, `0.25`, `0.5`, `1.0`, `2.5`, `5.0`, `10.0`, `+Inf` | Tool execution duration in seconds. | | `ag2_human_input_duration_seconds` | `agent`, `outcome`, `error_type` | `1.0`, `5.0`, `10.0`, `30.0`, `60.0`, `120.0`, `300.0`, `600.0`, `1200.0`, `1800.0`, `3600.0`, `+Inf` | Human input wait duration in seconds. | Prometheus histograms also expose `_bucket`, `_count`, and `_sum` series for each metric. ## Label Reference | Label | Values | Notes | |---|---|---| | `agent` | Agent name or `unknown` | Derived from the running `Agent`. | | `provider` | LLM Provider name (for ex.`openai`, `anthropic`, `gemini`, etc.) | | | `model` | Model name or `unknown` | | | `outcome` | `success`, `error` | `error` is used for raised exceptions, tool error results, timeouts, and cancellations. | | `finish_reason` | Provider finish reason or `unknown` | | | `error_type` | Exception class name or empty string | Empty string means the operation succeeded. | | `token_type` | `input`, `output`, `total`, `cache_read_input`, `cache_creation_input`, `thinking` | `total` overlaps `input` + `output`. | | `tool` | Tool name | Derived from the `ToolCallEvent` name. | Missing or empty label values are normalized to `unknown`, except `error_type`, which is an empty string for successful operations. Token values of zero are not emitted, and missing token fields are not synthesized as zero. Which `token_type` values appear depends on what the provider reports (for example, `thinking` only for reasoning models and `cache_read_input` / `cache_creation_input` only when prompt caching is active). ## Configuration `MetricsMiddleware` accepts: | Parameter | Type | Default | Description | |---|---|---|---| | `registry` | `prometheus_client.CollectorRegistry` | Required | Prometheus registry where AG2 metrics are registered and later exposed. |
!!! warning The middleware registers all counters and histograms during construction. The same `CollectorRegistry` cannot be used to construct multiple `MetricsMiddleware` instances because Prometheus collectors must be registered once per registry.
## CollectorRegistry Lifecycle Create one `MetricsMiddleware` per `CollectorRegistry`, then share that middleware instance across agents that should emit to the same registry. ```python from prometheus_client import CollectorRegistry from ag2 import Agent from ag2.config import OpenAIConfig from ag2.middleware import MetricsMiddleware registry = CollectorRegistry() metrics = MetricsMiddleware(registry=registry) assistant = Agent( "assistant", config=OpenAIConfig(model="gpt-4o-mini"), middleware=[metrics], ) reviewer = Agent( "reviewer", config=OpenAIConfig(model="gpt-4o-mini"), middleware=[metrics], ) ``` ## Prometheus Integration For a standalone worker or service process, the simplest integration is `start_http_server()` from `prometheus_client`: ```python from prometheus_client import CollectorRegistry, start_http_server from ag2.middleware import MetricsMiddleware registry = CollectorRegistry() metrics = MetricsMiddleware(registry=registry) start_http_server(8000, registry=registry) ``` Prometheus can scrape the endpoint with a job like: ```yaml scrape_configs: - job_name: ag2 metrics_path: /metrics static_configs: - targets: - localhost:8000 ``` For ASGI applications, Prometheus Python client can expose the same `CollectorRegistry` as an ASGI app via `make_asgi_app(registry=registry)`. For example, you can serve it directly with `uvicorn`: ```python import uvicorn from prometheus_client import CollectorRegistry, make_asgi_app from ag2.middleware import MetricsMiddleware registry = CollectorRegistry() metrics = MetricsMiddleware(registry=registry) app = make_asgi_app(registry=registry) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000) ``` Then scrape `http://localhost:8000/`. If you prefer the CLI form, expose the same `app` object from a module and run `uvicorn your_module:app`. ## Grafana Dashboard Useful starting panels: LLM call rate by provider, model, and outcome: ```promql sum by (provider, model, outcome) (rate(ag2_llm_calls_total[5m])) ``` LLM token rate by token type: ```promql sum by (provider, model, token_type) (rate(ag2_llm_tokens_total[5m])) ``` 95th percentile LLM latency: ```promql histogram_quantile( 0.95, sum by (le, provider, model) (rate(ag2_llm_call_duration_seconds_bucket[5m])) ) ``` Agent turn success rate: ```promql sum(rate(ag2_agent_turns_total{outcome="success"}[5m])) / sum(rate(ag2_agent_turns_total[5m])) ``` Tool error rate by tool and error type: ```promql sum by (tool, error_type) (rate(ag2_tool_calls_total{outcome="error"}[5m])) ``` Create separate rows for agent turns, LLM calls, tools, human input, and token usage. Use `outcome` and `error_type` as dashboard filters when investigating failures. ## Best Practices - Keep `agent`, `tool`, `provider`, and `model` labels low-cardinality. Do not include user IDs, request IDs, session IDs, or other unbounded values in agent or tool names. - Use stable, descriptive agent names such as `support_triage` or `invoice_reviewer`. - Reuse a single `MetricsMiddleware` instance per registry across all agents in one process. - If you combine metrics with `RetryMiddleware` and want each retry attempt counted as a separate LLM call, register `RetryMiddleware` before `MetricsMiddleware`. - Streaming LLM durations are recorded after the final response arrives, so dashboards update when the streamed call completes. --- # Testing Source: https://docs.ag2.ai/docs/user-guide/testing/ AG2 provides a built-in `TestConfig` utility in the `ag2.testing` module to help you write unit tests for your agents. It allows you to mock LLM responses and simulate tool execution scenarios without making actual API calls. ## How to mock LLM answers To mock LLM answers, you can use `TestConfig` in place of a standard model configuration. Pass the expected responses as arguments to `TestConfig`. Each argument represents the mocked response for a sequential turn in the conversation. ```python import pytest from ag2 import Agent from ag2.testing import TestConfig @pytest.mark.asyncio async def test_mock_llm_answer(): # Provide a TestConfig with the mocked string response agent = Agent("test_agent") # Ask the agent, passing the TestConfig res = await agent.ask( "Hi!", config=TestConfig("This is a mocked response."), ) # The agent returns the mocked response assert res.body == "This is a mocked response." ``` ## How to test tool execution You can also use `TestConfig` to yield tool calls. This allows you to test both successful tool execution and error handling. By providing a `ToolCallEvent` as the first response and a string as the final response, you can simulate a complete agent-tool interaction loop. ### Success case To test a successful tool execution, pass a `ToolCallEvent` followed by the final answer you expect the LLM to provide after the tool executes. ```python import pytest from ag2 import Agent from ag2.events import ToolCallEvent from ag2.testing import TestConfig @pytest.mark.asyncio async def test_tool_success(): # Define a tool def my_tool() -> str: return "tool execution result" agent = Agent("test_agent", tools=[my_tool]) # Configure TestConfig to first return a ToolCallEvent, then a final string answer test_config = TestConfig( ToolCallEvent(name="my_tool"), "final result", ) res = await agent.ask("Please use my_tool", config=test_config) # After the tool is called and succeeds, the agent returns the second mocked event assert res.body == "final result" ``` ### Errors You can test how your agent reacts when a tool raises an exception, or when an unregistered tool is requested by the LLM. If a tool raises an exception during execution, it will propagate up to the `ask` method. You can catch and assert this exception in your tests. ```python import pytest from ag2 import Agent from ag2.events import ToolCallEvent from ag2.testing import TestConfig @pytest.mark.asyncio async def test_tool_raise_exc(): # Define a tool that raises an error def failing_tool() -> str: raise ValueError("Something went wrong") test_config = TestConfig( ToolCallEvent(name="failing_tool"), "result", ) agent = Agent( "test_agent", config=test_config, tools=[failing_tool], ) with pytest.raises(ValueError, match="Something went wrong"): await agent.ask("Hi!") ``` #### Tool not found If the LLM attempts to call a tool that hasn't been registered with the agent, a `ToolNotFoundError` is raised. ```python import pytest from ag2 import Agent from ag2.events import ToolCallEvent from ag2.exceptions import ToolNotFoundError from ag2.testing import TestConfig @pytest.mark.asyncio async def test_tool_not_found(): # Mock the LLM returning a tool call for "unregistered_tool" test_config = TestConfig(ToolCallEvent(name="unregistered_tool")) # Agent is created WITHOUT any tools agent = Agent("test_agent", config=test_config) with pytest.raises(ToolNotFoundError, match="Tool `unregistered_tool` not found"): await agent.ask("Hi!") ``` --- # The Agent Harness Source: https://docs.ag2.ai/docs/user-guide/agent_harness/ A bare `Agent` is just a model loop. The **harness** is the set of opt-in primitives you compose onto it to give it richer capabilities - context assembly, persistent knowledge, sub-task spawning, and the supporting middleware they wire in. This page is the configuration reference for those primitives. For the conversational entry point (`agent.ask()`, tools, HITL, observing events), see [Agent Communication](agents.md). ## Constructor ```python Agent( name: str, prompt: str | Callable | Iterable = (), *, config: ModelConfig | None = None, tools: Iterable = (), middleware: Iterable = (), observers: Iterable = (), dependencies: dict | None = None, variables: dict | None = None, response_schema: ResponseProto | type | None = None, hitl_hook: HumanHook | None = None, plugins: Iterable[Plugin] = (), assembly: Iterable[AssemblyPolicy] = (), knowledge: KnowledgeConfig | None = None, tasks: TaskConfig | Literal[False] | None = None, ) ``` The loop-related parameters (`config`, `tools`, `middleware`, `observers`, `prompt`, ...) are covered in [Agent Communication](agents.md) and the parameter-specific guides. The harness hooks are `assembly=`, `knowledge=`, and `tasks=`, each documented below. ## `assembly=` - context policies A list of [`AssemblyPolicy`](advanced/assembly.md) instances. When non-empty, the Agent wires an internal `AssemblerMiddleware` at the outermost position of the middleware chain so your policies transform `(prompts, events)` before every LLM call. ```python from ag2 import Agent from ag2.policies import ( AlertPolicy, SlidingWindowPolicy, WorkingMemoryPolicy, ) agent = Agent( "assistant", config=config, assembly=[ WorkingMemoryPolicy(), # inject /memory/working.md AlertPolicy(), # deliver ObserverAlerts, halt on FATAL SlidingWindowPolicy(max_events=50), # cap history footprint ], ) ``` Order matters - see the [ordering rule in the assembly doc](advanced/assembly.md#ordering-matters). `AssemblerMiddleware.validate_order()` will flag known problematic compositions. ## `knowledge=` - KnowledgeConfig Groups everything that involves the [`KnowledgeStore`](advanced/knowledge_store.md): the store itself, optional bootstrap, and optional compaction + aggregation strategies. ```python from dataclasses import dataclass @dataclass class KnowledgeConfig: store: KnowledgeStore expose_tool: bool = True write_event_log: bool = True compact: CompactStrategy | None = None compact_trigger: CompactTrigger | None = None aggregate: AggregateStrategy | None = None aggregate_trigger: AggregateTrigger | None = None bootstrap: StoreBootstrap | None = None ``` | Field | What it does | |---|---| | `store` | Registered in `context.dependencies[KnowledgeStore]` so policies like `WorkingMemoryPolicy` / `EpisodicMemoryPolicy` can read it. | | `expose_tool` | When `True` (default), the agent gets an auto-injected `knowledge` action-group tool that lets the LLM call `read` / `write` / `list` / `delete` on the store. Set to `False` when the store should be policy-only - the model never sees the tool, and the bootstrap SKILL.md text drops its "use the `knowledge` tool" sentence. | | `write_event_log` | When `True` (default), the agent persists its stream history to `/log/{stream_id}.jsonl` at the end of each `ask()`. Set to `False` to keep the store free of stream logs (e.g. when the store is purely user-facing memory). | | `compact` / `compact_trigger` | Wires a compaction middleware that fires [`compact()`](advanced/compaction.md) between turns when the trigger thresholds are exceeded. | | `aggregate` / `aggregate_trigger` | Wires an aggregation middleware that fires [`aggregate()`](advanced/aggregation.md) on the configured cadence. Failures emit `AggregationFailed` on the stream - see [Aggregation > Wiring onto an Agent](advanced/aggregation.md#wiring-onto-an-agent) for the full lifecycle event triple. | | `bootstrap` | Runs once on first use to seed the store. `None` falls back to `DefaultBootstrap(mention_tool=expose_tool)`, so the generated SKILL.md text matches whether the LLM can actually call the `knowledge` tool. | ```python from ag2 import Agent, KnowledgeConfig from ag2.aggregate import AggregateTrigger, ConversationSummaryAggregate from ag2.compact import CompactTrigger, TailWindowCompact from ag2.knowledge import DiskKnowledgeStore from ag2.policies import WorkingMemoryPolicy from pathlib import Path store = DiskKnowledgeStore(Path("./knowledge")) agent = Agent( "assistant", config=main_config, knowledge=KnowledgeConfig( store=store, compact=TailWindowCompact(target=100), compact_trigger=CompactTrigger(max_events=200), aggregate=ConversationSummaryAggregate(config=summarizer_config), aggregate_trigger=AggregateTrigger(every_n_turns=10, on_end=True), ), assembly=[WorkingMemoryPolicy()], ) ``` The compaction and aggregation middleware are opt-in per field: passing `compact=` without `compact_trigger=` still works (a default `CompactTrigger()` with all thresholds disabled is used). Omit a strategy entirely and the corresponding middleware is not wired. ## `tasks=` - TaskConfig Sub-task delegation is **off by default** - a bare Agent has no `run_subtask` / `run_subtasks` tools. Pass `tasks=TaskConfig(...)` to opt in, and the Agent will auto-inject the pair of sub-task tools that let the LLM spawn isolated child Agents to handle self-contained work. `TaskConfig` configures how those children are built. ```python from dataclasses import dataclass @dataclass class TaskConfig: config: ModelConfig | None = None prompt: str = ( "You are a task agent. Complete the assigned task thoroughly and " "concisely. Return only the result." ) include_tools: Iterable[str] | None = None exclude_tools: Iterable[str] = () extra_tools: Iterable[Callable | Tool] = () max_concurrency: int | None = None ``` | Field | What it does | |---|---| | `config` | The `ModelConfig` used for sub-task Agents. Falls back to the parent Agent's `config`. | | `prompt` | Default system prompt for sub-task Agents. | | `include_tools` | Allowlist of parent-tool names to inherit. `None` means "inherit all". | | `exclude_tools` | Blocklist of parent-tool names to drop. Applied after `include_tools`. | | `extra_tools` | Additional tools given to sub-tasks that the parent does not have. | | `max_concurrency` | Optional cap shared by all sub-tasks spawned by this Agent. Extra work waits for a slot. `None` keeps concurrency unbounded. | By default a sub-task Agent inherits **all** of the parent's user-supplied tools. Sub-tasks are themselves constructed with `tasks=False` (the Agent default), so they have no `run_subtask` / `run_subtasks` tools - recursive delegation is structurally impossible and no depth limit is needed. ```python from ag2 import Agent, TaskConfig agent = Agent( "orchestrator", config=main_config, tools=[search, fetch_url, summarize], tasks=TaskConfig( config=worker_config, # cheaper model for sub-tasks prompt="You are a focused worker; one step only.", include_tools=["search", "fetch_url"], # don't expose summarize to children max_concurrency=4, # bound model/API fan-out ), ) ``` ### `tasks=False` - the default `tasks=False` is the Agent default, so a bare Agent never spawns children. You only need to pass it explicitly to be self-documenting; otherwise just omit `tasks=` entirely. ```python focused = Agent( "summarizer", prompt="Summarise the input. Do not delegate.", config=main_config, # tasks=False is the default - no run_subtask / run_subtasks tools. ) ``` ## `run_subtask` / `run_subtasks` - auto-injected tools When you opt in via `tasks=TaskConfig(...)`, the Agent exposes two tools to the LLM: - `run_subtask(task: str)` - spawn one sub-task Agent. Useful when the LLM has a single self-contained piece of work to delegate. - `run_subtasks(tasks: list[str], parallel: bool = True)` - spawn multiple sub-tasks in one tool call. Defaults to running them concurrently with `asyncio.gather`; pass `parallel=False` only when later tasks depend on earlier results. The LLM is told (via the tool descriptions) that it can call `run_subtask` multiple times in parallel within a single response, and that `run_subtasks` is the deliberate fan-out form. Each child gets a fresh `MemoryStream` and the parent's tools (filtered by `TaskConfig`). Set `TaskConfig(max_concurrency=N)` when that fan-out needs a model, API, or resource limit. The cap is shared by both tools on the Agent, so several parallel `run_subtask` calls and a concurrent `run_subtasks` call cannot bypass one another. For a more explicit, named delegate where the parent LLM sees a tool like `task_researcher` instead of generic `run_subtask`, use [`Agent.as_tool()`](#agentas_tool). The two patterns can coexist: a coordinator can have both auto-injected sub-tasks and a named `task_researcher` tool. See [Subagents](subagents.md) for the full sub-task delegation guide - context flow and custom streams for self-delegation via `as_tool()`. ## Agent.as_tool() Expose any Agent as a `FunctionTool` so another Agent can invoke it like any other tool: ```python child = Agent( "researcher", prompt="Answer the objective concisely.", config=main_config, ) parent = Agent( "lead", config=main_config, tools=[child.as_tool(description="Delegate fact-finding to a researcher.")], ) reply = await parent.ask("Find out where Melbourne is.") ``` `as_tool()` returns a `FunctionTool` named `task_{child.name}` that accepts an `objective` parameter and forwards it into the child's stream. See [Subagents](subagents.md) for sub-task streams, depth limiting, and stream factories. ## Reading an agent's composition An Agent reports what it is made of, so you can compare two agents, assert on composition in a test, or log it. Each constructor argument has a matching read-only property: | Property | Reports | |---|---| | `name` | agent name | | `system_prompt` | static prompt fragments, in order | | `dynamic_prompt` | dynamic prompt hooks, in registration order | | `tools` | attached tools | | `middleware` | agent-level middleware, in registration order | | `dependencies` / `variables` | injected values, keyed as supplied | | `config` / `response_schema` / `hitl_hook` | model config, schema, human-input hook | | `observers` / `assembly` / `tasks` | observers, assembly policies, `TaskConfig` | Sequences come back as tuples and mappings as read-only views, so an agent cannot be changed through them. ```python def fingerprint(agent: Agent) -> tuple: return ( agent.name, agent.system_prompt, tuple(t.name for t in agent.tools), tuple(m.description for m in agent.middleware), tuple(sorted(map(str, agent.dependencies))), ) assert fingerprint(built_one_way) == fingerprint(built_another_way) ``` ### Middleware entries `agent.middleware` and `tool.middleware` yield entries pairing each middleware with its [description](middleware.md#describing-middleware): ```python for entry in agent.middleware: entry.description # what it is and how it was configured entry.middleware # the object itself ``` Both are needed. A description says what a middleware *is*; identity says whether two tools *share* one. A single rate limiter across ten tools and ten separate ones configured identically produce equal descriptions, so `is` on `entry.middleware` is the only thing that tells them apart: ```python first, second = agent.tools first.middleware[0].middleware is second.middleware[0].middleware # True when both tools hold one instance, so they share its state ``` !!! note Entries are built when you read the property, so `agent.middleware[0] is agent.middleware[0]` is `False`. Compare `entry.middleware`, never the entry itself. ## Turn lifecycle Each `await agent.ask(...)` runs through the middleware chain in this order (outermost -> innermost): ``` 1. AssemblerMiddleware (if assembly=[...]) 2. _HaltCheckMiddleware (if assembly=[...] - watches for HaltEvent) 3. _CompactionMiddleware (if knowledge.compact configured) 4. _AggregationMiddleware (if knowledge.aggregate configured) 5. User-provided middleware (retry, rate-limit, logging, ...) 6. LLM client (innermost) ``` The internal harness middleware (`_AssemblerMiddleware`, `_HaltCheckMiddleware`, `_CompactionMiddleware`, `_AggregationMiddleware`) are assembled conditionally - you only pay for what you turn on. Lifecycle events emitted during a turn include `ObserverStarted` / `ObserverCompleted`, `CompactionCompleted`, `AggregationCompleted`, and `HaltEvent`. Subscribe to any of them via an [Observer](advanced/observers.md) or a stream subscriber. --- # Tasks Source: https://docs.ag2.ai/docs/user-guide/tasks/ A `Task` is a framework-core wrapper any `Agent` can use to give a unit of work a trackable lifecycle. While the task is active, the framework emits `TaskStarted`, `TaskProgress`, `TaskCompleted`, `TaskFailed`, and `TaskExpired` events on a stream - so observers (UIs, watchers, mirrors, test harnesses) can follow along without participating in execution. !!! note Tasks are **agent-owned**. The framework does not assign or schedule them. Standalone usage requires no observers - events fly past harmlessly if nothing subscribes. ## When to Use a Task Use a Task whenever a unit of work has a beginning, an end, and observable progress that you want to surface beyond your own function's return value: - **Long-running pipelines** where downstream consumers want progress checkpoints. - **HITL approvals** where a UI needs to know the task is waiting on a human. - **Test harnesses** that assert a sequence of lifecycle events. For lightweight LLM-driven sub-agent delegation see [Subagents](subagents.md) - that's a different feature that wraps an Agent in a `run_subtask` tool. ## Quick Start ```python from ag2 import Agent from ag2.config import AnthropicConfig agent = Agent("indexer", config=AnthropicConfig(model="claude-sonnet-5")) async with agent.task("index documents") as task: await task.progress({"stage": "discover", "files": 12}) await task.progress({"stage": "index", "indexed": 12}) await task.complete({"indexed": 12}) print(task.state) # TaskState.COMPLETED print(task.metadata.result) # {'indexed': 12} ``` The `async with` block opens the lifecycle. On clean exit the task auto-completes with `result=None` if you didn't call `complete()` or `fail()` yourself. ## Lifecycle States ``` CREATED -> RUNNING -> COMPLETED (terminal, success) -> FAILED (terminal, exception or explicit fail) -> EXPIRED (terminal, TTL elapsed) ``` | State | Meaning | |---|---| | `CREATED` | The Task object exists but `__aenter__` has not run. `task.task_id` and `task.metadata` raise. | | `RUNNING` | Inside the `async with` block. Progress events allowed. | | `COMPLETED` | Reached `complete()` or clean block exit. | | `FAILED` | Reached `fail()` or block exited via exception. | | `EXPIRED` | TTL elapsed; emitted by an external observer (e.g. a network hub's TTL sweeper). | The terminal states are immutable - once set, further `complete() / fail() / progress()` calls are silent no-ops. The set is exported as `ag2.task.TERMINAL_TASK_STATES`. ## API Reference ### `Agent.task(...)` ```python agent.task( title: str, *, description: str = "", payload: dict[str, Any] | None = None, capability: str | None = None, ttl_seconds: int | None = None, context: ConversationContext | None = None, ) -> Task ``` | Parameter | Type | Description | |---|---|---| | `title` | `str` | Short objective shown on every event. | | `description` | `str` | Optional longer description. | | `payload` | `dict[str, Any] \| None` | Initial payload merged into `TaskSpec`. | | `capability` | `str \| None` | Tags the task with a capability name; used by network mirrors. | | `ttl_seconds` | `int \| None` | Sets `metadata.expires_at`. The Task does not self-expire - an external observer must call `task.expire()` when the TTL elapses. | | `context` | `ConversationContext \| None` | If supplied, events flow on `context.stream` and `ag2.task` is stamped into `context.dependencies` for the duration of the block. If omitted, the Task creates a private `MemoryStream` on entry. | Returns an unentered `Task`. Use as `async with agent.task(...) as task:`. ### `Task` instance methods | Method | Description | |---|---| | `await task.progress(payload)` | Emits `TaskProgress`; merges `payload` into `metadata.progress` and stamps `last_progress_at`. No-op if already terminal. | | `await task.complete(result=None)` | Terminal. Emits `TaskCompleted`; sets `metadata.result` and `state = COMPLETED`. | | `await task.fail(error)` | Terminal. Accepts a string (wrapped in `RuntimeError`) or any `BaseException`. Emits `TaskFailed`; sets `state = FAILED`. | | `await task.expire()` | Terminal. Emits `TaskExpired`; sets `state = EXPIRED`. Called by external TTL observers. | ### Properties | Property | Available before `__aenter__`? | |---|---| | `task.state` | Yes - returns `TaskState.CREATED`. | | `task.task_id` | No - raises `RuntimeError`. | | `task.metadata` | No - raises `RuntimeError`. | | `task.context` | No - raises `RuntimeError`. | ## Bound Context vs. Standalone ```python from ag2.context import ConversationContext from ag2.stream import MemoryStream ctx = ConversationContext(stream=MemoryStream()) async with agent.task("with-ctx", context=ctx) as task: ... ``` Passing a `ConversationContext` shares the stream with the rest of your agent's run, so observers and middleware already attached to that stream see the lifecycle events. Without a context, the Task creates a private stream on entry. Events still fire - but only observers attached to that private stream see them. Useful for one-off background work that doesn't need to surface anywhere. ## Auto-Complete and Auto-Fail The `async with` block has these guarantees: - **Clean exit, no terminal call** -> auto `complete(result=None)`. - **Exception inside the block** -> auto `fail(exc)`, then the exception propagates. - **Already terminal at exit time** -> nothing further happens. ```python try: async with agent.task("flaky") as task: raise ValueError("boom") except ValueError as exc: print(task.state) # TaskState.FAILED print(task.metadata.error) # 'boom' ``` ## Observing the Lifecycle Subscribe directly on the bound stream to capture every lifecycle event in order. ```python from ag2.events import TaskCompleted, TaskProgress, TaskStarted stream = MemoryStream() ctx = ConversationContext(stream=stream) stream.subscribe( lambda ev: print(type(ev).__name__, getattr(ev, "payload", "")), sync_to_thread=False, ) async with agent.task("watched", context=ctx) as task: await task.progress({"step": "fetch"}) await task.complete({"ok": True}) ``` !!! note `TaskProgress` is marked transient - it is delivered live to subscribers but **not** persisted to the stream's storage. Subscribe before the events fire to capture them. `TaskStarted`, `TaskCompleted`, `TaskFailed`, and `TaskExpired` are persisted normally. ## Reading the Active Task with `TaskInject` Inside an `async with agent.task(...)` block, the framework stamps the active Task into `context.dependencies["ag2.task"]`. Two ways to read it: ### Direct access ```python async with agent.task("work", context=ctx) as task: active = ctx.dependencies["ag2.task"] assert active is task ``` ### `TaskInject` annotation `TaskInject` is a fast_depends-resolvable annotation that injects the active Task into any function the dependency-injection machinery resolves - most usefully a `@tool` body. ```python from ag2 import tool from ag2.task import TaskInject @tool async def report(message: str, task: TaskInject) -> str: if task is None: return "no active task" await task.progress({"tool_message": message}) return f"reported on task {task.task_id}" ``` The injection has `default=None`, so always treat `task` as possibly `None` and null-check before use. ## TTL and Expiry Setting `ttl_seconds=N` populates `metadata.expires_at` but **does not** start a timer. The Task primitive itself never self-expires - that's by design, so a standalone Task with no observer doesn't spawn a background task. Instead, an external observer (e.g. a network hub's TTL sweeper, a periodic watch) checks `expires_at` and calls `task.expire()` when due. For self-contained TTL behaviour, wire up a sweeper in your application: ```python async def sweep(task: Task, deadline: datetime) -> None: while task.state == TaskState.RUNNING: if datetime.now(timezone.utc) >= deadline: await task.expire() return await asyncio.sleep(1.0) ``` ## TaskSpec and TaskMetadata Two small dataclasses surface around a Task: - `TaskSpec` - what the task is doing: `title`, `description`, `payload`, optional `capability`. Created by `Agent.task(...)`. - `TaskMetadata` - mutable lifecycle record updated on each transition: `task_id`, `owner_id`, `spec`, `state`, ISO-8601 timestamps, `progress`, `result`, `error`, optional `session_id`. ```python async with agent.task("survey", description="probe upstream", payload={"region": "us"}) as task: print(task.metadata.spec.title) # 'survey' print(task.metadata.spec.payload) # {'region': 'us'} print(task.metadata.owner_id) # 'researcher' print(task.metadata.started_at) # ISO 8601 string ``` --- # Subagents Source: https://docs.ag2.ai/docs/user-guide/subagents/ Subagents let agents delegate work to other agents through tool calling. The calling agent's LLM decides when and what to delegate, and each sub-task runs on its own isolated stream with independent history. ## Why Use Subagents Breaking work across multiple agents gives you: - **Separation of concerns** - each agent has a focused prompt, tools, and config tuned for its role. - **Independent context** - sub-tasks run on fresh streams, so history doesn't grow unboundedly. - **LLM-driven orchestration** - the calling agent decides when to delegate, what context to pass, and how to use the result. !!! note When the LLM returns multiple tool calls in a single response, the framework dispatches them concurrently. Each concurrent sub-task gets its own copy of variables, so they don't interfere with each other. For auto-injected sub-tasks, `TaskConfig(max_concurrency=N)` bounds the total fan-out across both `run_subtask` and `run_subtasks`. !!! tip For lightweight self-delegation where the parent doesn't need a *named* delegate, opt in to the auto-injected `run_subtask` / `run_subtasks` tools by passing `tasks=TaskConfig(...)` - see [`tasks=` in The Agent Harness](agent_harness.md#tasks-taskconfig). Use `Agent.as_tool()` (below) when you want a distinct, purpose-named tool exposed to the LLM. ## Subagents API Use `Agent.as_tool()` to make one agent available as a tool for another. ```python from ag2 import Agent from ag2.config import AnthropicConfig config = AnthropicConfig("claude-sonnet-5") researcher = Agent( "researcher", prompt="You are a thorough researcher. Provide concise factual findings.", config=config, tools=[search_tool], ) writer = Agent( "writer", prompt="You are a skilled writer. Turn research into clear prose.", config=config, ) coordinator = Agent( "coordinator", prompt="First delegate research, then pass findings to the writer.", config=config, tools=[ researcher.as_tool(description="Research a topic and return findings."), writer.as_tool(description="Write an article. Pass research notes in the context parameter."), ], ) reply = await coordinator.ask("Write a short article about the history of Python.") print(await reply.content()) ``` The coordinator's LLM sees two tools - `task_researcher` and `task_writer` - and calls them as needed. Each call spawns the target agent on a fresh stream, runs it to completion, and returns the result. The calling agent's LLM sees a tool named `task_{agent.name}` with `objective` (required) and `context` (optional) parameters. The `context` tool parameter is how the calling LLM shares relevant information with the sub-task: ```python task_writer( objective="Write an article about Python's history", context="Key findings: Created by Guido van Rossum in 1991. Named after Monty Python." ) ``` `as_tool()` accepts these parameters: | Parameter | Type | Description | |---|---|---| | `description` | `str` | Tool description shown to the LLM (required) | | `name` | `str | None` | Override the default `task_{agent.name}` tool name | | `stream` | `StreamFactory | None` | Factory to create custom streams for sub-tasks (see [Sub-Task Streams](#sub-task-streams)) | | `middleware` | `Iterable[ToolMiddleware]` | Tool middleware applied to the delegate tool (e.g., `approval_required`) | You can also use `subagent_tool()` directly for more control: ```python from ag2.tools.subagents import subagent_tool coordinator = Agent( "coordinator", config=config, tools=[ subagent_tool(researcher, description="Research a topic."), ], ) ``` ## Background Subagents `as_tool()` and `subagent_tool()` are **blocking**: the tool call doesn't return until the sub-task finishes, so the calling LLM waits for the result before doing anything else. `background_agent_tool()` is **fire-and-forget** - it starts the sub-task, returns a task id immediately, and lets the parent LLM keep working in the same turn. The result is delivered later as a follow-up message to the parent. ```python from ag2.tools.subagents import background_agent_tool coordinator = Agent( "coordinator", prompt="Kick off long-running research in the background, then keep helping the user.", config=config, tools=[ background_agent_tool(researcher, description="Research a topic in the background."), ], ) reply = await coordinator.ask("Start deep research on Rust async runtimes, then outline the article.") ``` The calling LLM sees a tool named `background_task_{agent.name}` with the same `objective` (required) and `context` (optional) parameters as `as_tool()`. Calling it returns `"Background task started: {task_id}"` right away, so the LLM can issue other tool calls or continue reasoning while the sub-task runs. How it behaves: - The sub-task runs concurrently via `run_task` on its own stream (a fresh `MemoryStream` by default, or one built by the `stream` factory). - The parent `Agent.ask` loop **keeps running and will not return** until the background task finishes. Once it does, its result is pushed back into the parent's inbox as a follow-up turn (via `context.enqueue`), so the parent LLM can react to it. - On success the follow-up message reports the task result; on failure it reports the error - the exception does not propagate to the parent. `background_agent_tool()` accepts these parameters: | Parameter | Type | Description | |---|---|---| | `agent` | `Agent` | The agent to run as a background sub-task (positional) | | `description` | `str` | Tool description shown to the LLM (required) | | `name` | `str | None` | Override the default `background_task_{agent.name}` tool name | | `stream` | `StreamFactory | None` | Factory to create custom streams for sub-tasks (see [Sub-Task Streams](#sub-task-streams)) | | `middleware` | `Iterable[ToolMiddleware]` | Tool middleware applied to the delegate tool (e.g., `approval_required`) | !!! note Background sub-tasks are for work the parent can fire off and revisit later within the *same* turn - the `ask` call still awaits their completion before returning. They are not detached jobs that outlive the turn. ## Self-Delegation An agent can delegate to itself to break complex work into independent sub-tasks. Each sub-task runs as a fresh copy of the agent with its own stream and history. ```python analyst = Agent( "analyst", prompt=( "You have search and sub_task tools. " "Only use sub_task when the task has clearly independent parts. " "Otherwise handle it directly with search." ), config=config, tools=[search_tool], ) analyst.add_tool( analyst.as_tool( description="Break work into a focused sub-task for independent analysis.", name="sub_task", ) ) reply = await analyst.ask("Compare Python vs Rust for web APIs: performance, DX, and ecosystem.") ``` The analyst's LLM may call `sub_task` multiple times - one per aspect - then synthesise the results. ## Dynamic Agents `dynamic_agent()` lets the calling LLM **construct** an ephemeral agent at runtime - picking a name, system prompt, and a subset of available tools per objective - instead of pre-defining each delegate as a named `as_tool()`. Use it when the orchestrator should compose a focused worker on demand for each sub-task. ```python from ag2 import Agent from ag2.tools.dynamic import dynamic_agent from ag2.config import OpenAIConfig config = OpenAIConfig(model="gpt-4o-mini") orchestrator = Agent( "orchestrator", config=config, prompt=( "You orchestrate sub-tasks by calling create_and_run_agent. " "For each task, invent a focused agent name and system prompt, " "include only the tools the child genuinely needs, " "and pass a clear objective." ), tools=[ dynamic_agent(available_tools=[calc, web_search], config=config), ], ) reply = await orchestrator.ask("What is 17 * 25 + 4? Use a child agent.") ``` The orchestrator's LLM sees one tool, `create_and_run_agent(spec, objective)`. Each call spawns an ephemeral child `Agent` on a fresh stream, runs the objective via `run_task`, and returns the reply string. `dynamic_agent()` accepts these parameters: | Parameter | Type | Description | |---|---|---| | `available_tools` | `Iterable[Tool | Callable[..., Any]]` | Pool of tools the spawned child may pick from by name | | `config` | `ModelConfig` | Model configuration used for every spawned child | | `middleware` | `Iterable[ToolMiddleware]` | Tool middleware applied to `create_and_run_agent` | ### The AgentSpec the LLM constructs The LLM builds an `AgentSpec` on every call. It is JSON-serializable and captures the declarative parts of the child: | Field | Type | Purpose | |---|---|---| | `name` | `str` | Display name of the spawned child | | `prompt` | `list[str]` | System prompt for the child | | `tool_names` | `list[str]` | Subset of `available_tools` names to give the child | | `response_schema` | `ResponseSchemaSpec | None` | Optional structured output schema | Unrecognised keys are ignored, as Pydantic would anyway, but they raise an `UnknownSpecFieldWarning` naming the key and suggesting the closest valid field - so a misspelled `tool_nammes` no longer silently produces a child with no tools. To make it fatal, promote the warning: ```python import warnings from ag2.spec import UnknownSpecFieldWarning warnings.simplefilter("error", UnknownSpecFieldWarning) ``` ### How the LLM discovers available tool names The pool's names and descriptions are **rendered into the `create_and_run_agent` tool description automatically** when the factory is built. The calling agent's system prompt does not need to enumerate them - the LLM discovers the valid names from the tool schema. !!! tip If the LLM picks a name not in the pool, the framework returns `Error: unknown tools [...]. Available: [...]` as a recoverable string. The LLM reads the hint and retries with a corrected spec - no exception propagates to the caller. The auto-rendered menu prevents this in the common case. !!! warning Spawned children are themselves constructed **without** `dynamic_agent`, so they cannot recursively spawn further dynamic agents. Recursion is structurally impossible - no depth limit needed. ## Sub-Task Streams ### Default Behavior By default, each sub-task creates a fresh `MemoryStream`. The sub-task's history is isolated - it doesn't carry over between invocations. It means that subagent has no information about previous calls or results. It just sees the current call and the context. | What | Behavior | Why | |---|---|---| | **Dependencies** | Copied | Isolated - child mutations don't affect parent | | **Variables** | Copied; synced back on success | Concurrent-safe - user variable mutations propagate back | | **History** | Fresh stream | Clean context - the LLM passes relevant info via `context` parameter | | **Depth counter** | Incremented in child; excluded from sync-back | Internal bookkeeping - never leaks to parent | | **Agent prompt, tools, config** | Inherited | The sub-agent brings its own capabilities | ### Persistent Stream `persistent_stream()` gives the same agent a consistent stream across multiple invocations within a context. The sub-task's history accumulates across calls rather than starting fresh each time: ```python from ag2.tools.subagents import persistent_stream researcher.as_tool( description="Research a topic", stream=persistent_stream(), ) ``` It stores the stream ID in `context.dependencies` keyed by `f"ag:{agent.name}:stream"` and reuses the parent stream's storage backend. This is useful when the sub-agent benefits from seeing its own prior work - for example, a researcher that should avoid repeating searches. ### Shared Stream Instance Pass a `Stream` instance directly to keep a handle on the sub-agent's events. Every delegation runs against that same stream, so you can read its history - or subscribe to it - from the calling code: ```python from ag2 import MemoryStream sub_stream = MemoryStream() researcher.as_tool( description="Research a topic", stream=sub_stream, ) # after the run events = await sub_stream.history.get_events() ``` !!! warning Reusing one stream also makes the sub-agent **stateful**: because the history is shared, each delegation sees every earlier delegation's turns, and token cost grows with them. This is the same trade-off as `persistent_stream()`. To observe events *without* accumulating history, use a [custom factory](#custom-factory) that returns a fresh stream per call and subscribe to each one. Unlike `persistent_stream()`, an instance you construct yourself carries its own storage backend unless you pass the parent's - `MemoryStream(storage=parent_stream.history.storage)`. ### Custom Factory For full control, pass any callable matching `StreamFactory = Callable[[Agent, Context], Stream]`: ```python from ag2 import Agent, Context from ag2.streams.redis import RedisStream def make_redis_stream(agent: Agent, ctx: Context) -> RedisStream: return RedisStream(MY_REDIS_URL, prefix=f"ag2:sub:{agent.name}") researcher.as_tool( description="Research a topic", stream=make_redis_stream, ) ``` !!! note `stream=` accepts exactly these three shapes - `None`, a `Stream` instance, or a factory. Anything else raises `TypeError` when the tool is built, not when it first runs. Passing the class (`stream=MemoryStream`) instead of an instance (`stream=MemoryStream()`) is rejected the same way. The same rules apply to `background_agent_tool()`. --- # Coding with AI Assistants Source: https://docs.ag2.ai/docs/user-guide/coding_with_ai/ Set up your AI coding assistant (Claude Code, Cursor, Copilot, Codex, Windsurf, or any agent) so it can build **AG2** apps with you using current, accurate APIs and examples. !!! tip "Point your agent at this page" Paste this page's link into your assistant and ask it to follow the setup. It can install the AG2 skills and configure itself. Everything below is written to be runnable copy-paste. ## Why this matters AG2 (`ag2`) is an async, protocol-driven API. Models were largely trained on the older `autogen` / `pyautogen` surface, so out of the box an assistant will reach for stale patterns such as synchronous `ConversableAgent`, `initiate_chat`, and the like. The setup on this page gives your assistant three things it otherwise lacks: **AG2-specific skills**, an up-to-date **docs reference**, and **project rules** that keep it on the current AG2 API. !!! warning "Current API only" The earlier synchronous `ConversableAgent` / `initiate_chat` API has been removed from the package. Models were trained on it, so point your assistant at the current AG2 docs and skills, not legacy examples it may have memorized. ## Step 1: Install the AG2 Skills The most critical and dev-accelerating step. [ag2-skills](https://github.com/ag2ai/ag2-skills) is a catalog of [Agent Skills](https://agentskills.io/), on-demand instruction packs that teach an assistant how to build with AG2. Each skill loads only its name and description until it's relevant, then pulls in the full recipe. Skills cover quickstart, custom tools, the multi-agent network, middleware, memory, structured output, evaluation, and more. The fastest path uses the [`skills` CLI](https://skills.sh): === "All skills" ```bash # Install the full AG2 skill catalog npx skills add ag2ai/ag2-skills ``` === "One skill" ```bash # Install just the quickstart (good first taste) npx skills add ag2ai/ag2-skills@ag2-quickstart ``` === "Manual (Claude Code)" ```bash # Clone and copy individual skills into your user skills directory git clone https://github.com/ag2ai/ag2-skills.git cp -r ag2-skills/skills/ag2-overview ~/.claude/skills/ cp -r ag2-skills/skills/ag2-quickstart ~/.claude/skills/ ``` !!! tip "Where to start" After installing, tell your assistant to load **`ag2-overview`** (a map of AG2 capabilities) and **`ag2-quickstart`** (a minimal working agent). From there it can pull in the specific skill it needs, such as `ag2-add-custom-tool` or `ag2-network-quickstart`. ## Step 2: Point your assistant at the AG2 docs Skills teach patterns; the docs keep your assistant honest about the *exact* current signatures. The biggest risk is your assistant falling back on the **classic** `autogen` API it was trained on (`ConversableAgent`, `initiate_chat`, `GroupChat`) - now removed from the package. Give it current ground truth: - **Live AG2 docs:** [`https://docs.ag2.ai/docs/user-guide/agents/`](https://docs.ag2.ai/docs/user-guide/agents/) - point your agent at this section (most assistants can fetch a URL). - **AG2 docs source (Markdown):** [`ag2ai/ag2/website/docs/user-guide`](https://github.com/ag2ai/ag2/tree/main/website/docs/user-guide) - the raw `.mdx` your agent can read directly from GitHub. - **AG2 `llms.txt`:** [`https://docs.ag2.ai/llms.txt`](https://docs.ag2.ai/llms.txt) - a machine-readable index of the AG2 docs following the [llms.txt standard](https://llmstxt.org/), plus [`llms-full.txt`](https://docs.ag2.ai/llms-full.txt) for the entire AG2 docs in one file. Both are AG2-scoped, so they never point your agent at the classic API. Then anchor your prompt: *"Build with `ag2` only. If a signature is unfamiliar, check the AG2 docs before writing code - do not use the classic `autogen` API."* !!! warning "Prefer the official skills and docs" Generic docs-MCP servers and code indexers may surface outdated `ConversableAgent` / `initiate_chat` examples cached from older AG2 releases. Rely on the AG2 Skills and the docs links above, which always reflect the current API. ## Step 3: Add project rules to your repo A rules file pins AG2 conventions for every session in your project. The open [AGENTS.md](https://agents.md/) standard is read by Cursor, Copilot, Codex, Gemini CLI, Windsurf, and others; Claude Code reads `CLAUDE.md`. Drop one (or both, a symlink works) at your repo root: ```markdown # AGENTS.md This project is built on **AG2** (`ag2`). Follow these rules. ## API surface - Import only from `ag2` and its submodules (`ag2.config`, `ag2.tools`, ...). Do NOT use the legacy `autogen` / `pyautogen` API (`ConversableAgent`, `initiate_chat`) - it has been removed. - Agents are async. Use `await agent.ask(...)` and `await reply.ask(...)`. ## Conventions - Do not use `from __future__ import annotations`. - Public signatures accept `str | os.PathLike[str]`; use `pathlib.Path` internally. - Prefer top-level imports; no imports inside functions. ## Docs & skills - Install and use the AG2 skills: `npx skills add ag2ai/ag2-skills`. - AG2 docs: https://docs.ag2.ai/docs/user-guide/agents/ (machine-readable index at /llms.txt). - When unsure of a signature, check the AG2 docs before writing code. - Do NOT trust generic docs indexers - they surface the retired classic API. ``` === "Claude Code" ```bash # Claude Code reads CLAUDE.md - symlink it to a single source of truth ln -s AGENTS.md CLAUDE.md ``` === "Cursor" ```bash # Cursor reads AGENTS.md, or project rules under .cursor/rules/ mkdir -p .cursor/rules cp AGENTS.md .cursor/rules/ag2.md ``` === "Copilot" ```bash # GitHub Copilot reads .github/copilot-instructions.md mkdir -p .github cp AGENTS.md .github/copilot-instructions.md ``` ## Per-tool summary | Assistant | Skills | AG2 docs | Project rules | |---|---|---|---| | **Claude Code** | `npx skills add ag2ai/ag2-skills` or copy to `~/.claude/skills/` | point at the docs URL | `CLAUDE.md` (symlink to `AGENTS.md`) | | **Cursor** | `npx skills add ag2ai/ag2-skills` | point at the docs URL | `AGENTS.md` or `.cursor/rules/` | | **Copilot / Codex / Windsurf** | `npx skills add ag2ai/ag2-skills` | point at the docs URL | `AGENTS.md` (Copilot: `.github/copilot-instructions.md`) | | **Any agent** | paste `SKILL.md` contents into context | paste the docs (or `llms-full.txt`) | paste the rules above into your prompt | ## Tips & caveats - **Start from the quickstart.** Have your assistant scaffold from `ag2-quickstart` rather than inventing structure - see [Agent Communication](agents.md) and [Model Configuration](model_configuration.md). - **Prefer the documented high-level API** (`ag2`, `ag2.tools`, `ag2.config`) over reaching into internals. - **Always review generated code.** Models still drift toward legacy `autogen` patterns; verify imports come from `ag2` and that calls are `await`ed. - **Run the tests.** AG2 is async throughout - encourage your assistant to write and run [tests](testing.md) with `TestConfig` instead of hitting a live model. !!! note "Going further" The skill catalog mirrors these docs section-for-section - `ag2-network-quickstart` for [multi-agent networks](network/overview.md), `ag2-structured-output` for [typed responses](structured_output.md), `ag2-evaluation` for the [eval framework](evaluation/evaluation.md), and more. Install the set once and your assistant can pull in whichever it needs. --- # Multimodal Inputs Source: https://docs.ag2.ai/docs/user-guide/multimodal/inputs/ # Multimodal Inputs AG2 agents can process images, audio, video, and documents alongside text. The input event system provides a unified API across providers - you create inputs the same way regardless of which model you use. ## Input Types | Factory Function | Creates | Description | | :--- | :--- | :--- | | `ImageInput(...)` | Image input | JPEG, PNG, GIF, WebP | | `AudioInput(...)` | Audio input | WAV, MP3, OGG, FLAC, AAC | | `VideoInput(...)` | Video input | MP4, WebM, MOV, MKV, MPEG | | `DocumentInput(...)` | Document input | PDF, TXT, HTML, Markdown, CSV, JSON, Office formats | Each factory function supports multiple ways to provide the data: ```python from ag2.events import ImageInput, AudioInput, VideoInput, DocumentInput # From a URL image = ImageInput("https://example.com/photo.jpg") # From a local file path image = ImageInput(path="photo.jpg") # From raw bytes image = ImageInput(data=raw_bytes, media_type="image/png") # From a pre-uploaded file ID (provider-specific) image = ImageInput(file_id="file-abc123") ``` --- ## Using Inputs with Agents Pass inputs directly to `agent.ask()` as positional arguments alongside text: ```python from ag2 import Agent from ag2.config import GeminiConfig from ag2.events import ImageInput agent = Agent( "vision_agent", "You are a helpful assistant that describes images.", config=GeminiConfig(model="gemini-3-flash-preview"), ) image = ImageInput("https://example.com/photo.jpg") reply = await agent.ask("Describe this image in detail.", image) print(reply.body) ``` You can pass multiple inputs in a single request: ```python image1 = ImageInput("https://example.com/before.jpg") image2 = ImageInput("https://example.com/after.jpg") reply = await agent.ask("Compare these two images.", image1, image2) ``` --- ## Provider Support Not all providers support all input types. The table below shows what each provider accepts: | Input Type | OpenAI | OpenAI Responses | Gemini | Anthropic | xAI | Mistral | Bedrock | | :--- | :---: | :---: | :---: | :---: | :---: | :---: | :---: | | **Text** | Yes | Yes | Yes | Yes | Yes | Yes | Yes | | **Image (URL)** | Yes | Yes | Yes | Yes | Yes | Yes | - | | **Image (binary)** | Yes | Yes | Yes | Yes | Yes | Yes | Yes | | **Audio (URL)** | - | - | Yes | - | - | - | - | | **Audio (binary)** | Yes | - | Yes | - | - | - | - | | **Video (URL)** | - | - | Yes | - | - | - | - | | **Video (binary)** | - | - | Yes | - | - | - | Yes | | **Document (URL)** | - | Yes | Yes | Yes | Yes | Yes | - | | **Document (binary)** | - | - | Yes | Yes | Yes | Yes | Yes | | **File ID** | - | Yes | - | Yes | Yes | Yes | - | If you pass an unsupported input type to a provider, an `UnsupportedInputError` is raised with a clear message indicating what is not supported and by which provider. --- ## Provider-Specific Details ### Gemini Gemini has the broadest multimodal support - it accepts images, audio, video, and documents in all forms (URL, binary, and local file path). **YouTube URLs** are supported directly: ```python from ag2.events import VideoInput video = VideoInput("https://www.youtube.com/watch?v=dQw4w9WgXcQ") reply = await agent.ask("Summarize this video.", video) ``` **Google Files API** - for large files (>20MB), upload via the Google Files API first and pass the returned URI: ```python from google import genai from ag2.events import VideoInput client = genai.Client() uploaded = client.files.upload(file="large_video.mp4") # Wait for processing to complete import time while uploaded.state.name == "PROCESSING": time.sleep(2) uploaded = client.files.get(name=uploaded.name) video = VideoInput(uploaded.uri) reply = await agent.ask("Describe this video.", video) ``` #### Vendor Metadata Gemini supports provider-specific settings via `vendor_metadata` on binary inputs. These map to Gemini Part fields: | Key | Type | Description | | :--- | :--- | :--- | | `media_resolution` | `str` \| `dict[str, Any]` | Controls token allocation per image/video frame | | `video_metadata` | `dict[str, Any]` | Video clipping (`start_offset`, `end_offset`) and frame rate (`fps`) | | `display_name` | `str` | Display name for the file | **Media resolution** - control quality vs cost tradeoff for images and video frames: ```python from ag2.events import ImageInput # Lower resolution = fewer tokens = lower cost image = ImageInput( data=raw_bytes, media_type="image/jpeg", vendor_metadata={"media_resolution": "MEDIA_RESOLUTION_LOW"}, ) ``` Available values: `MEDIA_RESOLUTION_LOW`, `MEDIA_RESOLUTION_MEDIUM`, `MEDIA_RESOLUTION_HIGH`, `MEDIA_RESOLUTION_ULTRA_HIGH`. A `dict` is also accepted and maps straight onto Gemini's `PartMediaResolution`, whose `level` and `num_tokens` keys are mutually exclusive - set exactly one: ```python from ag2.events import ImageInput image = ImageInput( data=raw_bytes, media_type="image/jpeg", vendor_metadata={"media_resolution": {"level": "MEDIA_RESOLUTION_MEDIUM"{{ "}}" }}, ) ``` !!! warning "Setting both keys is rejected" Passing `{"level": ..., "num_tokens": ...}` together makes Gemini reject the request with `400 INVALID_ARGUMENT` (`oneof field 'value' is already set`). **Video clipping and frame rate** - process only a portion of a video or adjust the sampling rate: ```python from ag2.events import VideoInput video = VideoInput( path="lecture.mp4", vendor_metadata={ "video_metadata": { "start_offset": "60s", "end_offset": "120s", "fps": 0.5, }, }, ) reply = await agent.ask("Summarize this section of the video.", video) ``` **Display name** - attach a name to the file for reference: ```python from ag2.events import DocumentInput doc = DocumentInput( path="report.pdf", vendor_metadata={"display_name": "Q4 Financial Report"}, ) ``` ### OpenAI OpenAI supports images via both the Completions and Responses APIs. Audio binary input (WAV, MP3) is supported in the Completions API. The Responses API additionally supports file IDs and document URLs. #### Vendor Metadata OpenAI supports `vendor_metadata` for image detail control: ```python from ag2.events import ImageInput image = ImageInput( data=raw_bytes, media_type="image/png", vendor_metadata={"detail": "low"}, # "low", "high", or "auto" ) ``` ### Anthropic Anthropic supports images (JPEG, PNG, GIF, WebP) and documents (PDF) via URL, base64, or File ID. Audio and video are not supported. **File ID** - upload files via the Anthropic Files API (beta) and reference by ID: ```python import anthropic from ag2.events import ImageInput, DocumentInput client = anthropic.Anthropic() # Upload an image uploaded = client.beta.files.upload( file=("photo.jpg", open("photo.jpg", "rb"), "image/jpeg"), ) # Reference by file_id - filename determines block type (image vs document) image = ImageInput(file_id=uploaded.id, filename="photo.jpg") reply = await agent.ask("Describe this image.", image) ``` #### Vendor Metadata Anthropic supports `vendor_metadata` for prompt caching on content blocks: ```python from ag2.events import DocumentInput doc = DocumentInput( path="report.pdf", vendor_metadata={"cache_control": {"type": "ephemeral"{{ "}}" }}, ) ``` ### xAI xAI supports images (URL and binary), documents (URL and binary), and pre-uploaded file IDs. Audio and video are not currently supported - passing them raises `UnsupportedInputError`. **File ID** - reference a file previously uploaded via the xAI Files API: ```python from ag2.events import ImageInput, DocumentInput image = ImageInput(file_id="file-abc123", filename="photo.jpg") doc = DocumentInput(file_id="file-xyz789", filename="report.pdf") ``` #### Vendor Metadata xAI reads `detail` for image quality control from two **different** attributes depending on the input source - `vendor_metadata` for binary, `metadata` for URL. Mixing them up means the value is silently ignored and xAI falls back to `"auto"`. **Binary image** - set `detail` via `vendor_metadata`: ```python from ag2.events import BinaryInput, BinaryType image = BinaryInput( raw_bytes, media_type="image/png", kind=BinaryType.IMAGE, vendor_metadata={"detail": "low"}, # "low", "high", or "auto" ) ``` **URL image** - set `detail` via `metadata` (not `vendor_metadata`): ```python from ag2.events import UrlInput, BinaryType image = UrlInput( "https://example.com/photo.jpg", kind=BinaryType.IMAGE, metadata={"detail": "low"}, ) ``` !!! note The factory `ImageInput(url=...)` does not forward `metadata`. To configure `detail` on a URL image, construct `UrlInput` directly as shown above. **Document filename** - xAI requires a filename for binary documents. When sending raw bytes, either provide one via `vendor_metadata={"filename": ...}`, or rely on the auto-derived fallback (`file.` from the media type, e.g. `file.pdf` for `application/pdf`): ```python from ag2.events import BinaryInput, BinaryType doc = BinaryInput( pdf_bytes, media_type="application/pdf", kind=BinaryType.DOCUMENT, vendor_metadata={"filename": "Q4-report.pdf"}, ) ``` ### Mistral Mistral accepts images and documents by URL, binary, or file ID. Audio and video raise `UnsupportedInputError`. Plain text counts as a document, so `DocumentInput(data=b"...", media_type="text/plain")` works alongside PDFs. **Files API** - uploads must use the `ocr` purpose to be referenceable from a message. That is the default in AG2, so a plain upload is ready to use: ```python from ag2.config import MistralConfig from ag2.events import FileIdInput from ag2.files import FilesAPI config = MistralConfig(model="mistral-small-latest") files = FilesAPI(config) uploaded = await files.upload(data=pdf_bytes, filename="report.pdf") # Reference it as many times as you like - the bytes are never resent reply = await agent.ask("Summarise this report.", FileIdInput(file_id=uploaded.file_id)) ``` The other purposes are content-validated: `batch` and `fine-tune` require newline-delimited JSON in a specific schema and reject anything else with HTTP 422. !!! note "Remote URLs are fetched by Mistral" URL inputs are retrieved server-side, so the host must serve them to Mistral. Some hosts refuse, and the API returns "File could not be fetched from url". Pass the bytes instead when that happens. #### Vendor Metadata Mistral reads `detail` for image quality from `vendor_metadata` for binary images and `metadata` for URL images - the same split as xAI: ```python from ag2.events import BinaryInput, BinaryType image = BinaryInput( raw_bytes, media_type="image/png", kind=BinaryType.IMAGE, vendor_metadata={"detail": "low"}, # "low", "high", or "auto" ) ``` **Document name** - taken from `metadata={"filename": ...}` for URL documents and `vendor_metadata={"filename": ...}` for binary ones (set automatically when using `path=`). ### Amazon Bedrock The Bedrock Converse API accepts **binary sources only** - images (JPEG, PNG, GIF, WebP), documents (PDF, CSV, DOC, DOCX, XLS, XLSX, HTML, TXT, Markdown), and video (MP4, WebM, MOV, MKV, and more; Amazon Nova models). URL inputs and file IDs raise `UnsupportedInputError` - Bedrock has no Files API, so source data from a URL must be downloaded and passed as bytes: ```python from ag2 import Agent from ag2.config import BedrockConfig from ag2.events import DocumentInput, ImageInput agent = Agent( "vision_agent", "You describe images and summarize documents.", config=BedrockConfig(model="us.amazon.nova-lite-v1:0", region_name="us-east-1"), ) image = ImageInput(path="photo.jpg") doc = DocumentInput(data=pdf_bytes, media_type="application/pdf") reply = await agent.ask("Describe the image and summarize the document.", image, doc) ``` !!! note Modality support also depends on the **model** behind the Converse API: Amazon Nova models accept images, documents, and video; many others (e.g. DeepSeek) are text-only and return a `ValidationException` from AWS for non-text blocks. The provider raises `UnsupportedInputError` only for inputs the Converse API itself cannot carry. **Document name** - Converse requires a name for document blocks. It is taken from `vendor_metadata={"filename": ...}` (set automatically when using `path=`), sanitized to the characters Converse allows (alphanumerics, single spaces, hyphens, parentheses, brackets), and falls back to `"document"` when absent: ```python from ag2.events import BinaryInput, BinaryType doc = BinaryInput( pdf_bytes, media_type="application/pdf", kind=BinaryType.DOCUMENT, vendor_metadata={"filename": "Q4 report.pdf"}, ) ``` --- # Image Generation Source: https://docs.ag2.ai/docs/user-guide/multimodal/image_generation/ # Image Generation Some providers can produce images as part of an agent's reply. Generated images are always returned the same way - as a list of `BinaryResult` objects on `reply.files` - regardless of which provider produced them. AG2 exposes two different mechanisms, because the providers expose two different APIs: | Provider | Mechanism | How to enable | | :--- | :--- | :--- | | OpenAI | `ImageGenerationTool` (server-side tool, Responses API) | Add the tool to the agent | | Mistral | `ImageGenerationTool` (server-side tool, chat completions) | Add the tool to the agent | | Gemini | `IMAGE` response modality on an image model | Set `response_modalities=["TEXT", "IMAGE"]` | !!! note Gemini's image modality is **not** available on OpenAI or Mistral, and neither of their tools works on Gemini. Each provider uses its own mechanism below. ## Reading generated images Every generated image is a [`BinaryResult`](inputs.md) on `reply.files`. It carries the raw bytes and a `metadata` dict; the image's media type is stored under the `media_type` key. ```python reply = await agent.ask("Generate an image of a red bicycle on a beach.") for index, image in enumerate(reply.files): media_type = image.metadata.get("media_type", "image/png") extension = media_type.split("/")[-1] with open(f"image_{index}.{extension}", "wb") as file: file.write(image.data) ``` `reply.files` is empty when the model returns only text, so it is safe to iterate even when no image was produced. ## OpenAI Add `ImageGenerationTool` to an agent configured with the **Responses API** (`OpenAIResponsesConfig`). The model decides when to call the tool and the generated image is appended to `reply.files`. ```python from ag2 import Agent from ag2.config import OpenAIResponsesConfig from ag2.tools import ImageGenerationTool agent = Agent( "designer", config=OpenAIResponsesConfig(model="gpt-4.1"), tools=[ ImageGenerationTool( quality="high", size="1024x1024", output_format="png", background="transparent", ), ], ) reply = await agent.ask("Generate a logo for a coffee shop.") image = reply.files[0] ``` | Parameter | Description | | :--- | :--- | | `quality` | `"low"`, `"medium"`, `"high"`, or `"auto"` | | `size` | e.g. `"1024x1024"`, `"1536x1024"`, or `"auto"` | | `background` | `"transparent"`, `"opaque"`, or `"auto"` | | `output_format` | `"png"`, `"jpeg"`, or `"webp"` | | `output_compression` | 0-100, for jpeg/webp only | | `partial_images` | 1-3, number of partial images to stream | !!! warning `ImageGenerationTool` requires the Responses API. Using it with the Chat Completions API (`OpenAIConfig`) raises an `UnsupportedToolError`. ## Mistral Add `ImageGenerationTool` to an agent configured with `MistralConfig`. Unlike OpenAI, this works on the ordinary chat-completions endpoint - no separate API. ```python from ag2 import Agent from ag2.config import MistralConfig from ag2.tools import ImageGenerationTool agent = Agent( "designer", config=MistralConfig(model="mistral-medium-latest"), tools=[ImageGenerationTool()], ) reply = await agent.ask("Generate a logo for a coffee shop.") image = reply.files[0] ``` Mistral's tool takes no configuration, so `quality`, `size`, `background`, and `output_format` are accepted on the AG2 tool but ignored. The generated image is returned to AG2 as a short-lived signed URL rather than inline bytes; AG2 downloads it so `reply.files` stays consistent with the other providers. The originating URL is kept on the image's `metadata` under the `url` key. !!! note Mistral runs the whole exchange server-side and bills every internal turn, so a generated image costs far more tokens than a plain reply. The model often returns the image with no accompanying text, leaving `reply.body` empty - check `reply.files` rather than the body to detect success. Mistral's other server-side tools (`web_search`, `code_interpreter`, `document_library`) belong to its Agents API and raise `UnsupportedToolError` on chat completions. ## Gemini Gemini does not use a tool for image generation. Instead, you select an **image-capable model** and request the `IMAGE` response modality via `response_modalities`. The model returns the image inline and AG2 surfaces it on `reply.files`. ```python from ag2 import Agent from ag2.config import GeminiConfig config = GeminiConfig( model="gemini-3.1-flash-image", response_modalities=["TEXT", "IMAGE"], ) agent = Agent("designer", config=config) reply = await agent.ask("Generate an image of a friendly robot waving hello.") image = reply.files[0] ``` !!! note Image output requires an image-capable Gemini model (for example `gemini-3.1-flash-image`). Requesting the `IMAGE` modality on a text-only model returns no image. Include `"TEXT"` alongside `"IMAGE"` so the model can still return any accompanying text in `reply.body`. `response_modalities` is also available on `VertexAIConfig` for Gemini models served through Vertex AI. ### Controlling size and aspect ratio Gemini does not take a pixel `size` string like OpenAI. Instead, pass a `types.ImageConfig` through `image_config` to set the aspect ratio and a resolution tier: ```python from google.genai import types from ag2.config import GeminiConfig config = GeminiConfig( model="gemini-3.1-flash-image", response_modalities=["TEXT", "IMAGE"], image_config=types.ImageConfig(aspect_ratio="16:9", image_size="2K"), ) ``` | Field | Values | | :--- | :--- | | `aspect_ratio` | e.g. `"1:1"`, `"4:3"`, `"3:4"`, `"16:9"`, `"9:16"`, `"21:9"` | | `image_size` | resolution tier - `"1K"`, `"2K"` (higher tiers are model-dependent) | `image_config` is a full passthrough of the SDK's `types.ImageConfig`, so any other field it supports (such as `person_generation`) is available too. It is also accepted on `VertexAIConfig`. ## Editing an existing image To edit an image instead of generating one from scratch, pass it in as an [`ImageInput`](inputs.md) alongside your instruction. The edited image is returned on `reply.files`, exactly like a freshly generated one - so the same image can be sent back in for further rounds of editing. === "OpenAI" ```python linenums="1" from ag2 import Agent from ag2.config import OpenAIResponsesConfig from ag2.events import ImageInput from ag2.tools import ImageGenerationTool agent = Agent( "editor", config=OpenAIResponsesConfig(model="gpt-4.1"), tools=[ImageGenerationTool(size="1024x1024", output_format="png")], ) reply = await agent.ask( "Put a party hat on the robot. Keep everything else the same.", ImageInput(path="robot.png"), ) edited = reply.files[0] ``` === "Gemini" ```python linenums="1" from ag2 import Agent from ag2.config import GeminiConfig from ag2.events import ImageInput config = GeminiConfig(model="gemini-3.1-flash-image", response_modalities=["TEXT", "IMAGE"]) agent = Agent("editor", config=config) reply = await agent.ask( "Put a party hat on the robot. Keep everything else the same.", ImageInput(path="robot.png"), ) edited = reply.files[0] ``` `ImageInput` also accepts raw bytes - `ImageInput(data=image.data, media_type="image/png")` - which lets you feed a generated image straight back in for another edit without writing it to disk. --- # Agent Tools Source: https://docs.ag2.ai/docs/user-guide/tools/tools/ # Agent Tools Tools allow agents to interact with the outside world. By providing tools, you enable your agents to perform actions such as executing code, fetching data from APIs, querying databases, or performing complex calculations. Under the hood, a tool is a standard Python function accompanied by a schema that describes its purpose, inputs, and outputs to the underlying Large Language Model (LLM). ## Creating Agent Tools The easiest way to create a tool is by using the `@tool` decorator. This decorator automatically parses your function's signature, type hints, and docstring to generate a schema that the LLM can understand. For the best results, **always provide clear type hints and a descriptive docstring**. The LLM relies heavily on these to know when and how to invoke your tool. ```python from ag2 import tool @tool def calculate_shipping_cost(destination: str, weight_kg: float) -> str: """Calculates the shipping cost for a package based on its destination and weight. """ return "$15.00" ``` Once defined, you can equip an agent with this capability by passing the tool to the `tools` list during the agent's initialization. ```python from ag2 import Agent agent = Agent(name="ShippingAssistant", tools=[calculate_shipping_cost]) ``` !!! note For simpler use cases, you can pass an undecorated Python function directly to the agent's `tools` list. The framework will automatically convert it into a fully-fledged tool under the hood, extracting the schema from the signature and docstring just like the decorator would. ```python linenums="1" from ag2 import Agent def get_weather(location: str) -> str: """Returns the current weather for a given location.""" return "Sunny, 22°C" # get_weather is automatically converted to a tool agent = Agent(name="WeatherBot", tools=[get_weather]) ``` ### Registering Tools via a Decorator Alternatively, you can register a tool directly with an agent instance using its `@my_agent.tool` decorator. This approach is particularly useful when you need to dynamically add capabilities to an agent after it has been created, or when you are logically organizing your code by attaching specific tools to specific agent instances. ```python from ag2 import Agent agent = Agent(name="CalculatorBot") @agent.tool def multiply(a: int, b: int) -> int: """Multiplies two integers and returns the result.""" return a * b ``` ### Tool middleware To run async hooks around **one** tool (e.g. argument normalization, result redaction, bundled auditing), pass `middleware=[...]` with `ToolMiddleware` callables. See the dedicated [Tool middleware](tool_middleware.md) page. ## Synchronous and Asynchronous Tools Agent interactions are naturally asynchronous. To support this, tools are executed within an asynchronous event loop by default. However, you have the flexibility to define your tool functions as either synchronous (`def`) or asynchronous (`async def`). To ensure that heavy computational tasks or blocking I/O operations do not freeze the entire application, **synchronous tools are automatically executed in a separate thread** by default. ```python # This synchronous tool runs in a separate thread to prevent blocking @tool def fetch_data_sync(url: str) -> str: """Fetches data from a URL using a blocking request library.""" import requests return requests.get(url).text # This native asynchronous tool runs directly in the main event loop @tool async def fetch_data_async(url: str) -> str: """Fetches data from a URL using an async request library.""" import aiohttp async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.text() ``` ### Disabling Threaded Execution If you have a synchronous function that executes very quickly (e.g., simple string manipulation or math) and you want to avoid the minor overhead of thread creation, you can disable threaded execution by passing `sync_to_thread=False` to the `@tool` decorator. ```python # This tool runs synchronously in the main event loop, # without a separate thread @tool(sync_to_thread=False) def format_name(first_name: str, last_name: str) -> str: """Formats a full name.""" return f"{last_name.upper()}, {first_name.capitalize()}" ``` !!! warning When `sync_to_thread=False` is set, the synchronous tool runs directly within the asynchronous context. If the function performs time-consuming operations (like network requests or large loops), **it will block the entire event loop** until it finishes, preventing other agents or asynchronous tasks from making progress. ## Customizing Tool Schemas LLMs perform best when they have precise constraints and detailed instructions. The framework automatically generates a tool's schema from its function signature and docstring. For more granular control, you can define minimum/maximum values, enforce specific formats, or override the tool's name. You can override the basic properties directly in the decorator: ```python @tool( name="custom_math_tool", description="Performs advanced mathematical operations.", ) def math_op(a: int, b: int) -> int: return a + b ``` !!! note Explicitly setting the name and description overrides the automatically generated values. ### Deep Schema Customization with Pydantic Under the hood, arguments are serialized and validated using [Pydantic](https://docs.pydantic.dev/latest/) schemas. This means you can use standard `pydantic.Field` annotations to deeply customize individual schema parameters, providing the LLM with strict guidelines on what values are acceptable. ```python from typing import Annotated from pydantic import Field @tool def set_temperature( temp: Annotated[ int, Field( ..., description="The target temperature.", ge=10, le=30, ), ], mode: Annotated[ str, Field( ..., description="The thermostat mode.", pattern="^(heat|cool|auto)$", ), ] ) -> str: """Sets the thermostat to a specific temperature and mode.""" return f"Set to {temp}°C in {mode} mode." ``` ### Complete Custom Schema Example A complete example combining custom tool properties and strict parameter validation looks like this: ```python from typing import Annotated from pydantic import Field from ag2 import tool @tool( name="create_user_profile", description="Creates a new user profile in the database.", ) def create_profile( username: Annotated[ str, Field( ..., description="The chosen username. Must be alphanumeric.", min_length=3, max_length=20, ) ], age: Annotated[ int, Field( ..., description="The user's age. Must be 18 or older.", ge=18, ), ], ) -> str: return f"Profile for {username} created." ``` This configuration generates the following detailed JSON schema, ensuring the LLM understands exactly what inputs are required and valid: ```json { "description": "Creates a new user profile in the database.", "name": "create_user_profile", "parameters": { "properties": { "username": { "description": "The chosen username. Must be alphanumeric.", "maxLength": 20, "minLength": 3, "title": "Username", "type": "string" }, "age": { "description": "The user's age. Must be 18 or older.", "minimum": 18, "title": "Age", "type": "integer" } }, "required": [ "username", "age" ], "type": "object" } } ``` ## Execution Context Tools often need access to the broader execution context, such as injected dependencies, variables, or mechanisms for human-in-the-loop interactions. The AG2 framework supports these features natively. !!! note Under the hood, these contextual capabilities are powered by the [FastDepends](https://github.com/Lancetnik/FastDepends) library, ensuring robust and FastAPI-like dependency management. To access the execution context from within your tool, you can simply type-hint an argument with the `Context` object. ```python from ag2 import Context, tool @tool async def my_tool(context: Context) -> str: # Access context variables or dependencies here return f"Execution context: {context}" ``` For more detailed information on specific context features, see [Dependency Injection](../context/inject.md), [Context Variables](../context/variables.md), [Depends](../depends.md), [Human-in-the-loop](../context/human_in_the_loop.md). ## Returning Rich Tool Results By default, returning a plain `str` from a tool is the simplest option - the framework wraps it in a `TextInput` automatically. When you need more control over the returned content, you can use typed `Input` classes or compose multiple outputs with `ToolResult`. ### Input Types The framework provides typed input classes for text, structured data, images, audio, video, documents, and raw binary payloads. See [Multimodal Inputs](../multimodal/inputs.md) for the full reference including provider support and factory variants. Return any input type directly from a tool function just like you would a string: ```python from ag2 import DataInput, TextInput, tool @tool def get_status(task_id: str) -> TextInput: """Returns a human-readable status update.""" return TextInput(f"Task {task_id} is in progress.") @tool def get_user_profile(user_id: str) -> DataInput: """Returns a structured user profile.""" return DataInput({"id": user_id, "name": "Alice", "role": "admin"}) ``` ### Returning Images and Binary Data Use `ImageInput` when a tool needs to hand an image back to the model for further reasoning. The factory accepts a URL, a local file path, a pre-uploaded file ID, or raw bytes: ```python from ag2 import ImageInput, ToolResult, tool @tool def fetch_chart(chart_id: str) -> ImageInput: """Fetches a chart image by ID and returns it for visual analysis.""" url = f"https://charts.example.com/{chart_id}.png" return ImageInput(url) @tool def capture_screenshot(page: str) -> ImageInput: """Takes a screenshot of a page and returns it.""" import subprocess raw = subprocess.check_output(["screenshot-cli", page]) return ImageInput(data=raw, media_type="image/png") ``` When you have raw bytes of an arbitrary format, use `BinaryInput` directly and set the media type explicitly: ```python from ag2 import BinaryInput, ToolResult, tool @tool def export_pdf(report_id: str) -> ToolResult: """Exports a report as a PDF and returns it alongside a summary.""" pdf_bytes = _render_pdf(report_id) return ToolResult( f"Report {report_id} exported successfully.", BinaryInput(pdf_bytes, media_type="application/pdf"), ) ``` !!! note Not all LLM providers support every media type. Check your provider's documentation for the list of accepted MIME types and file formats. The framework passes the binary payload through as-is; format validation is the provider's responsibility. ### Returning Multiple Inputs Use `ToolResult` to combine multiple inputs into a single tool response. Each positional argument becomes a separate part the model receives: ```python from ag2 import ImageInput, ToolResult, tool @tool def analyze_product(product_id: str) -> ToolResult: """Returns a product image alongside its structured metadata.""" return ToolResult( ImageInput(f"https://cdn.example.com/products/{product_id}.jpg"), {"id": product_id, "name": "Widget Pro", "stock": 42}, ) ``` ## Returning a Final Tool Result By default, a tool result is sent back to the model so the agent can decide what to say next. When the tool itself already knows the exact final answer, you can return `ToolResult(..., final=True)` to end the turn immediately without another model round-trip. `ToolResult` accepts any `str` or `Input` as its first positional argument: ```python from ag2 import Agent, DataInput, TextInput, ToolResult, tool @tool def handoff_to_human(ticket_id: str) -> ToolResult: """Escalates a request and returns the final user-facing message.""" return ToolResult( f"Ticket {ticket_id} was escalated to a human agent.", final=True, ) @tool def get_exchange_rate(currency: str) -> ToolResult: """Returns the current exchange rate as structured data.""" return ToolResult( DataInput({"currency": currency, "rate": 1.23, "base": "USD"}), final=True, ) agent = Agent(name="SupportBot", tools=[handoff_to_human]) reply = await agent.ask("I need help with my ticket #123456.") print(reply.body) # Output: "Ticket 123456 was escalated to a human agent." ``` !!! note A `ToolResult` with `final=True` must contain **exactly one** part - either a `TextInput` or a `DataInput`. The content is returned as-is: `TextInput` produces the text directly, `DataInput` is JSON-serialized. This is especially useful for tools that: - perform an authoritative action and already know the exact reply - return a message that should not be paraphrased by the model - want to skip an extra LLM call for latency or cost reasons --- # Toolkits Source: https://docs.ag2.ai/docs/user-guide/tools/toolkits/ # Toolkits A `Toolkit` groups related tools into a single, reusable unit. Instead of passing individual tools one by one, you can bundle them into a toolkit and pass the whole collection to an agent. This is useful for organizing domain-specific capabilities (e.g., all database tools, all file-system tools) and sharing them across multiple agents. ```python from ag2 import Agent from ag2.tools import Toolkit def search_orders(query: str) -> str: """Searches the order database.""" return "Order #123" def cancel_order(order_id: str) -> str: """Cancels an order by its ID.""" return f"Order {order_id} cancelled." support_tools = Toolkit(search_orders, cancel_order) agent = Agent(name="SupportBot", tools=[support_tools]) ``` A toolkit accepts both plain functions and `@tool`-decorated functions in its `tools` list. Plain functions are automatically converted, just like when passing them directly to an agent. ## Registering Tools via a Decorator You can also add tools to a toolkit using the `@toolkit.tool` decorator, following the same pattern as `@agent.tool`: ```python from ag2.tools import Toolkit inventory = Toolkit() @inventory.tool def check_stock(item_id: str) -> int: """Returns the current stock count for an item.""" return 42 @inventory.tool( name="reorder_item", description="Places a reorder for a low-stock item.", ) def reorder(item_id: str, quantity: int) -> str: return f"Reordered {quantity} of {item_id}." ``` The toolkit can then be passed to any number of agents: ```python from ag2 import Agent warehouse_agent = Agent(name="WarehouseBot", tools=[inventory]) sales_agent = Agent(name="SalesBot", tools=[inventory]) ``` ## Combining Toolkits with Standalone Tools You can freely mix toolkits and individual tools in an agent's `tools` list: ```python from ag2 import tool @tool def escalate(reason: str) -> str: """Escalates the conversation to a human agent.""" return "Escalated." agent = Agent( name="SupportBot", tools=[support_tools, inventory, escalate], ) ``` ## Toolkit middleware Pass `middleware=[...]` to the `Toolkit` constructor to apply hooks to **every** tool in the set - both tools passed to the constructor and tools added later via `@toolkit.tool`: ```python from ag2 import Context from ag2.events import ToolCallEvent, ToolResultEvent from ag2.middleware import ToolExecution from ag2.tools import Toolkit async def log_calls( call_next: ToolExecution, event: ToolCallEvent, context: Context, ) -> ToolResultEvent: print(f"Calling {event.name}") return await call_next(event, context) support_tools = Toolkit(search_orders, cancel_order, middleware=[log_calls]) ``` Toolkit middleware is the **outermost** layer: it runs before any per-tool middleware defined with `@tool(middleware=[...])`. ### Per-tool middleware `@toolkit.tool` also accepts the same `middleware=[...]` option as `@tool` and `@agent.tool`. These per-tool hooks run **inside** the toolkit-level middleware. See [Tool middleware](tool_middleware.md). --- # Common Tools Source: https://docs.ag2.ai/docs/user-guide/tools/common_toolkits/ # Common Tools AG2 ships with ready-made tools and toolkits that bundle related function tools into a single `Toolkit`. Unlike [built-in provider tools](builtin_tools.md), these run locally as regular Python functions and work with **every** provider. ## FilesystemToolkit `FilesystemToolkit` gives an agent the ability to read, write, update, delete, and search files within a sandboxed directory. All paths are resolved relative to a configurable `base_path`, and a path-traversal guard prevents access outside it. ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import FilesystemToolkit fs = FilesystemToolkit(base_path="/tmp/workspace") agent = Agent( "assistant", config=AnthropicConfig(model="claude-sonnet-5"), tools=[fs], ) ``` ### Available tools | Tool | Description | | :--- | :--- | | `read_file` | Read the contents of a file | | `write_file` | Create or overwrite a file (creates parent directories automatically) | | `update_file` | Replace the first occurrence of a string in a file | | `delete_file` | Delete a file | | `find_files` | Search for files matching a glob pattern (supports recursive `**` patterns) | ### Read-only mode Pass `read_only=True` to expose only `read_file` and `find_files`: ```python fs = FilesystemToolkit(base_path="./docs", read_only=True) ``` ### Using individual tools Every tool is available as an attribute on the toolkit instance. You can pass individual tools to an agent instead of the whole set: ```python fs = FilesystemToolkit(base_path="/tmp/workspace") agent = Agent( "reader", config=AnthropicConfig(model="claude-sonnet-5"), tools=[fs.read_file(), fs.find_files()], ) ``` ### Using a temporary directory For throwaway workspaces, use `tempfile.TemporaryDirectory` so the directory and all its contents are automatically cleaned up when the context manager exits: ```python import tempfile from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import FilesystemToolkit async def main() -> None: with tempfile.TemporaryDirectory() as tmpdir: fs = FilesystemToolkit(base_path=tmpdir) agent = Agent( "assistant", config=AnthropicConfig(model="claude-sonnet-5"), tools=[fs], ) await agent.ask("Create a hello.py file that prints 'Hello, World!'") ``` !!! tip Prefer `tempfile.TemporaryDirectory` over hardcoded `/tmp` paths. It guarantees a unique directory per run and cleans up after itself, avoiding leftover files and collisions between concurrent executions. ### Path safety All paths are resolved relative to `base_path`. Any attempt to escape the base directory (e.g. `../../etc/passwd`) raises a `PermissionError`: ```python fs = FilesystemToolkit(base_path="/tmp/sandbox") # The agent can access /tmp/sandbox/data.txt # but NOT /tmp/sandbox/../../etc/passwd ``` --- ## DuckDuckSearchTool `DuckDuckSearchTool` gives an agent the ability to search the web using DuckDuckGo. No API key is required. !!! note Requires the `ddgs` extra: `pip install ag2[ddgs]` ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import DuckDuckSearchTool agent = Agent( "researcher", config=AnthropicConfig(model="claude-sonnet-5"), tools=[DuckDuckSearchTool()], ) ``` ### Configuration ```python tool = DuckDuckSearchTool( max_results=10, # default: 5 region="uk-en", # default: "us-en" safesearch="strict", # default: "moderate" - options: "on", "moderate", "off" ) ``` All parameters accept a `Variable` for dynamic values resolved at execution time. --- ## PerplexitySearchToolkit `PerplexitySearchToolkit` gives an agent two related tools powered by [Perplexity](https://www.perplexity.ai): raw web search via the Search API and LLM-grounded answers with citations via Sonar - sharing a single client. !!! note Requires the `perplexity` extra and an API key: `pip install ag2[perplexity]` ```python import os from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import PerplexitySearchToolkit agent = Agent( "researcher", config=AnthropicConfig(model="claude-sonnet-5"), tools=[PerplexitySearchToolkit(api_key=os.environ["PERPLEXITY_API_KEY"])], ) ``` If `api_key` is omitted, the Perplexity SDK reads the `PERPLEXITY_API_KEY` environment variable automatically. ### Tools | Tool | Description | | :--- | :--- | | `perplexity_search` | Raw web search via the [Search API](https://docs.perplexity.ai/docs/search/quickstart) - ranked title/url/snippet/date results, no LLM hop | | `perplexity_answer` | LLM-generated answer with citations via [Sonar Chat Completions](https://docs.perplexity.ai/docs/sonar/openai-compatibility) - also returns search results, citations, and optional images | ### Picking a subset of tools Each tool is exposed as a factory method on the toolkit (`toolkit.search()`, `toolkit.answer()`). Call the method to get a ready-to-use tool, then pass only the ones you need to the agent: ```python toolkit = PerplexitySearchToolkit(api_key=...) agent = Agent( "researcher", config=config, tools=[toolkit.search()], ) ``` ### Per-tool configuration Per-call parameters live on the factory methods, not on the toolkit itself: ```python toolkit = PerplexitySearchToolkit(api_key=...) search_tool = toolkit.search( max_results=10, max_tokens_per_page=512, search_domain_filter=["arxiv.org", "-medium.com"], # prefix '-' to exclude search_recency_filter="week", # "hour" | "day" | "week" | "month" | "year" search_after_date_filter="1/1/2025", # MM/DD/YYYY search_before_date_filter="12/31/2025", ) answer_tool = toolkit.answer( model="sonar-pro", # "sonar" | "sonar-pro" | "sonar-reasoning" | "sonar-reasoning-pro" | "sonar-deep-research" - default: "sonar" max_tokens=2000, # default: 1000 search_context_size="high", # "low" | "medium" | "high" - default: "high" search_mode="academic", # "web" | "academic" | "sec" search_recency_filter="month", # "hour" | "day" | "week" | "month" | "year" return_images=True, # include image URLs in the response return_related_questions=True, # include suggested follow-up questions search_domain_filter=["arxiv.org", "nature.com"], ) agent = Agent("researcher", config=config, tools=[search_tool, answer_tool]) ``` ### HTTP and SDK options The toolkit constructor accepts options for the underlying `httpx.AsyncClient` and the Perplexity SDK client. Any extra keyword arguments are forwarded directly to `AsyncPerplexity(...)` (e.g. `base_url`, `max_retries`, `default_headers`): ```python toolkit = PerplexitySearchToolkit( api_key=..., proxy="http://proxy.company.com:8080", # passed to httpx.AsyncClient verify=False, # disable TLS verification (httpx) timeout=30.0, # httpx timeout in seconds # extra kwargs below are forwarded to AsyncPerplexity base_url="https://custom.perplexity.example", max_retries=5, default_headers={"X-Trace-Id": "abc-123"}, ) ``` ### Result Both tools return a `PerplexitySearchResponse` with these fields: | Field | Description | | :--- | :--- | | `query` | The original search query | | `results` | List of `PerplexitySearchResult` (`title`, `url`, `snippet`, `date`) | | `content` | LLM-generated answer (filled by `perplexity_answer`; empty for `perplexity_search`) | | `citations` | URLs the model cited inline (filled by `perplexity_answer`) | | `images` | List of `PerplexityImageMeta` when `return_images=True` on `perplexity_answer` | When `return_images=True`, image URLs are also surfaced as `ImageInput` parts on the tool result so the next model turn receives them as proper image inputs. !!! tip Use `perplexity_search` when the agent only needs raw ranked URLs (cheaper, no LLM hop). Use `perplexity_answer` when a grounded answer with citations is helpful. !!! tip `search_domain_filter` on `perplexity_answer` is a Pro-tier feature on the Perplexity API; see [usage tiers](https://docs.perplexity.ai/guides/usage-tiers). --- ## TavilySearchTool `TavilySearchTool` gives an agent advanced web search capabilities via the [Tavily](https://tavily.com) API. Results include relevance scores and optional LLM-generated answers, raw page content, and images. !!! note Requires the `tavily` extra and an API key: `pip install ag2[tavily]` ```python import os from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import TavilySearchTool agent = Agent( "researcher", config=AnthropicConfig(model="claude-sonnet-5"), tools=[TavilySearchTool(api_key=os.environ["TAVILY_API_KEY"])], ) ``` If `api_key` is omitted, Tavily reads the `TAVILY_API_KEY` environment variable automatically. ### Configuration ```python tool = TavilySearchTool( max_results=5, search_depth="advanced", # "basic" | "advanced" | "fast" | "ultra-fast" topic="news", # "general" | "news" | "finance" include_answer=True, # add an LLM-generated summary to the response include_raw_content=True, # include full page text alongside the snippet include_images=True, # include image URLs in the response time_range="week", # "day" | "week" | "month" | "year" start_date="2024-01-01", # YYYY-MM-DD end_date="2024-12-31", # YYYY-MM-DD days=7, include_domains=["reuters.com", "bbc.com"], exclude_domains=["example.com"], country="US", # ISO country code for localized results auto_parameters=True, # let Tavily auto-tune query parameters include_favicon=True, # include result favicons in the response ) ``` All search parameters accept a `Variable` for dynamic values resolved at execution time. ### HTTP and SDK options The constructor also accepts options for the underlying `httpx.AsyncClient` and the Tavily SDK client. Any extra keyword arguments are forwarded directly to `AsyncTavilyClient(...)` (e.g. `api_base_url`, `company_info_tags`, `project_id`): ```python tool = TavilySearchTool( api_key=..., proxy="http://proxy.company.com:8080", # passed to httpx.AsyncClient verify=False, # disable TLS verification (httpx) timeout=30.0, # httpx timeout in seconds # extra kwargs below are forwarded to AsyncTavilyClient api_base_url="https://custom.tavily.example", company_info_tags=("news", "finance"), ) ``` --- ## SandboxShellTool `SandboxShellTool` gives an agent the ability to run shell commands inside an environment you choose. With no argument it uses a `LocalEnvironment` with a temporary working directory that is cleaned up on process exit. See the [Sandbox Shell page](local_shell.md) for the full guide. !!! warning A `LocalEnvironment` executes arbitrary shell commands on your machine. Use `allowed`, `blocked`, or `readonly` to restrict what the agent can run - or a `DockerEnvironment`, `DaytonaEnvironment`, or `TenkiEnvironment` for real isolation. ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import SandboxShellTool, LocalEnvironment agent = Agent( "engineer", config=AnthropicConfig(model="claude-sonnet-5"), tools=[SandboxShellTool(LocalEnvironment("/tmp/my_project"))], ) ``` The first argument is the environment; the backend (where commands run) is configured there. Passing nothing uses a temporary local directory. ### Restricting commands Command policy lives on the tool: ```python from ag2.tools import SandboxShellTool, LocalEnvironment # Allow only specific commands sh = SandboxShellTool(LocalEnvironment("/tmp/my_project"), allowed=["git", "python", "pip"]) # Block dangerous commands sh = SandboxShellTool(LocalEnvironment("/tmp/my_project"), blocked=["rm -rf", "curl", "wget"]) # Read-only mode - agent can inspect but not modify sh = SandboxShellTool(LocalEnvironment("/tmp/my_project"), readonly=True) # Hide sensitive files from the agent sh = SandboxShellTool(LocalEnvironment("/tmp/my_project"), ignore=["**/.env", "*.key", "secrets/**"]) ``` --- # MCP Servers Source: https://docs.ag2.ai/docs/user-guide/tools/mcp_servers/ # MCP Servers [MCP (Model Context Protocol)](https://modelcontextprotocol.io/) is a protocol introduced by Anthropic that aims to standardize how tools and prompts are exposed to LLMs. It can be thought of as a superset of regular [Tools](tools.md), created to solve two problems: 1. Each LLM provider had its own tool schema and types, making tools non-portable across providers. 2. There was no standard way to give an LLM a scoped, context-optimized surface to invoke APIs, RPCs, and similar remote capabilities. !!! note "Consuming, not serving" This page is about wiring an existing MCP server into an agent. For the inverse - exposing an AG2 `Agent` **as** an MCP server other clients connect to - see [Serving an Agent as an MCP Server](serving_mcp.md). ## Quick start ```python from ag2 import Agent from ag2.tools import MCPToolkit agent = Agent( name="Weather bot", tools=[MCPToolkit("https://my-mcp-url.example.com")], ) ``` That is the whole of the common case. `MCPToolkit` connects to the server, discovers its tools lazily, and exposes each one to the agent as an ordinary function tool. Nothing of yours is handed to the server unless you say so. ### Which of the two connection styles you want AG2 supports both ways MCP servers are typically wired into an agent: - **Client-side connection** (`MCPToolkit`) - AG2 connects to the MCP server itself, discovers the tools, and executes them locally. The LLM only ever sees ordinary function tools. Works with every provider. Supports both **remote** servers (HTTP / streamable-http) and **local** servers (subprocess speaking MCP over stdin/stdout). - **Provider-side connection** (`MCPServerTool`) - the MCP server URL and credentials are forwarded to the LLM provider, which connects to the server and invokes the tools on its end. Only works with providers that natively support it (e.g. Anthropic). | | `MCPToolkit` (client-side) | `MCPServerTool` (provider-side) | |---|---|---| | Who connects to the MCP server | AG2 | LLM provider | | Who executes tool calls | AG2 | LLM provider | | Works with any LLM provider | Yes | No - provider must support MCP passthrough | | Supports local stdio servers | Yes | No - provider only accepts URLs | | Credentials leave your infra | No | Yes - forwarded to the LLM provider | | Custom middleware on tool calls | Yes | No | | Lifecycle / connection pooling | Handled by AG2 | Handled by provider | !!! tip Pick `MCPToolkit` when you want provider-agnostic behavior, local control over tool execution, when you need to run a local stdio MCP server, or when your MCP credentials must stay inside your infrastructure. Pick `MCPServerTool` when you're only targeting a provider that supports it and you'd rather let the provider manage the MCP lifecycle for you. --- ## Recipes ### Authenticate to a remote server Most real-world MCP servers require authentication. `MCPServerConfig` is keyword-only, so every setting names itself: ```python from ag2 import Agent from ag2.tools import MCPServerConfig, MCPToolkit agent = Agent( name="Weather bot", tools=[ MCPToolkit( MCPServerConfig( server_url="https://my-mcp-url.example.com", authorization_token="XXXXXX", ) ) ], ) ``` ### Launch a local server over stdio Many MCP servers ship as CLIs that speak MCP over their own stdin/stdout - `npx -y @modelcontextprotocol/server-filesystem`, `uvx some-mcp-server`, a Python script in your repo. `MCPStdioServerConfig` launches one as a subprocess and pipes the protocol through its stdio: ```python from ag2 import Agent from ag2.tools import MCPStdioServerConfig, MCPToolkit agent = Agent( name="GitHub bot", tools=[ MCPToolkit( MCPStdioServerConfig( command="uvx", args=["mcp-server-github"], env={"GITHUB_TOKEN": "ghp_XXXXXX"}, cwd="/srv/workspace", allowed_tools=["list_issues", "create_issue"], server_label="github", ) ) ], ) ``` !!! note The subprocess is launched lazily, on the first tool-discovery / tool-call. A short-lived MCP session is opened for each operation, so there's no persistent process to manage from your code. ### Connect several servers at once `MCPToolkit` is just a [Toolkit](toolkits.md), so you can register as many as you need - and freely mix remote and local ones: ```python from ag2 import Agent from ag2.tools import MCPStdioServerConfig, MCPToolkit agent = Agent( name="Mixed bot", tools=[ MCPToolkit("https://my-mcp-url.example.com"), MCPToolkit("https://my-mcp-url2.example.com"), MCPToolkit( MCPStdioServerConfig( command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"], ) ), ], ) ``` ### Namespace the tool names When two servers expose the same tool name, set `tool_name_prefix` to keep them apart locally. The prefix is what the model sees; calls sent to the server keep the original name: ```python from ag2 import Agent from ag2.tools import MCPServerConfig, MCPToolkit agent = Agent( name="Mixed bot", tools=[ MCPToolkit( MCPServerConfig( server_url="https://github.example.com/mcp", tool_name_prefix="github_", ) ), MCPToolkit( MCPServerConfig( server_url="https://docs.example.com/mcp", tool_name_prefix="docs_", ) ), ], ) ``` Two remote `search` tools are then exposed locally as `github_search` and `docs_search`. `allowed_tools` and `blocked_tools` still match the original remote names. Like the other config fields, `tool_name_prefix` accepts a `Variable`, so the namespace can be resolved from the context at discovery time: ```python from ag2 import Variable from ag2.tools import MCPServerConfig, MCPToolkit toolkit = MCPToolkit( MCPServerConfig( server_url="https://my-mcp-url.example.com", tool_name_prefix=Variable("tenant_prefix"), ) ) ``` !!! warning Namespacing is opt-in. Two servers exposing the same tool name **without** distinct prefixes still collide: both proxies answer the same tool call, so the tool runs twice for one request. ### Declare what a server may ask you for A tool call can come back asking for input instead of returning a result. `answering=` is the one place you declare which of your own resources a server may use, and **everything in it is off by default**: ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.tools import MCPAnswerPolicy, MCPServerConfig, MCPToolkit agent = Agent( name="researcher", config=AnthropicConfig(model="claude-haiku-4-5-20251001"), hitl_hook=lambda event: input(event.content), # where a server's question goes tools=[ MCPToolkit( MCPServerConfig( server_url="https://my-mcp-url.example.com", protocol_mode="auto", # needed for a modern-era server; see below ), answering=MCPAnswerPolicy( elicitation="ask", # route questions to this agent's human sampling=True, # let the server borrow this agent's model roots=["/srv/project"], # directories to report max_rounds=10, # how often it may come back for more ), ) ], ) ``` `MCPToolkit` answers what you have enabled and retries the call - the whole loop happens inside the one operation, so nothing is held between calls: the pause is on the remote server and your end is simply waiting. With no human-input hook configured, a question surfaces the usual "human input was requested but not provided" failure rather than a silent decline: an absent channel is not a refusal, and reporting it as one would hand the server a decline you never made. ### Reach a server on the modern protocol era Pass `protocol_mode="auto"`. It probes `server/discover` and falls back to the handshake, which is what a server on revision 2026-07-28 needs - only that era can return a question as the *result* of a call. The default `"legacy"` performs the handshake only. --- ## Concepts ### Protocol era A **protocol era** is which family of MCP revisions a connection speaks, and therefore how a request for input travels. `protocol_mode` chooses how the connection settles on one: | | `"legacy"` (default) | `"auto"` | |---|---|---| | on connect | the `initialize` handshake, nothing else | probes `server/discover`, falls back to the handshake | | a request for input arrives | as a standalone request on the back-channel | as the **result** of the call, which is then retried | | cost | none - byte-identical to previous behaviour | one probe round trip per connection | The default stays `"legacy"` so upgrading AG2 changes no existing connection. Choosing the modern era is a visible line in your code. ### Paused run A **paused run** is a turn held mid-flight while someone is asked for something - and on this side of the protocol there never is one. The answer/retry loop runs inside the single operation that opened the session, so nothing is held between your calls: the pause is the remote server's, and your end is simply waiting. Nothing about sticky routing or restarts applies here. The [serving guide](serving_mcp.md) covers the side that does hold one. ### Resolved parameter A server may declare a **resolved parameter** - a tool parameter whose value comes from asking you rather than from the model's arguments. You do not write these; you decide whether to answer them, which is what the answer policy is for. The [serving guide](serving_mcp.md) covers writing one. ### What the MCP answer policy hands over | field | what it hands over | default | |---|---|---| | `elicitation` | your user's attention - the question goes to `context.input()`, and so to the agent's `hitl_hook` | `"decline"` | | `sampling` | **your model budget** - the completion runs on this agent's own model, and you pay for it | `False` | | `roots` | your filesystem layout - plain paths only, no `Variable`, since these are deployment configuration rather than a runtime value | `()` | | `max_rounds` | nothing; it bounds how many times a server may come back before the call is abandoned | `10` | `elicitation` reuses the same two-valued `ElicitationPolicy` as `ACPConfig.elicitation_policy`, so the word means one thing across AG2's protocol integrations. `sampling` is named for the protocol operation rather than "model" on purpose: this side **lends** your model, while [serving](serving_mcp.md)'s `client_model=` **borrows** the caller's - opposite directions that must not share a word. !!! warning "MCP has deprecated sampling and roots" SEP-2577 reached Final status on 2026-04-14 and deprecates sampling, roots and logging - not elicitation. The deprecation is annotation-only: each stays fully functional for a year past the release of every subsequent specification version. For `sampling`, the alternative SEP-2577 recommends is the server integrating an LLM provider API of its own. For `roots`, it recommends tool parameters, resource URIs, or the server's own configuration. Both fields stay here because a server that asks for one today needs an answer. ### A capability is advertised only when you enabled it So a conforming server never asks for what you would refuse. There is nothing to keep in step by hand: the client derives what it declares from which answering callbacks are supplied, and not enabling one *is* how it goes unadvertised. ### Refusal is asymmetric A server that asks anyway is refused, and the two refusals do not look alike: - **Elicitation has a `decline` action on the wire.** A question this agent will not answer is declined, and the server can degrade deliberately. - **Sampling and roots have no such arm.** The error returned for one of those ends the client session's request loop, so the **tool call** fails with that message rather than the server hearing an answer. ### A question has to fit a one-line answer `context.input()` is one string in, one string out, so that is the only shape of question this side can answer. A form-mode elicitation with exactly one property is put to your human and answered on that property, carrying the text they typed verbatim - a server that declared the property as a number receives that text, not a number. Anything else is declined without your human ever seeing it: a URL-mode elicitation, because a text channel cannot confirm that an out-of-band browser flow happened, and a form with more than one property, because splitting one free-text answer across fields would be fabricating data. ### What is *not* forwarded to your model The server's `max_tokens`, `temperature`, `stop_sequences`, `model_preferences` and `include_context` are all ignored: your configuration governs a call you are paying for, and a third party does not get to pick your model or redirect your spending. Only the messages and the system prompt it sent are used. The completion is one call against the model client, not a turn of your agent - your tools, history and response schema stay out of it, and any tool declarations the request carried are dropped with them, so a borrowed model cannot reach back into the agent that lent it. --- ## Operations ### What a failing answer looks like - **A server exhausts `max_rounds`.** The tool call fails; the agent sees a tool error it can act on. Raise the bound only if a server legitimately needs that many rounds - an unbounded one could loop the agent. - **A server asks for something you did not enable.** Elicitation is declined on the wire; sampling and roots fail the tool call. See [Refusal is asymmetric](#refusal-is-asymmetric). - **A server asks a question and you configured no human-input hook.** The call surfaces the usual "human input was requested but not provided" failure rather than a silent decline. - **A server asks a question this side cannot shape an answer for.** A URL-mode elicitation, or a form with more than one property, is declined without your human seeing it. See [A question has to fit a one-line answer](#a-question-has-to-fit-a-one-line-answer). - **A server is only reachable on the modern era.** `"legacy"` performs the handshake and nothing else, so a server that announces itself through `server/discover` is never met there. Set `protocol_mode="auto"`. ### Connections are per-operation A short-lived MCP session is opened for each operation - discovery, and each tool call - and closed after it. There is no persistent process or pool to manage from your code, and a stdio server's subprocess is launched lazily on the first one. That is also why the callbacks your `answering` policy implies are supplied per tool call rather than installed once on the toolkit: your human and your model are reachable only from the live context a call runs in. Discovery is opened without them, so nothing can be asked of you while the server's tools are being listed. ### The `mcp` SDK is a hard dependency of this path `MCPToolkit` imports two private `mcp` modules at module scope, on the eager `ag2.tools` import path. A rename in a minor `mcp` release therefore breaks `import ag2.tools` for everyone with the `mcp` extra installed, not only users of `protocol_mode="auto"`. AG2 pins both with a test so an SDK upgrade fails in CI rather than at your import - but pin your `mcp` version if you deploy from a floating range. --- ## Constructor reference ### `MCPToolkit(server, *, middleware=(), answering=None)` | parameter | what it is for | |---|---| | `server` | a URL string, an `MCPServerConfig`, or an `MCPStdioServerConfig` | | `middleware` | tool middleware applied to every call through this toolkit | | `answering` | an `MCPAnswerPolicy`; with none passed, nothing is advertised and nothing is answered | ### `MCPServerConfig` (keyword-only) | field | default | what it is for | |---|---|---| | `server_url` | - | where the server listens, including the MCP endpoint path | | `authorization_token` | `None` | bearer token sent on every request | | `headers` | `None` | extra HTTP headers | | `connection_timeout` | `30.0` | seconds to wait on the server | | `proxy` | `None` | HTTP proxy to route through | | `verify` | `True` | verify the server's TLS certificate | | `protocol_mode` | `"legacy"` | `"auto"` probes for a modern-era peer and falls back | | `server_label` | `""` | name the toolkit reports itself under | | `description` | `None` | what this server is for | | `allowed_tools` | `None` | tool names to expose; all of them when unset | | `blocked_tools` | `None` | tool names to hide, applied after `allowed_tools` | | `tool_name_prefix` | `""` | prefix put in front of the agent-visible tool names | ### `MCPStdioServerConfig` (keyword-only) | field | default | what it is for | |---|---|---| | `command` | - | the executable to launch | | `args` | `[]` | arguments passed to it | | `env` | `None` | subprocess environment; inherits this process's when unset | | `cwd` | `None` | working directory | | `encoding` | `"utf-8"` | encoding of the stdio pipes | | `protocol_mode` | `"legacy"` | `"auto"` probes for a modern-era peer and falls back | | `server_label` / `description` / `allowed_tools` / `blocked_tools` / `tool_name_prefix` | as above | as above | Every field of both configs except `connection_timeout`, `proxy`, `verify`, `encoding` and `protocol_mode` can be a `Variable` if the value is only known at runtime - handy for injecting per-conversation tokens or workspace paths. ### `MCPAnswerPolicy` `elicitation` (`"decline"`), `sampling` (`False`), `roots` (`()`), `max_rounds` (`10`) - see [What the MCP answer policy hands over](#what-the-mcp-answer-policy-hands-over). --- ## Provider-side: `MCPServerTool` `MCPServerTool` does not open a connection - it ships the server URL and credentials to the LLM provider as part of the request. The provider is then responsible for connecting to the MCP server and dispatching tool calls. Because the provider only accepts a URL, this path does **not** support local stdio servers - use `MCPToolkit` with `MCPStdioServerConfig` for those. ```python from ag2 import Agent from ag2.tools import MCPServerTool agent_with_mcp = Agent( name="Weather bot", tools=[ MCPServerTool( server_url="https://my-mcp-url.example.com", server_label="weather", authorization_token="XXXXXX", ), ], ) ``` `MCPServerTool` also accepts `description`, `allowed_tools`, `blocked_tools`, and `headers`. All constructor parameters can be a `Variable` if the value is only known at runtime. !!! warning "`blocked_tools` is not enforceable on every provider" OpenAI's and xAI's remote-MCP tools take an allow-list and nothing else, so a block cannot be expressed in the request they accept. AG2 raises `BlockedToolsUnsupportedError` instead of sending a request that would permit what you asked to block. Pass `allowed_tools` naming the tools you do want, or connect the server as an MCP toolkit (above) so AG2 executes and filters its tools itself - that path enforces `blocked_tools` on any provider. !!! warning With provider-side connections, your MCP credentials are sent to the LLM provider on every request. Only use this path with providers and servers you trust with those credentials. --- # Serving an Agent as an MCP Server Source: https://docs.ag2.ai/docs/user-guide/tools/serving_mcp/ # Serving an Agent as an MCP Server `ag2.mcp.MCPServer` exposes an AG2 `Agent` **as** an MCP server: any MCP client - Claude Desktop, Cursor, the MCP Inspector, another agent framework - can list its tools and talk to your agent. This is the inverse of [consuming an MCP server as tools](mcp_servers.md) with `MCPToolkit` / `MCPServerTool`. ## Quick start ```bash pip install "ag2[mcp]" ``` An `MCPServer` **is** an ASGI3 application. It serves MCP over streamable HTTP and manages its own lifespan, so a standalone `uvicorn` run just works: ```python import uvicorn from ag2 import Agent from ag2.config import AnthropicConfig from ag2.mcp import MCPServer agent = Agent( name="assistant", prompt="You are a helpful assistant.", config=AnthropicConfig(model="claude-haiku-4-5-20251001"), ) app = MCPServer(agent) if __name__ == "__main__": uvicorn.run(app, host="127.0.0.1", port=8000) ``` That is the whole of the common case. The agent is exposed as one tool - `ask` - that runs `Agent.ask()` and returns the reply; conversations are kept, and questions the agent raises reach the human behind the calling client. Everything below is opt-in on top of it. The server is also mountable into a host Starlette/FastAPI app, since it is an ordinary ASGI app. --- ## Recipes ### Serve over stdio instead For a locally-launched client, serve over stdin/stdout. The HTTP parameters (`path`, `stateless`, `json_response`, `security`) are ignored on this transport: ```python import asyncio from ag2 import Agent from ag2.config import AnthropicConfig from ag2.mcp import MCPServer agent = Agent(name="assistant", config=AnthropicConfig(model="claude-haiku-4-5-20251001")) server = MCPServer(agent) if __name__ == "__main__": asyncio.run(server.run_stdio()) ``` ### Rename the tool, or return a schema The conversational tool takes a required `message` and an optional `context` string prepended to it. Rename it with `tool_name=` and reword its description with `tool_description=`. If the agent has a `response_schema`, that schema is advertised verbatim as the tool's `outputSchema` and validated replies come back as `structuredContent`: ```python from pydantic import BaseModel from ag2 import Agent from ag2.config import AnthropicConfig from ag2.mcp import MCPServer class Weather(BaseModel): city: str temp_c: float agent = Agent( name="weather", config=AnthropicConfig(model="claude-haiku-4-5-20251001"), response_schema=Weather, ) app = MCPServer(agent, tool_name="forecast") ``` Presentation metadata (`name`, `version`, `title`, `description`, `instructions`, `website_url`, `icons`) is never derived from the agent - `instructions` in particular is client-facing "how to use this server" guidance, not the agent's system prompt. Pass it explicitly when you want it. ### Ask the calling client's human a question A tool inside a served agent can call `context.input()`, and the question reaches the human behind the **calling client** - the same tool code that prompts a terminal locally: ```python from ag2 import Agent, Context from ag2.config import AnthropicConfig from ag2.mcp import MCPServer async def book_table(when: str, ctx: Context) -> str: """Book a table, confirming the party size with the user.""" size = await ctx.input(f"How many people for {when}?") return f"Booked {when} for {size}." agent = Agent(name="concierge", config=AnthropicConfig(model="claude-haiku-4-5-20251001"), tools=[book_table]) app = MCPServer(agent, elicitation_policy="ask") # "ask" is the default ``` `elicitation_policy="decline"` never asks a client at all. There is deliberately no `"auto"`: an arbitrary question has no answer AG2 could invent without fabricating data on the user's behalf. Only a client that advertised it can answer is ever asked. A client that cannot - or a policy of `"decline"` - falls through to the agent's own `hitl_hook`, and with none configured the turn fails with the usual `HumanInputNotProvidedError` and its instructional message. That loud failure is deliberate: silently returning a degraded result would hide from the caller that the question went nowhere. ### Run on the calling client's model A deployment holding no model credentials can still serve an agent that needs one: `client_model=True` runs the agent's own reasoning on the **calling client's** LLM, over MCP sampling. ```python from ag2 import Agent from ag2.mcp import MCPServer agent = Agent(name="assistant") # no config of its own app = MCPServer(agent, client_model=True) ``` It is off unless you ask for it, and a server that has not enabled it never sends a sampling request and never needs the capability of its clients. !!! warning "This moves three things to the caller" **Cost** - every turn the agent takes spends their budget. **Capability** - which model answers is a fact about the client, so the same agent gives different answers to different callers and its quality is no longer yours to control. **Reproducibility** - a trace cannot be re-run against a known model, because the model was the peer's. A client that advertised no sampling capability is never asked. What happens then is read off the agent, not off a second switch: one with a `config` of its own falls back to it, and one without fails the turn with `MCPSamplingUnavailableError`. So `client_model=` decides only whether the caller's budget may be spent at all - a deployment holding credentials it would rather not spend keeps them as the fallback simply by configuring them. A turn needing tools or a structured response refuses rather than losing them silently, since sampling carries neither here, and so does a completion that comes back carrying no text at all - read as an empty reply it would report success while the agent answered with nothing. !!! note "`MCPDeprecationWarning` on this path is expected" MCP deprecated sampling in revision 2026-07-28 (SEP-2577), and the `mcp` SDK raises `mcp.MCPDeprecationWarning` on every borrowed request. The warning comes from the SDK, not from AG2, and AG2 does not suppress it - it is true, and hiding it would leave you believing a mechanism with an expiry date is ordinary. The deprecation is annotation-only: sampling stays functional for a year past each subsequent specification release. The alternative SEP-2577 recommends is integrating an LLM provider directly, which is what `client_model=False` already does. ### Serve a deterministic tool that asks for something A `@mcp_tool` served alongside the agent can ask the calling client for a parameter's value, using the SDK's resolver mechanism. A parameter annotated `Annotated[T, Resolve(fn)]` is a **resolved parameter**: the framework runs `fn`, and if it returns a request marker - `Elicit`, `Sample` or `ListRoots` - puts that request to the client and injects the answer. ```python from typing import Annotated from pydantic import BaseModel from ag2 import Agent from ag2.mcp import Elicit, MCPServer, Resolve, mcp_tool class Colour(BaseModel): answer: str def pick_colour() -> Elicit[Colour]: return Elicit("What colour?", Colour) @mcp_tool def paint(room: str, colour: Annotated[Colour, Resolve(pick_colour)]) -> str: """Paint a room.""" return f"painted {room} {colour.answer}" app = MCPServer(Agent(name="painter"), tools=[paint]) ``` A resolved parameter is kept out of the advertised `inputSchema` - it is filled by its resolver, never by the caller. `Resolve`, `Elicit`, `Sample`, `ListRoots` and `RequestStateSecurity` are re-exported from `ag2.mcp` so AG2's own examples have one import location. Everything else - wire models, less common types - comes from `mcp` directly; AG2 does not mirror the SDK. ### Run both halves against each other `examples/mcp/server_elicitation_stdio.py` serves an agent whose tool asks a question, and `examples/mcp/client_elicitation.py` calls it from a second AG2 agent whose `hitl_hook` answers it: ```python # examples/mcp/server_elicitation_stdio.py - the served half import asyncio from ag2 import Agent, Context from ag2.config import AnthropicConfig from ag2.mcp import MCPServer from ag2.tools import tool @tool(description="Book a table, confirming the party size with the user.") async def book_table(when: str, context: Context) -> str: size = await context.input(f"How many people for {when}?") return f"Booked {when} for {size}." agent = Agent( name="concierge", prompt="You are a restaurant concierge. Use book_table to make a booking.", config=AnthropicConfig(model="claude-sonnet-5"), tools=[book_table], ) async def main() -> None: await MCPServer(agent, elicitation_policy="ask").run_stdio() if __name__ == "__main__": asyncio.run(main()) ``` ```python # examples/mcp/client_elicitation.py - the calling half import asyncio import os import sys from ag2 import Agent from ag2.config import AnthropicConfig from ag2.events import HumanInputRequest, HumanMessage from ag2.tools import MCPAnswerPolicy, MCPStdioServerConfig, MCPToolkit def hitl_hook(event: HumanInputRequest) -> HumanMessage: print(f"[the served agent asks] {event.content}") return HumanMessage(content="four") async def main() -> None: agent = Agent( name="diner", config=AnthropicConfig(model="claude-sonnet-5"), hitl_hook=hitl_hook, tools=[ MCPToolkit( MCPStdioServerConfig( command=sys.executable, args=["-m", "examples.mcp.server_elicitation_stdio"], env=dict(os.environ), ), answering=MCPAnswerPolicy(elicitation="ask"), ) ], ) reply = await agent.ask("Ask the concierge to book a table for Friday at 8pm.") print(await reply.content()) if __name__ == "__main__": asyncio.run(main()) ``` Run `python -m examples.mcp.client_elicitation` with `ANTHROPIC_API_KEY` set; it launches the served half as a subprocess. --- ## Concepts ### Served agent and calling client The **served agent** is the AG2 agent this server exposes; the **calling client** is whatever MCP client invokes it. The two are separate deployments with separate resources - separate humans, separate model credentials - and every setting on this page is about which of them supplies what. ### Elicitation, and what the policy governs An **elicitation** is a question raised while a turn is running and put to the human behind the calling client. `elicitation_policy` governs the **served agent's own** questions: a `context.input()` inside a tool the agent called, which the deployment may not have written. **It does not gate a resolved parameter.** An `Elicit` returned by a `Resolve(...)` resolver asks regardless of the policy, and that is deliberate on two grounds: - A resolved parameter is part of a tool's signature - written by whoever wrote that tool and put it in `tools=[...]`. It is not run-time agent behaviour for a policy to override. - It cannot be gated cleanly. The SDK's `Resolve` marker holds only a callable, so what a resolver will ask for is unknowable until it runs. Gating would therefore have to refuse `Sample` and `ListRoots` resolvers too - a setting named for elicitation breaking two unrelated mechanisms. A deployment with no human anywhere simply serves no such tool. ### Calling-client model The **calling-client model** is the model a client makes available for the served agent's reasoning. It is not a second AG2 model configuration: nothing selects a model, and only `max_tokens` travels, because it is the only generation parameter the protocol makes mandatory. AG2 fixes that internally at 4096 and exposes no knob for it - the remaining generation parameters belong to a model configuration, and a borrowed model has none here. ### Deterministic MCP tool and resolved parameter A **deterministic MCP tool** is a `@mcp_tool` served next to the agent: its declared inputs and its result do not depend on an agent turn. A **resolved parameter** is one whose value comes from asking the calling client, declared in the signature rather than raised at run time. Its contract is the **opposite** of the conversational tool's: | | the agent's `ask` tool | a `@mcp_tool` resolver | |---|---|---| | across a round trip | the run is **held**, and resumes where it stopped | the resolver body **re-runs**, with answers already collected supplied to it | | runs more than once? | never | once per round | | the tool body | is the turn | runs **once**, on the round where every resolved parameter is satisfied | | can ask for | a question for the client's human | `Elicit`, `Sample`, or `ListRoots` | The last row is worth noticing: the agent's own turn never asks for the caller's roots - it has no filesystem of its own to scope to them - but a deterministic tool may be exactly the code that wants them. Write resolvers **idempotently**: a side effect in one fires once per round. Nothing is held for a deterministic tool, so none of the paused-run operations below apply to it. ### Protocol era MCP has two eras, and which one a connection speaks decides how a request for input travels: | | handshake era (up to 2025-11-25) | modern era (2026-07-28) | |---|---|---| | how the question travels | a standalone `elicitation/create` request, awaited inline | back as the **result** of the call, in an `InputRequiredResult` | | what the client does | answers on the back-channel | retries the call carrying the answer | | what the server holds | nothing | **the paused run, in this process** | The modern revision defines no server-to-client request at all, so a question can only come back as the result of the call. ### Paused run A **paused run** is a served agent's turn held mid-flight while the calling client is asked for something. A conversational turn cannot be replayed to get back to where it was - re-running it would re-issue LLM calls, re-run tool side effects and re-spend tokens - so on the modern era the run is held in the serving process between the two calls. See [Operating a server that pauses](#operating-a-server-that-pauses). ### Conversations By default (`sessions=True`) the server keeps **conversation sessions**: a history that accumulates across `tools/call` invocations, so a caller can hold a multi-turn exchange instead of a series of unrelated questions. A caller names the conversation it wants to continue. The tool takes an optional `conversation` argument holding an opaque **conversation handle**: - **Omit it** and the call starts a new conversation. The server mints a handle and returns it. - **Pass a handle back** and that conversation continues. A blank handle - `""`, or whitespace - counts as omitting it, because it names nothing. That matters for the reader this channel is built for: a model asked for an optional string argument routinely sends an empty one instead of leaving the key out, and read as an unknown handle that would leave it unable to start a conversation at all. No minted handle is blank, so nothing that could name a conversation is affected. The handle comes back twice, for two different readers: - in a **text content block**, so the model driving the tool can read it and carry it into its next call without any help from the host; - in the result's **`_meta`**, under the key `ai.ag2/conversation`, for clients threading it programmatically. `structuredContent` is deliberately left alone: on this tool it is the agent's response schema and is advertised verbatim as `outputSchema`, which MCP requires structured content to conform to. !!! note "The handle comes back on every reply" Both readers get the handle whether or not the caller ever names a conversation, and in both protocol eras - so a handshake-era client that never touches the `conversation` argument still sees the extra content block and `_meta` key. A client that asserts on the exact shape of a reply will notice; one that reads the first text block, or `structuredContent`, will not. Threading a handle from the caller's side - here with the `mcp` SDK's own client against the HTTP server above: ```python import asyncio from mcp.client import Client async def main() -> None: async with Client("http://127.0.0.1:8000/mcp") as client: first = await client.call_tool("ask", {"message": "My name is Ada."}) handle = first.meta["ai.ag2/conversation"] second = await client.call_tool( "ask", {"message": "What is my name?", "conversation": handle}, ) print(second.content[0].text) asyncio.run(main()) ``` Handles are minted by the server and are version-4 UUIDs - opaque and unguessable. A caller-chosen string is never adopted as a new conversation name; that would let any caller push other callers' conversations out of the bounded registry. A handle the server does not know - expired, evicted, or never minted - comes back as a **tool execution error**: a result flagged `isError`, not a JSON-RPC protocol error. The protocol draws that line so the model can recover by starting a new conversation instead of failing the turn. It is never treated as a fresh conversation, and never falls back to the MCP session. Which conversation an unnamed call lands in depends on the era, since each sanctions a different mechanism: | a `conversation` passed? | handshake era (up to 2025-11-25) | modern era (2026-07-28) | |---|---|---| | yes | that conversation | that conversation | | no | the caller's **MCP session** keeps its own history (one per process over stdio) | a fresh conversation each call | A **handshake-era** client negotiates an `initialize` handshake that opens an **MCP session** - the transport's own notion of a connected client - so it gets implicit continuity without passing anything. It may still opt into explicit handles: the handle returned on its first call names the same conversation its MCP session does, so it can migrate ahead of changing protocol revision. A **modern-era** client has no MCP session. The revision states that connections do not represent conversations, that clients may interleave unrelated requests on one transport, and that servers must not use connection or process identity to establish context - so the handle is the only continuity mechanism available there, on every transport. `sessions=False` makes every call stateless, and the `conversation` argument disappears from the advertised tool - a client is never offered an argument that cannot work. A client that passes one anyway gets a tool error saying the server keeps no conversations, rather than a reply that quietly forgets. --- ## Operations ### Sizing the conversation registry `SessionConfig` bounds the registry so a long-lived server cannot leak memory: ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.mcp import MCPServer, SessionConfig agent = Agent(name="assistant", config=AnthropicConfig(model="claude-haiku-4-5-20251001")) app = MCPServer( agent, sessions=SessionConfig(max_sessions=256, ttl=3600.0), ) ``` - `max_sessions` - LRU cap; the least-recently-used conversation's history is dropped once the cap is exceeded. - `ttl` - idle expiry in seconds; a conversation untouched for longer has its history dropped (`None`, the default, means no expiry). - `storage` - the history backend. Both bounds are quoted in the `conversation` argument's own description, so a client can judge whether an old handle is still worth presenting. !!! note "Every unnamed call with no session to fall back on creates a conversation" A call that omits `conversation` and has no MCP session to key on gets a fresh conversation - and a handle for it - on every call. That is every modern-era call, and every handshake-era call on a `stateless=True` transport, so one-shot traffic occupies registry slots too. Size `max_sessions` for your call rate, and set a `ttl` so abandoned one-shot conversations expire instead of pushing real ones out. ### Where conversations live A conversation's history is written through the configured `Storage`; the registry that maps a conversation's *name* - a handle, or an MCP session id - to that history is held **in the serving process**. So a conversation lives in the replica that created it. A handle minted on one replica is unknown on another, and an MCP session id resolves to a different, empty history there. Behind a load balancer, route a caller's calls back to the replica that answered the first one, or run a single replica. Passing a shared backend keeps the history itself out of process memory, which is what makes a large `max_sessions` affordable: ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.mcp import MCPServer, SessionConfig from ag2.streams import RedisStorage agent = Agent(name="assistant", config=AnthropicConfig(model="claude-haiku-4-5-20251001")) app = MCPServer( agent, sessions=SessionConfig(storage=RedisStorage("redis://localhost:6379")), ) ``` !!! warning "A shared backend does not make a handle portable" The registry is per-process, so a shared `Storage` shares the stored history but not the mapping from a name to it. Cross-replica continuity still needs the caller pinned to one replica. ### What `stateless` governs `stateless=True` stops the **HTTP transport** issuing an `mcp-session-id`. That is a *handshake-era* switch: modern-era requests are self-contained single exchanges that never carry a session id in the first place, so the flag does not reach them. Pairing `stateless=True` with `sessions=True` is a valid configuration, not a contradiction - no transport session, conversations named by handle. ### Operating a server that pauses A modern-era paused run lives in exactly one process. Four facts decide a deployment: - **Sticky routing is required.** The retry that resumes a run must reach the process holding it. A pausing server must not sit behind plain round-robin. - **A pause does not survive a restart.** The retry is refused with a protocol error and the caller starts the call again. - **More than one replica needs a shared `requestState` key.** The default policy mints a process-local one, so state minted by one worker is rejected by another; name the signing keys yourself with `request_state_security=` (see [Advanced: replacing the request-state policy](#advanced-replacing-the-request-state-policy)). Necessary, and not sufficient - the run is still in exactly one process. - **A conversation with a paused run takes no other calls.** A call naming that same `conversation` comes back as a tool error telling the caller to answer the outstanding question or start a separate conversation. The pause deliberately lets go of the conversation's lock - the retry that resumes it would otherwise deadlock behind it - but a conversation's history is still one thread of work, so a second call would queue inside the agent behind a run waiting on a human, with nothing to time it out. Refusing it is the recoverable answer, and one a model can act on. !!! warning "Stateless serving and pausing pull against each other" A server that can pause a run cannot be freely stateless. The first two failures above are *intermittent* under a load balancer, and the symptom is an occasional "Invalid or expired requestState" with nothing pointing at the cause. ### How long a pause lives, and what an idle server retains Retention is bounded by the lifetime of the `requestState` token that names the run - one number, not two that can disagree. Unless you replace the policy that mints it, that lifetime is the SDK ephemeral default of **600 seconds**. Once the token has expired no client can resume, so the run is unreachable. **Reclamation is lazy.** An expired run is swept on the next registry operation, so a server that receives no further call keeps an unreachable task and its stream until the process exits. What an idle server therefore retains is at most `ag2.mcp.pause.MAX_PAUSED_RUNS` (256) runs' worth of task and stream, whatever their tokens' age; a server still taking traffic sweeps them as it goes. Process shutdown reclaims all of them. `context.input(timeout=)` keeps its own meaning - how long the caller waits for an answer - and is enforced independently; whichever bound elapses first ends the turn, and the other does not then report a second, contradictory failure. !!! note "The timeout spans the client's side of the round trip" Served this way, the wait covers the network and a human reading the question in another application, which it never did before. Size it for that, not for a local prompt. ### Authentication and who owns a conversation `security=` makes the server an OAuth 2.1 Resource Server: it advertises RFC 9728 Protected Resource Metadata at `/.well-known/oauth-protected-resource`, answers a missing or invalid token with `401` (pointing at that metadata) and an insufficient scope with `403`. ```python from ag2 import Agent from ag2.config import AnthropicConfig from ag2.mcp import MCPServer from ag2.mcp.security import oauth2_scheme, require agent = Agent(name="assistant", config=AnthropicConfig(model="claude-haiku-4-5-20251001")) app = MCPServer( agent, path="/mcp", security=require( oauth2_scheme(url="https://auth.example.com"), resource_url="https://agent.example.com/mcp", verifier=my_token_verifier, # your TokenVerifier implementation required_scopes=["mcp.read"], ), ) ``` With authentication configured, a conversation records the principal that created it - the access token's subject, falling back to its client id - and revalidates that on **every** call, not only at creation. A handle presented under a different principal gets the same error as an unknown one, so the error does not disclose that the handle exists. !!! warning "Without `security`, the handle is the only credential" With no authentication there is no principal to bind to, so anyone holding a handle can continue the conversation it names. The handle travels through readable content - the model's context, client logs, tracing in between - so decide whether that is acceptable for your deployment before serving unauthenticated. ### Advanced: replacing the request-state policy `request_state_security=` is an escape hatch, not part of the happy path. It replaces the ephemeral policy that seals the state a paused run is resumed with, and whose `ttl` (600 seconds by default) is how long a pause lives. Pass `ephemeral(ttl=)` to move that bound; name the signing `keys` when more than one replica has to accept state another one minted: ```python import os from ag2.mcp import MCPServer, RequestStateSecurity # A shorter pause lifetime, under a key this process alone knows. app = MCPServer(agent, request_state_security=RequestStateSecurity.ephemeral(ttl=120.0)) # A key every replica shares, so a retry can be verified by whichever one answers it. app = MCPServer(agent, request_state_security=RequestStateSecurity(keys=[os.environ["MCP_STATE_KEY"]])) ``` Each key must carry at least 32 bytes of secret randomness - generate one with `python -c "import secrets; print(secrets.token_hex(32))"` - and a shorter one is rejected at construction. Pass several to rotate: any of them verifies, the first one signs. The shared key only makes the state *verifiable* everywhere; the run it names is still held in one process, so sticky routing is still what gets the retry back to it. ### Other escape hatches - **`server`** - the underlying low-level `mcp` `Server`, for wiring the SDK reaches that AG2 does not. - **`lifespan=`** - an `mcp` server lifespan whose yielded state every `tools/call` can reach via `request_context.lifespan_context`. - **`context_provider=`** - build the agent's `ConversationContext` yourself per call. --- ## Constructor reference ### `MCPServer(agent, **options)` | option | default | what it is for | |---|---|---| | `name` / `version` / `title` / `description` / `instructions` / `website_url` / `icons` | derived / `None` | populate the `initialize` handshake; presentation only, never derived from the agent | | `cache_hints` | `None` | `ttlMs` / `cacheScope` freshness hints (SEP-2549); only 2026-07-28 clients see them | | `tool_name` | `"ask"` | the conversational tool's name | | `tool_description` | `None` | its description | | `stream_progress` | `True` | forward the agent's stream events as progress notifications and log messages | | `context_provider` | `None` | build the agent's context per call | | `lifespan` | `None` | an `mcp` server lifespan | | `sessions` | `True` | conversation history: `True`, a `SessionConfig`, or `False` for stateless | | `elicitation_policy` | `"ask"` | whether the served agent's questions reach the calling client's human | | `client_model` | `False` | run the agent's reasoning on the calling client's model | | `resources` / `resource_templates` / `prompts` | `()` | expose these alongside the tool; each capability is advertised only when non-empty | | `tools` | `()` | deterministic `@mcp_tool` tools served next to `ask` | | `apps` | `()` | MCP Apps whose tools and interactive documents are served together | | `extensions` | `None` | protocol extensions advertised by the server | | `path` | `"/mcp"` | the HTTP endpoint path | | `stateless` | `False` | stop the HTTP transport issuing an `mcp-session-id` | | `json_response` | `False` | answer with JSON rather than SSE | | `security` | `None` | OAuth 2.1 Resource Server requirements | | `request_state_security` | `None` | advanced: replace the ephemeral request-state policy | ### `SessionConfig` | field | default | what it is for | |---|---|---| | `max_sessions` | `1024` | LRU cap on remembered conversations | | `ttl` | `None` | idle expiry, in seconds | | `storage` | `None` | the history backend; in-memory when unset | ### `MCPFunctionTool` Usually produced by `@mcp_tool`. Constructed directly it takes `name`, `description`, `handler`, `input_schema`, `title`, `annotations`, `output_schema`, and `meta` - and nothing else; resolver metadata is the decorator's own state, not a constructor argument. ### Names re-exported from `ag2.mcp` `MCPServer`, `SessionConfig`, `MCPFunctionTool`, `mcp_tool`, `MCPApp`, `AppSandbox`, `Resource`, `ResourceTemplate`, `Prompt`, `PromptArgument`, `PromptMessage`, `AskContext`, `ContextProvider`, `ExtensionMap`, `client_extension`, `build_ask_tool`, and the five curated MCP SDK names `Resolve`, `Elicit`, `Sample`, `ListRoots`, `RequestStateSecurity`. Error classes live in `ag2.mcp.errors`, all under `MCPServerError` except the last: `MCPAgentConfigError`, `MCPToolNameConflictError`, `UnknownConversationError` (a handle naming no live conversation, reported to the caller as a tool error), `MCPResourceNotFoundError`, `MCPPromptNotFoundError`, and `MCPSamplingUnavailableError` / `MCPSamplingRefusedError` (both `MCPSamplingError`). `MCPElicitationDeclinedError` subclasses core's `HumanInputNotProvidedError` instead, so a question the client refused is the same failure as one nobody was there to answer. ## Beyond the conversational tool - **MCP-UI** - see [MCP-UI Resources](mcp_ui.md) for returning renderable UI from a `@mcp_tool`. - **MCP Apps** - pass `apps=[MCPApp(...)]` to serve an interactive document and its bound tools together; see [MCP Apps](mcp_apps.md). - **Structured custom tools** - a pydantic model or dataclass return annotation supplies `outputSchema`; its value is returned as `structuredContent`, while `__str__` supplies readable text. `output_schema=` can override derivation and `meta=` adds extension metadata. - **Resources and prompts** are exposed by passing `resources=[Resource(...)]`, `resource_templates=[ResourceTemplate(...)]` and `prompts=[Prompt(...)]` from `ag2.mcp`. - **Protocol extensions** are advertised with `extensions=` and inspected per request with `client_extension()`. --- # MCP-UI Resources Source: https://docs.ag2.ai/docs/user-guide/tools/mcp_ui/ # MCP-UI Resources [MCP-UI](https://mcpui.dev/) lets an MCP server return a **renderable UI** - raw HTML, an `iframe` URL, or a Remote-DOM script - from a tool result instead of plain text. A UI-capable client (e.g. the JS [`@mcp-ui/client`](https://mcpui.dev/)) renders it inline; a text-only client falls back to the embedded resource. AG2 builds these resources and the interactive callbacks (**UI Actions**) in `ag2.mcp_ui`. They are returned from the serving side of `ag2.mcp` - an AG2 `Agent` exposed as an MCP server - so this page starts with the minimum you need to serve one, then focuses on the UI helpers. !!! note "This is the serving side of MCP" `ag2.mcp.MCPServer` exposes an agent **as** an MCP server - see [Serving an Agent as an MCP Server](serving_mcp.md) for transports, conversations and security. This is the inverse of consuming an MCP server as tools with `MCPToolkit` / `MCPServerTool` - see [MCP Servers](mcp_servers.md) for that. !!! warning "MCP-UI and MCP Apps are alternatives - pick one" [MCP Apps](mcp_apps.md) - `ag2.mcp.apps`, the `io.modelcontextprotocol/ui` extension - is the other mechanism, and it inverts this one. There the document is registered as a resource and read by URI **in parallel with** the call, so it cannot be built from the call's arguments; the per-call data reaches it as the result's `structuredContent` instead. A host implementing MCP Apps will not render UI a handler returned. Choose by the host you are serving. ## Installation `ag2.mcp_ui` ships as an optional extra (it is **not** pulled in by `ag2[mcp]`): ```bash pip install "ag2[mcp-ui]" ``` Importing `ag2.mcp_ui` without it raises a clear install hint. ## Serving a UI resource `MCPServer` wraps an agent as a single conversational `ask` tool. To return a UI resource you add a **custom tool** with the `@mcp_tool` decorator: the client sees it in `tools/list`, and a `tools/call` runs your handler directly - the agent is **not** invoked, so the result is deterministic. The handler returns the MCP content block(s) verbatim, which is exactly what `raw_html(...)` produces. ```python import asyncio from typing import Any from ag2 import Agent from ag2.config import AnthropicConfig from ag2.mcp import MCPServer, mcp_tool from ag2.mcp_ui import raw_html @mcp_tool async def show_greeting(name: str = "world") -> Any: """Return an interactive greeting card.""" return raw_html("ui://ag2/greeting", f"

Hello, {name} 👋

") agent = Agent(name="assistant", config=AnthropicConfig(model="claude-haiku-4-5-20251001")) server = MCPServer(agent, tools=[show_greeting]) if __name__ == "__main__": asyncio.run(server.run_stdio()) ``` `@mcp_tool` (from `ag2.mcp`) derives the tool `name` from the function name, the `description` from its docstring, and the `inputSchema` from the typed signature - the same convention as `ag2.tools.tool`. The handler may be sync or async, and may return a plain string (wrapped in a text block) instead of content blocks. Point a UI-capable MCP client (or the MCP Inspector) at this stdio server, call `show_greeting`, and the HTML renders inline. Optional `title` and `annotations` keyword arguments are advertised in `tools/list`; `annotations` takes `mcp.types.ToolAnnotations` behavior hints (`readOnlyHint`, `destructiveHint`, ...) that hosts use to decide e.g. whether to confirm with the user before calling: ```python from typing import Any from mcp.types import ToolAnnotations from ag2.mcp import mcp_tool from ag2.mcp_ui import external_url @mcp_tool(annotations=ToolAnnotations(readOnlyHint=True)) async def show_docs() -> Any: """Return the AG2 docs embedded in an iframe.""" return external_url("ui://ag2/docs", "https://docs.ag2.ai/") ``` !!! tip "Why a custom tool and not the `ask` tool" The agent's `ask` tool replies with model-generated text/images, so it cannot emit a deterministic UI blob on its own. Always return UI resources from your own `@mcp_tool`. ### Accessing the request context Annotate a parameter (any name) with `ag2.mcp.tools.MCPRequestContext` to receive the live MCP request context - the session, client parameters, and lifespan state. The parameter is excluded from the advertised `inputSchema`, so clients never see it: ```python from ag2.mcp import mcp_tool from ag2.mcp.tools import MCPRequestContext @mcp_tool async def whoami(ctx: MCPRequestContext) -> str: """Report the calling client.""" params = ctx.session.client_params return f"client: {params.clientInfo.name if params else 'unknown'}" ``` ## UI resource types `ag2.mcp_ui` provides three builders, one per MCP-UI content type. Each returns an `mcp.types.EmbeddedResource` - a valid `tools/call` content block - and each `uri` **must** start with `ui://`. | Builder | Content type | Client renders it as | |---|---|---| | `raw_html(uri, html)` | `rawHtml` | an inline HTML string | | `external_url(uri, url)` | `externalUrl` | `url` embedded in an `iframe` | | `remote_dom(uri, script)` | `remoteDom` | a mounted Remote-DOM script | ```python from ag2.mcp_ui import external_url, raw_html, remote_dom # Inline HTML raw_html("ui://ag2/greeting", "

Hi

") # External site in an iframe external_url("ui://ag2/docs", "https://docs.ag2.ai/") # Remote-DOM script (framework="react" by default, or "webcomponents") remote_dom("ui://ag2/widget", "root.appendChild(el)", framework="react") ``` Every builder accepts `encoding="blob"` to base64-encode the payload for transports or content that don't survive as inline text (the default is `encoding="text"`): ```python raw_html("ui://ag2/report", large_html, encoding="blob") ``` ### Rendering metadata Two optional keyword arguments attach metadata to the resource: - `ui_metadata` - client-side rendering hints. Keys are prefixed with `mcpui.dev/ui-` so the client recognizes them. - `metadata` - written verbatim into the resource `_meta`. ```python raw_html( "ui://ag2/chart", "
Loading...
", ui_metadata={"preferred-frame-size": ["800px", "600px"]}, metadata={"title": "Quarterly sales"}, ) ``` ## UI Actions A rendered UI can carry interactive elements. When the user interacts, the element sends a message to the host with `window.parent.postMessage`, and the host's `onUIAction` handler reacts. `ag2.mcp_ui` builds those message payloads. !!! warning "UI Actions are dispatched by the host, not your server" Unlike `ag2.a2ui` actions, these are **not** server-side handlers. The [MCP-UI spec](https://mcpui.dev/guide/protocol-details) puts the host in control: a UI Action is dispatched client-side (`onUIAction`). A `tool` action just asks the host to call an ordinary MCP tool on your server - expose that tool the normal way; there is no server-side "action handler" to register. These are **authoring** helpers: embed the built action into the UI with `post_message(action)`, which returns an HTML-attribute-safe `window.parent.postMessage(...)` call (quotes and apostrophes in the payload are escaped, so they cannot break out of the attribute). ```python from ag2.mcp_ui import post_message, raw_html, tool_call # A `tool` action: on click, the host calls tools/call add_to_cart(good_id="42"). action = tool_call("add_to_cart", {"good_id": "42"}) html = f'' resource = raw_html("ui://ag2/shop", html) ``` The five action types map one-to-one to the spec: | Builder | Action | What the host does | |---|---|---| | `tool_call(name, params)` | `tool` | call the MCP tool `name(params)` (`tools/call`) | | `prompt(text)` | `prompt` | send `text` into the conversation as a follow-up | | `link(url)` | `link` | open `url` | | `intent(name, params)` | `intent` | emit a host-defined `name` intent with `params` | | `notify(message)` | `notify` | receive `message` as a notification / log | ### Closing the loop with a `tool` action Because a `tool` action asks the host to call `tools/call`, you close the interaction loop by exposing the target as another `@mcp_tool` on the same server: ```python from typing import Any from ag2.mcp import mcp_tool from ag2.mcp_ui import post_message, raw_html, tool_call @mcp_tool async def show_greeting(name: str = "world") -> Any: """Return a greeting card with an 'Add to cart' button.""" action = tool_call("add_to_cart", {"good_id": "42"}) html = f'

Hello, {name} 👋

' return raw_html("ui://ag2/greeting", html) @mcp_tool async def add_to_cart(good_id: str) -> Any: """Target of the greeting card's UI Action; returns an updated card.""" return raw_html("ui://ag2/cart", f"Added item {good_id} to cart ✓") ``` Wire both into `MCPServer(agent, tools=[show_greeting, add_to_cart])`. When the user clicks the button, the host turns the `tool` action into a `tools/call add_to_cart` back to your server, and the returned card replaces the UI. !!! tip "Runnable example" A complete stdio server - greeting card, an iframe of the docs, and the `add_to_cart` loop - lives in `examples/mcp/server_ui.py`. --- # MCP Apps Source: https://docs.ag2.ai/docs/user-guide/tools/mcp_apps/ # MCP Apps [MCP Apps](https://modelcontextprotocol.io/extensions/apps/overview) - the `io.modelcontextprotocol/ui` extension - lets a served agent hand a host a **real interface** instead of a wall of text. `ag2.mcp.apps` serves one as an **app**: a document plus the tools that render it. ## The constraint that shapes everything **The host reads the document in parallel with the call that references it.** It does not wait for your handler; it fetches the `ui://` resource as soon as it sees the tool being called, often before. So the document **cannot be built from the call's arguments**. It is a static body, registered as a resource and read by URI. The per-call data reaches it afterwards, as the result's `structuredContent`, which the host redelivers into the frame as a `ui/notifications/tool-result` notification. Everything below follows from that one fact: why the HTML is declared next to the tool rather than returned by it, why your return type matters so much, and why a document that serves two tools needs to be told which result just arrived. !!! warning "MCP Apps and MCP-UI are alternatives - pick one" [`ag2.mcp_ui`](mcp_ui.md) teaches the **opposite** pattern: its handler builds HTML from the arguments it was given and returns it inside the call result. That is exactly what a host implementing MCP Apps will not render, and the parallel read is why. The two modules coexist and neither replaces the other. `ag2.mcp_ui` targets [MCP-UI](https://mcpui.dev/) clients and ships as the `ag2[mcp-ui]` extra; `ag2.mcp.apps` targets MCP Apps hosts and needs nothing beyond `ag2[mcp]`. Choose by the host you are serving. ## A first app `ag2ui` in the HTML below is not imported and not a package you install. It is a global that `ag2.mcp` itself injects: the server places a `""" shop = MCPApp("ui://shop/card", CARD) @dataclass class Item: name: str price: int def __str__(self) -> str: return f"{self.name} costs ${self.price}." @shop.tool async def show_item(item_id: str) -> Item: """Show a product card.""" return Item(name="Espresso cup", price=12) agent = Agent(name="shopkeeper", config=AnthropicConfig(model="claude-sonnet-5")) server = MCPServer(agent, apps=[shop]) if __name__ == "__main__": asyncio.run(server.run_stdio()) ``` That is the whole thing. `MCPServer(agent, apps=[shop])` registers the document as a resource served with `text/html;profile=mcp-app` (the only MIME type a host will render a `ui://` resource under), advertises `_meta.ui.resourceUri` on `show_item`, and injects the document runtime that makes `ag2ui` exist. `examples/mcp/server_apps.py` is the runnable version, with a button that calls a second tool. ## The return type does three jobs `Item` above is doing more work than it looks. From one annotation the framework derives all three of the things MCP wants: | From `-> Item` | Becomes | Read by | |---|---|---| | its JSON schema | the tool's `outputSchema` | the client, to validate | | its dump | the result's `structuredContent` | your document | | its `__str__` | the result's text content | the model, and a text-only client | Which is why `__str__` is worth writing: a text-only client gets *"Espresso cup costs $12."* rather than a JSON dump. Both pydantic models and dataclasses work. This applies to any custom tool, not only a UI-bound one - see [Structured output](serving_mcp.md#structured-output). ### Three levels of control A handler returns whichever of these fits, in increasing order of control: ```python from mcp.types import CallToolResult, TextContent @shop.tool async def typed() -> Item: """A typed value: schema, structured content and text all derived.""" return Item(name="Espresso cup", price=12) @shop.tool async def mapping() -> dict: """A mapping: structured content verbatim, text a JSON rendering, no schema.""" return {"name": "Espresso cup", "price": 12} @shop.tool async def explicit() -> CallToolResult: """A full result: nothing derived. Also how you state text and data separately.""" return CallToolResult( content=[TextContent(type="text", text="One espresso cup, $12.")], structuredContent={"name": "Espresso cup", "price": 12}, ) ``` !!! note "The schema follows the annotation, not the value" `outputSchema` is decided once, when the tool is declared. Annotating `-> CallToolResult` is what opts a tool out of an advertised schema; returning one from a handler annotated `-> Item` still advertises `Item`'s schema, and MCP requires your `structuredContent` to conform to it. ## Inside the document: `ag2ui` A host **discards every message from a view that has not completed the `ui/initialize` handshake**, and tells it nothing. A hand-written button on a document without that handshake is silently dead - which is why the runtime is injected by default. It defines one global, `ag2ui`: ```javascript await ag2ui.ready; // the handshake completed ag2ui.onToolResult(fn); // every result for this document ag2ui.onToolResult("show_item", fn); // only that tool's results ag2ui.onToolInput(fn); // the call's arguments ag2ui.onToolInputPartial(fn); // arguments while the model is still writing them ag2ui.onHostContextChanged(fn); // theme, display mode, styles, locale... ag2ui.onCancelled(fn); await ag2ui.callTool("add_to_cart", { item_id: "42" }); await ag2ui.sendMessage("Show me the kettle"); // into the conversation await ag2ui.openLink("https://example.com"); await ag2ui.readResource("ui://shop/card"); await ag2ui.updateContext({ content: [{ type: "text", text: "Viewing item 42" }] }); await ag2ui.requestDisplayMode("fullscreen"); await ag2ui.downloadFile(contents); await ag2ui.sampling({ messages: [...] }); ag2ui.log("info", "rendered"); ag2ui.reportSize(); // force a size report; it is automatic otherwise ag2ui.host.capabilities; // what the host advertised ag2ui.host.context; // theme, styles, display mode, container size ag2ui.host.protocolVersion; // the dialect revision the host answered with ag2ui.host.info; // the host's name and version ``` Each subscription returns an unsubscribe function. The document's size is reported automatically via a `ResizeObserver`, so the host can size the frame without your writing one. **Subscribe whenever you like; call only after `ag2ui.ready`.** Registering a handler is local and is safe the moment the document parses - which is why the card above awaits nothing. Everything that *talks* to the host is refused until the handshake completes, so `await ag2ui.ready` before the first `callTool`, `sendMessage` or `readResource`; calling earlier throws rather than posting into a channel the host is still discarding. Errors arrive on the **same** channel as results - a failed call is a result carrying `isError`, matching how `ag2.mcp` already treats a tool error as a result rather than an exception. Calling something the host did not advertise **throws in JavaScript before anything is sent**, because a host's own answer to an unadvertised method is silence: ```javascript try { await ag2ui.downloadFile(contents); } catch (e) { // "the host did not advertise 'downloadFile'" - rather than a button that does nothing. } ``` `requestDisplayMode` refuses locally in the same way, against `ag2ui.host.context.availableDisplayModes` rather than against a capability: asking for a mode the host never offered throws instead of hanging on an answer that will not come. ### Which tool answered MCP Apps gives the document nothing to correlate with: the notification carrying a result names neither the call nor the tool. With one tool per document that does not matter. With two - a card that renders itself *and* updates after an action - it does. The server therefore stamps the answering tool's name into the result's `_meta` under `ai.ag2/tool`, and `ag2ui.onToolResult(name, fn)` routes on it: ```javascript ag2ui.onToolResult("show_item", renderCard); ag2ui.onToolResult("add_to_cart", showConfirmation); ``` Set that key yourself and your value travels instead, so you can route by your own scheme. This is the one respect in which a `CallToolResult` you assembled yourself is modified; every other key you set travels untouched alongside the stamp. ### Who may call a tool A tool the document calls through `ag2ui.callTool` need not be one the model can call. `visibility=` says where the host surfaces it - `"model"` in the model's tool list, `"app"` to the document: ```python @dataclass class CartLine: status: str def __str__(self) -> str: return self.status @shop.tool async def show_item(item_id: str) -> Item: """Show a product card.""" return Item(name="Espresso cup", price=12) @shop.tool(visibility=["app"]) async def add_to_cart(item_id: str) -> CartLine: """Add an item to the cart. Called by the card's button, not by the model.""" return CartLine(status="Added to your cart") ``` Omit it and no `visibility` is advertised at all, leaving the host to its own default. It is a hint about where to *surface* a tool, not an authorization boundary: the tool is served and dispatched exactly like any other, so anything that must not be called by the model has to refuse in the handler. ### Bringing your own bundle `inject_runtime=False` turns injection off and the body is served byte-for-byte as you wrote it. `app.runtime_script()` returns the same `