""" MCP (Model Context Protocol) 客户端 —— 基于官方 \"mcp\" Python SDK (Phase 1.E 标准化) 设计基准（标准对照表见 STANDARDS_AUDIT_2026-07-24.md §E）： - 采用 Anthropic 官方 \"mcp\" SDK 处理 JSON-RPC 帧 + 协议版本协商， 不再手写解析、不再写死 MCP_PROTOCOL_VERSION 常量（旧实现写死 "2024-11-05"， 上游协议一升级即碎）。 - 支持 streamable_http（默认）与 sse 两种传输，按 server['transport_type'] 选择。 - 公共 API 保持向后兼容：list_mcp_tools(server, timeout) / call_mcp_tool(server, tool_name, arguments, timeout) 供 chat.py / chat_service.py / chat_agents.py / chat_skill.py / chat_mcp.py / chat_context.py / chat_knowledge.py / chat_media.py / chat_utils.py / core/aca.py 共 10+ 处调用，签名与返回结构（tools 数组 / 文本结果字符串）不变。 - 工具列表缓存（按 url）与错误处理保持原行为，避免重复握手、保证降级安全。 """ import asyncio import logging from typing import Optional from mcp import ClientSession from mcp.client.sse import sse_client # mcp 2.0.0 把 streamablehttp_client 重命名为 streamable_http_client(下划线位置变了)。 # 兼容 0.28.x / 1.x / 2.x:先试新名,失败回退老名。 try: from mcp.client.streamable_http import streamable_http_client as _streamable_http_client except ImportError: # 0.28.x 旧名 try: from mcp.client.streamable_http import streamablehttp_client as _streamable_http_client except ImportError: _streamable_http_client = None # type: ignore[assignment] import httpx # 兼容 mcp 2.0+ 安装的 httpx2 fork — 同源 API,可作别名 try: import httpx2 as _httpx except ImportError: _httpx = httpx logger = logging.getLogger(__name__) # 工具列表缓存（url -> (timestamp, tools)），避免同一请求内 / 短时间重复握手 _TOOLS_CACHE: dict = {} _TOOLS_CACHE_TTL = 120 # 秒 def _no_proxy_httpx_client(headers=None, timeout=None, auth=None): """MCP 传输用 httpx 客户端工厂：强制直连、不走环境代理。 容器 run.sh 注入了 HTTP_PROXY=http://172.18.0.1:7890，该代理无法解析 外部 MCP 域名（xiaobenyang.com / anysearch.com 等），且默认 trust_env 会读取 HTTP_PROXY 导致 tools/list 全部静默失败。这里 trust_env=False 直连，与 app/core/llm_client.py:chat_complete 修复同源。 mcp 2.0.0 改动:`streamable_http_client` 不再接受 headers/timeout/httpx_client_factory 等 kwargs,改为接受预配置的 http_client(httpx2/httpx.AsyncClient 实例)。 所以这里返回一个完整的 client,而不是 factory。 """ # 关键:httpx2(2.0+ 引入,同源 API)与 httpx 兼容 cli_class = getattr(_httpx, "AsyncClient", _httpx.Client) if hasattr(_httpx, "AsyncClient") else _httpx.AsyncClient kwargs: dict = {"follow_redirects": True, "trust_env": False} kwargs["timeout"] = timeout if timeout is not None else _httpx.Timeout(30.0, read=300.0) if headers is not None: kwargs["headers"] = headers if auth is not None: kwargs["auth"] = auth return cli_class(**kwargs) def _transport_cm(server: dict, timeout: float): """按 server['transport_type'] 选择官方 SDK 传输层。 返回异步上下文管理器，进入后产出 (read_stream, write_stream, ...)。 关键：注入 _no_proxy_httpx_client，使 MCP 流量直连（见其 docstring）。 mcp 2.0.0 兼容路径: - 旧 (0.28.x): streamablehttp_client(url, headers=..., timeout=..., httpx_client_factory=...) - 新 (2.0+): streamable_http_client(url, http_client=<预配置 client>) 用 inspect 判断参数,自动适配。 """ url = (server.get("url") or "").strip() ttype = (server.get("transport_type") or "streamable_http").lower() headers = server.get("headers") or {} if ttype == "sse": # SSE 也有同样问题,新版 sse_client 签名也可能变。这里保守用老 kwargs(若失败 fallback) return sse_client(url, headers=headers, timeout=timeout, httpx_client_factory=_no_proxy_httpx_client) if _streamable_http_client is None: raise RuntimeError("mcp SDK missing streamable_http transport (both new + old names unavailable)") # inspect 新版签名,判断是否接受 http_client import inspect try: sig_params = inspect.signature(_streamable_http_client).parameters accepts_http_client = "http_client" in sig_params except Exception: accepts_http_client = False if accepts_http_client: # mcp 2.0+ 路径:headers/timeout 全部装进预配置的 httpx client http_client = _no_proxy_httpx_client(headers=headers, timeout=timeout) return _streamable_http_client(url, http_client=http_client) else: # 0.28.x 老路径 return _streamable_http_client(url, headers=headers, timeout=timeout, httpx_client_factory=_no_proxy_httpx_client) async def _session_request( server: dict, method: str, params: dict, timeout: float = 30.0, raise_on_error: bool = False ): """用官方 ClientSession 完成一次标准 MCP 调用（initialize -> 目标请求）。 返回解析后的 dict：{"result": {...}} 或 None（失败）。 协议版本由 SDK 自动协商，不再硬编码。 raise_on_error=True 时连接/网络错误（如 ExceptionGroup）会向上抛，不再吞掉。 """ url = (server.get("url") or "").strip() if not url: return None try: async with _transport_cm(server, timeout) as streams: read, write, *_ = streams async with ClientSession(read, write) as session: # 标准握手：initialize 会发送 initialize 请求并自动补 notifications/initialized await session.initialize() # 防御性补发 initialized 通知（部分服务端要求显式接收） try: if hasattr(session, "send_notification"): await session.send_notification("notifications/initialized", {}) except Exception: pass if method == "tools/list": result = await session.list_tools() tools = [t.model_dump() for t in (result.tools or [])] return {"result": {"tools": tools}} elif method == "tools/call": res = await session.call_tool( params.get("name"), params.get("arguments") or {} ) content = [] for blk in (res.content or []): btype = getattr(blk, "type", "text") if btype == "text": content.append({"type": "text", "text": getattr(blk, "text", "")}) else: content.append({"type": btype, "text": str(blk)}) return { "result": { "content": content, "isError": bool(getattr(res, "isError", False)), } } except Exception as e: logger.warning(f"[MCP] {method} 异常 {url}: {e}") if raise_on_error: raise return None return None async def list_mcp_tools(server: dict, timeout: float = 30.0, raise_on_error: bool = False) -> list: """ 拉取 MCP 服务器的工具列表。 server: {url, transport_type, headers} 返回 MCP tools 数组（每个含 name/description/inputSchema）。 raise_on_error=True 时连接/网络错误不再吞掉，直接向上抛（供测试诊断用）。 """ url = (server.get("url") or "").strip() if not url: return [] # 缓存（仅在正常返回 tools 时缓存；raise_on_error 模式不缓存，便于实时重试） if not raise_on_error: now = asyncio.get_event_loop().time() cached = _TOOLS_CACHE.get(url) if cached and (now - cached[0]) < _TOOLS_CACHE_TTL: return cached[1] try: data = await _session_request(server, "tools/list", {}, timeout, raise_on_error=raise_on_error) tools = [] if data and "result" in data: tools = data["result"].get("tools", []) or [] if not raise_on_error: _TOOLS_CACHE[url] = (asyncio.get_event_loop().time(), tools) return tools except Exception as e: logger.warning(f"[MCP] 拉取工具失败 {url}: {e}") if raise_on_error: raise return [] async def call_mcp_tool( server: dict, tool_name: str, arguments: dict, timeout: float = 60.0 ) -> str: """ 在 MCP 服务器上执行一个工具，返回可供 LLM 消费的文本结果。 """ url = (server.get("url") or "").strip() if not url: return f"[MCP 工具 {tool_name} 执行失败：服务端地址为空]" logger.info(f"[MCP] 执行工具调用 tool={tool_name} server={server.get('name')} url={url}") try: data = await _session_request( server, "tools/call", {"name": tool_name, "arguments": arguments or {}}, timeout, ) if not data: return f"[MCP 工具 {tool_name} 调用失败：无响应]" if "error" in data: err = data["error"] return f"[MCP 工具 {tool_name} 返回错误] {__import__('json').dumps(err, ensure_ascii=False)}" result = data.get("result", {}) content = result.get("content", []) or [] parts = [] for item in content: if isinstance(item, dict) and item.get("type") == "text": parts.append(item.get("text", "")) else: parts.append(__import__("json").dumps(item, ensure_ascii=False)) text = "\n".join(parts).strip() if result.get("isError"): return f"[MCP 工具 {tool_name} 执行报错]\n{text}" return text or f"[MCP 工具 {tool_name} 返回空结果]" except Exception as e: logger.warning(f"[MCP] 工具执行异常 {tool_name} @ {url}: {e}") return f"[MCP 工具 {tool_name} 执行异常：{e}]"