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 with MCPToolkit / MCPServerTool.

Quick start#

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

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.

1
2
3
4
5

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.

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.

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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 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())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# 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 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:

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.

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.

Threading a handle from the caller's side — here with the mcp SDK's own client against the HTTP server above:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

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 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:

1
2
3
4
5
6
7
8
9

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.

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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

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")),
)

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). 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.

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.

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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

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.

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:

1
2
3
4
5
6
7
8
9

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)#

SessionConfig#

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 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.
• 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().