PydanticAI ビジュアルガイド

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=[...] に入れる束 という統一感があります。

図を読み込み中…
図1. FunctionToolset と MCPServer の対比

概念: 3 種類のトランスポート

クラス通信方式用途
MCPServerStdioサブプロセス + stdin/stdout推奨。npm/pip で配布されるローカル MCP サーバー
MCPServerStreamableHTTPHTTP (Streamable)リモート MCP サーバー / 社内ホスト
MCPServerSSEHTTP SSEレガシー。新規開発では推奨されない

最も使うのは MCPServerStdiouvx (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 のようにプレフィックス付きでツール名が公開されるため、複数サーバーの併用が安全。

図を読み込み中…
図2. MCP サーバー選択の判断

観察: 設定ファイルから一括ロード

複数 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 / CallDeferredHuman-in-the-Loop外部ジョブ連携 を表現するパターンを扱います。

Colab で実際に動かす

本レッスンの内容を Google Colab 上で実行できるノートブックを用意しています。下のボタンから自分のColab環境に開けます (要 Google アカウント / GOOGLE_API_KEY)。

Open in Colab

notebooks/ch07/02-mcp-client.ipynb