【次世代AI開発：第1回】MCP（Model Context Protocol）の内部構造解剖と自作サーバー実装

【新連載：MCP・高度RAG・自律エージェントの現場設計論】

※本連載は、MCP（Model Context Protocol）、Context Engineering、GraphRAG、Self-Healing Agent、ローカルSLMなどの最先端技術スタックを用い、実務プロダクションで真に耐えうる次世代AIシステムを構築する重厚なハンズオン連載です。

1. イントロダクション（本記事のねらいと到達目標）

AI技術の主軸は、単にチャットUIで文章を生成させる「一問一答」の段階から、外部のシステム、データベース、ローカルツール、クラウドサービスと連携して自律的にタスクを遂行する「AIエージェントシステム」へと急速にシフトしています。

しかし、従来のAIシステム構築においては、OpenAI、Anthropic、Googleなど、モデルプロバイダーごとに異なる Function Calling や Tool 利用のインターフェースが存在し、システム側も独自APIの繋ぎ込みロジックを個別に実装する必要がありました。

この「N×M」の密結合なカオスを解決すべく、2024年末にAnthropicによって提唱され、業界標準プロトコルとして急速に普及しているのが「MCP (Model Context Protocol)」です。

本連載の記念すべき第1回では、MCPの基礎となるアーキテクチャ思想と、内部でやり取りされる JSON-RPC 2.0 メッセージプロトコルの仕組みを解剖します。そして、Pythonの最新SDKが提供する FastMCP フレームワークを用いて、Tools（道具）、Resources（文脈データ）、Prompts（定型指示）の3大要素を網羅したオリジナルのMCPサーバーを完全実装・起動することを目標とします。

前提条件と動作検証環境（Prerequisites）

Pythonバージョン: Python 3.10 以上
動作確認済み主要ライブラリ:
- mcp>=1.2.0（Anthropic公式 MCP SDK）
推奨開発環境:
- VS Code または Cursor
- ターミナル環境（Bash / zsh / PowerShell）

2. 【理論・背景】なぜMCPが必要なのか：従来のAPI連携の限界

従来のLLMアプリケーション開発において、モデルに外部ツールやデータベースを利用させる場合、以下のような構造的課題が存在していました。

1. 接続の「N × M」問題とコードの断片化

異なるLLM基盤（OpenAI, Claude, Gemini, ローカルLLM等）や、多様なデータソース（PostgreSQL, Slack, GitHub, ローカルファイル等）が存在する中で、それぞれの組み合わせごとに専用のプロンプト変換やAPIクライアントコードを記述する必要がありました。

【従来の密結合アプローチ (N × M)】
[Claude]  ---> (専用ツール定義) ---> [PostgreSQL]
[OpenAI]  ---> (専用Function)  ---> [GitHub API]
[Gemini]  ---> (専用ToolCall)  ---> [Slack API]

2. プラグ＆プレイ化を実現する「MCP (Model Context Protocol)」

MCPは、コンピュータにおける USB Type-C のような標準規格として機能します。LLMクライアント（Claude Desktop, Cursor, AI Agent等）とデータソース/ツール（MCP Server）の間に標準化された抽象化レイヤーを挟むことで、一度作成したMCPサーバーは、どんなMCP対応クライアントからも即座に共通利用（プラグ＆プレイ）できるようになります。

【MCPによる標準化アプローチ (N + M)】
[Claude Desktop] ---\                  /---> [PostgreSQL MCP Server]
[Cursor Editor]  ----[ MCP Protocol ]------> [GitHub MCP Server]
[Custom Agent]   ---/   (JSON-RPC)     \---> [Local Tool MCP Server]

3. 従来アプローチとMCPの機能比較

評価項目	従来の Function Calling 実装	MCP (Model Context Protocol)
結合度	特定LLM API / フレームワークへの密結合	クライアント・サーバー完全分離（スパース結合）
拡張性	新ツール追加のたびにクライアントコードを改修	MCPサーバーをプロセス追加するだけで自動認識
通信規格	各社独自のHTTP/JSON構造	JSON-RPC 2.0（Stdio または SSE / HTTP）
セキュリティ	アプリケーションプロセス内に鍵や処理を同居	ツール実行プロセスを別環境へ隔離可能
抽象化の範囲	関数呼び出し（Tool）のみ	Tools（関数）＋ Resources（データ）＋ Prompts（定型句）

3. 【仕様解剖】MCPプロトコル（JSON-RPC 2.0）の内部構造

MCPは、クライアントとサーバー間において JSON-RPC 2.0 メッセージ規格に基づく双方向メッセージングで動作します。トランスポート層としては、ローカル環境で標準入出力を用いる Stdio（Standard I/O）、およびネットワーク経由で非同期通信を行う SSE（Server-Sent Events） が定義されています。

1. MCPを構成する「3大抽象プリミティブ」

MCPサーバーは、クライアント（LLM）に対して以下の3つの機能を露出（Expose）させます。

