ドキュメントMCP（Model Context Protocol）統合ガイド
MCP の概要
MCP（Model Context Protocol）は、AI アシスタントと外部ツールやデータソースを接続するためのオープン標準プロトコルです。本プロジェクトでは統一された MCP 統合ソリューションを採用し、すべての MCP 機能を 1 つのクライアントファイルに統合することで、軽量ながら全機能を備えた MCP サポートを実現しています。
️ アーキテクチャ設計
アーキテクチャ概要
┌─────────────────────────────────────────────────────────────┐
│ ASR-LLM-TTS 主系统 │
├─────────────────────────────────────────────────────────────┤
│ ┌─────────────────┐ │
│ │ Turn Manager │ ◄─── 用户输入 │
│ └─────────┬───────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ Function Call │────► │ MCPClient │ │
│ │ Registry │ │ (统一客户端) │ │
│ └─────────────────┘ └─────────┬───────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ Local Functions │ │ MCP Connections │ │
│ │ (本地工具) │ │ (远程MCP工具) │ │
│ └─────────────────┘ └─────────┬───────┘ │
└─────────────────────────────────────┼───────────────────────┘
│
┌─────────────────┼─────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ MCP Server 1 │ │ MCP Server 2 │ │ MCP Server N │
│ (vision-arm) │ │ (face-tracker) │ │ (amap-maps) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
コアコンポーネント
| コンポーネント | ファイルパス | 役割 |
|---|---|---|
| MCPClient | utils/mcp/client/mcp_client.py |
すべての機能を統合した統一 MCP クライアント |
| MCPConnection | utils/mcp/client/mcp_client.py |
個々の MCP サーバーとの接続管理 |
| MCP Configuration | utils/mcp/config/mcp_servers.json |
MCP サーバーの設定ファイル |
MCPClient の詳細
主な特長
- 単一ファイルへの統合：すべての MCP 機能を 1 つのファイルに統合
- バックグラウンド監視：自動的な接続管理とヘルスチェック
- ツールの統一：ローカルツールと MCP ツールを統一インターフェースで扱う
- 指数バックオフ：頻繁なリトライを避けるインテリジェントな再接続の仕組み
- 非同期優先：完全な非同期実装で、メインサービスをブロックしない
主要クラスの定義
@dataclass
class MCPTool:
"""MCP工具描述"""
name: str
parameters: Dict[str, Any]
description: str
server_name: str
active: bool = True
@dataclass
class MCPServerState:
"""MCP服务器状态"""
name: str
config: dict
status: str = "disconnected" # disconnected, connecting, connected, failed
last_health_check: Optional[datetime] = None
retry_count: int = 0
next_retry_time: Optional[datetime] = None
tools: List[MCPTool] = None
error_message: Optional[str] = None
def should_retry_now(self) -> bool:
"""检查是否应该现在重试连接"""
if self.status == "connected":
return False
if self.next_retry_time is None:
return True
return datetime.now() >= self.next_retry_time
接続管理
class MCPConnection:
"""单个MCP连接封装"""
def __init__(self, name: str, config: dict):
self.name = name
self.config = config
self.session: Optional[mcp.ClientSession] = None
self.exit_stack = AsyncExitStack()
self.tools: List[mcp.Tool] = []
async def health_check(self, timeout: float = 3.0) -> bool:
"""快速健康检查，验证服务器可达性"""
if "url" not in self.config:
return True # stdio连接无法网络检查
try:
parsed_url = urlparse(self.config["url"])
host = parsed_url.hostname
port = parsed_url.port or (443 if parsed_url.scheme == "https" else 80)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
result = sock.connect_ex((host, port))
sock.close()
return result == 0
except Exception:
return False
バックグラウンド監視の仕組み
async def _background_monitor(self) -> None:
"""后台监控任务"""
while not self.shutdown_event.is_set():
try:
# 尝试连接或重连服务器
for name, server in self.servers.items():
if server.should_retry_now():
await self._attempt_connection(name, server)
# 健康检查已连接的服务器
await self._health_check_all()
# 等待30秒
await asyncio.wait_for(self.shutdown_event.wait(), timeout=30.0)
except asyncio.TimeoutError:
continue
except Exception as e:
logger.error(f"后台监控异常: {e}")
await asyncio.sleep(10)
設定管理
MCP サーバーの設定
設定ファイル：utils/mcp/config/mcp_servers.json
{
"mcpServers": {
"vision-arm-control": {
"active": true,
"type": "streamable-http",
"url": "http://192.168.1.9:8002/mcp/",
"transport": "streamable_http",
"timeout": 10,
"health_check_timeout": 2,
"sse_read_timeout": 300,
"terminate_on_close": true,
"headers": {
"Accept": "application/json, text/event-stream",
"Content-Type": "application/json"
},
"description": "智能视觉机械臂控制系统MCP服务器"
},
"face-tracker": {
"active": true,
"type": "streamable-http",
"url": "http://192.168.1.9:8000/mcp/",
"transport": "streamable_http",
"timeout": 10,
"health_check_timeout": 2,
"sse_read_timeout": 300,
"terminate_on_close": true,
"headers": {
"Accept": "application/json, text/event-stream",
"Content-Type": "application/json"
},
"description": "智能人脸追踪系统MCP服务器"
},
"amap-maps": {
"active": false,
"type": "sse",
"url": "https://mcp.api-inference.modelscope.net/14ffca24064c4d/sse",
"description": "高德地图MCP服务器（第三方）"
}
}
}
対応する接続タイプ
1. Streamable HTTP 接続（推奨）
{
"server-name": {
"active": true,
"type": "streamable-http",
"url": "http://host:port/mcp/",
"transport": "streamable_http",
"timeout": 10,
"health_check_timeout": 2,
"sse_read_timeout": 300,
"terminate_on_close": true,
"headers": {
"Accept": "application/json, text/event-stream",
"Content-Type": "application/json"
}
}
}
2. SSE 接続
{
"server-name": {
"active": true,
"type": "sse",
"url": "https://mcp-server.example.com/sse",
"transport": "sse",
"timeout": 5,
"sse_read_timeout": 300,
"headers": {
"Authorization": "Bearer your-token",
"Accept": "text/event-stream"
}
}
}
3. stdio 接続（ローカルプロセス）
{
"server-name": {
"active": true,
"type": "stdio",
"command": "python",
"args": ["-m", "mcp_server_module"],
"cwd": "/path/to/working/directory",
"env": {
"API_KEY": "your-api-key"
},
"timeout": 30
}
}
ツール呼び出しの流れ
1. 初期化の流れ
1. MCPClient.initialize()
2. 加载配置文件 mcp_servers.json
3. 创建MCPServerState对象
4. 启动后台监控任务 _background_monitor()
5. 异步连接MCP服务器
6. 获取工具列表并合并schema
2. ツールの検出と登録
async def get_all_tools_schema(self) -> List[Dict[str, Any]]:
"""获取所有工具的schema（本地+MCP）"""
# 获取本地工具
local_schemas = []
if LOCAL_TOOLS_AVAILABLE:
try:
local_schemas = get_all_schemas()
except Exception as e:
logger.warning(f"无法获取本地工具: {e}")
# 获取MCP工具
mcp_schemas = []
for server in self.servers.values():
if server.status == "connected":
for tool in server.tools:
if tool.active:
schema = {
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters
}
}
mcp_schemas.append(schema)
logger.info(f"获取工具schema: 本地 {len(local_schemas)} 个, MCP {len(mcp_schemas)} 个")
return local_schemas + mcp_schemas
3. ツール呼び出しの実行
async def call_tool(self, tool_name: str, **arguments) -> Any:
"""调用工具（自动判断本地或MCP）"""
# 先检查是否为MCP工具
for server in self.servers.values():
if server.status == "connected":
for tool in server.tools:
if tool.name == tool_name and tool.active:
conn = self.connections.get(server.name)
if conn:
return await conn.call_tool(tool_name, arguments)
# 尝试本地工具
if LOCAL_TOOLS_AVAILABLE:
try:
return await FunctionRegistry.call(tool_name, **arguments)
except Exception as e:
logger.warning(f"本地工具调用失败: {e}")
raise ValueError(f"工具 '{tool_name}' 不存在或不可用")
接続状態の管理
接続ステートマシン
再接続戦略
[disconnected] → [connecting] → [connected]
↑ ↓ ↓
└── [failed] ←─────────────────┘
↓
(指数退避等待)
↓
[disconnected]
async def _attempt_connection(self, name: str, server: MCPServerState) -> None:
"""尝试连接服务器"""
try:
server.status = "connecting"
server.retry_count += 1
logger.info(f"尝试连接MCP服务器: {name} (第{server.retry_count}次)")
# 建立新连接
conn = MCPConnection(name, server.config)
await conn.connect()
# 连接成功
self.connections[name] = conn
server.status = "connected"
server.retry_count = 0
server.next_retry_time = None
logger.info(f"MCP服务器 {name} 连接成功，工具: {[t.name for t in server.tools]}")
except Exception as e:
# 连接失败，计算下次重试时间（指数退避）
server.status = "failed"
server.error_message = str(e)
delay = min(300, 10 * (2 ** min(server.retry_count - 1, 5)))
delay *= random.uniform(0.5, 1.5) # 添加抖动
server.next_retry_time = datetime.now() + timedelta(seconds=delay)
logger.warning(f"MCP服务器 {name} 连接失败: {e}, 下次重试: {server.next_retry_time.strftime('%H:%M:%S')}")
ステータス監視
サーバーステータスの照会
def get_server_status(self) -> Dict[str, Any]:
"""获取服务器状态摘要"""
connected_count = sum(1 for s in self.servers.values() if s.status == "connected")
total_tools = sum(len(s.tools) for s in self.servers.values() if s.status == "connected")
servers_info = []
for server in self.servers.values():
info = {
"name": server.name,
"status": server.status,
"tools_count": len(server.tools),
"last_health_check": server.last_health_check.isoformat() if server.last_health_check else None,
"error_message": server.error_message
}
if server.next_retry_time:
info["next_retry_time"] = server.next_retry_time.isoformat()
servers_info.append(info)
return {
"mcp_available": MCP_AVAILABLE,
"total_servers": len(self.servers),
"connected_servers": connected_count,
"total_tools": total_tools,
"servers": servers_info
}
グローバル API
便利な関数の使用
# 获取全局MCP客户端实例
async def get_mcp_client() -> MCPClient:
"""获取全局MCP客户端实例"""
global _mcp_client
if _mcp_client is None:
_mcp_client = MCPClient()
await _mcp_client.initialize()
return _mcp_client
# 获取增强的工具schema
async def get_enhanced_function_schemas() -> List[Dict[str, Any]]:
"""获取增强的函数schema列表"""
client = await get_mcp_client()
return client.get_all_tools_schema()
# 增强的工具调用
async def enhanced_function_call(tool_name: str, **kwargs) -> Any:
"""增强的函数调用"""
client = await get_mcp_client()
return await client.call_tool(tool_name, **kwargs)
# 清理全局客户端
async def cleanup_mcp_client() -> None:
"""清理全局MCP客户端"""
global _mcp_client
if _mcp_client:
await _mcp_client.cleanup()
_mcp_client = None
開発ガイド
新しい MCP サーバーの追加
-
設定ファイルを更新します
-
サービスを再起動します
{
"mcpServers": {
"my-custom-server": {
"active": true,
"type": "streamable-http",
"url": "http://localhost:9000/mcp/",
"transport": "streamable_http",
"timeout": 10,
"description": "我的自定义MCP服务器"
}
}
}
- MCP クライアントが新しいサーバーを自動検出して接続します
接続の確認
- ログを確認し、接続状態とツールの読み込み状況を確認します
MCP 接続のデバッグ
️
# 获取MCP状态
import asyncio
from utils.mcp.client.mcp_client import get_mcp_client
async def debug_mcp():
client = await get_mcp_client()
status = client.get_server_status()
print(f"MCP可用: {status['mcp_available']}")
print(f"服务器总数: {status['total_servers']}")
print(f"已连接: {status['connected_servers']}")
print(f"工具总数: {status['total_tools']}")
for server in status['servers']:
print(f"\n服务器: {server['name']}")
print(f" 状态: {server['status']}")
print(f" 工具数: {server['tools_count']}")
if server.get('error_message'):
print(f" 错误: {server['error_message']}")
# 运行调试
asyncio.run(debug_mcp())
セキュリティ上の考慮事項
1. 接続のセキュリティ
-
HTTPS/TLS による暗号化接続に対応
-
API キーおよびトークンによる認証
- ホワイトリストにより接続可能なサーバーを制限
2. ツール呼び出しの制限
-
パラメーターの検証と型チェック
-
実行タイムアウトの制御（デフォルト 60 秒）
- リソース使用量の監視
3. エラー処理
-
機密情報のフィルタリング
-
接続の異常がメインサービスに影響しない
- 詳細なエラーログの記録
⚡ パフォーマンスの最適化
1. 接続管理
-
コネクションプールの再利用により頻繁な接続を回避
-
ヘルスチェックの仕組みで接続の問題を早期に発見
- 指数バックオフによる再接続で無駄な試行を削減
2. 非同期処理
-
すべての操作を完全に非同期化
-
バックグラウンドタスクがメインプロセスをブロックしない
- 複数の MCP ツールを同時に呼び出し可能
3. リソースのクリーンアップ
-
切断された接続を自動的にクリーンアップ
-
メモリリソースを適時解放
- サービスを適切に終了する処理
この MCP 統合ソリューションは、機能の完全性を保ちながら実装の複雑さを大幅に低減し、ASR-LLM-TTS システムに強力で信頼性の高いリモートツール呼び出し機能を提供します。