tool_router.py•7.36 kB
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
MCP 工具集成路由器
处理工具相关的 MCP 请求
"""
import os
import json
import logging
import asyncio
from typing import Dict, Any, Optional, List
from .base_router import MCPBaseRouter
logger = logging.getLogger("tool_router")
class ToolRouter(MCPBaseRouter):
"""工具集成路由器"""
def __init__(self):
"""初始化路由器"""
super().__init__()
self.tools = {}
self._register_methods()
self._register_tools()
def _register_methods(self):
"""注册方法处理器"""
self.register_method("tools/list", self.handle_list_tools)
self.register_method("tools/execute", self.handle_execute_tool)
self.register_method("tools/cancel", self.handle_cancel_tool)
def _register_tools(self):
"""注册工具"""
# 注册示例工具
self.tools["search"] = {
"id": "search",
"name": "搜索工具",
"description": "在代码库中搜索内容",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索查询"
},
"scope": {
"type": "string",
"description": "搜索范围",
"enum": ["all", "code", "docs"]
}
},
"required": ["query"]
},
"handler": self._search_tool_handler
}
self.tools["fileSystem"] = {
"id": "fileSystem",
"name": "文件系统工具",
"description": "访问和操作文件系统",
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"description": "操作类型",
"enum": ["read", "write", "list", "delete"]
},
"path": {
"type": "string",
"description": "文件或目录路径"
},
"content": {
"type": "string",
"description": "写入的内容（仅用于写入操作）"
}
},
"required": ["action", "path"]
},
"handler": self._file_system_tool_handler
}
def handle_list_tools(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""处理工具列表请求"""
logger.info("处理工具列表请求")
# 构建工具列表（不包含处理器）
tools_list = []
for tool_id, tool in self.tools.items():
tools_list.append({
"id": tool["id"],
"name": tool["name"],
"description": tool["description"],
"parameters": tool["parameters"]
})
return {
"tools": tools_list
}
async def handle_execute_tool(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""处理执行工具请求（异步）"""
if not params or "id" not in params:
raise ValueError("缺少工具 ID 参数")
tool_id = params["id"]
tool_params = params.get("params", {})
logger.info(f"处理执行工具请求，工具 ID: {tool_id}，参数: {tool_params}")
# 查找工具
if tool_id not in self.tools:
raise ValueError(f"工具 ID '{tool_id}' 不存在")
tool = self.tools[tool_id]
handler = tool["handler"]
# 执行工具
try:
if asyncio.iscoroutinefunction(handler):
result = await handler(tool_params)
else:
result = handler(tool_params)
return {
"id": tool_id,
"status": "success",
"result": result
}
except Exception as e:
logger.exception(f"执行工具 '{tool_id}' 时出错")
return {
"id": tool_id,
"status": "error",
"error": str(e)
}
def handle_cancel_tool(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""处理取消工具执行请求"""
if not params or "id" not in params:
raise ValueError("缺少工具 ID 参数")
tool_id = params["id"]
logger.info(f"处理取消工具执行请求，工具 ID: {tool_id}")
# 在实际应用中，这里会取消正在执行的工具
return {
"id": tool_id,
"cancelled": True
}
def _search_tool_handler(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""搜索工具处理器"""
query = params.get("query", "")
scope = params.get("scope", "all")
# 示例搜索结果
results = []
if scope in ["all", "code"]:
results.append({
"path": "/examples/example.py",
"snippet": "def hello_world():",
"relevance": 0.95
})
if scope in ["all", "docs"]:
results.append({
"path": "/docs/example.md",
"snippet": "# 示例文档",
"relevance": 0.85
})
return {
"query": query,
"scope": scope,
"results": results
}
async def _file_system_tool_handler(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""文件系统工具处理器（异步）"""
action = params.get("action")
path = params.get("path")
if not action or not path:
raise ValueError("缺少必要参数")
# 模拟异步操作
await asyncio.sleep(0.1)
# 示例实现
if action == "read":
# 模拟读取文件
content = "这是模拟的文件内容"
return {
"path": path,
"content": content,
"size": len(content)
}
elif action == "write":
# 模拟写入文件
content = params.get("content", "")
return {
"path": path,
"written": True,
"size": len(content)
}
elif action == "list":
# 模拟列出目录
return {
"path": path,
"items": [
{"name": "file1.txt", "type": "file", "size": 1024},
{"name": "file2.py", "type": "file", "size": 2048},
{"name": "subdir", "type": "directory"}
]
}
elif action == "delete":
# 模拟删除文件
return {
"path": path,
"deleted": True
}
else:
raise ValueError(f"不支持的操作: {action}")