MCPサーバのツール一覧を取得するPythonコード

概要

生成AIを絡めたテストをするためPythonでMCPと通信を行う。
今回は、PythonのrequestsでMCPサーバに接続して、tools/listによりツール一覧を取得する。

実行結果

実行結果は以下のスクショ。
MCPサーバに含まれるaddツールがtools/listにより取得できている。
今回のソースコードは、セッションIDなどの情報も出力している。

ソースコード

import requests
import json

def get_mcp_tools(url="http://localhost:8080/mcp"):
    # ==========================================
    # 1. 初期化 (initialize) してセッションIDを取得
    # ==========================================
    init_headers = {
        "Accept": "application/json, text/event-stream",
        "Content-Type": "application/json"
    }

    init_payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-11-25",
            "capabilities": {},
            "clientInfo": {
                "name": "python-script",
                "version": "1.0"
            }
        }
    }

    print(f"1. 送信中: initialize リクエスト... (URL: {url})")
    try:
        init_response = requests.post(url, headers=init_headers, json=init_payload)
        init_response.raise_for_status()
    except requests.exceptions.RequestException as e:
        print(f"初期化リクエストに失敗しました: {e}")
        return

    # レスポンスヘッダーから Mcp-Session-Id を抽出
    session_id = init_response.headers.get("Mcp-Session-Id")
    if not session_id:
        print("エラー: レスポンスヘッダーに 'Mcp-Session-Id' が見つかりませんでした。")
        return
    
    print(f"   => 取得した Session ID: {session_id}\n")

    # ==========================================
    # 2. 取得したセッションIDを使って tools/list を実行
    # ==========================================
    tools_headers = {
        "Accept": "application/json, text/event-stream",
        "Content-Type": "application/json",
        "Mcp-Session-Id": session_id
    }

    tools_payload = {
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/list",
        "params":{}
    }

    print("2. 送信中: tools/list リクエスト...")
    try:
        tools_response = requests.post(url, headers=tools_headers, json=tools_payload)
        tools_response.raise_for_status()
    except requests.exceptions.RequestException as e:
        print(f"tools/list リクエストに失敗しました: {e}")
        return

    # ==========================================
    # 3. SSEフォーマットから JSON (data: の行) を抽出
    # ==========================================
    if not tools_response.text.strip():
        print("\nエラー: レスポンスが空です。")
        return

    json_str = None
    # レスポンスのテキストを行ごとに分割して処理
    for line in tools_response.text.splitlines():
        if line.startswith("data:"):
            # "data:" の文字 (5文字) 以降を切り出して取得
            json_str = line[5:].strip()
            break
            
    if json_str:
        try:
            # 切り出した文字列をJSONとして解析
            response_json = json.loads(json_str)
            print("   => ツール一覧 (JSON結果):")
            print(json.dumps(response_json, indent=4, ensure_ascii=False))
        except json.JSONDecodeError:
            print(f"\nエラー: data行のJSON解析に失敗しました。\n対象文字列: {json_str}")
    else:
        print("\nエラー: レスポンスの中に 'data:' から始まる行が見つかりませんでした。")
        print(f"生レスポンス:\n{tools_response.text}")

if __name__ == "__main__":
    # URLを引数として渡して実行
    target_url = "http://localhost:8080/mcp"
    get_mcp_tools(target_url)