2025年6月: Pythonで理解するMCP（Model Context Protocol）¶
杉田 (@ane45) です。2025年6月の「Python Monthly Topics」では、LLMと外部ツールやデータソースを簡単に接続するためのプロトコル「MCP（Model Context Protocol）」を取り上げます。 Python製Web UIフレームワークである Gradio を活用し、MCPホスト・MCPクライアント・MCPサーバーをすべてPythonで自作することで、MCPの構成要素と全体像をわかりやすく解説します。
MCPとは¶
MCP（Model Context Protocol） は、Claude を開発した Anthropic社 によって提案された、大規模言語モデル（LLM）と外部のツールやデータソースを効率的に連携させるためのプロトコルです。 このプロトコルでは、外部ツールやデータの使用方法を 共通フォーマットで記述し、LLM に伝えることで、どのツールを使い、どのように情報を渡すべきかをモデル自身が判断できる仕組みになっています。2025年3月にOpenAIもMCPサポートを発表し、注目を集めました。
LLMは、基本的には学習済みデータに基づく情報しか扱えず、最新情報の取得や外部サービスの操作ができません。これまでは 「LLMごと × 外部ツールごと」に個別の連携方法を用意する必要がありました。しかし、MCPの登場によって、LLMと外部サービスの連携方法が標準化され、さまざまなツールやデータソースを簡単に利用できるようになりました。
MCPの構成要素¶
MCPはMCPホスト、MCPサーバー、MCPクライアントで構成されています。
mcp 構成要素¶
MCPサーバーは主に以下の3つの機能を提供します。
MCP Server の機能¶
通信手段と認証¶
MCPは標準で以下の通信手段をサポートし、JSON-RPC 2.0でメッセージを交換します。
通信手段 |
認証 |
説明 |
|---|---|---|
stdio |
環境変数経由認証 |
MCPサーバーをローカルに配置する際に最適 |
Streamable HTTP |
OAuthベース認証 |
MCPサーバーをリモートホスティングする際に最適 |
現在はstdio使用でのMCP Server をローカル環境で動作させることが主流となっています。しかしMCPサーバーをローカル環境で動作させる場合、環境への依存性が課題となることがあります。また、社内で独自のMCPサーバーを運用し、複数のクライアントから同じMCPサーバーに接続したいケースなど、リモートホスティングの需要が高まっています。 MCPサーバーをリモートでホスティングするプラットフォームの選択肢の一つとして、Cloudflare Workersが挙げられます。 Cloudflare Workersは、workers-oauth-providerライブラリを利用することで、OAuthフローを容易に構築できるようサポートされています。
詳しくは以下のリンクを参照してください。
また、Anthropicの「MCP Connector」などを使うことで、API経由でリモートMCPサーバーを呼び出すことも可能です。
PythonでMCPホスト・MCPクライアント・MCPサーバーを作成する¶
ここからは、MCPホスト・MCPクライアント・MCPサーバーを実際に作成し、それぞれの動作を確認していきます。 MCPを利用して複数のMCPサーバーと連携し、Claude APIと組み合わせたWebチャットアプリケーションを作成します。
アプリの説明¶
機能：ユーザーの質問内容を解析し、OS情報やディスク使用量の質問に対しては、最適なMCPサーバーのツールを自動で選択・実行し、その結果をチャット画面に表示する
作成する AI チャットボットアプリ¶
ファイル構成¶
├── app.py # MPCホスト,MCPクライアントの実装
├── server
│ ├── mcp_disk_usage.py # MPCサーバー
│ └── mcp_os_name.py # MPCサーバー
├── .env
├── requirements.txt
├── images # チャットで表示する画像
│ ├── m_.jpeg
│ └── robo.jpg
使用ライブラリ¶
主に使用している外部ライブラリは以下の通りです。
動作環境¶
Python 3.11
仮想環境とライブラリインストール¶
% cd mcp-host-with-gradio
% python3 -m venv venv
% source venv/bin/activate
(venv) % pip install gradio anthropic mcp dotenv
.envファイルの設定¶
AnthropicのAPIキーが必要です。APIキーの作成は、以下参考にしてください。 APIの利用には、料金がかかりますがAPI従量課金であれば、数ドルから始めることが可能です。
ANTHROPIC_API_KEY=xxxxxxxxxxxxxxxxxx
MCPサーバーの実装¶
MCPサーバーを作成します。
import json
import platform
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("mcp_os_name")
@mcp.tool()
async def get_os_name() -> str:
"""OSの名前を取得します。
"""
os_name = platform.system()
return json.dumps({
"type": "text",
"text": os_name,
})
if __name__ == "__main__":
mcp.run(transport='stdio')
FastMCP: FastAPIベースのMCPサーバー実装用ライブラリです。ここでは「mcp_os_name」という名前でMCPサーバーを作成しています。
@mcp.tool: このデコレータを関数につけるだけで、その関数をMCPツールとして公開できます。
他にも @mcp.resource や @mcp.prompt などのデコレータが利用可能です。
get_os_name: MCPツールとして公開される関数です。
関数のdocstring（例: """OSの名前を取得する関数です。"""）は、LLMがツールの機能を理解するための説明文として使われます。LLMはこの説明をもとに関数を呼び出すかどうかを判断します。
mcp.run: MCPサーバーとして起動します。transport='stdio' を指定すると、標準入出力を使ってクライアントとメッセージのやりとりを行います。
現在サポートされているトランスポート方式は
stdio、sse、streamable-httpの3種類です。
以下は、PCのディスク使用量を取得するためのツールを提供するMCPサーバーです。
import json
import shutil
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("mcp_disk_usage")
@mcp.tool()
async def get_disk_usage() -> str:
"""ディスク使用量情報を取得します。
"""
total, used, free = shutil.disk_usage("/")
total_gb = total / (1024**3)
used_gb = used / (1024**3)
usage_percent = (used / total) * 100
disk_info = {
"total_gb": round(total_gb, 2),
"used_gb": round(used_gb, 2),
"usage_percent": round(usage_percent, 2)
}
result_text = (
f"ディスク使用量:\n"
f" 総容量: {disk_info['total_gb']} GB\n"
f" 使用量: {disk_info['used_gb']} GB\n"
f" 使用率: {disk_info['usage_percent']}%"
)
return json.dumps({
"type": "text",
"text": result_text,
})
if __name__ == "__main__":
mcp.run(transport='stdio')
MCPホスト、MCPクライアントの実装¶
以下にMCPホストおよびMCPクライアントの実装例を示します。 Claude Desktopのような本格的なMCPホストを開発する場合は、エラー処理や状態管理など、さらに多くの考慮事項が必要となりますが、この記事ではMCPの構成要素と処理の流れを理解することを目的に、必要最低限な実装にとどめています。 なお、本番レベルのMCPホストを構築したい場合は、LLMアプリケーション開発フレームワークのLangChainや、複雑なワークフローを構築できるLangGraphなどの利用も検討するとよいでしょう。
主な処理の流れ¶
処理シーケンス¶
各処理の詳細に関しては、後述します。
import asyncio
import os
from contextlib import AsyncExitStack
from typing import Any
import gradio as gr
from anthropic import Anthropic
from dotenv import load_dotenv
from gradio.components.chatbot import ChatMessage
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
load_dotenv()
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
class MCPClient: # ③ 個別サーバー接続管理
"""個別のMCPサーバーとの接続を管理するクラス"""
def __init__(self, server_name: str):
self.server_name = server_name
self.session = None
self.exit_stack = None
self.tools = []
self.tool_server_map = {}
async def connect(self, server_path: str) -> str:
"""MCPサーバーに接続し、利用可能なツールを取得"""
if self.exit_stack:
await self.exit_stack.aclose()
self.exit_stack = AsyncExitStack()
server_params = StdioServerParameters(
command="python",
args=[server_path],
env={"PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"}
)
# サーバープロセスを起動し、標準入出力経由でMCPサーバーと非同期に接続しセッションを初期化
stdio_transport = await self.exit_stack.enter_async_context(
stdio_client(server_params)
)
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(
ClientSession(self.stdio, self.write)
)
await self.session.initialize()
# サーバーから利用可能なツール一覧を取得
response = await self.session.list_tools()
self.tools = [{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
} for tool in response.tools]
self.tool_server_map = {tool.name: self.server_name for tool in response.tools}
tool_names = [tool["name"] for tool in self.tools]
return f"{self.server_name}と接続しました。利用可能なツール: {', '.join(tool_names)}"
class MultiMCPManager: # ④ 統合管理とClaude API連携
"""複数のMCPサーバーを統合管理し、Claude APIとの連携を行うメインクラス"""
def __init__(self):
self.os_client = MCPClient("mcp_os_name")
self.disk_client = MCPClient("mcp_disk_usage")
self.anthropic = Anthropic()
self.all_tools = []
self.tool_to_client = {}
self.model_name = "claude-3-7-sonnet-20250219"
def initialize_servers(self) -> str:
"""全サーバーへの接続"""
return loop.run_until_complete(self._initialize_servers())
async def _initialize_servers(self) -> str:
servers = [
(self.os_client, "server/mcp_os_name.py"),
(self.disk_client, "server/mcp_disk_usage.py")
]
tasks = [
self._connect_client(client, path)
for client, path in servers
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return "\n".join(str(result) for result in results)
async def _connect_client(self, client: MCPClient, server_path: str) -> str:
"""個別のクライアント接続処理"""
try:
result = await client.connect(server_path)
self.all_tools.extend(client.tools)
for tool_name in client.tool_server_map:
self.tool_to_client[tool_name] = client
return result
except Exception as e:
return f"Failed to connect to {server_path} server: {str(e)}"
def process_message(
self,
message: str,
history: list[dict[str, Any] | ChatMessage]
) -> tuple:
"""
Args:
message (str): ユーザーが入力した新しいメッセージ（質問など）。
history (list[dict[str, Any] | ChatMessage]): これまでのチャット履歴（リスト形式）。
Returns:
tuple: 更新後のチャット履歴と、入力欄の状態
"""
new_messages = loop.run_until_complete(self._process_query(message, history))
# チャット履歴を更新
updated_history = history + [{"role": "user", "content": message}] + new_messages
textbox_reset = gr.Textbox(value="")
return updated_history, textbox_reset
async def _process_query(
self,
message: str,
history: list[dict[str, Any] | ChatMessage]
) -> list[dict[str, Any]]:
claude_messages = []
for msg in history:
if isinstance(msg, ChatMessage):
role, content = msg.role, msg.content
else:
role, content = msg.get("role"), msg.get("content")
if role in ["user", "assistant", "system"]:
claude_messages.append({"role": role, "content": content})
claude_messages.append({"role": "user", "content": message})
# ユーザーからの質問を使用可能なツール情報を含めて、Claude API用の形式に変換して送信
response = self.anthropic.messages.create(
model=self.model_name,
max_tokens=1024,
messages=claude_messages,
tools=self.all_tools
)
result_messages = []
# Claude APIからの応答を処理
for content in response.content:
if content.type == 'text':
result_messages.append({
"role": "assistant",
"content": content.text
})
elif content.type == 'tool_use':
tool_name = content.name
tool_args = content.input
client = self.tool_to_client.get(tool_name)
# Claude API から使用を提示されたツールを実行
client = self.tool_to_client.get(tool_name)
result = await client.session.call_tool(tool_name, tool_args)
result_text = str(result.content)
result_messages.append({
"role": "assistant",
"content": "```\n" + result_text + "\n```",
"metadata": {
"parent_id": f"result_{tool_name}",
"id": f"raw_result_{tool_name}",
"title": "Raw Output"
}
})
# ツールの実行結果を含めて再度Claude API 呼び出し
claude_messages.append({
"role": "user",
"content": (
f"Tool result for {tool_name}:\n"
f"{result_text}"
)
})
next_response = self.anthropic.messages.create(
model=self.model_name,
max_tokens=1024,
messages=claude_messages,
)
if next_response.content and next_response.content[0].type == 'text':
result_messages.append({
"role": "assistant",
"content": next_response.content[0].text
})
return result_messages
manager = MultiMCPManager()
def gradio_interface(): # ② Gradio UI構築
with gr.Blocks(title="MCP Host Demo") as demo:
gr.Markdown("# MCP Host Demo")
# MCPサーバーに接続し、接続状況を表示
gr.Textbox(
label="MCP Server 接続状況",
value=manager.initialize_servers(),
interactive=False
)
chatbot = gr.Chatbot(
value=[],
height=600,
type="messages",
show_copy_button=True,
avatar_images=("images/m_.jpeg", "images/robo.jpg"),
)
with gr.Row(equal_height=True):
msg = gr.Textbox(
label="質問してください。",
placeholder="Ask about OS information or disk usage",
scale=4
)
clear_btn = gr.Button("Clear Chat", scale=1)
msg.submit(manager.process_message, [msg, chatbot], [chatbot, msg])
clear_btn.click(lambda: [], None, chatbot)
return demo
if __name__ == "__main__": # ① アプリケーション起動
if not os.getenv("ANTHROPIC_API_KEY"):
print("Warning: ANTHROPIC_API_KEY を .env ファイルに設定してください。")
interface = gradio_interface()
interface.launch(debug=True)
主な処理内容¶
① 〜④ の番号は上記実装のコメントと対応しています。
① アプリケーション起動
環境変数の確認とGradioアプリケーションの起動を行います。
② Gradio UI構築
GradioのUIコンポーネントを構築し、チャットボットインターフェースを提供します。
チャット欄からメッセージ送信時
manager.process_messageを呼び出すコールバックを設定します。
③ 個別サーバー接続管理 (MCPClientクラス)
個別のMCPサーバーとの接続を管理します。connectメソッドでサーバープロセスを起動し、利用可能なツール一覧を取得して内部に保存します。
StdioServerParametersは、MCPクライアントが「サーバープロセス」を標準入出力（stdio）経由で起動・接続する際の「起動パラメータ（設定情報）」をまとめるためのクラスです。
stdio_clientは、StdioServerParameters で指定されたコマンド・引数・環境変数などを使い、サーバースクリプトをサブプロセスとして起動し、サーバーの標準出力（stdout）を読み取るストリーム、標準入力（stdin）に書き込むストリームを作成します。
各MCPClientインスタンスは「1つのサーバープロセスとの接続・リソース管理」を担当します。
④ 統合管理とClaude API連携
process_message(): GradioのチャットUIから送られてきたユーザーのメッセージを受け取り、Claude）とMCPツールサーバーを連携させて、チャット履歴と応答を返す「メインの処理関数」です。
Claude API の anthropic.messages.create() メソッドは、メッセージ履歴を messages 引数として渡します。 この引数は、辞書のリストで構成され、各辞書は次の2つのキーを持ちます。
role：そのメッセージの発信者を示します。指定できる値は以下の3種類です。
user：ユーザーからの入力assistant：Claudeからの応答system：システムメッセージ（Claudeへの振る舞い指示）
content：実際のメッセージ内容（文字列または構造化コンテンツ）です。
ClaudeAPIのmessageの仕様の詳細に関しては以下のドキュメントを参考にしてください。
今回実装した内容を、冒頭で紹介したMCP構成要素の図に対応させると、以下のようになります。
本記事で実装したMCP構成要素の対応関係¶
アプリの実行¶
以下のコマンドを実行すると、アプリが起動します。
(venv) % python app.py
[06/15/25 14:02:49] INFO Processing request of type ListToolsRequest server.py:551
[06/15/25 14:02:49] INFO Processing request of type ListToolsRequest server.py:551
* Running on local URL: http://127.0.0.1:7860
コンソールに表示されたURLにアクセスすると、チャットボットアプリを利用できます。実際にチャット欄に質問を入力すれば、PCのOS情報やディスク容量など、実際の環境に基づいた回答が得られるはずです。
セキュリティに関して¶
MCPはまだ発展途上のプロトコルであり、今後の普及や発展のためにはセキュリティ対策が非常に重要な課題となっています。MCPサーバーは外部リソースへのアクセス権を持つため、万が一悪意のあるコードが含まれていると、情報漏洩や不正操作などのリスクが生じます。特にサードパーティ製のMCPサーバーを利用する場合は、信頼できる提供元かどうかを十分に確認し、サーバーの内容をしっかりチェックすることが不可欠です。安全に利用するためにも、公式ドキュメントやコミュニティの情報を参考にし、慎重にサーバーを選択しましょう。 また、MCPサーバーのセキュリティチェックツール（例：MCP-Shield、MCP-Scanなど）も登場しています。こうしたツールを活用して、MCPサーバーの安全性を事前に確認することが重要になっています。
まとめ¶
本記事では、MCPホスト・MCPクライアント・MCPサーバーをすべてPythonで実装しながら、MCPの全体像とその構成要素について解説しました。 MCPは、多様なデータソースやツールと連携したAIアプリケーションの構築に非常に有用なプロトコルです。PythonやTypeScriptをはじめ、さまざまな言語向けのSDKが提供されており、対応言語も拡大し続けています。主要なサービスベンダーもMCPの実装を積極的に進めており、オープンソース化や製品への組み込みが進むことで、今後さらに多くのユーザーに利用されることが期待されます。企業にとっても、優れたMCPサーバーを維持・開発するインセンティブが高まっています。 今後のロードマップでは、レジストリの整備や認証機能の拡充など、開発者にとってより使いやすい環境が整備されていく予定です。今後のMCPの進化とエコシステムの広がりに、ぜひご注目ください。
参考資料¶
GitHubリポジトリ: https://github.com/modelcontextprotocol/servers
Gradioを使ったMCPクライアントの構築方法