From d2ad1000b6325984d186d745b478432368aef3bc Mon Sep 17 00:00:00 2001 From: shuiyu67 Date: Wed, 5 Aug 2026 17:40:45 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20MCP=20=E5=AE=A2=E6=88=B7=E7=AB=AF?= =?UTF-8?q?=EF=BC=88=E8=BF=9E=E6=8E=A5=E5=A4=96=E9=83=A8=20MCP=20server?= =?UTF-8?q?=EF=BC=8C=E5=A4=96=E9=83=A8=E5=B7=A5=E5=85=B7=E6=A1=A5=E6=8E=A5?= =?UTF-8?q?=E8=BF=9B=E7=BB=9F=E4=B8=80=E5=B7=A5=E5=85=B7=E4=BD=93=E7=B3=BB?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 mcp_client.py：McpClient（subprocess stdio JSON-RPC，reader 线程+响应表， initialize/list_tools/call_tool/shutdown 超时可控）+ McpClientManager（connect_all 读 mcpServers、聚合、call_tool、cleanup 幂等） - unified_tool_manager.py：initialize 末尾加载外部 MCP server，按 server__tool 前缀 register_tool 带真实 inputSchema mcp_definition，shutdown 经 manager.cleanup 幂等终止子进程 - fc_tools.py：mcp_to_openai_tools 新增 get_mcp_tools 保真分支（同时修现状 liugin 工具 schema 被压成 args 的坑） - config.json：顶层加 mcpServers 空段（用户自填，已被 gitignore 不入库，代码 .get 兼容） - 新增 tests/test_mcp_client.py（5 项，全链路+桥接+保真，不触真实引擎/网络） Co-Authored-By: WorkBuddy --- mcp_client.py | 306 +++++++++++++++++++++++++++++++++++++++ tests/test_mcp_client.py | 193 ++++++++++++++++++++++++ unified_tool_manager.py | 96 ++++++++++++ xcli_core/fc_tools.py | 13 +- 4 files changed, 606 insertions(+), 2 deletions(-) create mode 100644 mcp_client.py create mode 100644 tests/test_mcp_client.py diff --git a/mcp_client.py b/mcp_client.py new file mode 100644 index 0000000..9d32d06 --- /dev/null +++ b/mcp_client.py @@ -0,0 +1,306 @@ +""" +小狸 MCP 客户端 +=============== +连接外部 MCP (Model Context Protocol) server，将其工具桥接进小狸的统一工具体系。 + +- transport = stdio（subprocess + 逐行 JSON-RPC 2.0），与项目根 mcp_server.py（暴露侧）对称。 +- 同步 + 线程安全：契合小狸「同步主循环 + ThreadPoolExecutor」模型，不引入官方 mcp SDK + （其 asyncio 风格与现有同步调用链不搭）。 +- 单个 server 失败不阻断其他 server，也不阻断主程序启动。 + +后续可在此文件扩展 SSE / HTTP transport（McpClient 子类即可）。 +""" + +import os +import json +import time +import threading +import subprocess +import logging +from typing import Dict, List, Any, Optional, Tuple + +logger = logging.getLogger(__name__) + +PROTOCOL_VERSION = "2024-11-05" + + +class McpClientError(RuntimeError): + """MCP 客户端错误""" + + +class McpClient: + """单个 MCP server 的 stdio 客户端（同步、线程安全）""" + + def __init__(self, name: str, command: str, args: Optional[List[str]] = None, + env: Optional[Dict[str, str]] = None, cwd: Optional[str] = None, + timeout: float = 30.0): + self.name = name + self.command = command + self.args = list(args or []) + self.env = env or {} + self.cwd = cwd + self.timeout = timeout + + self.proc = None + self._lock = threading.RLock() + self._next_id = 1 + self._pending: Dict[int, Dict[str, Any]] = {} + self._reader_thread = None + self._started = False + + # ── 生命周期 ── + def start(self) -> None: + """启动子进程并开启 reader 线程""" + if self._started: + return + full_env = dict(os.environ) + full_env.update(self.env) + cmd = [self.command] + self.args + try: + self.proc = subprocess.Popen( + cmd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + cwd=self.cwd, + env=full_env, + text=True, + encoding="utf-8", + bufsize=1, + ) + except Exception as e: + raise McpClientError(f"启动 MCP server '{self.name}' 失败: {e}") + + self._reader_thread = threading.Thread(target=self._reader_loop, daemon=True) + self._reader_thread.start() + self._started = True + + def _reader_loop(self): + """后台读线程：逐行解析 JSON-RPC 响应，按 id 投递到 _pending 表""" + try: + for raw in self.proc.stdout: + line = raw.strip() + if not line: + continue + try: + msg = json.loads(line) + except json.JSONDecodeError: + # 容错：跳过 server 可能混入 stdout 的非 JSON 噪声 + continue + msg_id = msg.get("id") + if msg_id is None: + # 通知（如进度），忽略 + continue + with self._lock: + pend = self._pending.pop(msg_id, None) + if pend is not None: + if "error" in msg: + pend["error"] = msg["error"] + else: + pend["value"] = msg.get("result") + pend["event"].set() + except Exception: + pass + finally: + # 连接断开：唤醒所有等待中的请求，避免永久阻塞 + with self._lock: + pendings = list(self._pending.values()) + self._pending.clear() + for pend in pendings: + pend["error"] = {"code": -1, "message": "MCP server 连接断开"} + pend["event"].set() + + def _send(self, method: str, params: Any = None, + is_notification: bool = False) -> Optional[Any]: + """发送一个 JSON-RPC 请求；非通知则等待响应（带超时）""" + if not self._started or self.proc is None: + raise McpClientError(f"MCP client '{self.name}' 未启动") + + with self._lock: + if is_notification: + req = {"jsonrpc": "2.0", "method": method, "params": params or {}} + pend = None + req_id = None + else: + req_id = self._next_id + self._next_id += 1 + pend = {"event": threading.Event(), "value": None, "error": None} + self._pending[req_id] = pend + req = {"jsonrpc": "2.0", "id": req_id, "method": method, "params": params or {}} + + try: + self.proc.stdin.write(json.dumps(req, ensure_ascii=False) + "\n") + self.proc.stdin.flush() + except (BrokenPipeError, ValueError, OSError) as e: + raise McpClientError(f"向 MCP server '{self.name}' 写入失败: {e}") + + if is_notification: + return None + + if not pend["event"].wait(self.timeout): + with self._lock: + self._pending.pop(req_id, None) + raise McpClientError(f"MCP server '{self.name}' 响应超时: {method}") + if pend["error"]: + err = pend["error"] + raise McpClientError( + f"MCP server '{self.name}' 错误 [{err.get('code')}]: {err.get('message')}") + return pend["value"] + + # ── MCP 协议方法 ── + def initialize(self) -> Dict[str, Any]: + """握手：协商协议版本并通知初始化完成""" + result = self._send("initialize", { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": {"name": "xiaoli-cli", "version": "8.0"}, + }) + # 通知 server 初始化完成（无响应，无需等待） + self._send("notifications/initialized", {}, is_notification=True) + return result or {} + + def list_tools(self) -> List[Dict[str, Any]]: + """列出 server 提供的工具定义""" + result = self._send("tools/list", {}) + if isinstance(result, dict): + return result.get("tools", []) + return [] + + def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str: + """调用一个工具，返回文本结果""" + result = self._send("tools/call", {"name": tool_name, "arguments": arguments or {}}) + return self._extract_content(result) + + @staticmethod + def _extract_content(result: Any) -> str: + """从 tools/call 的 result 提取文本结果""" + if result is None: + return "" + if isinstance(result, str): + return result + if isinstance(result, dict): + content = result.get("content") + if isinstance(content, list): + parts = [] + for item in content: + if isinstance(item, dict): + if item.get("type") == "text" or "text" in item: + parts.append(str(item.get("text", ""))) + return "\n".join(p for p in parts if p) + if isinstance(result.get("content"), str): + return result["content"] + # 部分 server 直接把结果放 result 顶层 + if "isError" in result and result.get("isError"): + return f"[工具错误] {json.dumps(result, ensure_ascii=False)}" + return json.dumps(result, ensure_ascii=False) + return str(result) + + def ping(self) -> bool: + try: + self._send("ping", {}) + return True + except Exception: + return False + + def stop(self) -> None: + """优雅关闭子进程，失败则强杀""" + if self.proc is None: + return + try: + self._send("shutdown", {}) + except Exception: + pass + try: + self._send("exit", {}, is_notification=True) + except Exception: + pass + try: + self.proc.stdin.close() + except Exception: + pass + try: + self.proc.wait(timeout=3) + except Exception: + try: + self.proc.kill() + except Exception: + pass + self._started = False + + +class McpClientManager: + """管理多个外部 MCP server 客户端（按配置并发连接、聚合工具）""" + + def __init__(self): + self.clients: Dict[str, McpClient] = {} + self._tool_index: List[Tuple[str, Dict[str, Any]]] = [] # (server_name, tool_def) + self._lock = threading.RLock() + self._cleaned = False + + def connect_all(self, servers: Dict[str, Dict[str, Any]], + cwd: Optional[str] = None) -> None: + """连接所有已声明且未禁用的 server。单个失败不阻断其他。 + + servers: {name: {"command", "args", "env", "cwd", "disabled"}} + """ + for name, spec in servers.items(): + if not isinstance(spec, dict): + continue + if spec.get("disabled"): + logger.info(f"MCP server '{name}' 已禁用，跳过") + continue + command = spec.get("command") + if not command: + logger.warning(f"MCP server '{name}' 缺少 command，跳过") + continue + args = spec.get("args") or [] + env = spec.get("env") or {} + server_cwd = spec.get("cwd") or cwd + client = McpClient(name, command, args, env, server_cwd) + try: + client.start() + client.initialize() + tools = client.list_tools() + with self._lock: + self.clients[name] = client + for tool in tools: + self._tool_index.append((name, tool)) + logger.info(f"已连接 MCP server '{name}': {len(tools)} 个工具") + except Exception as e: + try: + client.stop() + except Exception: + pass + logger.error(f"连接 MCP server '{name}' 失败: {e}") + # 不向上抛，继续下一个 server + + def get_all_tools(self) -> List[Tuple[str, Dict[str, Any]]]: + """返回 [(server_name, tool_def), ...]""" + with self._lock: + return list(self._tool_index) + + def get_servers(self) -> List[str]: + with self._lock: + return list(self.clients.keys()) + + def call_tool(self, server_name: str, tool_name: str, arguments: Dict[str, Any]) -> str: + with self._lock: + client = self.clients.get(server_name) + if client is None: + raise McpClientError(f"未连接的 MCP server: {server_name}") + return client.call_tool(tool_name, arguments) + + def cleanup(self) -> None: + """幂等：终止所有子进程。每个桥接工具的 shutdown 都会调一次，故必须可重入。""" + with self._lock: + if self._cleaned: + return + self._cleaned = True + clients = list(self.clients.values()) + self.clients.clear() + self._tool_index.clear() + for client in clients: + try: + client.stop() + except Exception: + pass diff --git a/tests/test_mcp_client.py b/tests/test_mcp_client.py new file mode 100644 index 0000000..8d061d8 --- /dev/null +++ b/tests/test_mcp_client.py @@ -0,0 +1,193 @@ +""" +MCP 客户端测试 +============== +用 inline 的 fake MCP server（写入临时文件，subprocess 启动）验证： + +1. McpClient 全链路：start → initialize → list_tools → call_tool → stop（进程回收） +2. McpClientManager 多 server 并发连接、聚合、调用、cleanup 幂等 +3. 超时：沉默 server 不响应 → call_tool 应在 timeout 内抛错 +4. UnifiedToolManager 桥接：外部 MCP 工具按 {server}__{tool} 注册、schema 保真、可被 execute 调用、shutdown 清理子进程 +5. fc_tools.mcp_to_openai_tools 优先走 get_mcp_tools 保真分支（不把 schema 压成 args） + +全程不依赖真实 AI 引擎、网络或外部 npx。 +""" + +import os +import sys + +import pytest + +# 确保项目根在 sys.path（unified_tool_manager / mcp_client / config 均在根目录） +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if ROOT not in sys.path: + sys.path.insert(0, ROOT) + +from mcp_client import McpClient, McpClientManager, McpClientError # noqa: E402 +import unified_tool_manager as utm # noqa: E402 +from xcli_core.fc_tools import mcp_to_openai_tools # noqa: E402 + +# fake MCP server：实现 initialize / tools/list / tools/call / ping， +# 以及一个 echo 工具（inputSchema 保真）。与 mcp_server.py 协议对称。 +FAKE_SERVER_SRC = r''' +import sys, json + +def send(obj): + sys.stdout.write(json.dumps(obj, ensure_ascii=False) + "\n") + sys.stdout.flush() + +def main(): + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + req = json.loads(line) + except Exception: + continue + method = req.get("method", "") + mid = req.get("id") + if method == "notifications/initialized": + continue + if method == "initialize": + send({"jsonrpc": "2.0", "id": mid, "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "fake", "version": "1.0"}, + }}) + elif method == "tools/list": + send({"jsonrpc": "2.0", "id": mid, "result": {"tools": [{ + "name": "echo", + "description": "回声工具", + "inputSchema": { + "type": "object", + "properties": {"text": {"type": "string", "description": "要回声的文本"}}, + "required": ["text"], + }, + }]}}) + elif method == "tools/call": + params = req.get("params", {}) + name = params.get("name") + args = params.get("arguments", {}) + if name == "echo": + send({"jsonrpc": "2.0", "id": mid, "result": { + "content": [{"type": "text", "text": "echo: " + str(args.get("text", ""))}], + }}) + else: + send({"jsonrpc": "2.0", "id": mid, "error": {"code": -32601, "message": "unknown tool"}}) + elif method == "ping": + send({"jsonrpc": "2.0", "id": mid, "result": {}}) + else: + send({"jsonrpc": "2.0", "id": mid, "error": {"code": -32601, "message": "method not found"}}) + +if __name__ == "__main__": + main() +''' + + +@pytest.fixture +def fake_server_path(tmp_path): + p = tmp_path / "fake_mcp_server.py" + p.write_text(FAKE_SERVER_SRC) + return str(p) + + +def test_mcp_client_lifecycle(fake_server_path): + client = McpClient("fake", sys.executable, [fake_server_path]) + try: + client.start() + info = client.initialize() + assert info.get("protocolVersion") == "2024-11-05" + tools = client.list_tools() + assert any(t.get("name") == "echo" for t in tools) + out = client.call_tool("echo", {"text": "hi"}) + assert "hi" in out + finally: + client.stop() + # 进程应已被回收 + assert client.proc.poll() is not None + + +def test_manager_multi_server(fake_server_path): + mgr = McpClientManager() + mgr.connect_all({ + "a": {"command": sys.executable, "args": [fake_server_path]}, + "b": {"command": sys.executable, "args": [fake_server_path]}, + }) + assert set(mgr.get_servers()) == {"a", "b"} + assert len(mgr.get_all_tools()) == 2 + out = mgr.call_tool("a", "echo", {"text": "x"}) + assert "x" in out + mgr.cleanup() + mgr.cleanup() # 幂等 + + +def test_client_timeout(tmp_path): + silent = tmp_path / "silent.py" + silent.write_text("import time\nwhile True:\n time.sleep(1)\n") + client = McpClient("silent", sys.executable, [str(silent)], timeout=2) + try: + client.start() + with pytest.raises(Exception): + client.list_tools() # 沉默 server 不响应 → 超时 + finally: + client.stop() + + +def test_unified_manager_bridges_external_mcp(tmp_path, monkeypatch): + fake = tmp_path / "fake_mcp_server.py" + fake.write_text(FAKE_SERVER_SRC) + cfg = {"mcpServers": {"fake": {"command": sys.executable, "args": [str(fake)]}}} + + import xcli_core.config as xcfg + monkeypatch.setattr(xcfg, "load_config", lambda: cfg) + try: + import config as root_cfg + monkeypatch.setattr(root_cfg, "load_config", lambda: cfg) + except ImportError: + pass + + mgr = utm.UnifiedToolManager() + # 用不存在的插件/技能目录，避免加载真实插件 + mgr.initialize( + plugins_dir=str(tmp_path / "noplugins"), + skills_dir=str(tmp_path / "noskills"), + ) + try: + tool = mgr.get_tool("fake__echo") + assert tool is not None, "外部 MCP 工具应已桥接为 fake__echo" + assert tool["protocol"] == "mcp" + + names = [t["name"] for t in mgr.get_mcp_tools()] + assert "fake__echo" in names + + echo_def = next(t for t in mgr.get_mcp_tools() if t["name"] == "fake__echo") + assert "text" in echo_def["inputSchema"]["properties"] + + res = mgr.execute("fake__echo", '{"text": "hello"}') + assert res.success + assert "hello" in (res.data or "") + finally: + mgr.shutdown() + + +def test_fc_tools_uses_get_mcp_tools(): + class FakeMgr: + def get_mcp_tools(self): + return [{ + "name": "my_tool", + "description": "d", + "inputSchema": { + "type": "object", + "properties": {"q": {"type": "string"}}, + "required": ["q"], + }, + }] + + tools = mcp_to_openai_tools(FakeMgr()) + my = next(t for t in tools if t["function"]["name"] == "my_tool") + # schema 保真：不被压成 {"args": string} + assert "q" in my["function"]["parameters"]["properties"] + assert "args" not in my["function"]["parameters"]["properties"] + # 内置工具（memory / scheduler）仍被追加 + names = [t["function"]["name"] for t in tools] + assert "memory" in names and "scheduler" in names diff --git a/unified_tool_manager.py b/unified_tool_manager.py index 55d94c1..9d0fe88 100644 --- a/unified_tool_manager.py +++ b/unified_tool_manager.py @@ -12,6 +12,11 @@ from colorama import Fore, Style from xcli_core.tool_result import ToolResult, ErrorCode from xcli_core.verbose import vprint +try: + from mcp_client import McpClientManager +except Exception: + McpClientManager = None + logger = logging.getLogger(__name__) @@ -48,6 +53,9 @@ class UnifiedToolManager: # MCP 工具定义缓存 self._mcp_tools_cache: Optional[List[Dict]] = None + # 外部 MCP client 管理器（懒加载） + self._mcp_client_manager = None + def initialize(self, plugins_dir: str = "plugins", skills_dir: str = "skills") -> None: """初始化并加载所有工具""" self._plugins_dir = plugins_dir @@ -55,6 +63,7 @@ class UnifiedToolManager: self._load_plugins(plugins_dir) self._load_skills(skills_dir) self._build_mcp_cache() + self._load_mcp_servers() n_liugin = sum(1 for p in self._protocol_map.values() if p == 'liugin') n_skill = sum(1 for p in self._protocol_map.values() if p == 'skill') @@ -302,6 +311,93 @@ class UnifiedToolManager: } } + # ==================== 外部 MCP 客户端桥接 ==================== + + def _load_mcp_servers(self) -> None: + """连接 config.json 中声明的外部 MCP server，将其工具桥接进统一注册表。 + + 外部工具以 ``{server}__{tool}`` 命名注册，schema（inputSchema）保真， + 模型可像调用本地工具一样调用外部 MCP 工具。 + """ + if McpClientManager is None: + return + try: + from config import load_config + except ImportError: + try: + from xcli_core.config import load_config + except ImportError: + return + + try: + cfg = load_config() + servers = cfg.get("mcpServers") or {} + except Exception: + return + + if not servers: + return + + if self._mcp_client_manager is None: + self._mcp_client_manager = McpClientManager() + + # unified_tool_manager.py 位于项目根，故 dirname(__file__) 即项目根 + project_dir = os.path.dirname(os.path.abspath(__file__)) + + try: + self._mcp_client_manager.connect_all(servers, cwd=project_dir) + except Exception as e: + print(f"{Fore.YELLOW}加载外部 MCP server 失败: {e}{Style.RESET_ALL}") + return + + added = 0 + for server_name, tool_def in self._mcp_client_manager.get_all_tools(): + remote_name = tool_def.get("name") + if not remote_name: + continue + local_name = f"{server_name}__{remote_name}" + mcp_def = { + "name": local_name, + "description": tool_def.get("description", ""), + "inputSchema": tool_def.get("inputSchema", {"type": "object", "properties": {}}), + } + handler = self._make_external_handler(server_name, remote_name) + # instance 存 manager，shutdown 时统一 cleanup（McpClientManager.cleanup 幂等） + self.register_tool( + local_name, handler, + description=mcp_def["description"], + protocol="mcp", + mcp_definition=mcp_def, + instance=self._mcp_client_manager, + ) + added += 1 + + if added: + n_servers = len(self._mcp_client_manager.get_servers()) + print(f"{Fore.GREEN}已桥接 {added} 个外部 MCP 工具{Style.RESET_ALL}" + f"{Fore.BLACK}{Style.BRIGHT} (来自 {n_servers} 个 server){Style.RESET_ALL}") + + def _make_external_handler(self, server_name: str, remote_name: str): + """为某个外部 MCP 工具生成 handler：解析参数 → call_tool → ToolResult""" + def handler(args): + try: + if isinstance(args, dict): + arguments = args + elif isinstance(args, str) and args.strip().startswith("{"): + arguments = json.loads(args) + else: + arguments = {"args": args} + except (json.JSONDecodeError, TypeError): + arguments = {"args": args} + + try: + result_text = self._mcp_client_manager.call_tool(server_name, remote_name, arguments) + return ToolResult.ok(result_text, tool_name=f"{server_name}__{remote_name}") + except Exception as e: + return ToolResult.fail(str(e), ErrorCode.EXEC_FAILED, + tool_name=f"{server_name}__{remote_name}") + return handler + # ==================== 公共 API ==================== @property diff --git a/xcli_core/fc_tools.py b/xcli_core/fc_tools.py index d639cb0..1382e78 100644 --- a/xcli_core/fc_tools.py +++ b/xcli_core/fc_tools.py @@ -21,8 +21,17 @@ def mcp_to_openai_tools(plugin_manager) -> List[Dict]: if not plugin_manager: return tools - # 统一工具管理器 - if hasattr(plugin_manager, 'plugins'): + # 优先使用统一工具管理器的 MCP 定义（含外部 MCP client 桥接的工具，schema 保真） + if hasattr(plugin_manager, 'get_mcp_tools'): + try: + for mcp_def in plugin_manager.get_mcp_tools(): + fc_def = mcp_to_fc(mcp_def) + if fc_def: + tools.append(fc_def) + except Exception: + pass + # 统一工具管理器（兜底，旧实现未暴露 get_mcp_tools 时） + elif hasattr(plugin_manager, 'plugins'): for plugin in plugin_manager.plugins: fc_def = _extract_fc_def(plugin) if fc_def: