diff --git a/backend/requirements.txt b/backend/requirements.txt index d996468..b16b191 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -2,7 +2,9 @@ # webrtc -> SmallWebRTCTransport / SmallWebRTCConnection + aiortc # silero -> 本地 VAD(判断用户说话起止),语音必备 # openai -> OpenAI 兼容的 LLM/STT/TTS 客户端(DeepSeek、SenseVoice、CosyVoice 都走它) -pipecat-ai[webrtc,websocket,silero,openai,mcp]==1.7.0 +pipecat-ai[webrtc,websocket,silero,openai,mcp]==1.13.0 +# MCP 2.x 使用 httpx2；普通业务 HTTP 请求继续使用 httpx。 +httpx2>=2.10.0,<3 Pillow>=11.1.0,<13 # FastGPT 类型助手:本地 SDK(包 /api/v1/chat/completions 流式 + chatId 会话) diff --git a/backend/services/tools/mcp_client.py b/backend/services/tools/mcp_client.py index d52f7b0..1034f2b 100644 --- a/backend/services/tools/mcp_client.py +++ b/backend/services/tools/mcp_client.py @@ -10,13 +10,13 @@ from __future__ import annotations import hashlib import json from contextlib import asynccontextmanager -from datetime import timedelta from typing import Any, AsyncIterator -import httpx +import httpx2 as httpx from mcp import ClientSession from mcp.client.sse import sse_client from mcp.client.streamable_http import streamable_http_client +from mcp.types import PaginatedRequestParams from models import RuntimeMcpServer @@ -59,7 +59,7 @@ class McpToolClient: server.url, http_client=http_client, ) as streams: - read_stream, write_stream, _ = streams + read_stream, write_stream = streams yield read_stream, write_stream return @@ -88,7 +88,7 @@ class McpToolClient: async with ClientSession( read_stream, write_stream, - read_timeout_seconds=timedelta(seconds=timeout), + read_timeout_seconds=timeout, ) as session: await session.initialize() yield session @@ -105,9 +105,11 @@ class McpToolClient: async with self._session(server) as session: cursor: str | None = None while True: - result = await session.list_tools(cursor=cursor) + result = await session.list_tools( + params=PaginatedRequestParams(cursor=cursor) if cursor else None, + ) for tool in result.tools: - input_schema = dict(tool.inputSchema or {}) + input_schema = dict(tool.input_schema or {}) discovered.append( { "name": tool.name, @@ -116,7 +118,7 @@ class McpToolClient: "schema_hash": schema_hash(input_schema), } ) - cursor = result.nextCursor + cursor = result.next_cursor if not cursor: break return discovered @@ -132,14 +134,14 @@ class McpToolClient: result = await session.call_tool( remote_tool_name, arguments=arguments, - read_timeout_seconds=timedelta(seconds=server.timeout_seconds), + read_timeout_seconds=server.timeout_seconds, ) data = self._normalize_result(result) encoded = json.dumps(data, ensure_ascii=False, default=str).encode("utf-8") if len(encoded) > MAX_MCP_RESULT_BYTES: raise McpClientError("MCP 工具响应超过 1 MB 限制") - if bool(result.isError): + if bool(result.is_error): message = data.get("text") or "MCP 工具返回执行错误" raise McpClientError(str(message)[:2048]) return data @@ -171,6 +173,6 @@ class McpToolClient: return { "text": "\n".join(text_parts), - "structuredContent": result.structuredContent, + "structuredContent": result.structured_content, "content": content, } diff --git a/backend/tests/test_mcp_client.py b/backend/tests/test_mcp_client.py index 7488322..ec589bb 100644 --- a/backend/tests/test_mcp_client.py +++ b/backend/tests/test_mcp_client.py @@ -1,13 +1,89 @@ +import json import unittest from contextlib import asynccontextmanager from unittest.mock import patch +import httpx2 + from models import RuntimeMcpServer from schemas import McpServerUpsert -from services.tools.mcp_client import McpToolClient +from services.tools.mcp_client import McpClientError, McpToolClient class McpTransportTests(unittest.IsolatedAsyncioTestCase): + async def test_streamable_http_discovery_and_call_with_real_sdk(self): + requests = [] + + def respond(request): + self.assertEqual(request.headers["Authorization"], "Bearer test") + message = json.loads(request.content) + requests.append(message) + if "id" not in message: + return httpx2.Response(202) + if message["method"] == "initialize": + result = { + "protocolVersion": "2025-06-18", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "test", "version": "1"}, + } + elif message["method"] == "tools/list": + if (message.get("params") or {}).get("cursor") == "page2": + result = {"tools": []} + else: + result = { + "tools": [{ + "name": "get_order", + "inputSchema": {"type": "object", "properties": {}}, + }], + "nextCursor": "page2", + } + elif message["method"] == "tools/call": + result = { + "content": [{"type": "text", "text": "已发货"}], + "structuredContent": {"status": "shipped"}, + } + else: + self.fail(f"Unexpected MCP method: {message['method']}") + return httpx2.Response(200, json={ + "jsonrpc": "2.0", "id": message["id"], "result": result, + }) + + client_class = httpx2.AsyncClient + + def client_factory(**kwargs): + return client_class(transport=httpx2.MockTransport(respond), **kwargs) + + server = RuntimeMcpServer( + id="mcp_http", + url="https://mcp.example.com/mcp", + secret_headers={"Authorization": "Bearer test"}, + ) + client = McpToolClient() + with patch("services.tools.mcp_client.httpx.AsyncClient", client_factory): + tools = await client.list_tools(server) + listings = [item for item in requests if item["method"] == "tools/list"] + self.assertEqual(len(listings), 2) + self.assertEqual(listings[1]["params"]["cursor"], "page2") + result = await client.call_tool(server, "get_order", {"order_id": "A-100"}) + + self.assertEqual(tools[0]["name"], "get_order") + self.assertEqual(result["text"], "已发货") + self.assertEqual(result["structuredContent"], {"status": "shipped"}) + call = next(item for item in requests if item["method"] == "tools/call") + self.assertEqual(call["params"]["arguments"], {"order_id": "A-100"}) + + async def test_sdk_timeout_has_readable_error(self): + @asynccontextmanager + async def timed_out(*args): + raise httpx2.ReadTimeout("read timed out") + yield + + server = RuntimeMcpServer(id="mcp_timeout", url="https://mcp.example.com/mcp") + client = McpToolClient() + with patch.object(client, "_transport_streams", timed_out): + with self.assertRaisesRegex(McpClientError, "超时"): + await client.list_tools(server) + def test_server_schema_accepts_sse(self): server = McpServerUpsert( name="旧版 MCP 服务", diff --git a/backend/tests/test_service_factory.py b/backend/tests/test_service_factory.py index 8e7f099..52562f1 100644 --- a/backend/tests/test_service_factory.py +++ b/backend/tests/test_service_factory.py @@ -13,8 +13,8 @@ from services.pipecat.service_factory import ( ) -class LLMServiceFactoryTest(unittest.TestCase): - def test_openai_compatible_llm_converts_async_tool_developer_result(self): +class LLMServiceFactoryTest(unittest.IsolatedAsyncioTestCase): + async def test_openai_compatible_llm_converts_async_tool_developer_result(self): service = create_llm( AssistantConfig( llm_interface_type="openai-llm", @@ -34,7 +34,7 @@ class LLMServiceFactoryTest(unittest.TestCase): ] ) - params = service.get_llm_adapter().get_llm_invocation_params( + params = await service.get_llm_adapter().get_llm_invocation_params( context, convert_developer_to_user=not service.supports_developer_role, )