ag2-mcp

将 AG2 Agent 作为 MCP 服务提供

ag2.mcp.MCPServer 可将 AG2 Agent 转换为 Model Context Protocol 服务：MCP 客户端（Claude Desktop、Cursor、MCP Inspector 或任何支持 MCP 的应用）可以连接并将你的 Agent 作为工具调用。它还可以在 Agent 之外提供提示词和资源。

服务端与客户端——请先阅读

这里有两个相反的方向，而本技能只涵盖其中一种。

方向	你的需求……	使用
服务端（本技能）	让其他 MCP 客户端调用你的 AG2 Agent	ag2.mcp.MCPServer
客户端	让你的 AG2 Agent 调用外部 MCP 服务的工具	MCPServerTool / MCP 工具包——参见 ag2-use-builtin-tools

如果用户说“让 Claude Desktop 与我的 Agent 通信”“通过 MCP 发布我的 Agent”或“托管 MCP 端点”→ 使用本技能。如果用户说“给我的 Agent 提供 GitHub MCP 工具”或“连接到 MCP 服务”→ 使用 ag2-use-builtin-tools。

适用场景

• 暴露 AG2 Agent，以便外部 MCP 客户端（Claude Desktop、Cursor、IDE）调用它。
• 发布一个会运行 Agent.ask() 并返回回复的对话式 ask 工具。
• 通过 MCP 提供可复用的提示词（模板）和资源（文件／配置／动态数据）。
• 需要按客户端会话维护多轮历史记录、受 OAuth2 保护的 HTTP，或按请求注入上下文。

安装

pip install "ag2[mcp]"

必需。在交付代码前运行此安装命令。若没有 mcp extra，from ag2.mcp import MCPServer 会解析为一个存根，使用时会引发“缺少可选依赖”错误。

60 秒快速上手——通过 stdio 提供 Agent

这是本地 MCP 客户端（Claude Desktop、Cursor、MCP Inspector）所需的形式。服务通过 stdin/stdout 读取和写入 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"),
)

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

将它注册到客户端（此处展示 Claude Desktop 的 claude_desktop_config.json；Cursor／其他客户端使用相同的 command + args 结构）：

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

Agent 必须设置模型 config=。为没有配置的 Agent 提供服务时，第一次工具调用会引发 MCPAgentConfigError。

通过 HTTP 提供服务（可流式传输的 HTTP 传输）

MCPServer 本身就是一个 ASGI3 应用，可直接交给 uvicorn。它会管理自己的生命周期（运行可流式传输的 HTTP 会话管理器），因此独立运行即可正常工作。

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 传输选项（stdio 模式下忽略）：

参数	默认值	作用
path	"/mcp"	MCP 端点的服务 URL 路径。
stateless	False	为 True 时，传输不会签发 mcp-session-id，因此无论 sessions= 如何设置，每次调用都是无状态的。
json_response	False	响应返回纯 JSON，而非 SSE。
security	None	强制使用 OAuth2 bearer 令牌（见下文）。

自定义工具

默认情况下，工具名为 ask，并自动生成描述。可通过以下方式覆盖：

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

该工具始终需要一个必填的 message 和一个可选的 context 字符串（会添加到消息前面）。其结构与 Agent.as_tool() 相同。

提示词

暴露可复用的提示词模板（MCP prompts/list + prompts/get）。渲染器接收调用参数组成的 {name: value} 字典，并返回纯 str（会成为一条 user 消息）或 PromptMessage 列表。渲染器可以是同步或异步的。

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

只有传入非空列表时，才会公布 prompts MCP 能力。

资源

暴露静态资源和模板资源（MCP resources/list + resources/read，以及模板对应的 resources/templates/list）。read 返回 str（文本）或 bytes（二进制）；可以是同步或异步的。

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 为 None，则默认使用 MCP SDK 的对应类型（str 对应 text/plain，bytes 对应 application/octet-stream）。只有至少提供一个资源或模板时，才会公布 resources 能力。

会话 — 多轮历史记录

默认情况下（sessions=True），每个 MCP 会话（通过 HTTP 时以传输层的 mcp-session-id 为键，或通过 stdio 时使用每个进程唯一的键）都保留自己的对话历史记录， 并且会在多次 tools/call 调用之间累积。 使用 SessionConfig 调整，或设置 sessions=False 以禁用会话，实现完全无状态的调用。

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 接受任意 ag2.history.Storage（例如基于 Redis 的存储），以支持跨副本连续性。 注意：stateless=True 的 HTTP 传输不会分配会话 ID，因此无论 sessions= 如何设置， 都会保持无状态。

结构化输出 → structuredContent

如果 Agent 的 response_schema 是对象模式（Pydantic 模型 / dataclass / dict），MCPServer 会将其作为工具的 outputSchema 公布，并向客户端返回经过验证的 structuredContent。 标量 / 联合模式不会公布 — 这些回复会以纯文本形式返回。

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": ...}

每个请求的上下文 — AskContext / ContextProvider

