Claude Code サブエージェント & MCP 実装実践ガイド¶
title: "Claude Code サブエージェント & MCP 実装実践ガイド" description: "Claude Codeのサブエージェント機能とMCP統合を実際のプロジェクトで活用する実装ガイド。コピペで動くコード例とトラブルシューティングを含む完全なチュートリアル。" tags: ["Claude Code", "MCP", "サブエージェント", "AI開発", "実装ガイド"] category: "🤖 AI開発・自動化"
この記事のポイント¶
専門エージェント活用
特定タスクに特化したサブエージェントで開発効率を向上
外部システム統合
MCPプロトコルでAPI・データベース・ツール連携を自動化
実装テンプレート
コピペで動く実用的なコード例とベストプラクティス
本格運用
トラブルシューティングから本番環境まで対応
📖 サブエージェント & MCP の概要¶
サブエージェントとは¶
サブエージェントは、独自のコンテキストウィンドウで動作する専門化されたAIアシスタントです。カスタムシステムプロンプト、特定のツールアクセス、独立した権限設定を持ち、メインエージェントから Task ツールを通じて呼び出されます。
ビルトインサブエージェントの種類¶
Claude Codeには6種類のビルトインサブエージェントが用意されています。
graph TB
A[メインエージェント] --> B[Task Tool]
B --> C[Explore]
B --> D[Plan]
B --> E[general-purpose]
B --> F[Bash]
B --> G[statusline-setup]
B --> H[Claude Code Guide]
C --> C1[高速ファイル探索・コード検索]
D --> D1[調査・計画立案]
E --> E1[複雑な多段階タスク]
F --> F1[ターミナルコマンド実行]
G --> G1[ステータスライン設定]
H --> H1[Claude Code使用方法の質問応答]| サブエージェント | モデル | アクセス権 | 用途 |
|---|---|---|---|
| Explore | Haiku（高速） | 読み取り専用（Read, Glob, Grep等） | ファイル探索・コード検索。thoroughness パラメータで quick / medium / very thorough を指定可能 |
| Plan | 継承（親モデル） | 読み取り専用ツール | プランモード用のリサーチエージェント。設計・実装計画の立案 |
| general-purpose | 継承（親モデル） | フルツールアクセス | 複雑な多段階タスクの実行。ファイル編集を含む全操作が可能 |
| Bash | 継承（親モデル） | ターミナル操作 | 別コンテキストでのコマンド実行 |
| statusline-setup | Sonnet | 限定的 | /statusline コマンドの設定用 |
| Claude Code Guide | Haiku（高速） | 読み取り専用 | Claude Codeの使い方に関する質問応答 |
カスタムサブエージェントの定義¶
.claude/agents/ ディレクトリにMarkdownファイルを作成することで、プロジェクト固有のカスタムサブエージェントを定義できます。
---
name: code-reviewer
description: コードのセキュリティとベストプラクティスをレビュー
tools:
- Read
- Grep
- Glob
model: sonnet
permissionMode: plan
maxTurns: 10
---
# Code Reviewer Agent
セキュリティ脆弱性、パフォーマンス問題、ベストプラクティス違反を検出してください。
レビュー結果は以下の形式で報告してください：
- 重要度（Critical / Warning / Info）
- 該当ファイルと行番号
- 問題の説明と修正案
カスタムサブエージェントのフロントマターフィールド¶
| フィールド | 説明 | 値の例 |
|---|---|---|
name | エージェントの識別名 | code-reviewer |
description | エージェントの説明 | コードレビュー専門エージェント |
tools | 使用可能なツールのリスト | [Read, Grep, Glob, Bash] |
disallowedTools | 使用禁止のツール | [Write, Edit] |
model | 使用するモデル | sonnet / opus / haiku / inherit |
permissionMode | 権限モード | plan / full |
maxTurns | 最大ターン数 | 10 |
skills | 利用可能なスキル | スキル名のリスト |
mcpServers | 利用可能なMCPサーバー | サーバー名のリスト |
hooks | エージェント固有のフック | フック定義 |
memory | 追加コンテキスト情報 | メモリ設定 |
サブエージェントの制約事項¶
重要な制約
- サブエージェントは他のサブエージェントを生成できません（ネスト不可）
- バックグラウンドサブエージェントは、事前承認されていない権限要求を自動的に拒否します
- MCPツールはバックグラウンドサブエージェントでは利用できません
- サブエージェントの無効化:
permissions.denyにTask(agent-name)を追加することで特定のサブエージェントを無効化できます
MCP (Model Context Protocol) の役割¶
MCPは、Claude Codeと外部ツール間の標準化された通信プロトコルです。
graph LR
A[Claude Code] --> B[MCP Protocol]
B --> C[Database MCP]
B --> D[API MCP]
B --> E[File System MCP]
C --> F[PostgreSQL/MySQL]
D --> G[REST/GraphQL API]
E --> H[ローカル/リモートファイル]トランスポートタイプ¶
MCPは3種類のトランスポートを提供しています。
| タイプ | 用途 | 備考 |
|---|---|---|
| HTTP | リモートサーバー接続（推奨） | リモートMCPサーバーへの接続に推奨される標準方式 |
| SSE (Server-Sent Events) | リモートサーバー接続 | 非推奨（deprecated）。HTTPへの移行を推奨 |
| stdio | ローカルプロセス通信 | ローカルで実行するMCPサーバー向け。標準入出力で通信 |
MCPのスコープ¶
MCPサーバーの設定は3つのスコープで管理できます。
| スコープ | 設定ファイル | 適用範囲 |
|---|---|---|
| local（デフォルト） | .claude/mcp.json（プロジェクトローカル設定） | 現在のプロジェクトのみ |
| project | .mcp.json（プロジェクトルート） | プロジェクト全体で共有 |
| user | ~/.claude.json（グローバル） | 全プロジェクトで利用可能 |
マネージドMCP（エンタープライズ）¶
企業環境では managed-mcp.json をシステムディレクトリに配置して、組織全体のMCPサーバーを管理できます。ポリシーで allowedMcpServers / deniedMcpServers を設定し、利用可能なサーバーを制御します。
出力制限¶
- MCPツールの出力が 10,000トークンを超えると警告が表示されます
- デフォルトの最大出力は 25,000トークンです（
MAX_MCP_OUTPUT_TOKENS環境変数で変更可能）
ツール検索（Tool Search）¶
MCPツール数がコンテキストの10%を超える場合、ツール検索機能が自動的に有効化されます。手動で有効化する場合は環境変数 ENABLE_TOOL_SEARCH を設定します。
🚀 初心者向け: 基本セットアップ¶
1. サブエージェント基本使用法¶
最も簡単なサブエージェント呼び出しから始めましょう。
# basic_subagent_example.py
"""
基本的なサブエージェント使用例
Claude Code内でTask toolを使用してサブエージェントを呼び出す
"""
# Claude Code内での使用例（メタコード）
task_result = claude_code.use_tool("Task", {
"description": "コード検索実行",
"prompt": "プロジェクト内のすべてのPythonファイルでlogging設定を検索し、統一されたロギング戦略を提案してください",
"subagent_type": "general-purpose"
})
print(f"タスク結果: {task_result}")
2. MCP 基本設定¶
まず、シンプルなMCPサーバーを作成します。設定は .mcp.json（プロジェクトスコープ）または ~/.claude.json（ユーザースコープ）に記述します。
.mcp.json では ${VAR} および ${VAR:-default} 構文で環境変数を展開できます。
{
"mcpServers": {
"simple-file-server": {
"command": "python",
"args": ["${PROJECT_ROOT:-/path/to}/simple_mcp_server.py"],
"env": {
"WORKSPACE_PATH": "${WORKSPACE_PATH:-.}"
}
}
}
}
# simple_mcp_server.py
"""
シンプルなファイル操作MCPサーバー
初心者向けの基本実装例
"""
import json
import sys
import os
from typing import Dict, Any, List
class SimpleMCPServer:
def __init__(self):
self.workspace_path = os.getenv('WORKSPACE_PATH', '.')
def list_files(self, directory: str = ".") -> List[str]:
"""指定ディレクトリのファイル一覧を取得"""
try:
full_path = os.path.join(self.workspace_path, directory)
return os.listdir(full_path)
except Exception as e:
return [f"エラー: {str(e)}"]
def read_file(self, file_path: str) -> str:
"""ファイル内容を読み取り"""
try:
full_path = os.path.join(self.workspace_path, file_path)
with open(full_path, 'r', encoding='utf-8') as f:
return f.read()
except Exception as e:
return f"エラー: {str(e)}"
def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""MCPリクエストを処理"""
method = request.get('method')
params = request.get('params', {})
if method == "tools/list":
return {
"tools": [
{
"name": "list_files",
"description": "ディレクトリのファイル一覧を取得",
"inputSchema": {
"type": "object",
"properties": {
"directory": {"type": "string", "default": "."}
}
}
},
{
"name": "read_file",
"description": "ファイル内容を読み取り",
"inputSchema": {
"type": "object",
"properties": {
"file_path": {"type": "string"}
},
"required": ["file_path"]
}
}
]
}
elif method == "tools/call":
tool_name = params.get('name')
arguments = params.get('arguments', {})
if tool_name == "list_files":
result = self.list_files(arguments.get('directory', '.'))
return {"content": [{"type": "text", "text": str(result)}]}
elif tool_name == "read_file":
result = self.read_file(arguments['file_path'])
return {"content": [{"type": "text", "text": result}]}
return {"error": "未対応のメソッド"}
def main():
"""MCPサーバーのメインループ"""
server = SimpleMCPServer()
for line in sys.stdin:
try:
request = json.loads(line.strip())
response = server.handle_request(request)
print(json.dumps(response))
sys.stdout.flush()
except Exception as e:
error_response = {"error": str(e)}
print(json.dumps(error_response))
sys.stdout.flush()
if __name__ == "__main__":
main()
🎯 中級者向け: 実用的な実装パターン¶
1. データベース統合MCPサーバー¶
実際のプロジェクトでよく使用されるデータベース操作MCPサーバーです。
# database_mcp_server.py
"""
データベース操作用MCPサーバー
SQLite/PostgreSQL対応の実用的な実装
"""
import json
import sys
import sqlite3
import os
from typing import Dict, Any, List, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DatabaseMCPServer:
def __init__(self):
self.db_path = os.getenv('DATABASE_URL', 'app.db')
self.connection = None
self.connect()
def connect(self):
"""データベースに接続"""
try:
self.connection = sqlite3.connect(self.db_path)
self.connection.row_factory = sqlite3.Row
logger.info(f"データベースに接続: {self.db_path}")
except Exception as e:
logger.error(f"データベース接続エラー: {e}")
def execute_query(self, query: str, params: Optional[List] = None) -> List[Dict]:
"""SQL クエリを実行して結果を返す"""
try:
cursor = self.connection.cursor()
if params:
cursor.execute(query, params)
else:
cursor.execute(query)
if query.strip().upper().startswith('SELECT'):
rows = cursor.fetchall()
return [dict(row) for row in rows]
else:
self.connection.commit()
return [{"affected_rows": cursor.rowcount}]
except Exception as e:
return [{"error": str(e)}]
def get_table_schema(self, table_name: str) -> List[Dict]:
"""テーブルスキーマを取得"""
query = f"PRAGMA table_info({table_name})"
return self.execute_query(query)
def list_tables(self) -> List[str]:
"""データベース内のテーブル一覧を取得"""
query = "SELECT name FROM sqlite_master WHERE type='table'"
result = self.execute_query(query)
return [row['name'] for row in result]
def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""MCPリクエストを処理"""
method = request.get('method')
params = request.get('params', {})
if method == "tools/list":
return {
"tools": [
{
"name": "execute_sql",
"description": "SQLクエリを実行",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"params": {"type": "array", "items": {"type": "string"}}
},
"required": ["query"]
}
},
{
"name": "get_schema",
"description": "テーブルスキーマを取得",
"inputSchema": {
"type": "object",
"properties": {
"table_name": {"type": "string"}
},
"required": ["table_name"]
}
},
{
"name": "list_tables",
"description": "データベースのテーブル一覧を取得",
"inputSchema": {"type": "object", "properties": {}}
}
]
}
elif method == "tools/call":
tool_name = params.get('name')
arguments = params.get('arguments', {})
if tool_name == "execute_sql":
result = self.execute_query(
arguments['query'],
arguments.get('params')
)
return {"content": [{"type": "text", "text": json.dumps(result, indent=2)}]}
elif tool_name == "get_schema":
result = self.get_table_schema(arguments['table_name'])
return {"content": [{"type": "text", "text": json.dumps(result, indent=2)}]}
elif tool_name == "list_tables":
result = self.list_tables()
return {"content": [{"type": "text", "text": json.dumps(result, indent=2)}]}
return {"error": "未対応のメソッド"}
def main():
server = DatabaseMCPServer()
for line in sys.stdin:
try:
request = json.loads(line.strip())
response = server.handle_request(request)
print(json.dumps(response))
sys.stdout.flush()
except Exception as e:
error_response = {"error": str(e)}
print(json.dumps(error_response))
sys.stdout.flush()
if __name__ == "__main__":
main()
2. API 統合MCPサーバー¶
外部APIとの連携を行うMCPサーバーの実装例です。
# api_mcp_server.py
"""
外部API統合用MCPサーバー
REST API呼び出しとレスポンス処理の実装
"""
import json
import sys
import requests
import os
from typing import Dict, Any, Optional
import logging
class APIMCPServer:
def __init__(self):
self.base_url = os.getenv('API_BASE_URL', 'https://api.example.com')
self.api_key = os.getenv('API_KEY')
self.session = requests.Session()
if self.api_key:
self.session.headers.update({'Authorization': f'Bearer {self.api_key}'})
def make_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Dict:
"""API リクエストを実行"""
try:
url = f"{self.base_url}/{endpoint.lstrip('/')}"
response = self.session.request(method, url, json=data)
response.raise_for_status()
return {
"status": response.status_code,
"data": response.json() if response.content else None
}
except requests.exceptions.RequestException as e:
return {"error": str(e)}
def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""MCPリクエストを処理"""
method = request.get('method')
params = request.get('params', {})
if method == "tools/list":
return {
"tools": [
{
"name": "api_call",
"description": "外部APIを呼び出し",
"inputSchema": {
"type": "object",
"properties": {
"method": {
"type": "string",
"enum": ["GET", "POST", "PUT", "DELETE"]
},
"endpoint": {"type": "string"},
"data": {"type": "object"}
},
"required": ["method", "endpoint"]
}
}
]
}
elif method == "tools/call":
tool_name = params.get('name')
arguments = params.get('arguments', {})
if tool_name == "api_call":
result = self.make_request(
arguments['method'],
arguments['endpoint'],
arguments.get('data')
)
return {"content": [{"type": "text", "text": json.dumps(result, indent=2)}]}
return {"error": "未対応のメソッド"}
def main():
server = APIMCPServer()
for line in sys.stdin:
try:
request = json.loads(line.strip())
response = server.handle_request(request)
print(json.dumps(response))
sys.stdout.flush()
except Exception as e:
error_response = {"error": str(e)}
print(json.dumps(error_response))
sys.stdout.flush()
if __name__ == "__main__":
main()
🏆 上級者向け: 高度な統合パターン¶
1. マルチエージェント協調システム¶
複数のサブエージェントが連携して複雑なタスクを処理するシステムです。
# multi_agent_coordinator.py
"""
マルチエージェント協調システム
複数のサブエージェントとMCPサーバーを組み合わせた高度な実装
"""
import asyncio
import json
from typing import Dict, List, Any
from dataclasses import dataclass
from enum import Enum
class AgentType(Enum):
GENERAL_PURPOSE = "general-purpose"
STATUSLINE_SETUP = "statusline-setup"
OUTPUT_STYLE_SETUP = "output-style-setup"
class TaskStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class Task:
id: str
description: str
agent_type: AgentType
dependencies: List[str]
status: TaskStatus = TaskStatus.PENDING
result: Any = None
error: str = None
class MultiAgentCoordinator:
def __init__(self):
self.tasks: Dict[str, Task] = {}
self.mcp_connections: Dict[str, Any] = {}
def add_task(self, task: Task):
"""タスクを追加"""
self.tasks[task.id] = task
def add_mcp_server(self, name: str, server_config: Dict):
"""MCPサーバー接続を追加"""
self.mcp_connections[name] = server_config
async def execute_task(self, task_id: str) -> bool:
"""単一タスクを実行"""
task = self.tasks.get(task_id)
if not task:
return False
# 依存関係チェック
for dep_id in task.dependencies:
dep_task = self.tasks.get(dep_id)
if not dep_task or dep_task.status != TaskStatus.COMPLETED:
return False
task.status = TaskStatus.IN_PROGRESS
try:
# サブエージェント呼び出し（疑似コード）
if task.agent_type == AgentType.GENERAL_PURPOSE:
result = await self._call_general_purpose_agent(task)
elif task.agent_type == AgentType.STATUSLINE_SETUP:
result = await self._call_statusline_agent(task)
elif task.agent_type == AgentType.OUTPUT_STYLE_SETUP:
result = await self._call_output_style_agent(task)
task.result = result
task.status = TaskStatus.COMPLETED
return True
except Exception as e:
task.error = str(e)
task.status = TaskStatus.FAILED
return False
async def _call_general_purpose_agent(self, task: Task) -> Any:
"""汎用エージェント呼び出し"""
# Claude Code Task tool呼び出しの疑似実装
agent_prompt = f"""
タスク: {task.description}
利用可能なMCPサーバー: {list(self.mcp_connections.keys())}
このタスクを実行し、結果を構造化された形で返してください。
"""
# 実際の実装ではClaude Code APIを使用
return {"status": "completed", "data": "サンプル結果"}
async def _call_statusline_agent(self, task: Task) -> Any:
"""ステータスライン設定エージェント呼び出し"""
return {"status": "completed", "config": "ステータスライン設定"}
async def _call_output_style_agent(self, task: Task) -> Any:
"""出力スタイル設定エージェント呼び出し"""
return {"status": "completed", "style": "出力スタイル設定"}
async def execute_workflow(self, task_ids: List[str]) -> Dict[str, Any]:
"""ワークフロー全体を実行"""
results = {}
while True:
# 実行可能なタスクを特定
executable_tasks = [
task_id for task_id in task_ids
if self.tasks[task_id].status == TaskStatus.PENDING
and all(
self.tasks[dep_id].status == TaskStatus.COMPLETED
for dep_id in self.tasks[task_id].dependencies
)
]
if not executable_tasks:
break
# 並列実行
tasks = [self.execute_task(task_id) for task_id in executable_tasks]
await asyncio.gather(*tasks)
# 結果集計
for task_id in task_ids:
task = self.tasks[task_id]
results[task_id] = {
"status": task.status.value,
"result": task.result,
"error": task.error
}
return results
# 使用例
async def main():
coordinator = MultiAgentCoordinator()
# MCPサーバー追加
coordinator.add_mcp_server("database", {
"command": "python",
"args": ["database_mcp_server.py"]
})
coordinator.add_mcp_server("api", {
"command": "python",
"args": ["api_mcp_server.py"]
})
# タスク追加
coordinator.add_task(Task(
id="task1",
description="データベースからユーザーデータを取得",
agent_type=AgentType.GENERAL_PURPOSE,
dependencies=[]
))
coordinator.add_task(Task(
id="task2",
description="APIでデータを外部システムに送信",
agent_type=AgentType.GENERAL_PURPOSE,
dependencies=["task1"]
))
coordinator.add_task(Task(
id="task3",
description="処理状況をステータスラインに表示",
agent_type=AgentType.STATUSLINE_SETUP,
dependencies=["task2"]
))
# ワークフロー実行
results = await coordinator.execute_workflow(["task1", "task2", "task3"])
print(json.dumps(results, indent=2))
if __name__ == "__main__":
asyncio.run(main())
2. リアルタイム監視MCPサーバー¶
システムの状態をリアルタイムで監視するMCPサーバーです。
# monitoring_mcp_server.py
"""
リアルタイム監視MCPサーバー
システムメトリクス、ログ監視、アラート機能を統合
"""
import json
import sys
import psutil
import time
import threading
from typing import Dict, Any, List
from queue import Queue
import logging
from datetime import datetime
class MonitoringMCPServer:
def __init__(self):
self.metrics_queue = Queue()
self.alerts = []
self.monitoring_active = False
self.monitoring_thread = None
def start_monitoring(self, interval: int = 5):
"""監視を開始"""
self.monitoring_active = True
self.monitoring_thread = threading.Thread(
target=self._monitor_loop,
args=(interval,)
)
self.monitoring_thread.daemon = True
self.monitoring_thread.start()
def stop_monitoring(self):
"""監視を停止"""
self.monitoring_active = False
if self.monitoring_thread:
self.monitoring_thread.join()
def _monitor_loop(self, interval: int):
"""監視ループ"""
while self.monitoring_active:
try:
metrics = self._collect_metrics()
self.metrics_queue.put(metrics)
self._check_alerts(metrics)
time.sleep(interval)
except Exception as e:
logging.error(f"監視エラー: {e}")
def _collect_metrics(self) -> Dict[str, Any]:
"""システムメトリクスを収集"""
return {
"timestamp": datetime.now().isoformat(),
"cpu_percent": psutil.cpu_percent(),
"memory_percent": psutil.virtual_memory().percent,
"disk_usage": psutil.disk_usage('/').percent,
"network_io": psutil.net_io_counters()._asdict(),
"process_count": len(psutil.pids())
}
def _check_alerts(self, metrics: Dict[str, Any]):
"""アラート条件をチェック"""
alerts = []
if metrics["cpu_percent"] > 80:
alerts.append({
"level": "warning",
"message": f"CPU使用率が高い: {metrics['cpu_percent']}%",
"timestamp": metrics["timestamp"]
})
if metrics["memory_percent"] > 85:
alerts.append({
"level": "critical",
"message": f"メモリ使用率が高い: {metrics['memory_percent']}%",
"timestamp": metrics["timestamp"]
})
if metrics["disk_usage"] > 90:
alerts.append({
"level": "critical",
"message": f"ディスク使用率が高い: {metrics['disk_usage']}%",
"timestamp": metrics["timestamp"]
})
self.alerts.extend(alerts)
# アラート数を制限
if len(self.alerts) > 100:
self.alerts = self.alerts[-50:]
def get_latest_metrics(self, count: int = 10) -> List[Dict[str, Any]]:
"""最新のメトリクスを取得"""
metrics = []
temp_queue = Queue()
# キューから取得
while not self.metrics_queue.empty() and len(metrics) < count:
metric = self.metrics_queue.get()
metrics.append(metric)
temp_queue.put(metric)
# キューに戻す
while not temp_queue.empty():
self.metrics_queue.put(temp_queue.get())
return list(reversed(metrics))
def get_alerts(self, level: str = None) -> List[Dict[str, Any]]:
"""アラートを取得"""
if level:
return [alert for alert in self.alerts if alert["level"] == level]
return self.alerts.copy()
def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
"""MCPリクエストを処理"""
method = request.get('method')
params = request.get('params', {})
if method == "tools/list":
return {
"tools": [
{
"name": "start_monitoring",
"description": "システム監視を開始",
"inputSchema": {
"type": "object",
"properties": {
"interval": {"type": "integer", "default": 5}
}
}
},
{
"name": "stop_monitoring",
"description": "システム監視を停止",
"inputSchema": {"type": "object", "properties": {}}
},
{
"name": "get_metrics",
"description": "最新のシステムメトリクスを取得",
"inputSchema": {
"type": "object",
"properties": {
"count": {"type": "integer", "default": 10}
}
}
},
{
"name": "get_alerts",
"description": "システムアラートを取得",
"inputSchema": {
"type": "object",
"properties": {
"level": {
"type": "string",
"enum": ["warning", "critical"]
}
}
}
}
]
}
elif method == "tools/call":
tool_name = params.get('name')
arguments = params.get('arguments', {})
if tool_name == "start_monitoring":
self.start_monitoring(arguments.get('interval', 5))
return {"content": [{"type": "text", "text": "監視を開始しました"}]}
elif tool_name == "stop_monitoring":
self.stop_monitoring()
return {"content": [{"type": "text", "text": "監視を停止しました"}]}
elif tool_name == "get_metrics":
metrics = self.get_latest_metrics(arguments.get('count', 10))
return {"content": [{"type": "text", "text": json.dumps(metrics, indent=2)}]}
elif tool_name == "get_alerts":
alerts = self.get_alerts(arguments.get('level'))
return {"content": [{"type": "text", "text": json.dumps(alerts, indent=2)}]}
return {"error": "未対応のメソッド"}
def main():
server = MonitoringMCPServer()
for line in sys.stdin:
try:
request = json.loads(line.strip())
response = server.handle_request(request)
print(json.dumps(response))
sys.stdout.flush()
except Exception as e:
error_response = {"error": str(e)}
print(json.dumps(error_response))
sys.stdout.flush()
if __name__ == "__main__":
main()
🛠️ 統合設定とデプロイ¶
Claude Code 設定ファイル¶
完全な統合環境のための設定例です。.mcp.json では環境変数展開（${VAR:-default} 構文）が利用できます。
{
"mcpServers": {
"database": {
"command": "python",
"args": ["/project/mcp/database_mcp_server.py"],
"env": {
"DATABASE_URL": "${DATABASE_URL:-sqlite:///app.db}",
"LOG_LEVEL": "${LOG_LEVEL:-INFO}"
}
},
"api-integration": {
"command": "python",
"args": ["/project/mcp/api_mcp_server.py"],
"env": {
"API_BASE_URL": "https://api.example.com",
"API_KEY": "${API_KEY}"
}
},
"monitoring": {
"command": "python",
"args": ["/project/mcp/monitoring_mcp_server.py"],
"env": {
"MONITORING_INTERVAL": "5"
}
}
}
}
フック設定（hooks）¶
フックはプロジェクトの .claude/settings.json またはユーザー設定の ~/.claude/settings.json に設定します。公式のイベント名を使用してください。
{
"hooks": {
"PreToolUse": [
{
"matcher": "Bash",
"hooks": [
{
"type": "command",
"command": "echo 'ツール実行前: Bash'"
}
]
}
],
"PostToolUse": [
{
"matcher": "Bash",
"hooks": [
{
"type": "command",
"command": "echo 'ツール実行後: Bash'"
}
]
}
],
"SubagentStart": [
{
"matcher": "general-purpose",
"hooks": [
{
"type": "command",
"command": "echo 'サブエージェント開始: $(date)'"
}
]
}
],
"SubagentStop": [
{
"matcher": "general-purpose",
"hooks": [
{
"type": "command",
"command": "echo 'サブエージェント終了: $(date)'"
}
]
}
]
}
}
利用可能なフックイベント
| イベント名 | タイミング |
|---|---|
PreToolUse | ツール実行前 |
PostToolUse | ツール実行後 |
SubagentStart | サブエージェント開始時 |
SubagentStop | サブエージェント終了時 |
Notification | 通知発生時 |
Stop | セッション終了時 |
Docker コンテナ構成¶
MCPサーバーを含む完全な開発環境をDockerで構築します。
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# システム依存関係をインストール
RUN apt-get update && apt-get install -y \
sqlite3 \
curl \
&& rm -rf /var/lib/apt/lists/*
# Python依存関係をインストール
COPY requirements.txt .
RUN pip install -r requirements.txt
# MCPサーバーファイルをコピー
COPY mcp/ ./mcp/
COPY config/ ./config/
# Claude Code設定
RUN mkdir -p /root/.claude
COPY config/claude-config.json /root/.claude/config.json
# 実行権限を設定
RUN chmod +x mcp/*.py
EXPOSE 8000
CMD ["python", "-m", "http.server", "8000"]
# docker-compose.yml
version: '3.8'
services:
claude-mcp-environment:
build: .
volumes:
- .:/app
- claude-cache:/root/.claude
environment:
- DATABASE_URL=sqlite:///data/app.db
- API_KEY=${API_KEY}
- PYTHONPATH=/app
ports:
- "8000:8000"
depends_on:
- database
database:
image: postgres:14
environment:
- POSTGRES_DB=claude_mcp
- POSTGRES_USER=claude
- POSTGRES_PASSWORD=password
volumes:
- postgres_data:/var/lib/postgresql/data
ports:
- "5432:5432"
volumes:
claude-cache:
postgres_data:
🚨 トラブルシューティング¶
よくある問題と解決法¶
1. MCPサーバー接続エラー¶
症状: MCPサーバーに接続できない
# デバッグ手順
# 1. MCPサーバーが単独で動作するかテスト
echo '{"method": "tools/list", "params": {}}' | python mcp/simple_mcp_server.py
# 2. 設定ファイルの構文チェック
python -m json.tool config/claude-config.json
# 3. 環境変数の確認
env | grep -E "(DATABASE_URL|API_KEY)"
解決法: - パスとPython環境の確認 - 権限設定の修正
- 環境変数の正しい設定
2. サブエージェント応答なし¶
症状: サブエージェントが応答しない、または期待した動作をしない
# デバッグ用コード
def debug_subagent_call():
"""サブエージェント呼び出しのデバッグ"""
import logging
logging.basicConfig(level=logging.DEBUG)
# Task tool呼び出しのログを有効化
task_params = {
"description": "テストタスク - ファイル一覧表示",
"prompt": "現在のディレクトリのファイル一覧を表示してください。デバッグ情報も含めて詳細に報告してください。",
"subagent_type": "general-purpose"
}
print(f"呼び出しパラメータ: {task_params}")
# ここでClaude Code Task toolを実際に呼び出し
解決法: - プロンプトの明確化 - サブエージェントタイプの確認 - タイムアウト設定の調整
3. パフォーマンス問題¶
症状: 処理が遅い、メモリ使用量が多い
# パフォーマンス最適化例
import asyncio
import aiofiles
from concurrent.futures import ThreadPoolExecutor
class OptimizedMCPServer:
def __init__(self):
self.executor = ThreadPoolExecutor(max_workers=4)
async def handle_async_request(self, request):
"""非同期リクエスト処理"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
self.executor,
self.handle_sync_request,
request
)
def handle_sync_request(self, request):
"""同期リクエスト処理（既存実装）"""
# 既存のhandle_requestロジック
pass
解決法: - 非同期処理の導入 - キャッシュの実装 - バッチ処理の最適化
📈 ベストプラクティス¶
1. エラーハンドリング¶
# 堅牢なエラーハンドリング例
import traceback
from functools import wraps
def mcp_error_handler(func):
"""MCPサーバー用エラーハンドラー"""
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
error_info = {
"error": str(e),
"type": type(e).__name__,
"traceback": traceback.format_exc()
}
return {"content": [{"type": "text", "text": json.dumps(error_info)}]}
return wrapper
class RobustMCPServer:
@mcp_error_handler
def handle_request(self, request):
# リクエスト処理ロジック
pass
2. ログ戦略¶
# 構造化ログの実装
import logging
import json
from datetime import datetime
class MCPLogger:
def __init__(self, name):
self.logger = logging.getLogger(name)
handler = logging.StreamHandler()
handler.setFormatter(self.JSONFormatter())
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
class JSONFormatter(logging.Formatter):
def format(self, record):
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"level": record.levelname,
"message": record.getMessage(),
"module": record.module,
"function": record.funcName
}
return json.dumps(log_entry)
def log_request(self, request, response):
"""リクエスト/レスポンスをログ記録"""
self.logger.info("MCP Request", extra={
"request": request,
"response_status": "success" if "error" not in response else "error"
})
3. テスト戦略¶
# MCPサーバーのテスト例
import unittest
import json
from io import StringIO
import sys
class TestMCPServer(unittest.TestCase):
def setUp(self):
self.server = SimpleMCPServer()
def test_tools_list(self):
"""tools/listメソッドのテスト"""
request = {"method": "tools/list", "params": {}}
response = self.server.handle_request(request)
self.assertIn("tools", response)
self.assertIsInstance(response["tools"], list)
def test_file_operations(self):
"""ファイル操作のテスト"""
# テストファイル作成
test_file = "test_file.txt"
with open(test_file, 'w') as f:
f.write("テストコンテンツ")
# ファイル読み取りテスト
request = {
"method": "tools/call",
"params": {
"name": "read_file",
"arguments": {"file_path": test_file}
}
}
response = self.server.handle_request(request)
self.assertIn("content", response)
self.assertIn("テストコンテンツ", response["content"][0]["text"])
# クリーンアップ
os.remove(test_file)
if __name__ == "__main__":
unittest.main()
🔗 関連記事¶
実装をさらに深めるための関連記事：
- Claude Code MCP 統合戦略 - 拡張性を極める
- Claude Code 応用編完全ガイド
- Claude Code ベストプラクティス集
- Claude Code GitHub Actions統合
📝 まとめ¶
Claude Codeのサブエージェント機能とMCP統合により、従来の開発フローを大幅に効率化できます。
実装の要点¶
- 段階的導入: 簡単なMCPサーバーから始めて、段階的に機能を拡張
- エラーハンドリング: 堅牢なエラー処理で本番環境での安定稼働を確保
- パフォーマンス最適化: 非同期処理とキャッシュで大規模プロジェクトに対応
- テスト自動化: 継続的な品質保証のためのテスト戦略
次のステップ¶
- 自分のプロジェクトでのMCPサーバー実装
- 複数サブエージェントによる協調システムの構築
- 本番環境でのモニタリングとアラート設定
この実装ガイドを参考に、AI支援開発環境の可能性を最大限に活用してください。