Go to Qiita Advent Calendar Top
LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@yutaka-tanaka(田中 裕)in株式会社豆蔵

MCPの仕様をサンプルコードで理解する

Posted at

MCPのインタフェース仕様（Streamable HTTP）を詳しく理解するための良さげなサンプルコード（Python）を探していたのですが、AIエージェントツールでうまく動作しなかったりと、あまり良いものが見つからなかったので、ChatGPTで生成したサンプルコードに若干手を加えたもので動作検証してシーケンス図に整理してみました。

こんな人向けの内容

自分のようにMCPの仕様書だけでは（むずかしくて）いまいち本質を理解できないような人や、文字よりもソースコードや図表を使って右脳で理解したい人向けの内容になっています

  • MCPサーバの動きを大まかに理解したい
  • MCPサーバの動作に最低限どのような実装が必要となるのか知りたい
  • MCPクライアントとMCPサーバ間の入出力データを目視確認したい

おことわり

  • あくまでMCPプロトコルの仕様を理解するためのサンプルコードになっているのでMCPのライブラリ（SDK）は使用しない実装となっています。実際のMCPサーバ開発で実装する場合はFastMCP等のライブラリを使用して実装することを強くオススメします

    https://gofastmcp.com/getting-started/welcome

  • シーケンス図はあくまで大まかに理解するための記述なので厳密性は度外視してます。MCPの仕様を精緻に理解したい人は公式サイトを参照してください

    https://modelcontextprotocol.io/docs/getting-started/intro

MCPサーバのサンプルコード: mcp-server.py

※実行方法は後述します

import argparse
import json
import os

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
import uvicorn

app = FastAPI()

shutdown_requested = False


def log_request(data):
    print("=== MCP Message Received ===")
    print(json.dumps(data, indent=2, ensure_ascii=False))
    print("============================")


def normalize_method(method: str) -> str:
    """
    notifications/xxx → xxx に正規化
    """
    if method and method.startswith("notifications/"):
        return method[len("notifications/") :]
    return method


@app.post("/mcp")
async def mcp_handler(request: Request):
    global shutdown_requested

    body = await request.json()
    log_request(body)

    raw_method = body.get("method")
    method = normalize_method(raw_method)
    req_id = body.get("id")  # None => notification

    # ===============================
    # Requests
    # ===============================
    if method == "initialize":
        return JSONResponse(
            {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "protocolVersion": body["params"]["protocolVersion"],
                    "capabilities": {"tools": {"list": True, "call": True}},
                    "serverInfo": {
                        "name": "MCP-Server-Sample",
                        "version": "1.0.0",
                    },
                },
            }
        )

    if method == "shutdown":
        shutdown_requested = True
        return JSONResponse({"jsonrpc": "2.0", "id": req_id, "result": None})

    if method == "tools/list":
        return JSONResponse(
            {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "tools": [
                        {
                            "name": "echo",
                            "description": "Echo input text",
                            "inputSchema": {
                                "type": "object",
                                "properties": {"text": {"type": "string"}},
                                "required": ["text"],
                            },
                        }
                    ]
                },
            }
        )

    if method == "tools/call":
        args = body["params"]["arguments"]
        return JSONResponse(
            {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "content": [
                        {
                            "type": "text",
                            "text": args.get("text", "") + " [Echo from MCP Server]",
                        }
                    ]
                },
            }
        )

    # ===============================
    # Notifications
    # ===============================
    if req_id is None:
        if method == "initialized":
            print("MCP initialized notification received.")
            return Response(status_code=204)

        if method == "exit":
            if shutdown_requested:
                print("MCP exit notification received. Shutting down server.")
                os._exit(0)
            return Response(status_code=204)

        return Response(status_code=204)

    # ===============================
    # Unknown
    # ===============================
    return JSONResponse(
        {
            "jsonrpc": "2.0",
            "id": req_id,
            "error": {
                "code": -32601,
                "message": f"Unknown method: {raw_method}",
            },
        }
    )


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="MCP Server")
    parser.add_argument("--port", type=int, default=8000)
    args = parser.parse_args()

    uvicorn.run(app, host="0.0.0.0", port=args.port)