context_provider 是一个异步钩子，会针对每个请求运行。它接收经过身份验证的 bearer token （一个 mcp.server.auth.provider.AccessToken，未进行身份验证时为 None），并返回一个 AskContext，其中非 None 字段会直接传入 Agent.ask()。 使用它注入每个主体对应的变量、工具或提示词 — 这是无状态执行器通常会省略的上下文。

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 字段：variables: dict | None、tools: list | None、 prompt: list[str] | str | None。任何设为 None 的字段都会被省略，因此会保留默认的（无状态）行为。

安全 — OAuth2 bearer（仅限 HTTP）— 需要外部配置

对于 HTTP，可使用 OAuth 2.1 bearer 身份验证保护端点。MCP 服务仅充当 资源服务：它通过 RFC 9728 受保护资源元数据，在 /.well-known/oauth-protected-resource 公布受信任的授权服务器，并验证所提供的令牌。令牌仍由你的外部授权服务器签发。

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 的路径部分必须等于 path（此处为 /mcp）， 否则 MCPServer 会引发 ValueError。
• 缺少令牌 / 令牌无效 → 401（附带指向元数据的 WWW-Authenticate 标头）；权限范围不足 → 403。
• verifier 需要自行提供，类型为 mcp.server.auth.provider.TokenVerifier。 oauth2_scheme(url=...) 会拒绝非 http(s) URL（OIDC 签发者的字符串不是可用的授权服务器 URL — 请传入完整 URL）。

需要外部配置：需要一个真实的授权服务器来签发令牌，并需要一个具体的 TokenVerifier。可在进程内测试未进行身份验证的路径（见下文测试）；令牌往返验证需要使用你的 OAuth 提供方。

进程内测试 — 无套接字、无子进程

ag2.mcp.testing 会完全在内存中启动服务。使用 connect() 获取低层级的 ClientSession（列出 / 调用工具、提示词、 资源），使用 serve() 获取通过 ASGI 传输的 httpx.AsyncClient（测试 HTTP 路径、状态码、元数据）。配合使用 ag2.testing 中的 TestConfig 来模拟 LLM — 无需 API 密钥。

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) 会将额外的 kwargs（例如 logging_callback / message_handler）传递给客户端会话 — 用于观察流式进度 / 日志通知。

多轮使用 TestConfig 的注意事项：TestConfig.create() 会为每轮创建一个新的响应迭代器，因此传入 TestConfig("a", "b") 不会在两次独立的 MCP call_tool 之间依次显示 "a"、"b" — 每次调用都会从第一条预设响应开始重放。这是模拟对象的特性，而非服务的问题：会话历史确实会累积（可通过检查自定义测试客户端收到的不断增长的消息列表进行验证，或使用真实模型）。

公共 API 参考

均可从 ag2.mcp 导入：

符号	类型	用途
MCPServer	类	将 Agent 封装为 MCP 服务（ASGI 应用 + run_stdio()）。
SessionConfig	数据类	用于多轮历史记录的 max_sessions、ttl、storage。
Prompt	数据类	name、render、description、arguments —— 提示词模板。
PromptArgument	数据类	name、description、required —— 已声明的提示词参数。
PromptMessage	数据类	role（"user"/"assistant"）、text —— 一条渲染后的消息。
Resource	数据类	uri、name、read、description、mime_type —— 静态资源。
ResourceTemplate	数据类	uri_template、name、read、... —— RFC 6570 动态资源。
AskContext	数据类	variables、tools、prompt —— 注入 ask() 的每请求上下文。
ContextProvider	类型别名	`async (AccessToken
build_ask_tool	函数	单独构建唯一的对话式 MCPTool（适用于高级用法/测试）。

从 ag2.mcp.security 导入：oauth2_scheme、require、Scheme、 Requirement。从 ag2.mcp.testing 导入：connect、serve。

常见问题

• 缺少 mcp extra —— 执行 pip install "ag2[mcp]"；否则导入的会是依赖占位项，使用时会报错。
• Agent 未配置模型 —— MCPServer 会接受该 Agent，但首次调用工具时会引发 MCPAgentConfigError。请设置 Agent(config=...)。
• 混淆服务端和客户端 —— MCPServer 为你的 Agent 提供服务。若要让 Agent 使用外部 MCP 服务的工具，请使用 MCPServerTool（参见 ag2-use-builtin-tools）。
• instructions= ≠ 系统提示词 —— instructions 是握手中面向客户端的“如何使用此服务”文本；它并非从 Agent 的提示词派生。请显式传入。
• stateless=True HTTP 会丢弃会话 —— 无状态 HTTP 传输不会发送 mcp-session-id，因此无法关联多轮历史记录。需要会话时请使用 stateless=False（默认值）。
• security.resource_url 路径不匹配 —— 其路径必须与 path 相同；否则 MCPServer.__init__ 会引发 ValueError。
• 非对象 response_schema —— 只有对象架构会得到 outputSchema/structuredContent；标量/联合类型会以文本形式返回。

深入了解

• 源码：ag2/mcp/{server,sessions,prompts,resources,executor,info,security,testing}.py
• 涵盖上述所有示例的可运行参考：references/test_server.py（使用 python references/test_server.py 运行，无需 API 密钥）
• MCP 规范：https://modelcontextprotocol.io
• 客户端侧（让 Agent 使用 MCP 服务）：技能 ag2-use-builtin-tools