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サポートを発表し、注目を集めました。[1]
LLMは、基本的には学習済みデータに基づく情報しか扱えず、最新情報の取得や外部サービスの操作ができません。これまでは各LLMがそれぞれ外部ツールとの連携方法を個別に用意する必要がありました。しかし、MCPの登場によって、LLMと外部サービスの連携方法が標準化されつつあり、さまざまなツールやデータソースを簡単に利用できるようになりました。
MCPの構成要素¶
MCPは、MCPホスト、MCPクライアント、MCPサーバーの3つの要素で構成されています。
MCPホスト：LLMアプリケーション本体であり、複数のクライアントインスタンスを生成・管理します。
MCPクライアント：ホストによって生成され、特定のMCPサーバーと1対1で接続し、その接続を維持します。
MCPサーバー：リソース、ツール、プロンプトを外部に公開します。ローカルプロセスまたは、リモートサービスとしても動作可能です。
MCP 構成要素¶
MCPサーバーは主に以下の3つの機能を提供します。
リソース : ユーザーやLLMが利用する文脈やデータ（text・画像・動画）
ツール : LLMが呼び出す実行可能な関数(外部API呼び出し・DB操作・ファイルの作成/編集)
プロンプト : 定義済のプロンプトテンプレート (引数で受け取った値を使ってプロンプトを生成)
通信手段と認証・認可¶
MCPは標準で以下の通信手段をサポートし、JSON-RPC 2.0でメッセージを交換します。
通信手段 |
認証・認可 |
説明 |
|---|---|---|
Standard Input/Output (stdio) |
環境変数による認証情報の供給 |
MCPサーバーをローカルに配置する際に最適 |
Streamable HTTP |
OAuthベース認証 |
MCPサーバーをリモートホスティングする際に最適 |
stdio を使用する場合、環境変数に設定されたクレデンシャルを用いて、MCPサーバーにリクエストを送信します。 一方、HTTPベースの通信では OAuth による認可フローがサポートされています。2025年6月18日のMCP仕様改訂によりこのフローが強化されています。詳細は以下を参照してください。
現在はstdio使用でのMCPサーバーをローカル環境で動作させることが主流となっています。しかし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 チャットボットアプリ¶
このアプリの実装内容を、冒頭で紹介したMCP構成要素の図に対応させると、下図のようになります。
実装するMCP構成要素¶
ファイル構成¶
├── app.py # MCPホスト,MCPクライアントの実装
├── server
│ ├── mcp_disk_usage.py # MCPサーバー
│ └── mcp_os_name.py # MCPサーバー
├── .env
├── images # チャットで表示する画像
│ ├── m_.jpeg
│ └── robo.jpg
使用ライブラリ¶
主に使用している外部ライブラリは以下の通りです。
ライブラリ |
概要 |
|---|---|
PythonのWeb UIフレームワーク。チャットボット、フォーム、ダッシュボードなどを簡単に構築できる |
|
Claude AIモデルにアクセスするための公式Python SDK。テキスト生成、ツール呼び出し、会話管理機能を提供 |
|
LLMと外部ツールやデータソースと連携するためのMCPプロトコルのPython実装 |
|
.env ファイルから環境変数を読み込みアプリケーションで利用できるようにする |
動作環境¶
Python 3.12
ライブラリの使用バージョン
gradio 5.34.2
anthropic 0.54.0
mcp 1.9.4
python-dotenv 1.1.0
仮想環境とライブラリインストール¶
% 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従量課金であれば、5ドルから始めることが可能です。
ANTHROPIC_API_KEY=xxxxxxxxxxxxxxxxxx
MCPサーバーの実装¶
以下はOSの名前を取得する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: 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' を指定すると、標準入出力を使ってクライアントとメッセージのやりとりを行います。
現在 FastMCPクラスのrunメソッドで指定できるtransportは
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クライアントの実装例を示します。 この記事ではMCPの構成要素と処理の流れを理解することを目的に、必要最低限な実装にとどめています。 Claude Desktopのような本格的なMCPホストを開発する場合は、エラー処理や状態管理など、さらに多くのことを考慮する必要があります。 なお、本番レベルのMCPホストを構築したい場合は、LLMアプリケーション開発フレームワークのLangChain や、複雑なワークフローを構築できる LangGraph などの利用も検討するとよいでしょう。
主な処理の流れ¶
処理シーケンス¶
処理は大きく以下の4つのパートに分かれています。
① アプリケーション起動
環境変数の読み込みとGradioアプリケーションの起動を行うmain処理
② Gradio UI構築
GradioのUIコンポーネントを構築し、チャットボットインターフェースを提供する関数
③ MCPClientクラス
個別のサーバー接続を管理するクラス。当クラスのインスタンスはMCPクライアントに相当します。
④ MultiMCPManagerクラス
複数のMCPサーバーを管理し、Claude APIとの連携を行うメインクラスで、 MCPホストに相当します。
それぞれのパートごとの処理内容を詳しく見ていきます。
①アプリケーション起動¶
必要なライブラリと環境変数の読み込みを行います。
Gradioアプリケーションの起動を行います。
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)
... 中略 ...
if __name__ == "__main__":
if not os.getenv("ANTHROPIC_API_KEY"):
print("Warning: ANTHROPIC_API_KEY を .env ファイルに設定してください。")
interface = gradio_interface()
interface.launch(debug=True)
② Gradio UI構築¶
GradioのUIコンポーネントを構築し、チャットボットインターフェースを提供します。
チャット欄からメッセージ送信時
manager.process_messageを呼び出すコールバックを設定します。
def gradio_interface():
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=500,
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
③ MCPClientクラス¶
各MCPClientインスタンスは「1つのサーバープロセスとの接続・リソース管理」を担当します。
StdioServerParametersは、MCPクライアントが「サーバープロセス」を標準入出力（stdio）経由で起動・接続する際の「起動パラメータ（設定情報）」をまとめるためのクラスです。
stdio_clientは、StdioServerParameters で指定されたコマンド・引数・環境変数などを使い、サーバースクリプトをサブプロセスとして起動し、サーバーの標準出力（stdout）を読み取るストリーム、標準入力（stdin）に書き込むストリームを作成します。
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)}"
④ MultiMCPManagerクラス¶
複数のMCPサーバーを統合管理し、Claude APIとの連携を行うメインクラスです。
initialize_servers()メソッドで、すべてのMCPサーバーに接続し、利用可能なツール情報をまとめて取得します。
process_message()メソッドでチャット履歴とユーザーからのメッセージを受け取り、Claude APIに問い合わせます。
Claude APIが「ツールを使え」と指示した場合、該当するMCPサーバーのツールを実行し、その結果をAIに返します。
class MultiMCPManager:
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:
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()
Claude API の anthropic.messages.create() メソッドは、メッセージ履歴を messages 引数として渡します。 この引数は、辞書のリストで構成され、各辞書は次の2つのキーを持ちます。
role：そのメッセージの発信者を示します。指定できる値は以下の3種類です。
user：ユーザーからの入力assistant：Claudeからの応答system：システムメッセージ（Claudeへの振る舞い指示）
content：実際のメッセージ内容（文字列または構造化コンテンツ）です。
ClaudeAPIのmessageの仕様の詳細に関しては以下のドキュメントを参考にしてください。
チャットボットアプリの実行¶
以下のコマンドを実行すると、アプリが起動します。
(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情報やディスク容量など、実際の環境に基づいた回答が得られるはずです。
サンプルの全文は以下のリポジトリにあります。手元で実行したい場合は、こちらを参照してください。
GitHubリポジトリ：https://github.com/masakos/mcp-host-with-gradio
セキュリティに関して¶
MCPはまだ発展途上のプロトコルであり、今後の普及や発展のためにはセキュリティ対策が非常に重要な課題となっています。MCPサーバーは外部リソースへのアクセス権を持つため、万が一悪意のあるコードが含まれていると、情報漏洩や不正操作などのリスクが生じます。特にサードパーティ製のMCPサーバーを利用する場合は、信頼できる提供元かどうかを十分に確認し、サーバーの内容をしっかりチェックすることが不可欠です。安全に利用するためにも、公式ドキュメントやコミュニティの情報を参考にし、慎重にサーバーを選択しましょう。
また、MCPサーバーのセキュリティチェックツール（例：MCP-Shield、MCP-Scanなど）も登場しています。こうしたツールを活用して、MCPサーバーの安全性を事前に確認することが重要になっています。
まとめ¶
本記事では、MCPホスト・MCPクライアント・MCPサーバーをすべてPythonで実装しながら、MCPの仕組みや各構成要素の役割を体験的に理解することを目指しました。自作を通じて、MCPの拡張性、そして今後のAIアプリ開発における可能性を実感していただけたのではないでしょうか。
MCPは、多様なデータソースやツールと連携したAIアプリケーションの構築に非常に有用なプロトコルです。PythonやTypeScriptをはじめ、さまざまな言語向けのSDKが提供されており、対応言語も拡大し続けています。主要なサービスベンダーもMCPの実装を積極的に進めており、オープンソース化や製品への組み込みが進むことで、今後さらに多くのユーザーに利用されることが期待されます。企業にとっても、優れたMCPサーバーを維持・開発するインセンティブが高まっています。 今後のロードマップでは、レジストリの整備など開発者にとってより使いやすい環境が整備されていく予定です。今後のMCPの進化とエコシステムの広がりに、ぜひご注目ください。
参考資料¶
GitHubリポジトリ: https://github.com/modelcontextprotocol/servers
Gradioを使ったMCPクライアントの構築方法