ローカルLLMに目と手をつけてみる(9.MCP)

MCP前夜

LLMに外部データを参照にいってほしい、という考えは当然持ち上がります。念の為にRAGとの違いを書くと、RAGは事前情報を渡せますが、実行中にダイナミックにデータを読んだり、書いたりはできません。

それでもLLMと会話が終わるたびに、ベクトルデータベースにやり取りを記憶する、ということに使えます。

LLMモデルが外部とのやり取りをすることをFunction CallingとかTool Callingなどといいます。つまりLLMが外部へのアクセスの方法を知っていること、必要に応じてそれを使うことが期待されます。
Function Callingのフローについては、たとえば実験用に使っているMistralだとここにFunction Callingのメカニズムがものすごく丁寧に記載されています。ただし、JSONファイルで使いうる関数を定義しておかねばなりません。

このことは「モデルごとに関数を使わせる設定が違う」ことを意味しています。
それはちょっとやってられません。

現在、業界全体としての解決策は以下のような二種類が提示されています。

1. OpenAI API 互換レイヤー: ライブラリ側でこの複雑なタグ打ちや登録を代行する。
2. MCP (Model Context Protocol): 「道具の渡し方」そのものをモデルの文法から切り離し、標準化する。

1はChatGPTなどで外部関数を使わせる場合、OpenAI SDKを使う。陰で関数を使えるようにラッピングするので気付かないエンジニアもいるようです。

MCPとは

MCPはモデルごとに機能拡張を開発する必要のない共通の方法です。AI界の「USB-C規格」ともいえます。
LLM側がMCPをサポートしているか、ということの確認は必要となります。

三つの役割

MCPの世界には、3つの登場人物がいます。

1. MCPサーバー (The Tool): 「時計」「Google検索」「ファイル操作」など、特定のデータサービスをするサーバー
2. MCPホスト LLM本体や機能を呼び出すPythonプログラム。「道具を使いたい側」
3. MCPクライアント (The Bridge): ホスト（LLM）とサーバー（道具）を繋ぐケーブルの役割をします。

Anthropicが提供している FastMCPをインストールする。

pip install mcp

MCPサーバーを作ってみます。
ソースコードファイル名：mcp_clock_server.py

# pip install mcp 前提
# pip install mcp 前提
from mcp.server.fastmcp import FastMCP
from datetime import datetime

# サーバーに名前をつけます
mcp = FastMCP("Yuki-Clock")

# --- ツール1: 日付のみ ---
@mcp.tool()
def get_current_date() -> str:
    """現在の日付（年・月・日）のみを取得します。
    曜日は含まれません。今日が何日か知りたい時に使います。"""
    now = datetime.now()
    return now.strftime("%Y年%m月%d日")

# --- ツール2: 時刻のみ ---
@mcp.tool()
def get_current_time() -> str:
    """現在の時刻（時・分・秒）のみを取得します。
    日付の情報は含まれません。今が何時か知りたい時に使います。"""
    now = datetime.now()
    return now.strftime("%H時%M分%S秒")

if __name__ == "__main__":
    mcp.run()

これは単体でも動きますが、なにも表示されません。クライアントから突かれる存在だからです。
以下がこのサーバーを自動起動させるコートをふくんだクライアントプログラムです。
ソースコードファイル名：mcp_client_test.py

# pip install mcp 前提
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run():
    # 1. サーバーの起動条件を指定（上のサーバープログラム）
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_clock_server.py"], # サーバーのファイル名
    )

    # 2. サーバーに接続する
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初期化
            await session.initialize()

            # 3. 使えるツールの一覧を取得して表示してみる
            tools = await session.list_tools()
            print("--- 利用可能なツール ---")
            for tool in tools.tools:
                print(f"名前: {tool.name}")
                print(f"説明: {tool.description}")

            # 4. 実際にツールを呼び出してみる
            print("\n--- ツールを実行中... ---")
            result = await session.call_tool("get_current_time", arguments={})
            print(f"結果: {result.content[0].text}")

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

async/awaitですね。サーバー側の動きを待たずに他の処理が可能な宣言です。昔はコールバックという手法を使いましたが、それの進化系です。asyncは簡単ではないので、別途、他の資料にあたってください。

次はMCPを動かしている例です。関数リストをRAGのように事前に渡し、質問に応じて関数を呼び出します。
ソースコード：mcp_client_testv3.py

# pip install mcp 前提
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run():
    # 1. サーバー（時計）の起動設定
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_clock_server.py"], 
    )

    print("（システム：MCPサーバーを起動して接続します...）")
    
    # 2. サーバーと接続
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初期化（「握手」のようなもの）
            await session.initialize()

            # --- STEP 1: MCPサーバーから「道具の生データ」を取得 ---
            tools_response = await session.list_tools()
            print("\n=== [STEP 1] MCPから取得したツールの生データ ===")
            for tool in tools_response.tools:
                print(f"名前: {tool.name}")
                print(f"説明文（docstringより）: {tool.description}")
                print(f"引数の構造: {tool.inputSchema}")

            # --- STEP 2: ブローカーによる「翻訳（プロンプト化）」 ---
            # ここが、司さんの仰る「LLMに伝えるための形」を作る部分です
            print("\n=== [STEP 2] LLM（Mistral）へ送る『お品書き』への変換 ===")
            
            mistral_format_tools = []
            for tool in tools_response.tools:
                # サーバーの情報を、Mistralが期待するJSON構造に詰め替える
                mistral_format_tools.append({
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema
                    }
                })
            
            # Mistralの専用タグ [AVAILABLE_TOOLS] で囲む
            prompt_for_yuki = f"[AVAILABLE_TOOLS]\n{json.dumps(mistral_format_tools, indent=2, ensure_ascii=False)}\n[/AVAILABLE_TOOLS]"
            
            print("【生成されたプロンプト】")
            print(prompt_for_yuki)
            print("\n※この文字列が、実際の会話の裏側でシステムプロンプトと一緒にLLMへ送られます。")

            # --- STEP 3: 実行テスト（念のため） ---
            print("\n=== [STEP 3] 実際にツールを実行してみる ===")
            result = await session.call_tool("get_current_time", arguments={})
            print(f"実行結果: {result.content[0].text}")

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