pip install "ag2[mcp]"

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

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

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

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)

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

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']}"),
    ],
)

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(),
        ),
    ],
)

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

`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=`.

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)

undefined

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)

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)

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