MCP 工具包成普通 Tool,和本地工具走同一条管线
每个 MCP 服务器的工具被包成 MCPTool——它就是 Tool 的子类,带 McpToolCard 元数据,和本地函数工具共享参数规整、事件触发与错误包装。传输方式由 McpServerConfig.client_type 选择,默认 sse;client 目录下有 stdio、sse、streamable http、playwright、openapi 五种客户端,另有 reconnect.py 处理重连。代价是 MCP 在这里没有自己的命名空间策略,工具名冲突要靠上层注册表处理。
- 适合
- 希望 MCP 工具与本地工具完全同构、共享参数规整与事件的系统。
- 代价
- 没有自己的命名空间策略,工具名冲突要靠上层注册表处理。
class McpServerConfig(BaseModel):
server_id: str = Field(default_factory=lambda: uuid.uuid4().hex)
server_name: str
server_path: str
client_type: str = 'sse'
params: Dict[str, Any] = Field(default_factory=dict)
auth_headers: dict = Field(default_factory=dict)
auth_query_params: Dict[str, str] = Field(default_factory=dict)
# ...(114-145 行:McpToolResult 与 McpToolCard)
class MCPTool(Tool):
"""MCP Tool class that wraps MCP server tools for LLM modules"""
def __init__(self, mcp_client: Any, tool_info: McpToolCard): # McpClient or its subclasses
# ...(150-165 行:构造与 stream(MCP 工具不支持流式))
async def invoke(self, inputs: Input, **kwargs) -> Output:
try:
# Prepare arguments for MCP tool call
arguments = inputs if isinstance(inputs, dict) else {}
if self._card.input_params is not None:
# ...(171-185 行:按 ToolCard 的 schema 规整参数并触发解析事件)
result = await self._mcp_client.call_tool(tool_name=self._card.name, arguments=arguments)
if isinstance(result, McpToolResult):
return result
return {"result": result}