MCPクライアントのサンプルコード: mcp-client.py

※実行方法は後述します

import argparse
import json
import requests


def send_request(server_url, payload):
    print("\n=== Sending Message ===")
    print(json.dumps(payload, indent=2, ensure_ascii=False))
    print("=======================\n")

    try:
        response = requests.post(server_url, json=payload, timeout=5)
    except Exception as e:
        print(f"[ERROR] {e}")
        return None

    print("=== Server Response ===")

    if "application/json" not in response.headers.get("Content-Type", ""):
        print("[INFO] No JSON response (notification).")
        print("=======================\n")
        return response

    print(json.dumps(response.json(), indent=2, ensure_ascii=False))
    print("=======================\n")
    return response


def main(port):
    url = f"http://localhost:{port}/mcp"

    send_request(
        url,
        {
            "jsonrpc": "2.0",
            "id": 0,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-03-26",
                "capabilities": {},
                "clientInfo": {"name": "MCP-Client-Sample", "version": "1.0"},
            },
        },
    )

    send_request(
        url,
        {
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
        },
    )

    send_request(url, {"jsonrpc": "2.0", "id": 1, "method": "tools/list"})

    send_request(
        url,
        {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/call",
            "params": {
                "name": "echo",
                "arguments": {"text": "Hello from client"},
            },
        },
    )

    send_request(url, {"jsonrpc": "2.0", "id": 3, "method": "shutdown"})


# ===============================
# Exit (Commented out by default)
# ===============================
#    send_request(
#        url,
#        {
#            "jsonrpc": "2.0",
#            "method": "notifications/exit",
#        },
#    )

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--port", type=int, default=8000)
    main(parser.parse_args().port)

サンプルコードの補足

MCPライブラリ（SDK）等のリファレンス実装では具体的にどのように処理をしているのかわからなかったのですが、本サンプルコードではとりあえず以下のように実装にしています

  • 処理の判定基準は id がないものをnotificationと判定しています
  • tools/callの実装部（echo）は後述のシーケンス図のように責務を分割して別クラスや別メソッドで実装した方が良いと思いましたが、サンプルコードはできるだけシンプルにしたかったのでMCPサーバの処理内部に埋め込んでいます
  • MCPクライアントのnotifications/exitは実行するとMCPサーバのプロセスが終了してしまうため、コメントアウトしています

サンプルコードのシーケンス図

サンプルのソースコードの処理をざっくりシーケンス図で表すと以下の通りとなります

サンプルコード実行方法

各サンプルコード準備

$ mkdir ~/mcp-demo
# cd ~/mcp-demo
$ vi mcp-server.py
---
mcp-server.pyのソースコードをCopy&Paste
---

$ vi mcp-client.py
---
mcp-client.pyのソースコードをCopy&Paste
---

MCPサーバ起動手順

上記ファイルの準備が完了したらコンソールから以下のコマンドで起動します

# Pythonツールのインストール
$ sudo apt install pip python3.12-venv
$ vi ~/.bash_aliases
---
alias python="python3"
alias venv='python -m venv .venv; source .venv/bin/activate'
---

$ source ~/.bash_aliases
$ cd ~/mcp-demo

$ venv
$ pip install fastapi uvicorn requests

# MCP Server起動
# ポート指定(--port)なしの受信ポートは8080
$ python mcp-server.py --port 8000
# 正常に起動するとクライアントからの受信を待ち受ける

MCPクライアント実行手順

MCPサーバとは別のコンソールから以下のコマンドを実行します。正常に実行されるとMCPサーバとMCPクライアントのコンソールに、入出力データ（JSON）が出力されます

# MCP Client実行
# ポート指定なし(--port)の送信ポートは8080
$ python mcp-client.py --port 8080

