fix(client): surface streamable http transport errors

Author: go165

src/mcp/client/streamable_http.py

@@ -468,11 +468,19 @@ async def _handle_message(session_message: SessionMessage) -> None:
                         read_stream_writer=read_stream_writer,
                     )
 
-                    async def handle_request_async():
-                        if is_resumption:
-                            await self._handle_resumption_request(ctx)
-                        else:
-                            await self._handle_post_request(ctx)
+                    async def handle_request_async() -> None:
+                        try:
+                            if is_resumption:
+                                await self._handle_resumption_request(ctx)
+                            else:
+                                await self._handle_post_request(ctx)
+                        except httpx.HTTPError as exc:
+                            logger.exception("Transport error handling request")
+                            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 read_stream_writer.send(error_msg)
 
                     # If this is a request, start a new task to handle it
                     if isinstance(message, JSONRPCRequest):

tests/issues/test_915_streamable_http_unreachable.py

@@ -0,0 +1,50 @@
+from typing import cast
+
+import anyio
+import httpx
+import pytest
+
+from mcp.client.session_group import ClientSessionGroup, StreamableHttpParameters
+from mcp.shared.exceptions import MCPError
+
+pytestmark = pytest.mark.anyio
+
+
+def _contains_cancel_scope_error(exc: BaseException) -> bool:
+    if isinstance(exc, RuntimeError) and "Attempted to exit cancel scope" in str(exc):
+        return True
+
+    raw_grouped_exceptions = getattr(exc, "exceptions", ())
+    if isinstance(raw_grouped_exceptions, tuple):
+        grouped_exceptions = cast(tuple[BaseException, ...], raw_grouped_exceptions)
+        return any(_contains_cancel_scope_error(inner) for inner in grouped_exceptions)
+
+    return any(_contains_cancel_scope_error(inner) for inner in (exc.__cause__, exc.__context__) if inner is not None)
+
+
+async def test_session_group_streamable_http_connect_error_is_catchable(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def raise_connect_error(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("server unavailable", request=request)
+
+    def mock_http_client(
+        headers: dict[str, str] | None = None,
+        timeout: httpx.Timeout | None = None,
+        auth: httpx.Auth | None = None,
+    ) -> httpx.AsyncClient:
+        return httpx.AsyncClient(
+            auth=auth,
+            headers=headers,
+            timeout=timeout,
+            transport=httpx.MockTransport(raise_connect_error),
+        )
+
+    monkeypatch.setattr("mcp.client.session_group.create_mcp_http_client", mock_http_client)
+
+    async with ClientSessionGroup() as group:
+        with anyio.fail_after(5), pytest.raises(MCPError) as exc_info:
+            await group.connect_to_server(StreamableHttpParameters(url="http://example.invalid/mcp"))
+
+    assert "Transport error: server unavailable" in exc_info.value.error.message
+    assert not _contains_cancel_scope_error(exc_info.value)