📖

MCP（Model Context Protocol）のPython実践ガイド

2025/11/03に公開

 はじめにMCP（Model Context Protocol）は、LLMと外部ツールを接続するプロトコルです。この記事では、実際のコード例とともにMCPの使い方を段階的に解説します。

コード全文は以下のGitHubリポジトリにて公開しています。

https://github.com/M6saw0/mcp-client-sample
想定読了時間: 約10分

 1. MCPの基本概念MCPには3つの主要な機能があります：

ツール（Tool）: 関数呼び出しにより外部機能を実行する

リソース（Resource）: データや情報を提供する（静的・動的）

プロンプト（Prompt）: LLM向けのプロンプトテンプレートを提供する
MCPサーバーには主に2つのトランスポート方式があります。この記事ではstdioとstreamable-httpの2つを解説します。

stdio: 標準入出力で通信（同一プロセス内で使用）

streamable-http: HTTP経由で通信（ネットワーク経由）

 2. stdioでサーバーを作成するまずは最もシンプルなstdioトランスポートでサーバーを作成します。

 2.1 基本的なサーバーコードfrom mcp.server.fastmcp import FastMCP

# FastMCPインスタンスを作成
mcp = FastMCP(name="FastMCP Demo Server")

# ツールの定義
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

# リソースの定義（動的）
@mcp.resource("time://{zone}")
def get_time(zone: str) -> str:
    """Return ISO 8601 timestamp."""
    from datetime import datetime, timezone
    now = datetime.now(timezone.utc)
    return now.isoformat() if zone.lower() == "utc" else now.astimezone().isoformat()

# リソースの定義（静的）
@mcp.resource("info://server")
def get_server_info() -> str:
    """Return server metadata."""
    return "FastMCP demo server"

# プロンプトの定義
@mcp.prompt()
def greet_user(name: str, tone: str = "friendly") -> str:
    """Generate a greeting instruction."""
    return f"Craft a {tone} greeting addressed to {name}."

# サーバーを起動
if __name__ == "__main__":
    mcp.run(transport="stdio")

 2.2 非同期ツールとプログレス報告長時間かかる処理では、プログレスを報告できます：
from typing import Annotated
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
import asyncio

@mcp.tool()
async def countdown(
    start: int,
    ctx: Annotated[Context[ServerSession, None], "Injected by FastMCP"],
) -> list[int]:
    """Count down from start to zero."""
    sequence = []
    for step, value in enumerate(range(start, -1, -1), start=1):
        await ctx.report_progress(
            progress=step,
            total=start + 1,
            message=f"Counting value {value}",
        )
        sequence.append(value)
        await asyncio.sleep(0.2)
    return sequence

 3. stdioクライアントを作成するサーバーを呼び出すクライアントコードです：
import asyncio
from pathlib import Path
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    server_path = Path(__file__).with_name("server.py")
    server_params = StdioServerParameters(command="python", args=[str(server_path)])
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # ツールの一覧取得
            tools = await session.list_tools()
            print("Available tools:", [tool.name for tool in tools.tools])
            
            # ツールの呼び出し
            result = await session.call_tool("add", arguments={"a": 2, "b": 5})
            print("add result:", result.content)
            
            # リソースの読み取り
            resource = await session.read_resource("time://local")
            print("time://local:", resource.contents)
            
            # プロンプトの取得
            prompt = await session.get_prompt("greet_user", arguments={"name": "Alice"})
            print("prompt:", prompt.messages)

asyncio.run(main())
プログレスの報告を受け取るには、ClientSession.call_tool に progress_callback を渡します：
async def on_progress(progress: float, total: float | None, message: str | None) -> None:
    print(f"{progress}/{total or 0} - {message or ''}")

result = await session.call_tool(
    "countdown",
    arguments={"start": 3},
    progress_callback=on_progress,
)

 4. streamable-httpでサーバーを作成するstdioサーバーから、ほんの少しの変更でHTTPサーバーにできます：
変更点: FastMCPの初期化時にhostとportを指定し、transport="streamable-http"を指定します。
from mcp.server.fastmcp import FastMCP

# 変更点: hostとportを指定
mcp = FastMCP(
    name="FastMCP StreamableHTTP Demo",
    host="127.0.0.1",  # localhostのみ. 0.0.0.0でも可.
    port=8765,
)

# ツール・リソース・プロンプトの定義は同じ
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

# ...（他の定義も同じ）

# 変更点: transportを"streamable-http"に変更
if __name__ == "__main__":
    mcp.run(transport="streamable-http")

 5. streamable-httpクライアントを作成するHTTPクライアントも少しの変更でOKです：
変更点: stdio_clientの代わりにstreamablehttp_clientを使用し、URLを指定します。
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

STREAMABLE_HTTP_URL = "http://127.0.0.1:8765/mcp"

