Lesson CH07-L02
MCPクライアント
外部MCPサーバーのツールをそのままAgentに統合する。
- 読了目安
- 11 min
- Colab目安
- 18 min
- 合計
- 29 min
- 前提
- Toolset
関連: 公式ドキュメント
一行サマリ
MCPServerStdio('uvx', args=['mcp-server-fetch']) のように MCP サーバーをそのまま Agent(toolsets=[server]) に渡す だけで、外部サーバーが提供するツール群が PydanticAI ツールとして自動登録される。Toolset (Ch7-L01) と 完全に同じインターフェース で扱えるのがポイント。
ヒーロー: MCP は「Toolset の外部版」
Ch7-L01 の FunctionToolset は 「同じプロセスの関数を束にする」 仕組みでした。MCP (Model Context Protocol) は同じ抽象を 「別プロセスやリモート HTTP の外部サーバーに広げた」 もの。Agent から見れば どちらも toolsets=[...] に入れる束 という統一感があります。
概念: 3 種類のトランスポート
| クラス | 通信方式 | 用途 |
|---|---|---|
MCPServerStdio | サブプロセス + stdin/stdout | 推奨。npm/pip で配布されるローカル MCP サーバー |
MCPServerStreamableHTTP | HTTP (Streamable) | リモート MCP サーバー / 社内ホスト |
MCPServerSSE | HTTP SSE | レガシー。新規開発では推奨されない |
最も使うのは MCPServerStdio。uvx (Python) や npx (Node) で MCP サーバーをサブプロセスとして起動するシナリオが圧倒的に多いです。
コード: 3 つのパターン
パターン 1: 公開 MCP サーバー (mcp-server-fetch) を Stdio で起動
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio
from pydantic_ai.models.google import GoogleModel
# Python 製の MCP サーバー (URL を fetch するシンプルなもの)
server = MCPServerStdio('uvx', args=['mcp-server-fetch'])
agent = Agent(
GoogleModel('gemini-3-flash-preview'),
toolsets=[server],
instructions='ユーザーの URL を MCP の fetch ツールで取得し、要約してください。',
)
async def main():
async with agent:
result = await agent.run(
'https://example.com/ の見出しを取得して 3 行で要約'
)
print(result.output)
# import asyncio; asyncio.run(main())ポイント:
MCPServerStdio('uvx', args=[...])でサブプロセス起動Agent(toolsets=[server])で 関数のように 統合async with agent:で 接続のライフサイクル を管理 (推奨)- ツールの一覧と実行は MCP サーバーが内部で公開 したものが自動使用される
パターン 2: リモート HTTP MCP サーバーに接続
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
from pydantic_ai.models.google import GoogleModel
# 社内に立てた MCP サーバー (例)
server = MCPServerStreamableHTTP('http://localhost:8000/mcp')
agent = Agent(
GoogleModel('gemini-3-flash-preview'),
toolsets=[server],
)
async def main():
async with agent:
result = await agent.run('社内 ID 12345 の状態を教えて')
print(result.output)社内のドメイン特化ツール群を MCP として配信し、複数チームの Agent から共通利用するのが典型ユースケース。
パターン 3: 複数 MCP サーバーを tool_prefix で衝突回避
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio
from pydantic_ai.models.google import GoogleModel
fs = MCPServerStdio('npx', args=['-y', '@modelcontextprotocol/server-filesystem', '/tmp/work'], tool_prefix='fs')
git = MCPServerStdio('uvx', args=['mcp-server-git', '--repository', '/tmp/work'], tool_prefix='git')
agent = Agent(
GoogleModel('gemini-3-flash-preview'),
toolsets=[fs, git],
instructions='ファイル操作と git 操作を両方使って質問に答えてください。',
)
async def main():
async with agent:
result = await agent.run('/tmp/work の README.md を読んで、最終更新コミットを確認して')
print(result.output)MCP は Ch7-L01 の Toolset と同じ感覚 で tool_prefix を渡せます。fs_read_file git_log のようにプレフィックス付きでツール名が公開されるため、複数サーバーの併用が安全。
観察: 設定ファイルから一括ロード
複数 MCP サーバーを使うなら、設定ファイル (Claude Desktop と同じ JSON 形式) から load_mcp_servers で一括読み込みすると便利。
{
"mcpServers": {
"fs": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/work"]
},
"git": {
"command": "uvx",
"args": ["mcp-server-git", "--repository", "/tmp/work"]
}
}
}from pydantic_ai.mcp import load_mcp_servers
servers = load_mcp_servers('mcp_config.json')
agent = Agent('google-gla:gemini-3-flash-preview', toolsets=servers)まとめ
- MCP は 「Toolset の外部版」 —
Agent(toolsets=[server])で統一的に統合 MCPServerStdio(推奨) /MCPServerStreamableHTTP/MCPServerSSE(非推奨)async with agent:で接続ライフサイクルを管理tool_prefixで複数サーバーの名前衝突を回避- 設定ファイル +
load_mcp_serversで複数サーバーの一括ロード
次レッスンでは Deferred Tools — 実行を保留する requires_approval=True / CallDeferred で Human-in-the-Loop や 外部ジョブ連携 を表現するパターンを扱います。
Colab で実際に動かす
本レッスンの内容を Google Colab 上で実行できるノートブックを用意しています。下のボタンから自分のColab環境に開けます (要 Google アカウント / GOOGLE_API_KEY)。
notebooks/ch07/02-mcp-client.ipynb