杉田
Python製Web UIフレームワークであるGradioを活用し、MCPホスト・
MCPとは
MCP
このプロトコルは、外部ツールやデータの使用方法を共通フォーマットで記述してLLMに伝えることで、どのツールを使い、どのように情報を渡すべきかをモデル自身が判断できる仕組みになっています。2025年3月にOpenAIもMCPサポートを発表し、注目を集めました[1]。
LLMは、基本的には学習済みデータに基づく情報しか扱えず、最新情報の取得や外部サービスの操作ができません。これまでは各LLMがそれぞれ外部ツールとの連携方法を個別に用意する必要がありました。しかし、MCPの登場によって、LLMと外部サービスの連携方法が標準化されつつあり、さまざまなツールやデータソースを簡単に利用できるようになりました。
MCPの構成要素
MCPは、MCPホスト、MCPクライアント、MCPサーバーの3つの要素で構成されています。
- MCPホスト:LLMアプリケーション本体であり、複数のクライアントインスタンスを生成・
管理します。 - MCPクライアント:ホストによって生成され、特定のMCPサーバーと1対1で接続し、その接続を維持します。
- MCPサーバー:リソース、ツール、プロンプトを外部に公開します。ローカルプロセスまたは、リモートサービスとしても動作可能です。

MCPサーバーは主に以下の3つの機能を提供します。
- リソース : ユーザーやLLMが利用する文脈やデータ
(text・ 画像・ 動画) - ツール : LLMが呼び出す実行可能な関数
(外部API呼び出し・ DB操作・ ファイルの作成/編集) - プロンプト : 定義済のプロンプトテンプレート
(引数で受け取った値を使ってプロンプトを生成)
通信手段と認証・認可
MCPは標準で以下の通信手段をサポートし、JSON-RPC 2.
通信手段 | 認証・ |
説明 |
---|---|---|
Standard Input/ |
環境変数による認証情報の供給 | MCPサーバーをローカルに配置する際に最適 |
Streamable HTTP | OAuthベース認証 | MCPサーバーをリモートホスティングする際に最適 |
stdioを使用する場合、環境変数に設定されたクレデンシャルを用いて、MCPサーバーにリクエストを送信します。一方、HTTPベースの通信ではOAuthによる認可フローがサポートされています。2025年6月18日のMCP仕様改訂により、このフローが強化されています。詳細は以下を参照してください。
現在はstdio使用でのMCPサーバーをローカル環境で動作させることが主流となっています。しかしMCPサーバーをローカル環境で動作させる場合、環境への依存性が課題となることがあります。また、社内で独自のMCPサーバーを運用し、複数のクライアントから同じMCPサーバーに接続したいケースなど、リモートホスティングの需要が高まっています。
MCPサーバーをリモートでホスティングするプラットフォームの選択肢の1つとして、Cloudflare Workersが挙げられます。Cloudflare Workersは、workers-oauth-providerライブラリを利用することで、OAuthフローを容易に構築できるようサポートされています。詳しくは以下のリンクを参照してください。
また、Anthropicの
PythonでMCPホスト・MCPクライアント・MCPサーバーを作成する
ここからは、MCPホスト・
アプリの説明
- 機能:ユーザーの質問内容を解析し、OS情報やディスク使用量の質問に対して、最適なMCPサーバーのツールを自動で選択・
実行し、その結果をチャット画面に表示する
このアプリの実装内容を、冒頭で紹介したMCP構成要素の図に対応させると、下図のようになります。

ファイル構成
├── app.py # MCPホスト,MCPクライアントの実装
├── server
│ ├── mcp_disk_usage.py # MCPサーバー
│ └── mcp_os_name.py # MCPサーバー
├── .env
├── images # チャットで表示する画像
│ ├── m_.jpeg
│ └── robo.jpg
使用ライブラリ
主に使用している外部ライブラリは以下の通りです。
ライブラリ | 概要 |
---|---|
gradio | PythonのWeb UIフレームワーク。チャットボット、フォーム、ダッシュボードなどを簡単に構築できる |
anthropic | Claude AIモデルにアクセスするための公式Python SDK。テキスト生成、ツール呼び出し、会話管理機能を提供 |
mcp | LLMと外部ツールやデータソースと連携するためのMCPプロトコルのPython実装 |
python-dotenv | .env ファイルから環境変数を読み込みアプリケーションで利用できるようにする |
動作環境
- Python 3.
12 - ライブラリの使用バージョン
- gradio 5.
34. 2 - anthropic 0.
54. 0 - mcp 1.
9.4 - python-dotenv 1.
1.0
- gradio 5.
仮想環境とライブラリインストール
% 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.
- 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ホスト・
MCPは、多様なデータソースやツールと連携したAIアプリケーションの構築に非常に有用なプロトコルです。PythonやTypeScriptをはじめ、さまざまな言語向けのSDKが提供されており、対応言語も拡大し続けています。主要なサービスベンダーもMCPの実装を積極的に進めており、オープンソース化や製品への組み込みが進むことで、今後さらに多くのユーザーに利用されることが期待されます。企業にとっても、優れたMCPサーバーを維持・
今後のロードマップでは、レジストリの整備など開発者にとってより使いやすい環境が整備されていく予定です。今後のMCPの進化とエコシステムの広がりに、ぜひご注目ください。
参考資料
- MCP
(Model Context Protocol):https:// modelcontextprotocol. io - Model Context Protocol Servers -GitHub
(GitHubリポジトリ) - Building An Mcp Client With Gradio
(Gradioを使ったMCPクライアントの構築方法)