更新: 2026年03月 · 14 分で読める · 6,809 文字
MCPサーバーを自作してツール連携を実現する実践ガイド
Model Context Protocol（MCP）サーバーを自作することで、AIアシスタントに業務特化のツール機能を直接統合できます。この記事では、実装から本番デプロイまでの手順を、すぐに使えるコード例とともに解説します。
MCPサーバーとは何か
MCPサーバーはAnthropicが開発したプロトコルで、Claude等のLLMと外部ツール・システムを安全に接続する標準的な仕組みです。従来のAPI直叩きとは異なり、LLMが構造化された方法でツールを呼び出せるため、誤った操作やセキュリティリスクを減らせます。
自作MCPサーバーの利点は以下の通りです：
- 社内データベースやレガシーシステムへの直接アクセスをAIに許可
- 独自ビジネスロジックをAIのコンテキストに組み込み
- 複数のツールを一つのプロトコルで統一管理
- 監査ログやアクセス制御を集中管理可能
MCPサーバーの基本構造と実装方法
最小限の実装例
Pythonで最小限のMCPサーバーを実装します。このサーバーは「顧客情報を検索する」というツールを提供します。
#!/usr/bin/env python3
import json
import sys
from typing import Any
# MCPサーバーの最小実装
class SimpleMCPServer:
def __init__(self):
self.tools = {
"search_customer": {
"description": "顧客情報を顧客IDで検索",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "検索対象の顧客ID"
}
},
"required": ["customer_id"]
}
}
}
def handle_request(self, request: dict) -> dict:
"""MCPリクエストを処理"""
if request["type"] == "tools/list":
return self.list_tools()
elif request["type"] == "tool/call":
return self.execute_tool(request["name"], request["params"])
else:
return {"error": "Unknown request type"}
def list_tools(self) -> dict:
"""利用可能なツール一覧を返す"""
return {
"type": "tools",
"tools": [
{
"name": name,
"description": config["description"],
"inputSchema": config["inputSchema"]
}
for name, config in self.tools.items()
]
}
def execute_tool(self, tool_name: str, params: dict) -> dict:
"""ツールを実行"""
if tool_name == "search_customer":
customer_id = params.get("customer_id")
# ダミーデータベース照会
customers = {
"C001": {"name": "田中太郎", "email": "tanaka@example.com", "plan": "premium"},
"C002": {"name": "鈴木花子", "email": "suzuki@example.com", "plan": "basic"}
}
if customer_id in customers:
return {
"type": "tool_result",
"content": customers[customer_id],
"isError": False
}
else:
return {
"type": "tool_result",
"content": f"顧客ID {customer_id} は見つかりません",
"isError": True
}
return {"error": f"Unknown tool: {tool_name}"}
# メイン処理
if __name__ == "__main__":
server = SimpleMCPServer()
# stdoutでJSONリクエストを受け取りJSONレスポンスを返す
for line in sys.stdin:
try:
request = json.loads(line)
response = server.handle_request(request)
print(json.dumps(response, ensure_ascii=False))
except json.JSONDecodeError:
print(json.dumps({"error": "Invalid JSON"}))
except Exception as e:
print(json.dumps({"error": str(e)}))実装の重要なポイント
上記のコード例で押さえるべき点：
- tools/list：LLMが呼び出し可能なツール一覧をスキーマ付きで返す必須エンドポイント
- tool/call：実際のツール実行リクエストを処理。パラメータはスキーマに従って検証される
- inputSchema：JSON Schema形式で入力パラメータを定義。LLMが正しいパラメータを生成するための重要な指標
- JSON Lines形式：stdoutを通じてリクエスト・レスポンスをやり取り。改行区切りでエラーハンドリングが容易
実践例：複数ツールの統合と認証
APIキー認証を含む実装
本番環境では認証と複数ツールの管理が必須です。以下は社内ダッシュボード連携を想定した実装です。
#!/usr/bin/env python3
import json
import sys
import os
import hashlib
from datetime import datetime
from typing import Optional, Dict, Any
class ProductionMCPServer:
def __init__(self):
self.api_key = os.getenv("MCP_API_KEY", "")
self.audit_log = []
self.tools = {
"get_sales_metrics": {
"description": "月間売上実績を集計期間で取得",
"inputSchema": {
"type": "object",
"properties": {
"year": {"type": "integer", "description": "年"},
"month": {"type": "integer", "description": "月"}
},
"required": ["year", "month"]
}
},
"create_alert": {
"description": "ダッシュボードにアラートを作成",
"inputSchema": {
"type": "object",
"properties": {
"severity": {
"type": "string",
"enum": ["low", "medium", "high"],
"description": "アラートの重要度"
},
"message": {
"type": "string",
"description": "アラートメッセージ"
}
},
"required": ["severity", "message"]
}
},
"query_customer_history": {
"description": "顧客の購買履歴を検索",
"inputSchema": {
"type": "object",
"properties": {
"customer_id": {"type": "string"},
"limit": {
"type": "integer",
"default": 10,
"description": "取得件数"
}
},
"required": ["customer_id"]
}
}
}
def verify_request(self, request: dict) -> tuple[bool, Optional[str]]:
"""リクエストの認証と検証"""
auth_header = request.get("auth", "")
if not auth_header:
return False, "認証ヘッダーが必要です"
# API キーの検証（簡易版：実装では暗号化を使用）
if auth_header != self.api_key:
return False, "無効なAPIキー"
return True, None
def log_audit(self, tool_name: str, params: dict, status: str):
"""監査ログに記録"""
self.audit_log.append({
"timestamp": datetime.now().isoformat(),
"tool": tool_name,
"params": params,
"status": status
})
def handle_request(self, request: dict) -> dict:
"""メインリクエストハンドラ"""
# 認証確認
is_valid, error = self.verify_request(request)
if not is_valid:
return {"error": error, "type": "auth_error"}
request_type = request.get("type")
if request_type == "tools/list":
return self.list_tools()
elif request_type == "tool/call":
tool_name = request.get("name")
params = request.get("params", {})
# 入力パラメータの基本検証
if tool_name not in self.tools:
return {"error": f"ツール '{tool_name}' は存在しません", "isError": True}
result = self.execute_tool(tool_name, params)
self.log_audit(tool_name, params, "success" if not result.get("isError") else "error")
return result
return {"error": "Unknown request type"}
def list_tools(self) -> dict:
return {
"type": "tools",
"tools": [
{
"name": name,
"description": config["description"],
"inputSchema": config["inputSchema"]
}
for name, config in self.tools.items()
]
}
def execute_tool(self, tool_name: str, params: dict) -> dict:
"""ツール実行ロジック"""
try:
if tool_name == "get_sales_metrics":
year = params.get("year")
month = params.get("month")
# ダミーデータ
metrics = {
"revenue": 5200000,
"orders": 342,
"average_order_value": 15204,
"growth_rate": 12.5
}
return {
"type": "tool_result",
"content": {
"period": f"{year}年{month}月",
"metrics": metrics
},
"isError": False
}
elif tool_name == "create_alert":
severity = params.get("severity")
message = params.get("message")
# アラート作成（実装では実際のダッシュボードAPI呼び出し）
return {
"type": "tool_result",
"content": {
"alert_id": "ALT_" + hashlib.md5(message.encode()).hexdigest()[:8],
"severity": severity,
"created_at": datetime.now().isoformat(),
"status": "created"
},
"isError": False
}
elif tool_name == "query_customer_history":
customer_id = params.get("customer_id")
limit = params.get("limit", 10)
# ダミー履歴データ
history = [
{"order_id": "ORD001", "date": "2025-01-15", "amount": 45000},
{"order_id": "ORD002", "date": "2025-01-10", "amount": 12300}
]
return {
"type": "tool_result",
"content": {
"customer_id": customer_id,
"total_records": len(history),
"history": history[:limit]
},
"isError": False
}
except KeyError as e:
return {
"type": "tool_result",
"content": f"必須パラメータが不足しています: {str(e)}",
"isError": True
}
except Exception as e:
return {
"type": "tool_result",
"content": f"ツール実行エラー: {str(e)}",
"isError": True
}
if __name__ == "__main__":
server = ProductionMCPServer()
for line in sys.stdin:
try:
request = json.loads(line)
response = server.handle_request(request)
print(json.dumps(response, ensure_ascii=False))
sys.stdout.flush()
except json.JSONDecodeError:
print(json.dumps({"error": "Invalid JSON"}))
except Exception as e:
print(json.dumps({"error": str(e), "type": "server_error"}))MCPサーバーの設定と起動
Claude Desktop での設定
作成したMCPサーバーをClaude Desktopで利用するには、設定ファイルを編集します。
{
"mcpServers": {
"sales-dashboard": {
"command": "python",
"args": ["/path/to/mcp_server.py"],
"env": {
"MCP_API_KEY": "your-secure-api-key-here"
},
"disabled": false,
"alwaysAllow": []
}
}
}設定ファイルの配置場所：
- macOS:
~/Library/Application Support/Claude/claude_desktop_config.json - Windows:
%APPDATA%\Claude\claude_desktop_config.json - Linux:
~/.config/Claude/claude_desktop_config.json
Docker コンテナでの運用
本番環境ではDockerコンテナで実行することをお勧めします。
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY mcp_server.py .
ENV MCP_API_KEY=changeme
CMD ["python", "mcp_server.py"]# requirements.txt
python-dotenv==1.0.0
おすすめAIリソース
-
Anthropic Claude API Docs
Official Claude API reference. Essential for implementation.
-
OpenAI Platform
Official GPT series API documentation with pricing details.
-
Hugging Face
Open-source model hub with many free models to try.
おすすめAIリソース
- Anthropic Claude API Docs Official Claude API reference. Essential for implementation.
- OpenAI Platform Official GPT series API documentation with pricing details.
- Hugging Face Open-source model hub with many free models to try.