Model Context Protocol (MCP)

Model Context Protocol（MCP）

Model Context Protocol（MCP）は、LLM と外部ツールやデータソース間の通信チャネルを確立するために設計されたプロトコルです。クライアント・サーバーアーキテクチャに従い、メッセージングには JSON-RPC 2.0 を使用します。 MCP は Anthropic によって登場してからまだ数か月しか経っていませんが、AI コミュニティで急速に大きな注目を集めています。 以下のセクションでは、MCP の仕様を説明し、Python の例をいくつか紹介します。

---

MCP サーバー

MCP サーバーは、クライアントに次の 3 つの主要な機能を提供します。

機能	説明
ツール	ロジックを実行する（例：計算機）、副作用を引き起こす（例：環境を変更する）、または CPU 負荷の高い処理を行う関数やサービスです。
リソース	画像、ファイルなど、あらゆる種類のデータです。
プロンプト	クライアントが使用できるプロンプトテンプレートです。

MCP サーバーは、FastMCP または Low-Level API を使用して実装できます。 HTTP 経由の SSE（Server-Sent Events）または stdin による通信に対応しています。 ローカル通信には stdin を、ネットワーク通信には SSE を使用します。 この通信方式は、MCP 仕様ではトランスポート層と呼ばれています。

FastMCP

LLM ベースの英語学習アプリケーション向けに MCP サーバーを実装するとします。 この MCP サーバーは、評価クイズ（リソース）、学習者のレベルを判定するトリアージ関数（ツール）、およびプロンプトテンプレート（プロンプト）を提供します。

from mcp.server.fastmcp import FastMCP

# You can customize important parameters by passing them to FastMCP.
# In this example, all key parameters are set to their default values,
# but you can modify them according to your needs.
# sse_url = http://0.0.0.0:8000/sse
mcp = FastMCP(
    name="MCPServer",
    debug=True,
    host="0.0.0.0",
    port=8000,
    sse_path="/sse",
    message_path="/messages/",
    log_level="DEBUG",
)


@mcp.resource(
    uri="https://quiz.xyz",
    name="GetQuiz",
    description="Provides a link to an online English level assessment quiz.",
)
def get_quiz() -> str:
    return "Link to online quiz: https://quiz.xyz"


@mcp.tool(
    name="FindLevel",
    description="Determines the student's English level based on their quiz score.",
)
def find_level(grade: int) -> str:
    if grade < 50:
        return "Beginner"
    if grade < 75:
        return "Intermediate"
    return "Expert"


@mcp.prompt(
    name="GetPrompt",
    description="Generates a prompt to ask an LLM to teach English based on the student's level.",
)
def get_prompt(name: str, level: str) -> str:
    return f"Teach {name} English based on this level: {level}."

FastMCP を SSE 経由で提供する方法はいくつかあります。sse_app、run_sse_async、または run を使用できます。 最初の方法では、ASGI フレームワーク（例：Starlette）を使用します。 2 つ目の方法では、Python の asyncio を使用します。 3 つ目は、内部で run_sse_async を使用する MCP インスタンスの組み込み関数です。

from starlette.applications import Starlette
from starlette.routing import Mount
import uvicorn


if __name__ == "__main__":
    app = Starlette(debug=True, routes=[
        Mount("/", mcp.sse_app()),
    ])
    uvicorn.run(app)

import asyncio

async def main():
    await mcp.run_sse_async()


if __name__ == "__main__":
    asyncio.run(main())

# Pass transport=stdio for stdio server
if __name__ == "__main__":
    mcp.run(transport="sse")

Low Level API

Low-Level API を使ったサーバーの実装には時間がかかります。 Server クラスをインスタンス化し、すべての機能を一覧表示して呼び出すためのメソッドを多数実装する必要があります。 実装方法の詳細なガイドは、mcp python-sdk の公式リポジトリのこちらにあります。

---

MCP クライアント

サーバーの公開方法に応じて、クライアントの実装方法は sdio-client と sse-client の 2 通りあります。

SSE クライアント

fastmcp-server.py でトランスポートを sse に設定します。

import asyncio
import logging
from typing import Union
from urllib.parse import urlparse

from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream

import mcp.types as types
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
from mcp.shared.session import RequestResponder

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("client")


async def message_handler(
    message: Union[
        RequestResponder[types.ServerRequest, types.ClientResult],
        types.ServerNotification,
        Exception,
    ],
) -> None:
    if isinstance(message, Exception):
        logger.error("Error: %s", message)
        return
    logger.info("Received message from server: %s", message)


async def run_session(
    read_stream: MemoryObjectReceiveStream,
    write_stream: MemoryObjectSendStream,
) -> None:
    async with ClientSession(
        read_stream,
        write_stream,
        message_handler=message_handler,
    ) as session:
        initialize_response = await session.initialize()
        logger.info(f"Initialized response: {initialize_response}")
        tools = await session.list_tools()
        tools = tools.tools
        logger.info(f"List of tools: {[tool.name for tool in tools]}")
        response = await session.call_tool("FindLevel", {"grade": 86})
        logger.info(f"Find level response: {response}")


