diff --git a/CHANGELOG.md b/CHANGELOG.md index 1986db2..b677edf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ How to update: ### Fixed +- Fixed `MCPServerStreamableHttp` so Docker MCP Gateway style streaming endpoints that redirect `/mcp/` to `/mcp` and return JSON-RPC inside SSE `data:` frames can complete the HTTP handshake. - Fixed OpenAI-compatible tool schema export so `Any` and `**kwargs` no longer produce invalid `{"type": "any"}` entries in native `tools=` payloads. - Fixed `ReActTextParser` so common ReAct variants such as `Action Input` blocks, XML-style action tags, and fenced JSON tool calls no longer fall into parser repair loops. diff --git a/README.md b/README.md index f74c2ca..097a89e 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ QitOS core is the small framework. Product-grade applications and showcase agent ## Latest Updates +- **MCP Streamable HTTP compatibility**: `MCPServerStreamableHttp` now follows gateway redirects and parses JSON-RPC responses wrapped in SSE `data:` frames. - **Native tool schema hardening**: OpenAI-compatible `tools=` payloads no longer export invalid `type: any` schemas for `Any` or `**kwargs` parameters. - **ReAct parser compatibility**: `ReActTextParser` now accepts common `Action Input`, XML-style action tags, and fenced JSON tool-call variants that some OpenAI-compatible models emit. diff --git a/README.zh.md b/README.zh.md index 6cb09cc..a7f1757 100644 --- a/README.zh.md +++ b/README.zh.md @@ -18,6 +18,7 @@ QitOS 主仓库是小而清晰的核心框架。产品级 / 展示级应用会 ## 最新进展 +- **MCP Streamable HTTP 兼容性**：`MCPServerStreamableHttp` 现在会跟随 gateway 重定向，并解析 SSE `data:` 帧中封装的 JSON-RPC 响应。 - **原生工具 schema 加固**：OpenAI-compatible `tools=` payload 不再为 `Any` 或 `**kwargs` 参数导出非法的 `type: any` schema。 - **ReAct 解析兼容性**：`ReActTextParser` 现在可以解析部分 OpenAI-compatible 模型常见的 `Action Input`、XML 风格 action 标签和围栏 JSON tool-call 变体。 diff --git a/docs/tutorials/mcp-integration.mdx b/docs/tutorials/mcp-integration.mdx index 940b277..250d79e 100644 --- a/docs/tutorials/mcp-integration.mdx +++ b/docs/tutorials/mcp-integration.mdx @@ -31,6 +31,10 @@ server = MCPServerStreamableHttp( await server.connect() ``` +The HTTP transport follows redirects from gateways such as Docker MCP Gateway +and accepts JSON-RPC responses returned either as plain JSON or as Server-Sent +Events `data:` frames. + Both transports perform the MCP initialization handshake automatically during `connect()`. After connecting, you can list available tools: ```python diff --git a/docs/zh/tutorials/mcp-integration.mdx b/docs/zh/tutorials/mcp-integration.mdx index 970b039..54e4087 100644 --- a/docs/zh/tutorials/mcp-integration.mdx +++ b/docs/zh/tutorials/mcp-integration.mdx @@ -31,6 +31,9 @@ server = MCPServerStreamableHttp( await server.connect() ``` +HTTP 传输会跟随 Docker MCP Gateway 等网关返回的重定向，并支持普通 +JSON 响应以及封装在 Server-Sent Events `data:` 帧中的 JSON-RPC 响应。 + 两种传输都会在 `connect()` 期间自动完成 MCP 初始化握手。连接成功后，可以列出可用工具： ```python diff --git a/qitos/mcp/http.py b/qitos/mcp/http.py index 13f15a6..0c29990 100644 --- a/qitos/mcp/http.py +++ b/qitos/mcp/http.py @@ -70,6 +70,7 @@ async def connect(self) -> None: base_url=self._url, headers={**self._headers, "Content-Type": "application/json"}, timeout=httpx.Timeout(30.0), + follow_redirects=True, ) # MCP initialization handshake @@ -149,7 +150,7 @@ async def _send_request(self, method: str, params: Dict[str, Any]) -> Dict[str, response = await self._client.post("", json=payload) response.raise_for_status() - data = response.json() + data = self._decode_response(response) logger.debug("MCP HTTP <- %s", json.dumps(data)[:500]) if "error" in data: @@ -160,6 +161,58 @@ async def _send_request(self, method: str, params: Dict[str, Any]) -> Dict[str, return data.get("result", {}) + def _decode_response(self, response: Any) -> Dict[str, Any]: + content_type = str(response.headers.get("Content-Type", "")).lower() + if "text/event-stream" in content_type: + return self._decode_sse_response(response.text) + return response.json() + + def _decode_sse_response(self, text: str) -> Dict[str, Any]: + data_lines: List[str] = [] + last_error: Optional[json.JSONDecodeError] = None + + for raw_line in text.splitlines(): + line = raw_line.rstrip("\r") + if not line: + if data_lines: + parsed = self._parse_sse_data(data_lines) + if parsed is not None: + return parsed + try: + json.loads("\n".join(data_lines)) + except json.JSONDecodeError as exc: + last_error = exc + data_lines = [] + continue + if line.startswith(":"): + continue + if line.startswith("data:"): + data_lines.append(line[5:].lstrip()) + + if data_lines: + parsed = self._parse_sse_data(data_lines) + if parsed is not None: + return parsed + try: + json.loads("\n".join(data_lines)) + except json.JSONDecodeError as exc: + last_error = exc + + if last_error is not None: + raise ValueError( + "MCP HTTP SSE data frame did not contain valid JSON" + ) from last_error + raise ValueError("MCP HTTP SSE response did not include a data frame") + + def _parse_sse_data(self, data_lines: List[str]) -> Optional[Dict[str, Any]]: + try: + data = json.loads("\n".join(data_lines)) + except json.JSONDecodeError: + return None + if not isinstance(data, dict): + raise ValueError("MCP HTTP SSE data frame did not contain a JSON object") + return data + async def _send_notification(self, method: str, params: Optional[Dict[str, Any]] = None) -> None: """Send a JSON-RPC notification via HTTP POST.""" if self._client is None: diff --git a/tests/mcp/test_mcp.py b/tests/mcp/test_mcp.py index 0ca1449..f7598db 100644 --- a/tests/mcp/test_mcp.py +++ b/tests/mcp/test_mcp.py @@ -22,6 +22,7 @@ import pytest from qitos.core.tool import FunctionTool, ToolSpec +import qitos.mcp.http as http_transport from qitos.mcp import ( MCPServer, MCPToolInfo, @@ -603,6 +604,114 @@ async def test_operations_without_connect_raise(self) -> None: class TestMCPServerStreamableHttp: + def _patch_http_client(self, monkeypatch, handler): + httpx = http_transport.httpx + if httpx is None: + pytest.skip("httpx not installed") + + transport = httpx.MockTransport(handler) + real_async_client = httpx.AsyncClient + + def make_client(*args, **kwargs): + kwargs["transport"] = transport + return real_async_client(*args, **kwargs) + + monkeypatch.setattr(httpx, "AsyncClient", make_client) + return httpx + + def _sse_response(self, httpx, request_id: int, result: Dict[str, Any]): + payload = {"jsonrpc": "2.0", "id": request_id, "result": result} + body = ( + "event: message\n" + f"data: {json.dumps(payload)}\n\n" + ) + return httpx.Response( + 200, + headers={"Content-Type": "text/event-stream"}, + text=body, + ) + + def test_streamable_http_parses_sse_json_rpc_responses(self, monkeypatch) -> None: + def handler(request): + payload = json.loads(request.content or b"{}") + request_id = payload.get("id") + if request_id is None: + return httpx.Response(202) + if payload.get("method") == "initialize": + result = { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "sse-repro", "version": "0.0.1"}, + } + elif payload.get("method") == "tools/list": + result = { + "tools": [ + { + "name": "nmap_scan", + "description": "Run nmap", + "inputSchema": {"type": "object"}, + } + ] + } + else: + result = {} + return self._sse_response(httpx, request_id, result) + + httpx = self._patch_http_client(monkeypatch, handler) + server = MCPServerStreamableHttp(url="http://localhost:8080/mcp") + + async def exercise(): + try: + await server.connect() + return await server.list_tools() + finally: + await server.cleanup() + + tools = asyncio.run(exercise()) + assert [tool.name for tool in tools] == ["nmap_scan"] + + def test_streamable_http_follows_redirect_before_sse_response( + self, monkeypatch + ) -> None: + paths: List[str] = [] + + def handler(request): + paths.append(request.url.path) + if request.url.path == "/mcp/": + return httpx.Response(307, headers={"Location": "/mcp"}) + payload = json.loads(request.content or b"{}") + request_id = payload.get("id") + if request_id is None: + return httpx.Response(202) + if payload.get("method") == "initialize": + result = { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "redirect-repro", "version": "0.0.1"}, + } + elif payload.get("method") == "tools/list": + result = {"tools": []} + else: + result = {} + return self._sse_response(httpx, request_id, result) + + httpx = self._patch_http_client(monkeypatch, handler) + server = MCPServerStreamableHttp( + url="http://localhost:8080/mcp", + headers={"Authorization": "Bearer reprotoken123"}, + ) + + async def exercise(): + try: + await server.connect() + return await server.list_tools() + finally: + await server.cleanup() + + assert asyncio.run(exercise()) == [] + assert "/mcp/" in paths + assert "/mcp" in paths + def test_construction(self) -> None: try: server = MCPServerStreamableHttp(