✨ 支持 MCP Streamable HTTP 传输协议 #28

nonebot_plugin_llmchat/mcpclient.py

@@ -3,8 +3,10 @@ from contextlib import AsyncExitStack
 from time import monotonic
 from typing import Any, cast
 
+import httpx
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.sse import sse_client
+from mcp.client.streamable_http import streamable_http_client
 from mcp.client.stdio import stdio_client
 from nonebot import logger
 
@@ -88,9 +90,42 @@ class MCPClient:
         config = self.server_config[server_name]
         session_stack = AsyncExitStack()
         if config.url:
-            transport = await session_stack.enter_async_context(
-                sse_client(url=config.url, headers=config.headers)
-            )
+            transport_type = config.transport
+            if transport_type == "streamable_http":
+                logger.debug(f"服务器[{server_name}]使用 streamable_http 传输协议")
+                http_client = await session_stack.enter_async_context(
+                    httpx.AsyncClient(headers=config.headers or {})
+                )
+                read, write, _ = await session_stack.enter_async_context(
+                    streamable_http_client(url=config.url, http_client=http_client)
+                )
+                transport = (read, write)
+            elif transport_type == "sse":
+                logger.debug(f"服务器[{server_name}]使用 sse 传输协议")
+                transport = await session_stack.enter_async_context(
+                    sse_client(url=config.url, headers=config.headers)
+                )
+            else:
+                # 未指定协议，自动探测：先尝试 streamable_http，失败则回退到 sse
+                logger.debug(f"服务器[{server_name}]未指定传输协议，开始自动探测")
+                probe_stack = AsyncExitStack()
+                try:
+                    http_client = await probe_stack.enter_async_context(
+                        httpx.AsyncClient(headers=config.headers or {})
+                    )
+                    read, write, _ = await probe_stack.enter_async_context(
+                        streamable_http_client(url=config.url, http_client=http_client)
+                    )
+                    await session_stack.enter_async_context(probe_stack)
+                    transport = (read, write)
+                    logger.debug(f"服务器[{server_name}]自动探测成功: 使用 streamable_http 传输协议")
+                except Exception as e:
+                    await probe_stack.aclose()
+                    logger.debug(f"服务器[{server_name}]streamable_http 探测失败({e})，回退到 sse")
+                    transport = await session_stack.enter_async_context(
+                        sse_client(url=config.url, headers=config.headers)
+                    )
+                    logger.debug(f"服务器[{server_name}]自动探测成功: 使用 sse 传输协议")
         elif config.command:
             stdio_params: dict[str, Any] = {
                 "command": config.command,