AIエージェントとの接続

任意のAIエージェントツールを使ってサンプルのMCPサーバとの接続を試してみてください。問題なく接続できるとMCPサーバのコンソールに呼び出し元から受信した入力データ（JSON）が出力されます。色々なツールを使って検証したわけではないので、サンプルソース側の問題で接続がうまくいかない場合もあると思います。コンソールの出力内容を確認しながら、適宜コードの方を修正してみてください（生成AIにお願いするのが早いかも）

尚、自分はDifyを使って接続を試したので、参考までの設定手順を残しておきます

1)「ツール」にある「MCP」のタブを開く
2)「MCP サーバー（HTTP）を追加」押下して以下の設定を入力
---
サーバーURL: http://<IP>:8000/mcp
名前とアイコン: 任意
サーバー識別子: 任意
※その他の設定はデフォルト値のまま
---

サンプルコードを使って検証した結果

<検証結果1>

サンプルコードを実際に動かすことで、MCP内部で具体的にどのような入出力データ発生しているかを確認できたことで、MCPサーバの実装には必須のメソッドと任意のメソッドがあることが理解できました。また、ビジネスロジックを実装するtoolsメソッドが存在しないとMCPサーバを構築する意味を成さないため事実上の必須メソッドと捉えて良いものと理解しました

〇必須のメソッド

  • initialize
  • shutdown
  • notifications/initialized (or initialized)
  • notifications/exit (or exit)

〇事実上必須のメソッド

  • tools/list
  • tools/call

※任意メソッドを含めた主要メソッドについては後ろでまとめています

<検証結果2>

AIエージェントツールのDifyを使ってMCP接続機能を使用して接続検証したところ notifications/initializedが渡されていることを確認しましたが、MCPの仕様（JSON-RPCの仕様上？）ではinitializedのみで呼び出しができるようにする必要もあるようなので、とりあえずnotifications/ が付いていたら正規化し、どちらでも処理が動くようにサンプルコードを修正しました（この辺りは何が正しい実装なのかいまだに良くわかってない）

参考: MCPの主要メソッドと処理内容

サンプルソースコードやシーケンス図はあくまで動作に最低限必要な範囲のみしています。MCPの仕様ではメソッドは数多く存在しており、すべてを記載することは難しいことから主要なメソッド（ネット上で良く見かけたメソッド）を以下のようにまとめてみました

※全メソッドを網羅したい場合は公式サイトを参照してください

Requestメソッド（同期系）の一覧

カテゴリ メソッド 方向 必須区分 処理概要
lifecycle initialize Client → Server 必須 プロトコル初期化・能力確認交換
lifecycle shutdown Client → Server 必須 正常終了要求（exit前段）
tools tools/list Client → Server 任意 利用可能ツール一覧取得
tools tools/call Client → Server 任意 ツール実行
resources resources/list Client → Server 任意 リソース一覧取得
resources resources/read Client → Server 任意 リソース内容取得
resources resources/subscribe Client → Server 任意 リソース更新購読
resources resources/unsubscribe Client → Server 任意 購読解除
prompts prompts/list Client → Server 任意 プロンプト一覧取得
prompts prompts/get Client → Server 任意 プロンプト取得
sampling sampling/createMessage Server → Client 任意 LLM生成要求（Server→Host唯一のrequest）
utility
(cancel)		 cancelRequest Client → Server 任意 実行中requestのキャンセル（中断）

Notificationメソッド（非同期系）の一覧

カテゴリ メソッド 方向 必須区分 処理概要
lifecycle notifications/initialized
(initialized)		 Client → Server 必須 初期化完了通知
lifecycle notifications/exit
(exit)		 Client → Server 必須 プロセス終了通知
utility notifications/progress
(progress)		 Server → Client 任意 進捗通知
utility notifications/logging/message
(logging/message)		 Server → Client 任意 ログ通知
resources notifications/resources/updated
(resources/updated)		 Server → Client 任意 リソース更新通知
0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?