Tools（ツール）:
- LLMが呼び出す「実行可能な関数」。副作用（Side-effect）を伴うアクション（例: DB書き込み、外部API発行、計算など）を実行します。
Resources（リソース）:
- URI（例: file:///path/to/data.json や db://users/profile）で指定される「読み取り専用の文脈データ」。ファイルの中身やログなど、LLMが知識として参照する情報を提供します。
Prompts（プロンプト）:
- サーバー側で事前定義された「再利用可能なプロンプトテンプレート」。ユーザーやシステムが複雑なタスクを呼び出すための定型フレーズを提供します。

2. MCP初期化およびツール実行のシーケンス

4. 【完全実装】FastMCPを用いた独自MCPサーバーの構築

それでは、実際に Python の最新SDKに含まれる高レベルフレームワーク FastMCP を使用して、Tools、Resources、Prompts の全要素をカバーした独自のMCPサーバー MyFirstMCPServer を構築してみましょう。

手元の環境で新しい Python ファイル mcp_server_demo.py を作成し、以下のコードを記述します。

# 動作確認済みライブラリバージョン: mcp>=1.2.0
import os
import sys
import datetime
from typing import Dict, Any
from mcp.server.fastmcp import FastMCP

# 1. MCPサーバーインスタンスの初期化
# サーバー名「EnterpriseAssistantServer」として定義します
mcp = FastMCP("EnterpriseAssistantServer")


# =====================================================================
# 2. Tools の定義 (@mcp.tool デコレータ)
# LLMが計算やアクションを実行するための関数です。
# =====================================================================

@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> str:
    """体重(kg)と身長(m)を受け取り、BMI値と肥満度の判定結果を算出します。

    Args:
        weight_kg (float): 体重（キログラム）
        height_m (float): 身長（メートル）
    """
    if height_m <= 0:
        return "エラー: 身長は0より大きい数値を指定してください。"
    
    bmi = weight_kg / (height_m ** 2)
    bmi_rounded = round(bmi, 2)
    
    if bmi < 18.5:
        category = "低体重 (Underweight)"
    elif 18.5 <= bmi < 25.0:
        category = "普通体重 (Normal weight)"
    elif 25.0 <= bmi < 30.0:
        category = "肥満 1度 (Overweight)"
    else:
        category = "肥満 2度以上 (Obese)"
        
    return f"BMI数値: {bmi_rounded} (判定: {category})"


@mcp.tool()
def fetch_system_status() -> str:
    """現在のサーバー稼働状況と時刻情報を取得します。"""
    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    python_version = sys.version.split()[0]
    return f"サーバー状態: 正常稼働中 | 現在時刻: {now} | Python環境: v{python_version}"


# =====================================================================
# 3. Resources の定義 (@mcp.resource デコレータ)
# LLMが文脈として読み取るためのURIベースの静的・動的データです。
# =====================================================================

@mcp.resource("system://info")
def get_system_info() -> str:
    """システム仕様メタデータを返すリソース"""
    return """
[システム仕様定義書]
- システム名: 次世代AI統合プラットフォーム
- バージョン: v2.4.0-production
- 動作モード: エンタープライズ・セキュアモード
- サポートプロトコル: MCP v1.0 / JSON-RPC 2.0
"""


@mcp.resource("user://profile/{user_id}")
def get_user_profile(user_id: str) -> str:
    """ユーザーIDに応じた動的プロファイルを返すリソース"""
    # 疑似データベースからの検索処理
    profiles: Dict[str, Dict[str, Any]] = {
        "usr_101": {"name": "山田 太郎", "role": "リードエンジニア", "department": "AI推進部"},
        "usr_102": {"name": "佐藤 花子", "role": "プロダクトマネージャー", "department": "企画部"},
    }
    
    user_data = profiles.get(user_id)
    if not user_data:
        return f"ユーザープロファイルエラー: ID '{user_id}' は見つかりませんでした。"
        
    return f"【ユーザープロファイル情報】\n- ID: {user_id}\n- 氏名: {user_data['name']}\n- 役職: {user_data['role']}\n- 部署: {user_data['department']}"


# =====================================================================
# 4. Prompts の定義 (@mcp.prompt デコレータ)
# 再利用可能なプロンプトテンプレートを事前定義します。
# =====================================================================

@mcp.prompt()
def code_review_prompt(language: str, code_snippet: str) -> str:
    """指定されたプログラミング言語のコードに対するレビュープロンプトを構築します"""
    return f"""以下の{language}言語で記述されたソースコードを詳細にレビューしてください。

【評価ポイント】
1. パフォーマンス上のボトルネックや無駄なループ
2. セキュリティ上の脆弱性（SQLインジェクション、メモリ漏洩等）
3. コードの可読性と命名規則

【対象コード】:
```{language}
{code_snippet}

"""

=====================================================================

5. エントリポイント（サーバーの起動）

