Claude CodeとMCP（Model Context Protocol）サーバーの連携は、AIアシスタントの機能を大幅に拡張する強力な手段だと思います。本ガイドでは、MCPサーバーの基礎知識から実装、トラブルシューティングまで、段階的に解説しているので、自社システムやツールとClaudeを統合したい開発者の方には参考になるはずです。
MCPサーバーとは｜基本概念の理解
MCPサーバーは、Model Context Protocolに準拠した外部ツールやリソースへのアクセスポイントです。Claude Codeを使用する際、このプロトコルを通じてデータベース、API、ファイルシステムなど様々なリソースにアクセスできるわけですね。
従来のAIアシスタントは学習データの範囲内でしか応答できませんでしたが、MCPサーバーを活用することで、以下のようなことが実現できます：
- リアルタイムデータへのアクセス
- カスタムビジネスロジックの実行
- セキュアな認証情報の管理
- 複数のシステムへのシームレスな連携
MCPサーバー構築の前準備
MCPサーバーを構築する前に、環境を整えておくことが重要です。実際のプロジェクトに進む前に、きちんと確認しておきましょう。
必要な環境とツール
- Python 3.8以上（またはNode.js 14以上）
- pip または npm パッケージマネージャー
- Claude APIキー
- MCPサーバーライブラリ（mcp-server-python など）
- テスト用のHTTPクライアント（curlやPostman）
MCPプロトコルの理解
MCPサーバーとClaudeの通信は、JSON-RPCベースのプロトコルに従っています。リクエスト・レスポンス形式で情報がやり取りされ、タイムアウトやエラーハンドリングが組み込まれているという構造ですね。調べてみたら、この設計が複数システム連携を想定した作りになっていることが分かりました。
シンプルなMCPサーバーの実装
最初の一歩として、PythonでシンプルなMCPサーバーを実装してみます。ここから始めると、MCPサーバーの基本的な流れが掴みやすいと思います。
基本的なサーバー実装
from mcp_server_python import Server, Tool
import json
from datetime import datetime
# MCPサーバーインスタンスの作成
server = Server("my-mcp-server")
# ツール定義：現在時刻を取得
@server.tool(
name="get_current_time",
description="現在の日時を取得します",
input_schema={
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "タイムゾーン (例: Asia/Tokyo)"
}
},
"required": []
}
)
def get_current_time(timezone: str = "Asia/Tokyo") -> str:
"""指定されたタイムゾーンの現在時刻を返す"""
try:
import pytz
tz = pytz.timezone(timezone)
current_time = datetime.now(tz)
return current_time.isoformat()
except Exception as e:
return f"エラー: {str(e)}"
# ツール定義：テキスト処理
@server.tool(
name="process_text",
description="テキストを大文字に変換します",
input_schema={
"type": "object",
"properties": {
"text": {
"type": "string",
"description": "処理対象のテキスト"
}
},
"required": ["text"]
}
)
def process_text(text: str) -> str:
"""テキストを大文字に変換"""
return text.upper()
if __name__ == "__main__":
server.run()
サーバーの起動方法
pip install mcp-server-python pytz
python mcp_server.py
サーバーが起動すると、デフォルトでhttp://localhost:3000でリッスンを開始します。ここまで来たら、サーバーが正常に動作しているかを確認することが次のステップになります。
Claude CodeとMCPサーバーの連携
Claude CodeでMCPサーバーを活用するには、サーバーのエンドポイントを適切に設定する必要があります。ここが実際のプロジェクトでよくハマるポイントだと思います。
設定ファイルの準備
{
"mcpServers": {
"my-mcp-server": {
"url": "http://localhost:3000",
"authentication": {
"type": "bearer",
"token": "your-api-key-here"
},
"timeout": 30000,
"retryAttempts": 3
}
}
}
Claude APIでのMCP実行
import anthropic
client = anthropic.Anthropic(api_key="your-claude-api-key")
# MCPサーバーを使用するメッセージ送信
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
tools=[
{
"type": "mcp_tool",
"name": "get_current_time",
"mcp_server": "my-mcp-server"
},
{
"type": "mcp_tool",
"name": "process_text",
"mcp_server": "my-mcp-server"
}
],
messages=[
{
"role": "user",
"content": "現在の東京時刻を教えてください。それから、『Hello World』を大文字で処理してください。"
}
]
)
print(response.content)
実践的なMCPサーバー実装例
データベースアクセスやAPI連携など、ビジネスロジックを含む実装例を紹介します。こういった複雑な連携こそが、MCPサーバーの本来の価値だと思います。
データベース連携サーバー
from mcp_server_python import Server
import sqlite3
import json
server = Server("db-mcp-server")
DB_PATH = "./app.db"
@server.tool(
name="query_database",
description="SQLクエリを実行してデータを取得",
input_schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "実行するSELECTクエリ"
}
},
"required": ["query"]
}
)
def query_database(query: str) -> dict:
"""データベースをクエリ実行"""
try:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute(query)
results = [dict(row) for row in cursor.fetchall()]
conn.close()
return {
"success": True,
"data": results,
"count": len(results)
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
@server.tool(
name="insert_record",
description="新しいレコードをデータベースに挿入",
input_schema={
"type": "object",
"properties": {
"table": {
"type": "string",
"description": "テーブル名"
},
"data": {
"type": "object",
"description": "挿入するデータ（キー・バリュー形式）"
}
},
"required": ["table", "data"]
}
)
def insert_record(table: str, data: dict) -> dict:
"""新規レコード挿入"""
try:
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
columns = ", ".join(data.keys())
placeholders = ", ".join(["?" for _ in data])
query = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
cursor.execute(query, list(data.values()))
conn.commit()
conn.close()
return {
"success": True,
"message": f"レコードが正常に挿入されました"
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
if __name__ == "__main__":
server.run()
セキュリティベストプラクティス
MCPサーバーは外部システムへのアクセスポイントになるため、セキュリティ対策が欠かせません。これを後から追加しようとすると大変なので、最初からきちんと実装することをお勧めします。
実装すべきセキュリティ対策
- 認証・認可：APIキーやOAuth 2.0を使用した認証の実装
- 入力値検証：SQLインジェクションやコマンドインジェクション対策
- レート制限：過度なリクエストからの保護
- ログ記録：すべてのリクエスト・レスポンスを監査ログに記録
- TLS/SSL暗号化：通信経路の暗号化
- タイムアウト設定：長時間実行の防止
セキュアな実装例
from mcp_server_python import Server
from functools import wraps
import hmac
import hashlib
import logging
server = Server("secure-mcp-server")
API_KEY = "your-secure-api-key"
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def require_auth(func):
"""認証デコレーター"""
@wraps(func)
def wrapper(*args, api_key: str = None, **kwargs):
if not api_key or not hmac.compare_digest(api_key, API_KEY):
logger.warning("認証失敗")
raise PermissionError("無効なAPIキー")
logger.info(f"ツール実行: {func.__name__}")
return func(*args, **kwargs)
return wrapper
@server.tool(
name="secure_operation",
description="認証が必要な操作",
input_schema={
"type": "object",
"properties": {
"data": {"type": "string"},
"api_key": {"type": "string"}
},
"required": ["data", "api_key"]
}
)
@require_auth
def secure_operation(data: str, api_key: str = None) -> dict:
"""セキュアな操作"""
return {"status": "success", "data": data}
if __name__ == "__main__":
server.run()
トラブルシューティング
MCPサーバー連携時に発生しやすい問題と解決方法を紹介します。本番環境で動かしてみると、開発環境では見えなかった問題が浮かび上がることもあるため、ここは重要です。
よくあるエラーと対処法
- 接続タイムアウト：サーバーが起動しているか確認、ファイアウォール設定を見直す
- 認証エラー：APIキーが正しく設定されているか、有効期限を確認
- メモリリーク：接続を適切にクローズしているか、ガベージコレクション設定を確認
- レスポンス遅延：データベースクエリの最適化、キャッシング導入
- JSON解析エラー：レスポンスフォーマットが正しいか検証
デバッグ方法
ローカルで開発時は、詳細なログを有効にして動作を追跡することが重要です。公式ドキュメントを読んでみたら、ログレベルの設定がかなり細かく制御できることに気づきました：
import logging
# ログレベルを設定
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('mcp_server.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
logger.debug("デバッグ情報")
logger.info("通常の情報")
logger.warning("警告メッセージ")
logger.error("エラーメッセージ")
まとめ
Claude CodeとMCPサーバーの連携は、AIアシスタントの可能性を大きく広げます。本ガイドで解説した要点をまとめると以下のようになります：
- MCPプロトコルの理解：JSON-RPCベースの通信方式を把握することが基礎になる
- 段階的な実装：シンプルなツールから始めて、複雑なロジックへと進める
- セキュリティ優先：認証、入力値検証、ログ記録を必ず実装する
- 適切なエラーハンドリング：例外処理とタイムアウト設定でロバストなシステムを構築
- 継続的な監視：ログとメトリクスで本番環境の問題を素早く検出
- ドキュメント整備：ツール仕様を明確に記述することで保守性が向上する
MCPサーバーを活用することで、Claude Codeは単なるテキスト生成ツールから、実際のビジネス価値を生み出すAIシステムへと進化します。本ガイドを参考に、ぜひ実装を進めてみてください。
よくある質問（FAQ）
MCPサーバーとAPIの違いは何ですか？
MCPサーバーはClaudeとの連携を前提に設計されたプロトコルで、JSON-RPCベースの構造になっています。一方、通常のRESTful APIは汎用的な設計のため、MCPサーバーはAIとの対話的な通信に最適化されているという点が大きく異なります。簡単に言えば、MCPはClaudeのためのAPI仕様だと考えると分かりやすいと思います。
複数のMCPサーバーを同時に使用できますか？
はい、複数のMCPサーバーを同時に使用することができます。設定ファイルのmcpServersセクションに複数のサーバーを記述することで、Claudeが複数のツールセットにアクセスできるようになります。ただし、サーバーの数が増えるとレスポンス時間に影響する可能性があるため、タイムアウト設定に注意が必要です。
MCPサーバーのスケーリングはどのように行いますか？
スケーリング方法は、負荷の種類によって異なります。データベースアクセスが多い場合はコネクションプーリングを導入し、計算負荷が高い場合はマルチプロセッシングやワーカープールを使用します。本番環境では、複数のMCPサーバーインスタンスをロードバランサーの背後に配置する構成が一般的だと思います。
ローカル環境からのテストと本番環境での動作に違いがありますか？
大きな違いが2つあります。1つはネットワーク遅延で、ローカルではほぼゼロですが、本番環境ではネットワークレイテンシーが発生します。もう1つはセキュリティで、ローカルではHTTPでもテストできますが、本番環境ではTLS/SSL暗号化が必須になります。本番環境にデプロイする前に、必ずステージング環境での十分なテストを行うことをお勧めします。
MCPサーバーのログを効果的に活用するコツはありますか？
ログの活用では、レベル分けが重要です。DEBUGレベルは開発時のみ、本番環境ではINFO以上に設定するのが一般的です。また、すべてのエラーに関連する情報（リクエストID、ユーザー情報、タイムスタンプなど）をログに含めることで、後からトラブル原因の特定が容易になります。さらに、ログファイルは定期的にローテーションして、ディスク容量の圧迫を防ぐことも忘れずに。
