fix(client): surface non-2xx HTTP responses to the waiting caller

A 4xx/5xx response to a message POST hit a bare `response.raise_for_status()`
inside the POST's background task. Nothing caught the exception and nothing
resolved the request, so the caller sat until its read timeout fired: a 401 was
indistinguishable from a slow server, and the escaping error tore down the
transport's task group with it.

Non-2xx statuses are now answered rather than raised, generalising the handling
404 already had: a `JSONRPCError` stamped with the original request's id, sent
as a `SessionMessage` over the read stream, so it resolves through the normal
response-correlation path. A spec-compliant JSON-RPC error in the body is
preferred over the status-derived stand-in. The connection stays usable.

This matches the behaviour already shipped on the 2.0 line, with one deliberate
difference: 404 is answered ahead of the body check. A terminated session 404s
*with* a JSON-RPC error body, so preferring it would replace this line's
`Session terminated` / code 32600 — which callers match on for session
expiry — with the server's own wording.

Callers still cannot tell a terminal 401 from a retryable 503; both remain
`-32603 Server returned an error response`, as on 2.0. Carrying the status
would be a follow-up on the 2.0 line.

Github-Issue: #3091
Reported-by: afterrburn

Author: himanshu-commits

src/mcp/client/streamable_http.py

@@ -20,6 +20,7 @@
 from anyio.abc import TaskGroup
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from httpx_sse import EventSource, ServerSentEvent, aconnect_sse
+from pydantic import ValidationError
 from typing_extensions import deprecated
 
 from mcp.shared._httpx_utils import (
@@ -28,6 +29,7 @@
 )
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.types import (
+    INTERNAL_ERROR,
     ErrorData,
     InitializeResult,
     JSONRPCError,
@@ -347,15 +349,14 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
                 logger.debug("Received 202 Accepted")
                 return
 
-            if response.status_code == 404:  # pragma: no branch
+            if response.status_code >= 400:
+                # Raising here would only throw into this POST's background task, where nothing
+                # catches it: the waiting caller would never be told, and would sit until its
+                # read timeout fired. Answer it instead, on the id it is waiting for.
                 if isinstance(message.root, JSONRPCRequest):
-                    await self._send_session_terminated_error(  # pragma: no cover
-                        ctx.read_stream_writer,  # pragma: no cover
-                        message.root.id,  # pragma: no cover
-                    )  # pragma: no cover
-                return  # pragma: no cover
+                    await self._send_http_status_error(response, ctx.read_stream_writer, message.root.id)
+                return
 
-            response.raise_for_status()
             if is_initialization:
                 self._maybe_extract_session_id_from_response(response)
 
@@ -507,19 +508,46 @@ async def _handle_unexpected_content_type(
         logger.error(error_msg)  # pragma: no cover
         await read_stream_writer.send(ValueError(error_msg))  # pragma: no cover
 
-    async def _send_session_terminated_error(
+    async def _send_http_status_error(
         self,
+        response: httpx.Response,
         read_stream_writer: StreamWriter,
         request_id: RequestId,
     ) -> None:
-        """Send a session terminated error response."""
-        jsonrpc_error = JSONRPCError(
-            jsonrpc="2.0",
-            id=request_id,
-            error=ErrorData(code=32600, message="Session terminated"),
-        )
-        session_message = SessionMessage(JSONRPCMessage(jsonrpc_error))
-        await read_stream_writer.send(session_message)
+        """Surface a non-2xx response to the caller that is waiting on `request_id`.
+
+        The error is routed as a `JSONRPCError` stamped with the original request's id, because
+        that id is what the session correlates a reply against; a bare exception carries none and
+        so can never resolve the caller.
+        """
+        if response.status_code == 404:
+            # Answered ahead of the body, unlike every other status: a terminated session 404s
+            # *with* a JSON-RPC error body, and preferring it would rewrite the message and code
+            # this release line already reports. Callers match on both, so 404 stays verbatim.
+            error_data = ErrorData(code=32600, message="Session terminated")
+            await read_stream_writer.send(
+                SessionMessage(JSONRPCMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data)))
+            )
+            return
+
+        # A spec-correct server may put the JSON-RPC error in the body of a non-2xx (e.g. 400 for
+        # INVALID_PARAMS). Prefer it: the server's account of why it refused beats our stand-in.
+        if response.headers.get(CONTENT_TYPE, "").lower().startswith(JSON):
+            try:
+                body = await response.aread()
+                parsed = JSONRPCMessage.model_validate_json(body)
+                if isinstance(parsed.root, JSONRPCError):
+                    # Re-stamped with this request's id rather than trusting the body's: a server
+                    # echoing someone else's id would otherwise resolve the wrong caller.
+                    reply = JSONRPCError(jsonrpc="2.0", id=request_id, error=parsed.root.error)
+                    await read_stream_writer.send(SessionMessage(JSONRPCMessage(reply)))
+                    return
+            except (httpx.StreamError, ValidationError):
+                logger.debug("Non-2xx body was not a JSON-RPC error; using the status-derived fallback")
+
+        error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response")
+        jsonrpc_error = JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data)
+        await read_stream_writer.send(SessionMessage(JSONRPCMessage(jsonrpc_error)))
 
     async def post_writer(
         self,

tests/shared/test_streamable_http.py

@@ -2353,3 +2353,193 @@ async def test_streamablehttp_client_deprecation_warning(basic_server: None, bas
                 await session.initialize()
                 tools = await session.list_tools()
                 assert len(tools.tools) > 0
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("status", [401, 403, 500, 503])
+async def test_non_2xx_status_reaches_the_waiting_caller(status: int) -> None:
+    """A 4xx/5xx to a request is answered as a JSON-RPC error rather than raised into the void.
+
+    Regression test for #3091: the status used to escape `raise_for_status()` inside the POST's
+    background task, where nothing caught it and nothing resolved the caller, so the request hung
+    until its read timeout — a 401 was indistinguishable from a slow server. The `fail_after` below
+    is what pins that: without the fix this test hangs rather than failing an assertion.
+    """
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(status)
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client,
+            streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _),
+            read,
+            write,
+        ):
+            request = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={})
+            await write.send(SessionMessage(types.JSONRPCMessage(request)))
+            reply = await read.receive()
+
+            assert isinstance(reply, SessionMessage)
+            assert isinstance(reply.message.root, types.JSONRPCError)
+            assert reply.message.root.id == 1
+            assert reply.message.root.error == types.ErrorData(
+                code=types.INTERNAL_ERROR, message="Server returned an error response"
+            )
+
+
+@pytest.mark.anyio
+async def test_404_still_reports_session_terminated() -> None:
+    """The 404 spelling is unchanged by #3091: same message, same (positive) code callers match on."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(404)
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client,
+            streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _),
+            read,
+            write,
+        ):
+            request = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={})
+            await write.send(SessionMessage(types.JSONRPCMessage(request)))
+            reply = await read.receive()
+
+            assert isinstance(reply, SessionMessage)
+            assert isinstance(reply.message.root, types.JSONRPCError)
+            assert reply.message.root.error == types.ErrorData(code=32600, message="Session terminated")
+
+
+@pytest.mark.anyio
+async def test_non_2xx_prefers_the_servers_own_jsonrpc_error_body() -> None:
+    """When the server explains itself in the body, the caller gets that, not our stand-in.
+
+    The reply is re-stamped with the request's id rather than trusting the body's, so a server that
+    echoes a different id cannot resolve the wrong caller (or, echoing none, no caller at all).
+    """
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        error = {"code": types.INVALID_PARAMS, "message": "Unknown tool: nope"}
+        return httpx.Response(400, json={"jsonrpc": "2.0", "id": 99, "error": error})
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client,
+            streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _),
+            read,
+            write,
+        ):
+            request = types.JSONRPCRequest(jsonrpc="2.0", id=7, method="tools/call", params={})
+            await write.send(SessionMessage(types.JSONRPCMessage(request)))
+            reply = await read.receive()
+
+            assert isinstance(reply, SessionMessage)
+            assert isinstance(reply.message.root, types.JSONRPCError)
+            assert reply.message.root.id == 7
+            assert reply.message.root.error == types.ErrorData(code=types.INVALID_PARAMS, message="Unknown tool: nope")
+
+
+@pytest.mark.anyio
+async def test_a_failed_request_leaves_the_session_usable() -> None:
+    """The connection survives a 500: the error resolves one call, it does not tear down the transport.
+
+    In v1 the escaping `HTTPStatusError` cancelled the transport's task group, so a single 500 took
+    the whole session with it.
+    """
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        request_id = json.loads(request.content)["id"]
+        if request_id == 1:
+            return httpx.Response(500)
+        return httpx.Response(200, json={"jsonrpc": "2.0", "id": request_id, "result": {}})
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client,
+            streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _),
+            read,
+            write,
+        ):
+            failing = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={})
+            await write.send(SessionMessage(types.JSONRPCMessage(failing)))
+            first = await read.receive()
+
+            following = types.JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/list", params={})
+            await write.send(SessionMessage(types.JSONRPCMessage(following)))
+            second = await read.receive()
+
+            assert isinstance(first, SessionMessage)
+            assert isinstance(first.message.root, types.JSONRPCError)
+            assert isinstance(second, SessionMessage)
+            assert isinstance(second.message.root, types.JSONRPCResponse)
+
+
+@pytest.mark.anyio
+async def test_non_2xx_to_a_notification_resolves_nobody_and_keeps_the_session_alive() -> None:
+    """A notification has no id and no waiter, so a 4xx to one cannot be surfaced — only survived.
+
+    The following request still gets its answer: the rejected notification must not take the
+    transport down with it.
+    """
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = json.loads(request.content)
+        if "id" not in body:
+            return httpx.Response(500)
+        return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}})
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client,
+            streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _),
+            read,
+            write,
+        ):
+            notification = types.JSONRPCNotification(jsonrpc="2.0", method="notifications/progress")
+            await write.send(SessionMessage(types.JSONRPCMessage(notification)))
+
+            request = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={})
+            await write.send(SessionMessage(types.JSONRPCMessage(request)))
+            reply = await read.receive()
+
+            assert isinstance(reply, SessionMessage)
+            assert isinstance(reply.message.root, types.JSONRPCResponse)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("body", "case"),
+    [
+        ({"jsonrpc": "2.0", "id": 1, "result": {}}, "valid JSON-RPC, but not an error"),
+        ({"detail": "gateway rejected the request"}, "JSON, but not JSON-RPC at all"),
+    ],
+)
+async def test_a_json_body_that_is_not_a_jsonrpc_error_falls_back_to_the_stand_in(
+    body: dict[str, Any], case: str
+) -> None:
+    """`Content-Type: application/json` is not a promise of a JSON-RPC error body.
+
+    Proxies and gateways send JSON that is nothing of the sort. Neither shape may be handed to the
+    caller as their answer, and neither may be allowed to escape: both fall back to the stand-in.
+    """
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, json=body)
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client,
+            streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _),
+            read,
+            write,
+        ):
+            request = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={})
+            await write.send(SessionMessage(types.JSONRPCMessage(request)))
+            reply = await read.receive()
+
+            assert isinstance(reply, SessionMessage)
+            assert isinstance(reply.message.root, types.JSONRPCError)
+            assert reply.message.root.error == types.ErrorData(
+                code=types.INTERNAL_ERROR, message="Server returned an error response"
+            )