Python Monthly Topics

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の構成要素

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サーバーをリモートでホスティングするプラットフォームの選択肢の1つとして、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構成要素
実装する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

仮想環境とライブラリインストール

% 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ドルから始めることが可能です。

.env
ANTHROPIC_API_KEY=xxxxxxxxxxxxxxxxxx

MCPサーバーの実装

以下はOSの名前を取得するMCPサーバーです。

mcp_os_name.py
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はstdiossestreamable-httpの3種類があります。

ツールが返す結果は、テキスト形式に加えてスキーマで定義された構造化データを返すことも可能です。

以下は、PCのディスク使用量を取得するためのツールを提供するMCPサーバーです。

mcp_disc_usage.py
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アプリケーションの起動を行います。
app.py:①アプリケーション起動
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を呼び出すコールバックを設定します。
app.py:②Gradio UI構築
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)に書き込むストリームを作成します。
app.py:③MCPClientクラス
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に返します。
app.py:④MultiMCPManagerクラス
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情報やディスク容量など、実際の環境に基づいた回答が得られるはずです。

サンプルの全文は以下のリポジトリにあります。手元で実行したい場合は、こちらを参照してください。

セキュリティに関して

MCPはまだ発展途上のプロトコルであり、今後の普及や発展のためにはセキュリティ対策が非常に重要な課題となっています。MCPサーバーは外部リソースへのアクセス権を持つため、万が一悪意のあるコードが含まれていると、情報漏洩や不正操作などのリスクが生じます。特にサードパーティ製のMCPサーバーを利用する場合は、信頼できる提供元かどうかを十分に確認し、サーバーの内容をしっかりチェックすることが不可欠です。安全に利用するためにも、公式ドキュメントやコミュニティの情報を参考にし、慎重にサーバーを選択しましょう。

また、MCPサーバーのセキュリティチェックツール(例:MCP-ShieldMCP-Scanなど)も登場しています。こうしたツールを活用して、MCPサーバーの安全性を事前に確認することが重要になっています。

まとめ

本記事では、MCPホスト・MCPクライアント・MCPサーバーをすべてPythonで実装しながら、MCPの仕組みや各構成要素の役割を体験的に理解することを目指しました。自作を通じて、MCPの拡張性、そして今後のAIアプリ開発における可能性を実感していただけたのではないでしょうか。

MCPは、多様なデータソースやツールと連携したAIアプリケーションの構築に非常に有用なプロトコルです。PythonやTypeScriptをはじめ、さまざまな言語向けのSDKが提供されており、対応言語も拡大し続けています。主要なサービスベンダーもMCPの実装を積極的に進めており、オープンソース化や製品への組み込みが進むことで、今後さらに多くのユーザーに利用されることが期待されます。企業にとっても、優れたMCPサーバーを維持・開発するインセンティブが高まっています。

今後のロードマップでは、レジストリの整備など開発者にとってより使いやすい環境が整備されていく予定です。今後のMCPの進化とエコシステムの広がりに、ぜひご注目ください。

参考資料

おすすめ記事

記事・ニュース一覧