MCP (Model Context Protocol) サーバー開発ガイド

AIモデルサービングの効率性と柔軟性を最大化できるMCP (Model Context Protocol) サーバーの開発方法についてご紹介します。MCPは、モデル推論時に必要なコンテキスト情報を効率的に管理・伝達するためのプロトコルであり、これを活用することで、複雑なAIサービスもすっきりと拡張性高く構築できます。

1. MCPとは何か？

MCPは、モデルが推論を実行するために必要な追加情報（コンテキスト）を定義し、伝達する方式を標準化したプロトコルです。例えば、大規模言語モデル（LLM）の場合、以前の会話履歴、ユーザー設定、システムメッセージなどがコンテキストとなり得ます。MCPは、これらのコンテキストをモデル入力と分離して管理することで、以下の利点を提供します。

効率的なコンテキスト管理: 重複するコンテキスト送信を減らし、ネットワーク負荷を軽減します。
柔軟なモデルサービング: 同じモデルでも異なるコンテキストを注入することで、多様なシナリオに対応できます。
拡張性: コンテキストロジックとモデルロジックを分離することで、システムの拡張が容易になります。
再現性: コンテキストを明確に定義し管理することで、推論結果を容易に再現できます。

2. MCPサーバー開発のための基本構造

MCPサーバーは、主に以下の構成要素で成り立っています。

MCP Handler: MCPリクエストをパースし、コンテキストを処理する中核ロジックです。
Context Store: コンテキスト情報を保存し、参照するためのストレージです。（例：Redis, NoSQL DB, In-memory Cache）
Model Inference Engine: 実際のAIモデル推論を実行する部分です。（例：PyTorch, TensorFlow Serving, ONNX Runtime）
API Endpoint: クライアントからMCPリクエストを受け取り、処理するHTTP/gRPCエンドポイントです。

3. 開発段階別ガイド

3.1. プロトコル定義 (Protobuf推奨)

MCPの核は「コンテキスト」をどのように定義するかです。プロトコルバッファ (Protobuf) を使用して、コンテキストの構造を明確に定義することを強く推奨します。

例: model_context.proto

syntax = "proto3";

package mcp;

message UserContext {
    string user_id = 1;
    repeated string preferences = 2;
    map<string, string> custom_data = 3;
}

message ConversationContext {
    string conversation_id = 1;
    repeated Message messages = 2; // 以前の会話履歴
}

message Message {
    string sender = 1;
    string text = 2;
    int64 timestamp = 3;
}

message ModelContext {
    oneof context_type {
        UserContext user_ctx = 1;
        ConversationContext conv_ctx = 2;
        // 他の種類のコンテキストを追加可能
    }
    // 共通で必要なメタデータ
    string model_version = 10;
    string request_id = 11;
}

message InferenceRequest {
    string model_name = 1;
    bytes input_data = 2; // モデル入力データ (serialized)
    ModelContext context = 3; // MCPコンテキスト
}

message InferenceResponse {
    bytes output_data = 1; // モデル出力データ (serialized)
    string request_id = 2;
    string model_name = 3;
}

上記のProtobuf定義により、InferenceRequestにModelContextフィールドを含めてコンテキスト情報を一緒に送信できます。

3.2. コンテキストストアの実装

コンテキストストアは、MCPサーバーのパフォーマンスに重要な影響を与えます。どのコンテキストをどの周期で保存・管理するかによって、適切なストレージを選択する必要があります。

Redis: 高速な読み書き速度を提供し、リアルタイムのコンテキスト管理に適しています。揮発性データやTTL (Time-To-Live) が必要なコンテキストに優れています。
NoSQL DB (MongoDB, Cassandraなど): 柔軟なスキーマと水平スケーリングが可能で、多様な種類のコンテキストデータを保存・管理するのに適しています。
リレーショナルDB: 構造化されたコンテキストデータに適しており、ACID特性が必要な場合に検討できます。
In-memory Cache: 非常に短時間のみ保持されるコンテキストや、キャッシング層としての役割で活用できます。

例 (Python + Redis):

import redis
import json

class ContextStore:
    def __init__(self, host='localhost', port=6379, db=0):
        self.r = redis.Redis(host=host, port=port, db=db)

    def get_context(self, context_id: str) -> dict:
        data = self.r.get(f"mcp:context:{context_id}")
        return json.loads(data) if data else {}

    def set_context(self, context_id: str, context_data: dict, ttl_seconds: int = None):
        self.r.set(f"mcp:context:{context_id}", json.dumps(context_data))
        if ttl_seconds:
            self.r.expire(f"mcp:context:{context_id}", ttl_seconds)

    def delete_context(self, context_id: str):
        self.r.delete(f"mcp:context:{context_id}")

# 使用例
# store = ContextStore()
# store.set_context("user-123", {"preferences": ["dark_mode", "notifications"], "last_login": "2025-06-17"})
# user_ctx = store.get_context("user-123")

3.3. MCP Handlerの実装

MCP Handlerは、クライアントから受け取ったInferenceRequestからModelContextを抽出し、これに基づいて必要なコンテキスト情報をコンテキストストアから参照したり生成したりして、モデル推論に注入する役割を担います。

中核ロジック:

Requestパース: Protobufで定義されたInferenceRequestを逆シリアル化します。
Context抽出: request.contextフィールドからModelContextを抽出します。
Context組み合わせ:
- ModelContext自体に含まれるコンテキストを使用します。
- ModelContext内のIDを使用して、コンテキストストアから追加のコンテキストをロードします。
- (必要に応じて) デフォルトのコンテキストまたはキャッシュされたコンテキストを適用します。
モデル入力準備: 組み合わせたコンテキストとrequest.input_dataを合わせて、モデルが要求する形式に変換します。
モデル推論呼び出し: 準備された入力をモデル推論エンジンに渡し、推論を実行します。
Response生成: モデルの出力とリクエストIDなどを含めてInferenceResponseを生成します。

