【次世代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が知識として参照する情報を提供します。
- URI(例:
-
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サーバーの実践開発へ進みましょう。