ag2-mcp

AG2 エージェントを MCP サーバーとして提供する

ag2.mcp.MCPServer は AG2 の Agent を Model Context Protocol サーバーにします。MCP クライアント（Claude Desktop、Cursor、MCP Inspector、または MCP 対応アプリ）は接続し、エージェントをツールとして呼び出せます。エージェントと併せて プロンプトやリソースを公開することもできます。

サーバー側とクライアント側 — まずはこちらをお読みください

方向は反対のものが 2 つあり、このスキルが扱うのはそのうちの 1 つだけです。

方向	目的…	使用するもの
サーバー（このスキル）	他の MCP クライアントから自分の AG2 エージェントを呼び出す	ag2.mcp.MCPServer
クライアント	AG2 エージェントから外部 MCP サーバーのツールを呼び出す	MCPServerTool / MCP ツールキット — ag2-use-builtin-tools を参照

ユーザーが「Claude Desktop から自分のエージェントとやり取りさせたい」「自分のエージェントを MCP 経由で公開したい」「MCP エンドポイントをホストしたい」と言った場合は、このスキルを使います。「自分のエージェントに GitHub MCP ツールを使わせたい」または「MCP サーバーに接続したい」と言った場合は、ag2-use-builtin-tools を使います。

使用する場面

• AG2 エージェントを公開し、外部 MCP クライアント（Claude Desktop、Cursor、IDE）から呼び出せるようにします。
• Agent.ask() を実行して応答を返す、単一の会話型 ask スタイルのツールを公開します。
• 再利用可能なプロンプト（テンプレート）やリソース（ファイル／設定／動的データ）を MCP 経由で提供します。
• クライアントセッションごとの複数ターンの履歴、OAuth2 で保護された HTTP、またはリクエストごとのコンテキスト注入が必要な場合に使います。

インストール

pip install "ag2[mcp]"

必須です。コードを提供する前に、このインストールを実行してください。mcp extra がない場合、from ag2.mcp import MCPServer はスタブに解決され、使用時に 「missing optional dependency」エラーが発生します。

60 秒でできるレシピ — stdio 経由でエージェントを提供する

これはローカル 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-..." }
    }
  }
}

エージェントには必ずモデルの config= を設定してください。config がないエージェントを提供すると、 最初のツール呼び出し時に 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 では無視されます）。

Param	Default	Effect
path	"/mcp"	MCP エンドポイントを提供する URL パスです。
stateless	False	True の場合、トランスポートは mcp-session-id を発行しないため、sessions= の値にかかわらず、すべての呼び出しがステートレスになります。
json_response	False	応答を SSE ではなくプレーンな JSON で返します。
security	None	OAuth2 ベアラー認証を必須にします（以下を参照）。

ツールのカスタマイズ

デフォルトでは、ツール名は 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）。少なくとも 1 つのリソースまたはテンプレートが指定された場合に限り、resources ケイパビリティが通知されます。

セッション — 複数ターンの履歴

デフォルト（sessions=True）では、各 MCP セッション（HTTP の場合はトランスポートの mcp-session-id、stdio の場合はプロセスごとに 1 つのキーで識別されます）が、 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

エージェントにオブジェクトスキーマ（Pydantic モデル／dataclass／dict）の response_schema がある場合、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 は、リクエストごとに実行される非同期フックです。認証済みのベアラートークン（mcp.server.auth.provider.AccessToken。未認証の場合は None）を受け取り、AskContext を返します。その 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 ベアラー（HTTP のみ）— 外部設定が必要

HTTP では、OAuth 2.1 ベアラー認証でエンドポイントを保護します。MCP サーバーは純粋なリソースサーバーとして機能します。RFC 9728 Protected Resource Metadata を /.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 を使うと、サーバー全体をメモリ内で起動できます。 低レベルの ClientSession（ツール、プロンプト、リソースの一覧表示／呼び出し）には connect() を使用し、httpx.AsyncClient を ASGI トランスポート経由で利用して HTTP 経路、ステータスコード、メタデータをテストするには serve() を使用します。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) は追加のキーワード引数（例：logging_callback／message_handler）をクライアントセッションに転送します。これは、ストリーミングされる進行状況／ログ通知を確認する方法です。

複数ターンにおける TestConfig の注意点： TestConfig.create() はターンごとに新しいレスポンスイテレーターを作成するため、TestConfig("a", "b") を指定しても、2 回の別々の 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 エクストラが不足している — pip install "ag2[mcp]" を実行してください。そうしないと、インポートされるのは使用時に例外を発生させる依存関係のスタブです。
• Agent にモデル設定がない — MCPServer は受け付けますが、最初のツール呼び出しで MCPAgentConfigError が発生します。Agent(config=...) を設定してください。
• サーバーとクライアントを混同している — MCPServer はエージェントを提供します。エージェントから外部 MCP サーバーのツールを利用するには、MCPServerTool を使ってください（ag2-use-builtin-tools を参照）。
• instructions= は system prompt ではありません — instructions はハンドシェイクでクライアントに提示する「このサーバーの使い方」のテキストであり、エージェントのプロンプトから生成されるものではありません。明示的に渡してください。
• 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
• クライアント側（エージェントから MCP サーバーを利用）：スキル ag2-use-builtin-tools