Cover malformed MCP response timeout with e2e proof

Constraint: Reproduce malformed JSON-RPC response before changing SDK behavior.

Rejected: Unit-only coverage | it did not exercise the MCP transport/context-manager path.

Confidence: high

Scope-risk: narrow

Directive: Keep malformed MCP response handling bounded at the SDK MCP boundary; services must still return valid JSON-RPC errors.

Tested: uv run pytest tests/e2e/test_mcp_malformed_response.py -q; uv run pytest tests/unittests/tool/test_mcp.py -q; uv run pytest tests/unittests/tool -q; uv run ruff check agentrun/tool/api/mcp.py tests/unittests/tool/test_mcp.py tests/e2e/test_mcp_malformed_response.py; git diff --check

Change-Id: Icde49bbfd79f29eb64acdab904f1f5df8df47bcd
Not-tested: Full e2e suite against remote AgentRun services.

Author: congxiao-wxx

agentrun/tool/api/mcp.py

@@ -170,6 +170,28 @@ async def _wait_for_mcp_request(
                 f" {self.endpoint}"
             ) from exc
 
+    def _find_mcp_timeout_error(
+        self, exc: BaseException
+    ) -> Optional[TimeoutError]:
+        if isinstance(exc, TimeoutError) and str(exc).startswith("MCP "):
+            return exc
+
+        nested_exceptions = getattr(exc, "exceptions", None)
+        if not nested_exceptions:
+            return None
+
+        for nested_exc in nested_exceptions:
+            timeout_error = self._find_mcp_timeout_error(nested_exc)
+            if timeout_error is not None:
+                return timeout_error
+
+        return None
+
+    def _raise_mcp_timeout_if_present(self, exc: BaseException) -> None:
+        timeout_error = self._find_mcp_timeout_error(exc)
+        if timeout_error is not None:
+            raise timeout_error
+
     def _build_ram_auth(self, url: str) -> tuple:
         """当目标是 agentrun-data 域名时，改写 URL 并返回 httpx Auth handler。
 
@@ -270,6 +292,9 @@ async def list_tools_async(self) -> List[ToolInfo]:
                 "mcp package is not installed. Install it with: pip install mcp"
             )
             return []
+        except Exception as exc:
+            self._raise_mcp_timeout_if_present(exc)
+            raise
 
     def list_tools(self) -> List[ToolInfo]:
         """同步获取工具列表 / Get tool list synchronously
@@ -347,6 +372,9 @@ async def call_tool_async(
                 "mcp package is required for MCP tool calls. "
                 "Install it with: pip install mcp"
             )
+        except Exception as exc:
+            self._raise_mcp_timeout_if_present(exc)
+            raise
 
     def call_tool(
         self,

tests/e2e/test_mcp_malformed_response.py

@@ -0,0 +1,87 @@
+"""E2E regression tests for malformed MCP streamable-http responses."""
+
+import asyncio
+import socket
+import threading
+import time
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+import httpx
+import pytest
+import uvicorn
+
+from agentrun.tool.api.mcp import ToolMCPSession
+from agentrun.utils.config import Config
+
+
+def _find_free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+def _build_malformed_mcp_app() -> FastAPI:
+    app = FastAPI()
+
+    @app.get("/health")
+    async def health():
+        return {"ok": True}
+
+    @app.post("/mcp")
+    async def mcp_endpoint(request: Request):
+        payload = await request.json()
+        return JSONResponse(
+            {
+                "jsonrpc": "2.0",
+                "id": payload.get("id"),
+                "error": {
+                    "code": -32000,
+                    "message": None,
+                },
+            }
+        )
+
+    return app
+
+
+@pytest.fixture
+def malformed_mcp_server():
+    app = _build_malformed_mcp_app()
+    port = _find_free_port()
+    config = uvicorn.Config(
+        app, host="127.0.0.1", port=port, log_level="warning"
+    )
+    server = uvicorn.Server(config)
+
+    thread = threading.Thread(target=server.run, daemon=True)
+    thread.start()
+
+    base_url = f"http://127.0.0.1:{port}"
+    for _ in range(50):
+        try:
+            httpx.get(f"{base_url}/health", timeout=0.2)
+            break
+        except Exception:
+            time.sleep(0.1)
+    else:
+        raise RuntimeError("malformed MCP server did not start")
+
+    yield f"{base_url}/mcp"
+
+    server.should_exit = True
+    thread.join(timeout=5)
+
+
+@pytest.mark.asyncio
+async def test_streamable_mcp_malformed_initialize_error_fails_fast(
+    malformed_mcp_server,
+):
+    session = ToolMCPSession(
+        endpoint=malformed_mcp_server,
+        session_affinity="MCP_STREAMABLE",
+        config=Config(timeout=0.05),
+    )
+
+    with pytest.raises(TimeoutError, match="MCP initialize timed out"):
+        await asyncio.wait_for(session.list_tools_async(), timeout=1)