qihuaiflow/backend/python/app/mcp_client.py

"""
MCP (Model Context Protocol) 客户端 —— 异步实现 streamable_http 传输。

支持：
- initialize / notifications/initialized / tools/list / tools/call 标准流程
- streamable_http：POST JSON-RPC，响应可能是 SSE（event: message / data: {...}）或纯 JSON
- SSE 解析兼容
- 直连外部端点（trust_env=False，不绕容器代理）

供 chat.py 在 AI 对话中把 MCP 工具暴露为 function calling，并代理执行 tools/call。
"""

import json
import logging
import asyncio
from typing import Optional

import httpx

logger = logging.getLogger(__name__)

MCP_PROTOCOL_VERSION = "2024-11-05"

# 工具列表缓存（url -> (timestamp, tools)），避免同一请求内 / 短时间重复握手
_TOOLS_CACHE: dict = {}
_TOOLS_CACHE_TTL = 120  # 秒


def _parse_sse(text: str):
    """从响应文本解析 JSON-RPC 结果。兼容 SSE（data: 行）与纯 JSON。"""
    if not text:
        return None
    text = text.strip()
    # SSE 格式：多行 "event: message" / "data: {...}"
    data_lines = []
    for line in text.splitlines():
        ls = line.strip()
        if ls.startswith("data:"):
            data_lines.append(ls[5:].strip())
    if data_lines:
        last = data_lines[-1]
        try:
            return json.loads(last)
        except Exception:
            pass
    # 纯 JSON
    try:
        return json.loads(text)
    except Exception:
        return None


async def _handshake_and_request(
    url: str,
    headers: dict,
    method: str,
    params: dict,
    timeout: float = 30.0,
):
    """
    完成一次完整 MCP 会话（initialize -> initialized -> 目标请求）。
    返回解析后的 JSON-RPC 响应 dict，或 None。
    """
    base_headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
    }
    base_headers.update(headers or {})

    async with httpx.AsyncClient(timeout=timeout, trust_env=False) as client:
        # 1) initialize
        r = await client.post(
            url,
            headers=base_headers,
            json={
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {
                    "protocolVersion": MCP_PROTOCOL_VERSION,
                    "capabilities": {},
                    "clientInfo": {"name": "aiflow", "version": "1.0"},
                },
            },
        )
        if r.status_code >= 400:
            logger.warning(f"[MCP] initialize 失败 {url} -> HTTP {r.status_code}")
            return None
        sid = r.headers.get("mcp-session-id") or r.headers.get("Mcp-Session-Id")
        req_headers = dict(base_headers)
        if sid:
            req_headers["Mcp-Session-Id"] = sid

        # 2) initialized 通知（不需要响应）
        try:
            await client.post(
                url,
                headers=req_headers,
                json={"jsonrpc": "2.0", "method": "notifications/initialized"},
            )
        except Exception:
            pass

        # 3) 目标请求
        rid = 2 if method != "tools/call" else 3
        r2 = await client.post(
            url,
            headers=req_headers,
            json={
                "jsonrpc": "2.0",
                "id": rid,
                "method": method,
                "params": params,
            },
        )
        if r2.status_code >= 400:
            logger.warning(f"[MCP] {method} 失败 {url} -> HTTP {r2.status_code}")
            return None
        return _parse_sse(r2.text)


async def list_mcp_tools(server: dict, timeout: float = 30.0) -> list:
    """
    拉取 MCP 服务器的工具列表。
    server: {url, transport_type, headers}
    返回 MCP tools 数组（每个含 name/description/inputSchema）。
    """
    url = (server.get("url") or "").strip()
    if not url:
        return []
    # 缓存
    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 _handshake_and_request(
            url, server.get("headers") or {}, "tools/list", {}, timeout
        )
        tools = []
        if data and "result" in data:
            tools = data["result"].get("tools", []) or []
        _TOOLS_CACHE[url] = (now, tools)
        return tools
    except Exception as e:
        logger.warning(f"[MCP] 拉取工具失败 {url}: {e}")
        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 _handshake_and_request(
            url,
            server.get("headers") or {},
            "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} 返回错误] {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(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}]"