ag2-mcp

Serving an AG2 agent as an MCP server

ag2.mcp.MCPServer turns an AG2 Agent into a Model Context Protocol server: MCP clients (Claude Desktop, Cursor, the MCP Inspector, or any MCP-speaking app) connect and call your agent as a tool. It can also expose prompts and resources alongside the agent.

Server side vs. client side — read this first

There are two opposite directions, and this skill is only one of them.

Direction	You want…	Use
Server (this skill)	other MCP clients to call your AG2 agent	ag2.mcp.MCPServer
Client	your AG2 agent to call an external MCP server's tools	MCPServerTool / MCP toolkits — see ag2-use-builtin-tools

If the user says "let Claude Desktop talk to my agent", "publish my agent over MCP", or "host an MCP endpoint" → this skill. If they say "give my agent the GitHub MCP tools" or "connect to an MCP server" → ag2-use-builtin-tools.

When to use

• Expose an AG2 agent so external MCP clients (Claude Desktop, Cursor, IDEs) can call it.
• Publish a single conversational ask-style tool that runs Agent.ask() and returns the reply.
• Serve reusable prompts (templates) and resources (files/config/dynamic data) over MCP.
• Need multi-turn history per client session, OAuth2-protected HTTP, or per-request context injection.

Installation

pip install "ag2[mcp]"

Required. Run this install before delivering the code. Without the mcp extra, from ag2.mcp import MCPServer resolves to a stub that raises a "missing optional dependency" error on use.

60-second recipe — serve an agent over stdio

This is the form local MCP clients (Claude Desktop, Cursor, MCP Inspector) expect. The server reads/writes MCP frames over stdin/stdout.

import asyncio

from ag2 import Agent
from ag2.config import OpenAIConfig
from ag2.mcp import MCPServer

agent = Agent(
    name="assistant",
    prompt="You are a helpful assistant.",
    config=OpenAIConfig(model="gpt-4o-mini"),
)

# The agent is exposed as ONE conversational tool, named "ask" by default,
# taking a required `message` and an optional `context` string.
server = MCPServer(
    agent,
    name="assistant-mcp",                 # serverInfo.name in the handshake
    instructions="Ask me anything.",       # client-facing usage hint (NOT the agent prompt)
)

if __name__ == "__main__":
    asyncio.run(server.run_stdio())

Register it with a client (Claude Desktop claude_desktop_config.json shown; Cursor / other clients use the same command + args shape):

{
  "mcpServers": {
    "assistant": {
      "command": "python",
      "args": ["/absolute/path/to/serve_stdio.py"],
      "env": { "OPENAI_API_KEY": "sk-..." }
    }
  }
}

The agent must have a model config= set. Serving an agent with no config raises MCPAgentConfigError on the first tool call.

Serve over HTTP (streamable HTTP transport)

MCPServer is itself an ASGI3 application — hand it straight to uvicorn. It manages its own lifespan (it runs the streamable-HTTP session manager), so a standalone run just works.

import uvicorn

from ag2 import Agent
from ag2.config import OpenAIConfig
from ag2.mcp import MCPServer

agent = Agent(name="assistant", prompt="You help users.", config=OpenAIConfig(model="gpt-4o-mini"))

app = MCPServer(agent, path="/mcp")  # MCP endpoint mounted at /mcp

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

HTTP transport knobs (ignored over stdio):

Param	Default	Effect
path	"/mcp"	URL path the MCP endpoint is served at.
stateless	False	When True the transport issues no mcp-session-id, so every call is stateless regardless of sessions=.
json_response	False	Return plain JSON instead of SSE for responses.
security	None	OAuth2 bearer enforcement (see below).

Customising the tool

By default the tool is named ask with an auto-generated description. Override:

server = MCPServer(
    agent,
    tool_name="consult_expert",
    tool_description="Consult the expert agent about a question.",
    stream_progress=True,   # forward agent stream events as MCP progress/log notifications (default True)
)

The tool always takes a required message and an optional context string (prepended to the message). Mirrors Agent.as_tool()'s shape.

Prompts

Expose reusable prompt templates (MCP prompts/list + prompts/get). A renderer receives the call arguments as a {name: value} dict and returns either a plain str (becomes one user message) or a list of PromptMessage. Renderers may be sync or async.

from ag2.mcp import MCPServer, Prompt, PromptArgument, PromptMessage

def render_review(args: dict[str, str]) -> list[PromptMessage]:
    return [
        PromptMessage(role="user", text=f"Review this {args['language']} code:"),
        PromptMessage(role="user", text=args.get("code", "")),
    ]

server = MCPServer(
    agent,
    prompts=[
        Prompt(
            name="code_review",
            description="Generate a code-review prompt.",
            render=render_review,
            arguments=(
                PromptArgument(name="language", description="Programming language", required=True),
                PromptArgument(name="code", description="The code to review", required=False),
            ),
        ),
        # A bare-string renderer becomes a single user message.
        Prompt(name="greet", render=lambda args: f"Say hello to {args['who']}"),
    ],
)