例 (Python + FastAPI):

from fastapi import FastAPI, Request
from google.protobuf.json_format import ParseDict, MessageToDict
import your_protobuf_pb2 as pb2 # Protobufコンパイル結果

app = FastAPI()
context_store = ContextStore() # 上で定義したContextStoreインスタンス

# 仮想のモデル推論関数
def run_inference(input_data: bytes, combined_context: dict) -> bytes:
    print(f"Running inference with input: {input_data[:20]}..., context: {combined_context}")
    # 実際のモデル推論ロジック (TensorFlow, PyTorchなど)
    # ここでは簡単に入力データを返す
    return b"processed_" + input_data

@app.post("/infer")
async def infer(request: Request):
    raw_body = await request.body()
    try:
        # Protobufバイナリ直接パース (実運用時)
        # inference_request = pb2.InferenceRequest()
        # inference_request.ParseFromString(raw_body)

        # JSONリクエストと仮定 (開発の便宜上)
        request_data = await request.json()
        inference_request = ParseDict(request_data, pb2.InferenceRequest())

    except Exception as e:
        return {"error": f"Invalid request format: {e}"}, 400

    model_name = inference_request.model_name
    input_data = inference_request.input_data

    combined_context = {}
    if inference_request.HasField("context"):
        # ModelContextフィールドが存在する場合
        if inference_request.context.HasField("user_ctx"):
            user_ctx_dict = MessageToDict(inference_request.context.user_ctx)
            combined_context.update(user_ctx_dict)
            # UserContext IDを利用してContextStoreから追加データをロード (例)
            if "user_id" in user_ctx_dict:
                stored_user_data = context_store.get_context(f"user:{user_ctx_dict['user_id']}")
                combined_context.update(stored_user_data) # 保存されたデータで更新

        if inference_request.context.HasField("conv_ctx"):
            conv_ctx_dict = MessageToDict(inference_request.context.conv_ctx)
            combined_context.update(conv_ctx_dict)
            # ConversationContext IDを利用してContextStoreから追加データをロード (例)
            if "conversation_id" in conv_ctx_dict:
                stored_conv_data = context_store.get_context(f"conv:{conv_ctx_dict['conversation_id']}")
                combined_context.update(stored_conv_data)

        # 共通メタデータ追加
        if inference_request.context.model_version:
            combined_context["model_version"] = inference_request.context.model_version
        if inference_request.context.request_id:
            combined_context["request_id"] = inference_request.context.request_id


    # モデル推論実行
    output_data = run_inference(input_data, combined_context)

    response = pb2.InferenceResponse(
        output_data=output_data,
        request_id=inference_request.context.request_id if inference_request.HasField("context") else "",
        model_name=model_name
    )

    # Protobufバイナリで応答 (実運用時)
    # return Response(content=response.SerializeToString(), media_type="application/x-protobuf")

    # JSON応答と仮定 (開発の便宜上)
    return MessageToDict(response)

# サーバー実行: uvicorn main:app --reload

3.4. モデル推論エンジン連携

実際のAIモデルをサービングする方法は様々です。

直接統合: PyTorch、TensorFlow、JAXなど、ディープラーニングフレームワークをMCPサーバー内に直接ロードして推論します。高速な応答時間が必要な場合に適しています。
モデルサービングフレームワーク活用: NVIDIA Triton Inference Server、TensorFlow Serving、TorchServeなどの専門的なモデルサービングフレームワークを使用します。モデル管理、バッチ処理、A/Bテストなどの高度な機能を活用できます。MCPサーバーは、これらのフレームワークのAPIを呼び出すゲートウェイの役割を果たします。
外部API呼び出し: OpenAI API、Google Cloud AI Platformなど、外部のAIモデルAPIを呼び出します。この場合、MCPサーバーはコンテキストを組み合わせて外部APIの入力形式に変換する役割を担います。

どの方式を選択しても、MCP Handlerで準備されたコンテキストと入力データを、モデルが要求する形式に変換して渡すことが重要です。

4. 高度な考慮事項

バージョン管理: MCPプロトコルとモデルコンテキストのスキーマ変更に対するバージョン管理戦略を確立する必要があります。
キャッシング戦略: 頻繁に使用されるコンテキストや計算コストの高いコンテキストは、キャッシングしてパフォーマンスを最適化できます。
エラーハンドリング: コンテキストロードの失敗、モデル推論エラーなど、さまざまなエラー状況に対する堅牢なエラーハンドリングを実装する必要があります。
ロギングとモニタリング: コンテキスト処理の過程、モデル推論のパフォーマンス、エラー発生状況などを詳細にロギングし、モニタリングすることで、問題発生時に迅速に診断・解決できるようにします。
セキュリティ: 機密性の高いコンテキスト情報（例：ユーザー個人情報）に対するアクセス制御、暗号化、安全な転送などを考慮する必要があります。
スケーラビリティ: トラフィック増加に備え、MCPサーバーを水平方向にスケールアウトできるように設計する必要があります。（ロードバランサー、Kubernetesなど活用）
非同期処理: コンテキストのロード、モデル推論などが非同期的に実行されるようにすることで、応答時間を短縮できます。

5. 結論

MCP (Model Context Protocol) は、AIモデルサービングの複雑さを軽減し、柔軟性を高める強力な方法論です。Protobufを活用した明確なプロトコル定義、効率的なコンテキストストア管理、そして堅牢なMCP Handlerの実装を通じて、あなたのAIサービスをさらに強力にすることができます。