fix issues #21

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.
 

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.
 

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 变体。
 

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

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

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:

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(