async def main(command_or_url: str) -> None:
    if urlparse(command_or_url).scheme in ("http", "https"):
        async with sse_client(command_or_url) as streams:
            await run_session(*streams)


if __name__ == "__main__":
    sse_url = "http://0.0.0.0:8000/sse"  # change url
    asyncio.run(main(sse_url))

Stdio クライアント

stdio クライアントを実行するにはサーバースクリプトを渡す必要があります（fastmcp-server.py でトランスポートを stdio に設定します）。

python stdio.py fastmcp-server.py

import asyncio
import sys
import logging
from contextlib import AsyncExitStack
from typing import Optional

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("client")


class MCPClient:
    def __init__(self) -> None:
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_server(self, server_script_path: str) -> None:
        if not server_script_path.endswith(".py"):
            raise ValueError("Server script must be a .py file")

        server_params = StdioServerParameters(
            command="python",
            args=[server_script_path],
            env=None,
        )

        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)
        )
        initialize_response = await self.session.initialize()
        logger.info(f"Initialized response: {initialize_response}")

    async def list_tools(self) -> None:
        if not self.session:
            raise RuntimeError("Session is not initialized.")
        response = await self.session.list_tools()
        tools = response.tools
        logger.info(f"List of tools: {[tool.name for tool in tools]}")

    async def find_level(self, grade: int) -> None:
        response = await self.session.call_tool("FindLevel", {"grade": grade})
        logger.info(f"Find level response: {response}")

    async def cleanup(self) -> None:
        await self.exit_stack.aclose()


async def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)

    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.list_tools()
        await client.find_level(86)
    except Exception as exc:
        print("Error:", exc)
    finally:
        await client.cleanup()


if __name__ == "__main__":
    asyncio.run(main())

---

プロトコル仕様

クライアントとサーバー間では、3 種類のメッセージがやり取りされます。 リモートプロシージャコールを可能にするため、すべてのメッセージは JSON-RPC 2.0 形式で記述されます。

• リクエスト：応答を必要とするリクエストです（JSONRPCRequest）。
• レスポンス：リクエストに対する成功時（エラーではない）の応答です（JSONRPCResponse）。
• 通知：応答を必要としない通知です（JSONRPCNotification）。

標準 JSON-RPC エラーコード：

• PARSE_ERROR = -32700
• INVALID_REQUEST = -32600
• METHOD_NOT_FOUND = -32601
• INVALID_PARAMS = -32602
• INTERNAL_ERROR = -32603

初期化

クライアントはメッセージを送信する前に、サーバーとの接続を正常に初期化する必要があります。

1. クライアントが初期化リクエスト（[InitializeRequest](https://github.com/modelcontextprotocol/python-sdk/blob/babb477dffa33f46cdc886bc885eb1d521151430/src/mcp/types.py#L283）を送信します。
2. サーバーがプロトコルのバージョンと機能（[InitializeResult](https://github.com/modelcontextprotocol/python-sdk/blob/babb477dffa33f46cdc886bc885eb1d521151430/src/mcp/types.py#L293）を返します。
3. クライアントが確認のため、初期化完了通知（[InitializedNotification](https://github.com/modelcontextprotocol/python-sdk/blob/babb477dffa33f46cdc886bc885eb1d521151430/src/mcp/types.py#L304C7-L304C30）を送信します。
4. 接続が確立されます。

検出

クライアントは次の操作を実行できます。

• ツールの一覧表示：ListToolsRequest/ ListToolsResult
• リソースの一覧表示：ListResourcesRequest/ ListResourcesResult
• プロンプトの一覧表示：ListPromptsRequest/ ListPromptsResult

呼び出し

クライアントは次の操作を実行できます。

• ツールの呼び出し：CallToolRequest/ CallToolResult
• プロンプトの取得：GetPromptRequest/ GetPromptResult
• リソースの読み取り：ReadResourceRequest /ReadResourceResult

---

Gradio + MCP

Gradio アプリで MCP サーバーを公開するのは簡単で、シームレスに連携できます。私のお気に入りの MCP 統合のひとつです。

1. MCP の追加機能を含めて Gradio をインストールします。

   pip install "gradio[mcp]"
   

2. mcp_server=True を指定してアプリを起動します。
3. MCP サーバーのアドレスがコンソールに表示されます。アプリページの下部から MCP のドキュメントにアクセスできます。

• 詳細については、Gradio MCP server guideをご覧ください。
• Hugging Face のデモ Spaceをお試しください。

---

参考資料

0. Model Context Protocol の紹介
1. MCP Python SDK/ Github
2. Model Context Protocol/ Github
3. Model Context Protocol ドキュメント
4. Hugging Face の MCP コース