async def main():
    # 変更点: streamablehttp_clientを使用
    async with streamablehttp_client(STREAMABLE_HTTP_URL) as (read, write, get_session_id):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # セッションIDを取得（HTTP特有）
            if (session_id := get_session_id()) is not None:
                print("Session ID:", session_id)
            
            # 以降の使い方はstdioと同じ
            tools = await session.list_tools()
            print("Available tools:", [tool.name for tool in tools.tools])
            
            result = await session.call_tool("add", arguments={"a": 2, "b": 5})
            print("add result:", result.content)

asyncio.run(main())

 6. 複数サーバーを扱うクライアント各サーバーへの接続後に確立した ClientSession を「保持して再利用する」ことで、複数サーバーを扱えます。
交通層（stdio/HTTP）を開く → ClientSession を initialize → 名前で引けるように辞書等に格納
# リポジトリ実装に合わせたパターン（async with）
sessions: dict[str, ClientSession] = {}

async with stdio_client(params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        sessions["stdio"] = session  # 必要なら保持して再利用
        # ここで何度でも call_tool を実行
        res = await session.call_tool("add", {"a": 1, "b": 2})
        print(res.content)
保持した ClientSession から call_tool を呼び出す
result = await sessions[server].call_tool(tool, args)
最後にまとめてクローズ（接続とセッションのライフサイクルを一元管理）
この「保持と再利用、ライフサイクル管理」を汎用化・整理したのが MultiServerClient です。実装や詳しい使い方は、リポジトリの multi-server/client.py を参照してください。特に次の関数/メソッドが要点です：

MultiServerClient.__init__: サーバー登録（stdio/http）

connect(): 登録済みサーバーへ接続を確立

_ensure_session(name): セッションの有無を確認し、なければ接続

_connect(name): セッション作成のエントリポイント（Future管理）

_session_task(name, future): 実接続とライフサイクル（async with stdio_client/streamablehttp_client → ClientSession → initialize → 待機 → クローズ）

session(name): 保持済み ClientSession を取り出すためのコンテキスト

list_tools(): 全サーバーのツール列挙

call_tool(server, tool, arguments): 指定サーバーへ委譲して実行

close(): 全セッションの安全な終了（シャットダウンイベントと待機）

 7. 生成AIから利用する
 7.1 どのスキーマに変換するか（Responses と Chat Completions）Responses API（例: client.responses.create）は次の形式：
{
  "type": "function",
  "name": "server__tool",
  "description": "...",
  "parameters": { "type": "object", "properties": {"a":{"type":"integer"}}, "required": ["a"], "additionalProperties": false },
  "strict": true
}
Chat Completions API（例: client.chat.completions.create）は次の形式：
{
  "type": "function",
  "function": {
    "name": "server__tool",
    "description": "...",
    "parameters": { "type": "object", "properties": {"a":{"type":"integer"}}, "required": ["a"] }
  },
  "strict": true
}

 7.2 必要情報をどう取得するか（inspect / docstring / MCP メタ）MCP サーバーの list_tools() から取得した情報を使ってスキーマを生成します。

description と inputSchema を使用
本リポジトリでは for-llm/mcp_client.py の MultiServerClient.get_tool_details() が、全サーバーの list_tools() を集約し、{"description","inputSchema"} を提供
ちなみに、Python 関数からスキーマを自動生成することも可能です。

inspect と型ヒント、docstring を用いて JSON Schema を合成
リポジトリの for-llm/tool_schemas.py には以下を実装：
function_to_responses_tool(func)
function_to_chat_tool(func)
build_responses_toolkit(*functions)
build_chat_toolkit(*functions)


 7.3 実装コードと具体的メソッド実際の実装はリポジトリの for-llm/run_llm.py を参照してください。
スキーマ変換ユーティリティ: for-llm/tool_schemas.py
MCP → Responses: mcp_tool_to_responses_schema(tool_name, description, input_schema, strict=True)
MCP → Chat: mcp_tool_to_chat_schema(tool_name, description, input_schema, strict=True)
Python関数 → Responses: function_to_responses_tool(func) / まとめ: build_responses_toolkit(...)
Python関数 → Chat: function_to_chat_tool(func) / まとめ: build_chat_toolkit(...)
ツールメタ取得: for-llm/mcp_client.py

MultiServerClient.get_tool_details() → { combined_name: {"server_name","tool_name","description","inputSchema"} }

 まとめMCPにはツール・リソース・プロンプトの3つの機能がある

stdioは同一プロセス内、streamable-httpはネットワーク経由で通信
サーバーとクライアントのコードは非常にシンプル
複数サーバーを統合管理できる
スキーマ変換+Toolsを利用して、生成AIから動的に利用することが可能
実際のコードはプロジェクトリポジトリを参照してください。

はじめに

1. MCPの基本概念

2. stdioでサーバーを作成する

2.1 基本的なサーバーコード

2.2 非同期ツールとプログレス報告

3. stdioクライアントを作成する

4. streamable-httpでサーバーを作成する

5. streamable-httpクライアントを作成する

6. 複数サーバーを扱うクライアント

7. 生成AIから利用する

7.1 どのスキーマに変換するか（Responses と Chat Completions）

7.2 必要情報をどう取得するか（inspect / docstring / MCP メタ）

7.3 実装コードと具体的メソッド

まとめ

Discussion