MCPとは？
Model Context Protocol (MCP) は、AIアシスタント（Claudeなど）と外部システムとの対話を可能にする標準化されたプロトコルです。以下を定義します：
- AIがツールを発見し呼び出す方法
- ツールが結果を返す方法
- 通信の標準フォーマット
コア概念
AI Assistant <---> MCP Protocol <---> Your System
(Claude) (標準化通信) (データベース/API等)MCP Server アーキテクチャ原理
1. 全体アーキテクチャ
┌─────────────────────────────────────────┐
│ MCP Client (AI) │
└────────────┬────────────────────────────┘
│ stdio/HTTP
▼
┌─────────────────────────────────────────┐
│ MCP Server メインプロセス │
│ ┌─────────────────────────────────┐ │
│ │ Protocol Handler │ │
│ │ (MCPプロトコルリクエスト/レスポンス処理) │ │
│ └──────────┬──────────────────────┘ │
│ │ │
│ ┌──────────▼──────────────────────┐ │
│ │ Tools Registry │ │
│ │ (ツール登録と管理) │ │
│ └──────────┬──────────────────────┘ │
│ │ │
│ ┌──────────▼──────────────────────┐ │
│ │ Business Logic │ │
│ │ (ビジネスロジック実装) │ │
│ └──────────┬──────────────────────┘ │
│ │ │
│ ┌──────────▼──────────────────────┐ │
│ │ External System Client │ │
│ │ (外部システムコネクタ) │ │
│ └─────────────────────────────────┘ │
└─────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────┐
│ External System │
│ (Database/API/Service) │
└─────────────────────────────────────────┘2. 通信フロー
# 1. AI がリクエストを送信
Request: {
"method": "tools/call",
"params": {
"name": "query_database",
"arguments": {"query": "SELECT * FROM users"}
}
}
# 2. Server が処理
Server -> Tool Handler -> Execute Query -> Return Results
# 3. レスポンスを返す
Response: {
"result": {
"content": [{"type": "text", "text": "クエリ結果..."}]
}
}コアコンポーネント詳解
1. Server メインクラス（コアスケジューラ）
Denodo MCP Server を例として：
# src/server.py
class DenodoMCPServer:
"""MCP Server のコアクラス"""
def __init__(self):
# 各コンポーネントを初期化
self.config = None # 設定管理
self.denodo_client = None # 外部システムクライアント
self.tools = None # ツールコレクション
self.server = Server("denodo-mcp-server") # MCPプロトコルサーバー2. ツール定義（Tools）
# src/tools.py
class DenodoTools:
def get_tools(self) -> List[Tool]:
"""利用可能なツールを定義"""
return [
Tool(
name="query_denodo",
description="SQL クエリを実行",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
}
)
]3. リクエストハンドラー
async def handle_tool_call(self, name: str, arguments: dict):
"""ツール呼び出しリクエストを処理"""
if name == "query_denodo":
result = await self._query_denodo(arguments)
return [TextContent(
type="text",
text=json.dumps(result)
)]4. 外部システムクライアント
# src/denodo_client.py
class DenodoClient:
def execute_query(self, query: str):
connection = self._get_connection()
cursor = connection.cursor()
cursor.execute(query)
return cursor.fetchall()Denodo MCP Server を例として
```python
# config.py
@dataclass
class ServerConfig:
"""サーバー設定"""
# 外部システム設定
host: str
username: str
password: str
# サーバー設定
log_level: str = "INFO"
@classmethod
def from_env(cls):
"""環境変数から読み込み"""
return cls(
host=os.getenv('HOST'),
username=os.getenv('USERNAME'),
password=os.getenv('PASSWORD')
)
```
### Step 3: ツールロジックの実装
```python
# tools.py
class MyTools:
def get_tools(self):
"""Step 1: ツールメタデータを定義"""
return [
Tool(
name="my_tool",
description="ツールの説明",
inputSchema={...}
)
]
async def handle_tool_call(self, name, arguments):
"""Step 2: ルーティングロジックを実装"""
if name == "my_tool":
return await self._my_tool_impl(arguments)
async def _my_tool_impl(self, arguments):
"""Step 3: 具体的なビジネス実装"""
# あなたのビジネスロジック
result = do_something(arguments)
return format_result(result)
```
### Step 4: メインサーバーの作成
```python
# server.py
class MyMCPServer:
def initialize(self):
"""すべてのコンポーネントを初期化"""
self.config = Config.load()
self.client = MyClient(self.config)
self.tools = MyTools(self.client)
# MCPプロトコルハンドラーを登録
self._register_handlers()
def _register_handlers(self):
"""リクエストハンドラーを登録"""
@self.server.list_tools()
async def list_tools():
return self.tools.get_tools()
@self.server.call_tool()
async def call_tool(name, arguments):
return await self.tools.handle_tool_call(name, arguments)
```
### Step 5: 起動エントリポイントの実装
```python
# server.py
async def main():
server = MyMCPServer()
server.initialize()
# stdioモードを使用（ローカル開発）
await server.run_stdio()
if __name__ == "__main__":
asyncio.run(main())
```
---
## ベストプラクティス
### 1. エラー処理
```python
async def handle_tool_call(self, name, arguments):
try:
# ツールロジックを実行
result = await self._execute_tool(name, arguments)
return success_response(result)
except ValidationError as e:
# パラメータ検証エラー
return error_response(f"パラメータエラー: {e}")
except ConnectionError as e:
# 接続エラー
return error_response(f"接続失敗: {e}")
except Exception as e:
# 未知のエラー
logger.error(f"ツール実行失敗: {e}")
return error_response("内部エラー")
```
### 2. ログ記録
```python
logger.info(f"ツール実行開始: {tool_name}")
logger.debug(f"ツールパラメータ: {arguments}")
logger.error(f"実行失敗: {error}")
```
### 3. 設定管理
```python
# 複数の設定ソースをサポート
class Config:
@classmethod
def load(cls):
if os.getenv('USE_CLOUD_CONFIG'):
return cls.from_cloud()
else:
return cls.from_env()
```
### 4. 非同期処理
```python
# 非同期を使用してパフォーマンスを向上
async def batch_query(self, queries):
tasks = [self.execute_query(q) for q in queries]
results = await asyncio.gather(*tasks)
return results
```
### 5. セキュリティ考慮事項
```python
# パラメータ検証
def validate_query(query: str):
# SQLインジェクション防止
if "DROP" in query.upper():
raise ValueError("危険な操作")
# パラメータ化クエリを使用
cursor.execute("SELECT * FROM users WHERE id = ?", [user_id])
```
---
## MCP Server のテスト
### 1. ユニットテスト
```python
# tests/test_tools.py
def test_list_tables():
client = MockClient()
tools = MyTools(client)
result = await tools._list_tables({})
assert result["success"] == True
```
### 2. MCP Inspector の使用
```bash
# Inspector をインストール
npx @modelcontextprotocol/inspector
# サーバーをテスト
npx @modelcontextprotocol/inspector python -m src.server
```
### 3. 統合テスト
```python
# 完全なリクエストフローをシミュレート
async def test_end_to_end():
server = MyMCPServer()
server.initialize()
# ツール呼び出しをシミュレート
result = await server.tools.handle_tool_call(
"query_database",
{"query": "SELECT 1"}
)
assert "success" in result