The prompts MCP capability is advertised only when a non-empty list is passed.

Resources

Expose static and templated resources (MCP resources/list + resources/read, plus resources/templates/list for templates). read returns str (text) or bytes (binary); sync or async.

from pathlib import Path
from ag2.mcp import MCPServer, Resource, ResourceTemplate

server = MCPServer(
    agent,
    resources=[
        Resource(
            uri="config://app",
            name="app-config",
            description="Static app config.",
            mime_type="application/json",
            read=lambda: '{"env": "prod"}',
        ),
    ],
    resource_templates=[
        # RFC 6570 templates: {var} matches one path segment, {+var} spans '/'.
        ResourceTemplate(
            uri_template="file:///{+path}",
            name="file",
            description="Read a file by path.",
            read=lambda vars: Path(vars["path"]).read_text(),
        ),
    ],
)

mime_type defaults per the MCP SDK (text/plain for str, application/octet-stream for bytes) when left None. The resources capability is advertised only when at least one resource or template is given.

Sessions — multi-turn history

By default (sessions=True) each MCP session (keyed by the transport's mcp-session-id over HTTP, or a single per-process key over stdio) keeps its own conversation history that accumulates across tools/call invocations. Tune it with SessionConfig, or disable with sessions=False for fully stateless calls.

from ag2.mcp import MCPServer, SessionConfig

server = MCPServer(
    agent,
    sessions=SessionConfig(
        max_sessions=1024,   # LRU cap; least-recently-used session's history is dropped past the cap
        ttl=3600,            # optional idle-expiry in seconds (None = never expire)
        storage=None,        # pluggable history backend; defaults to in-memory MemoryStorage
    ),
)

# Or stateless — every call independent:
stateless = MCPServer(agent, sessions=False)

storage accepts any ag2.history.Storage (e.g. a Redis-backed store) for cross-replica continuity. Note: a stateless=True HTTP transport issues no session id, so it stays stateless regardless of sessions=.

Structured output → structuredContent

If the agent has a response_schema that is an object schema (Pydantic model / dataclass / dict), MCPServer advertises it as the tool's outputSchema and returns validated structuredContent to clients. Scalar/union schemas aren't advertised — those replies flow back as plain text.

from pydantic import BaseModel
from ag2 import Agent
from ag2.config import OpenAIConfig
from ag2.mcp import MCPServer

class Weather(BaseModel):
    city: str
    temp_c: float

agent = Agent(name="weather", prompt="Report weather.", response_schema=Weather,
              config=OpenAIConfig(model="gpt-4o-mini"))
server = MCPServer(agent)
# tool.outputSchema is set; call results carry result.structuredContent == {"city": ..., "temp_c": ...}

Per-request context — AskContext / ContextProvider

A context_provider is an async hook that runs per request. It receives the authenticated bearer token (an mcp.server.auth.provider.AccessToken, or None when unauthenticated) and returns an AskContext whose non-None fields are passed straight into Agent.ask(). Use it to inject per-principal variables, tools, or prompt — context the stateless executor otherwise omits.

from typing import Any
from ag2.mcp import AskContext, ContextProvider, MCPServer

async def provide(token: Any) -> AskContext:
    # Resolve the caller from `token`, then scope the turn to them.
    tenant = "acme"  # e.g. token.scopes / a claims lookup
    return AskContext(
        variables={"tenant": tenant},   # -> Agent.ask(variables=...)
        tools=None,                      # -> Agent.ask(tools=...)  (None = leave default)
        prompt="Be concise.",           # -> Agent.ask(prompt=...)
    )

server = MCPServer(agent, context_provider=provide)

AskContext fields: variables: dict | None, tools: list | None, prompt: list[str] | str | None. Any field left None is omitted, so the default (stateless) behavior is preserved.

Security — OAuth2 bearer (HTTP only) — needs external setup

For HTTP, protect the endpoint with OAuth 2.1 bearer auth. The MCP server acts purely as a Resource Server: it advertises trusted authorization server(s) via RFC 9728 Protected Resource Metadata at /.well-known/oauth-protected-resource and verifies presented tokens. Issuing tokens stays with your external authorization server.

from ag2.mcp import MCPServer
from ag2.mcp.security import oauth2_scheme, require

security = require(
    oauth2_scheme(url="https://auth.example.com"),  # absolute http(s) issuer URL
    resource_url="https://api.example.com/mcp",     # this server's public endpoint
    verifier=my_token_verifier,                     # your mcp TokenVerifier implementation
    required_scopes=["mcp.read"],                   # a token must carry every scope
)

app = MCPServer(agent, path="/mcp", security=security)

• security.resource_url's path component must equal path (here /mcp), or MCPServer raises ValueError.
• Missing/invalid token → 401 (with a WWW-Authenticate header pointing at the metadata); insufficient scopes → 403.
• verifier is a bring-your-own mcp.server.auth.provider.TokenVerifier. oauth2_scheme(url=...) rejects non-http(s) URLs (an OIDC issuer string is not a usable authorization-server URL — pass the full URL).

Requires external setup: a real authorization server to mint tokens and a concrete TokenVerifier. Exercise the unauthenticated path in-process (see testing below); the token round-trip needs your OAuth provider.

Testing in-process — no sockets, no subprocess

ag2.mcp.testing stands the server up entirely in memory. Use connect() for a low-level ClientSession (list/call tools, prompts, resources) and serve() for an httpx.AsyncClient over the ASGI transport (exercise the HTTP path, status codes, metadata). Pair with TestConfig from ag2.testing to mock the LLM — no API keys needed.

import asyncio

from ag2 import Agent
from ag2.testing import TestConfig
from ag2.mcp import MCPServer, Resource
from ag2.mcp import testing

async def main() -> None:
    agent = Agent(name="assistant", prompt="p", config=TestConfig("Hello from the agent!"))
    server = MCPServer(
        agent,
        resources=[Resource(uri="config://app", name="cfg", read=lambda: '{"env": "prod"}')],
    )

    # In-memory MCP client/server pair (the MCP analog of an ASGI test client).
    async with testing.connect(server) as session:
        await session.initialize()

        tools = await session.list_tools()
        assert [t.name for t in tools.tools] == ["ask"]

        result = await session.call_tool("ask", {"message": "Hi"})
        assert "Hello from the agent" in result.content[0].text

        res = await session.read_resource("config://app")
        assert res.contents[0].text == '{"env": "prod"}'

    # Exercise the HTTP transport (initialize handshake, session id) in-memory:
    async with testing.serve(server) as client:
        resp = await client.post(
            "/mcp",
            headers={"Accept": "application/json, text/event-stream", "Content-Type": "application/json"},
            json={
                "jsonrpc": "2.0", "id": 1, "method": "initialize",
                "params": {"protocolVersion": "2025-06-18", "capabilities": {},
                           "clientInfo": {"name": "test", "version": "1.0"}},
            },
        )
        assert resp.status_code == 200
        assert "mcp-session-id" in resp.headers

    print("ok")

if __name__ == "__main__":
    asyncio.run(main())

testing.connect(server, raise_exceptions=..., **session_kwargs) forwards extra kwargs (e.g. logging_callback / message_handler) to the client session — how you observe streamed progress / log notifications.

TestConfig caveat for multi-turn: TestConfig.create() builds a fresh response iterator per turn, so giving it TestConfig("a", "b") will not show "a" then "b" across two separate MCP call_tools — each call replays from the first scripted response. That's a property of the mock, not the server: session history really does accumulate (verify it by inspecting the growing message list a custom test client receives, or use a real model).

Public API reference

All importable from ag2.mcp:

Symbol	Kind	Purpose
MCPServer	class	Wrap an Agent as an MCP server (ASGI app + run_stdio()).
SessionConfig	dataclass	max_sessions, ttl, storage for multi-turn history.
Prompt	dataclass	name, render, description, arguments — a prompt template.
PromptArgument	dataclass	name, description, required — a declared prompt arg.
PromptMessage	dataclass	role ("user"/"assistant"), text — one rendered message.
Resource	dataclass	uri, name, read, description, mime_type — static resource.
ResourceTemplate	dataclass	uri_template, name, read, ... — RFC 6570 dynamic resource.
AskContext	dataclass	variables, tools, prompt — per-request injection into ask().
ContextProvider	type alias	`async (AccessToken
build_ask_tool	function	Build the single conversational MCPTool standalone (advanced/tests).

From ag2.mcp.security: oauth2_scheme, require, Scheme, Requirement. From ag2.mcp.testing: connect, serve.

Common pitfalls

• Missing mcp extra — pip install "ag2[mcp]"; otherwise the imports are dependency stubs that raise on use.
• Agent has no model config — MCPServer accepts it, but the first tool call raises MCPAgentConfigError. Set Agent(config=...).
• Confusing server with client — MCPServer SERVES your agent. To CONSUME an external MCP server's tools from your agent, use MCPServerTool (see ag2-use-builtin-tools).
• instructions= ≠ system prompt — instructions is client-facing "how to use this server" text in the handshake; it is not derived from the agent's prompt. Pass it explicitly.
• stateless=True HTTP discards sessions — a stateless HTTP transport issues no mcp-session-id, so multi-turn history can't key. Use stateless=False (default) when you want sessions.
• security.resource_url path mismatch — its path must equal path; otherwise MCPServer.__init__ raises ValueError.
• Non-object response_schema — only object schemas get outputSchema/structuredContent; scalars/unions come back as text.

Going deeper

• Source: ag2/mcp/{server,sessions,prompts,resources,executor,info,security,testing}.py
• Runnable reference covering every sample above: references/test_server.py (run with python references/test_server.py, no API keys needed)
• MCP spec: https://modelcontextprotocol.io
• Client side (consuming MCP servers from an agent): skill ag2-use-builtin-tools