デフォルトで Stdio (標準入出力) トランスポートで起動します。

=====================================================================

if name == "main":
print("🚀 FastMCP サーバーを起動しています (Stdio モード)...", file=sys.stderr)
mcp.run()


---

## 5. コードの行別・ロジック詳細解説

実装した `mcp_server_demo.py` の内部ロジックについて、技術的なポイントを1つずつ深掘り解説します。

### 1. `FastMCP("EnterpriseAssistantServer")` によるインスタンス化
- `FastMCP` は、MCPの低レイヤーメッセージ（JSON-RPCのパース、セッションハンドシェイク、Stdioストリームの管理）を内部で全自動化してくれる高レベルラッパーです。
- 引数に渡したサーバー名は、MCPクライアントとの初期化通信時に `serverInfo.name` として相手へ伝達されます。

### 2. `@mcp.tool()` デコレータと Pydantic / 型ヒントによるスキーマ自動生成
- 関数に `@mcp.tool()` を付与すると、FastMCPは関数の **型ヒント（Type Hints: `weight_kg: float`）** と **Docstring（docstringのテキスト）** を解析し、自動的に `JSON Schema` を生成します。
- クライアントが `tools/list` を要求した際、この生成された型情報と説明文がそのまま送信されるため、LLMは「引数にどんな型の変数を渡せば良いか」を正確に理解できます。

### 3. `@mcp.resource("URIパターン")` による文脈データの提供
- `@mcp.resource()` は、URI形式のパラメータを受け取るデータ提供インターフェースです。
- `"user://profile/{user_id}"` のように波括弧 `{user_id}` を使用することで、動的なパスパラメータを関数の引数としてそのまま受け取ることができます。

### 4. `mcp.run()` による Stdio トランスポートの駆動
- スクリプトが直接実行された際、`mcp.run()` はプロセスの `sys.stdin`（標準入力）からの JSON-RPC リクエストを待機し、`sys.stdout`（標準出力）にレスポンスを出力する無限ループに入ります。
- ログ出力を行う場合は、`sys.stdout` を汚染してJSON-RPC通信を壊さないよう、必ず `sys.stderr`（標準エラー出力）へ出力するのがMCP開発の鉄則です。

---

## 6. プロダクション導入・セキュリティ・パフォーマンス最適化

自作のMCPサーバーを実際の運用環境や社内ツールに組み込む際の、ベストプラクティスと防衛線について解説します。

### 1. デバッグ手法：FastMCP Inspector の活用
作成したMCPサーバーが正しくJSON-RPCを返しているかをブラウザ上でグラフィカルにテストするには、MCP公式の Inspector ツールを利用するのが最も効率的です。

```bash
# FastMCP Inspector ツールをインストールしてローカル開発UIを起動
npx @modelcontextprotocol/inspector python mcp_server_demo.py

実行するとローカルサーバーが起動し、ブラウザ上でツールの実行テスト、リソースの取得テスト、プロンプトの展開テストを対話的に行うことができます。

2. Claude Desktop への組み込み設定

作成したサーバーを Anthropic 公式の Claude Desktop アプリに接続するには、設定ファイル（claude_desktop_config.json）に以下のJSONを追加します。

{
  "mcpServers": {
    "enterprise_assistant": {
      "command": "python",
      "args": [
        "C:/path/to/your/mcp_server_demo.py"
      ]
    }
  }
}

3. セキュリティと権限分離の原則

入力バリデーションの強制: LLMから渡される Tool の引数は、意図しない不正な文字列や数値が含まれる可能性があるため、関数内で必ず型の範囲チェックやサニタイズ（height_m <= 0 の拒否など）を実施してください。
最小権限の原則（Least Privilege）: データベース書き込みや削除など、副作用の大きい操作を行う Tool には、承認フロー（Human-in-the-Loop）を設けるか、読み取り専用の権限で動作させるのが安全です。

7. まとめと次回の展望

今回は、業界の共通プロトコルとして急拡大する MCP (Model Context Protocol) の理論背景、JSON-RPC 2.0 の内部動作メカニズム、そして FastMCP を用いたTools / Resources / Promptsの完全な実装コードを解説しました。

MCPを採用することで、特定のLLMベンダーに依存しない、極めて拡張性の高い「プラグ＆プレイなツール・ナレッジ接続」が可能になることが実感できたかと思います。

しかし、実際のエンタープライズ開発においては、単一のPythonファイルをローカルで叩くだけでなく、「社内のPostgreSQLデータベース」や「REST API」、「Dockerコンテナ環境」と型安全に接続し、複数ユーザーから安全に利用できるようにする必要があります。

次回、第2回。
今回構築した基礎をもとに、**『Claude・Cursorと自社DBをつなぐ実践MCPサーバー構築』**に進みます。

Cursor や Windsurf、Claude Desktop から自社の本番相当データベースを安全に操作する、真の実務向けMCPサーバーの実践開発へ進みましょう。