feat: MCP 客户端(连接外部 MCP server,外部工具桥接进统一工具体系)

- 新增 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 <workbuddy@tencent.com>
This commit is contained in:
shuiyu67andWorkBuddy committed 2026-08-05 17:40:45 +08:00
1 parent a3977ecc85
commit d2ad1000b6
4 files changed
+606 -2

No files matched your search

+306
View File
@@ -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
+193
View File
@@ -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
+96
View File
@@ -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
+11 -2
View File
@@ -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: