fix: surface streamable http request failures

Author: Genmin

src/mcp/client/streamable_http.py

@@ -429,6 +429,20 @@ async def _handle_reconnection(
             # Try to reconnect again if we still have an event ID
             await self._handle_reconnection(ctx, last_event_id, retry_interval_ms, attempt + 1)
 
+    async def _handle_request_error(self, ctx: RequestContext, exc: Exception) -> None:
+        """Report a request transport failure without crashing the transport task group."""
+        logger.debug("Error handling StreamableHTTP request", exc_info=True)
+
+        message = ctx.session_message.message
+        if isinstance(message, JSONRPCRequest):
+            error_data = ErrorData(code=INTERNAL_ERROR, message=f"Transport error: {exc}")
+            error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
+            with contextlib.suppress(anyio.BrokenResourceError, anyio.ClosedResourceError):
+                await ctx.read_stream_writer.send(error_msg)
+        else:
+            with contextlib.suppress(anyio.BrokenResourceError, anyio.ClosedResourceError):
+                await ctx.read_stream_writer.send(exc)
+
     async def post_writer(
         self,
         client: httpx.AsyncClient,
@@ -468,10 +482,13 @@ async def _handle_message(session_message: SessionMessage) -> None:
                     )
 
                     async def handle_request_async():
-                        if is_resumption:
-                            await self._handle_resumption_request(ctx)
-                        else:
-                            await self._handle_post_request(ctx)
+                        try:
+                            if is_resumption:
+                                await self._handle_resumption_request(ctx)
+                            else:
+                                await self._handle_post_request(ctx)
+                        except Exception as exc:
+                            await self._handle_request_error(ctx, exc)
 
                     # If this is a request, start a new task to handle it
                     if isinstance(message, JSONRPCRequest):

tests/client/test_notification_response.py

@@ -6,6 +6,7 @@
 
 import json
 
+import anyio
 import httpx
 import pytest
 from starlette.applications import Starlette
@@ -152,6 +153,19 @@ async def test_http_error_status_sends_jsonrpc_error() -> None:
                     await session.list_tools()
 
 
+async def test_transport_error_sends_jsonrpc_error() -> None:
+    """Verify request transport errors unblock the pending request with an MCPError."""
+
+    async def raise_connect_error(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("All connection attempts failed", request=request)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(raise_connect_error)) as client:
+        async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
+                with pytest.raises(MCPError, match="Transport error: All connection attempts failed"):
+                    await session.initialize()
+
+
 async def test_http_error_on_notification_does_not_hang() -> None:
     """Verify HTTP errors on notifications are silently ignored.
 
@@ -168,6 +182,23 @@ async def test_http_error_on_notification_does_not_hang() -> None:
                 await session.send_notification(RootsListChangedNotification(method="notifications/roots/list_changed"))
 
 
+async def test_transport_error_on_notification_does_not_crash_transport() -> None:
+    """Verify transport errors on notifications do not crash the transport task group."""
+
+    async def handle_request(request: httpx.Request) -> httpx.Response:
+        data = json.loads(request.content)
+        if data.get("method") == "initialize":
+            return httpx.Response(200, json={"jsonrpc": "2.0", "id": data["id"], "result": INIT_RESPONSE})
+        raise httpx.ConnectError("All connection attempts failed", request=request)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handle_request)) as client:
+        async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
+                await session.initialize()
+                await session.send_notification(RootsListChangedNotification(method="notifications/roots/list_changed"))
+                await anyio.sleep(0)
+
+
 def _create_invalid_json_response_app() -> Starlette:
     """Create a server that returns invalid JSON for requests."""
 

tests/client/test_session_group.py

@@ -385,3 +385,24 @@ async def test_client_session_group_establish_session_parameterized(
             # 3. Assert returned values
             assert returned_server_info is mock_initialize_result.server_info
             assert returned_session is mock_entered_session
+
+
+@pytest.mark.anyio
+async def test_client_session_group_streamable_http_connect_error_is_catchable() -> None:
+    async def raise_connect_error(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("All connection attempts failed", request=request)
+
+    def mock_client_factory(
+        headers: dict[str, str] | None = None,
+        timeout: httpx.Timeout | None = None,
+        auth: httpx.Auth | None = None,
+    ) -> httpx.AsyncClient:
+        return httpx.AsyncClient(transport=httpx.MockTransport(raise_connect_error))
+
+    group = ClientSessionGroup()
+    async with group:
+        with mock.patch("mcp.client.session_group.create_mcp_http_client", side_effect=mock_client_factory):
+            with pytest.raises(MCPError, match="Transport error: All connection attempts failed"):
+                await group.connect_to_server(StreamableHttpParameters(url="http://localhost:3001/mcp/"))
+
+    assert group.sessions == []