fix(client): surface non-2xx POST failures to the awaiting caller (#2110)

src/mcp/client/sse.py

@@ -115,21 +115,49 @@ async def sse_reader(task_status: TaskStatus[str] = anyio.TASK_STATUS_IGNORED):
                 finally:
                     await read_stream_writer.aclose()
 
+            async def _fail_request(message: types.JSONRPCMessage, exc: httpx2.HTTPError) -> None:
+                """Resolve the request this failed POST carried, so its caller raises instead of hanging.
+
+                Correlated by request id rather than fanned out across the session:
+                one rejected POST must not tear down unrelated in-flight requests.
+                A notification has no waiter, so there is nothing to resolve.
+                """
+                if not isinstance(message, types.JSONRPCRequest):
+                    return
+                status = exc.response.status_code if isinstance(exc, httpx2.HTTPStatusError) else None
+                error = types.ErrorData(
+                    code=types.INTERNAL_ERROR,
+                    message=f"Sending the request failed: {exc}",
+                    # The HTTP status is the whole diagnosis for 401/403 vs 5xx, and
+                    # the JSON-RPC code cannot carry it.
+                    data={"http_status": status} if status is not None else None,
+                )
+                reply = SessionMessage(types.JSONRPCError(jsonrpc="2.0", id=message.id, error=error))
+                try:
+                    await read_stream_writer.send(reply)
+                except (anyio.BrokenResourceError, anyio.ClosedResourceError):  # pragma: no cover
+                    logger.debug("read stream closed before request %r could be failed", message.id)
+
             async def post_writer(endpoint_url: str):
                 try:
                     async with write_stream_reader, write_stream:
 
                         async def _send_message(session_message: SessionMessage) -> None:
                             logger.debug(f"Sending client message: {session_message}")
-                            response = await client.post(
-                                endpoint_url,
-                                json=session_message.message.model_dump(
-                                    by_alias=True,
-                                    mode="json",
-                                    exclude_unset=True,
-                                ),
-                            )
-                            response.raise_for_status()
+                            try:
+                                response = await client.post(
+                                    endpoint_url,
+                                    json=session_message.message.model_dump(
+                                        by_alias=True,
+                                        mode="json",
+                                        exclude_unset=True,
+                                    ),
+                                )
+                                response.raise_for_status()
+                            except httpx2.HTTPError as exc:
+                                logger.exception("Error sending client message")
+                                await _fail_request(session_message.message, exc)
+                                return
                             logger.debug(f"Client message sent successfully: {response.status_code}")
 
                         async for session_message in write_stream_reader:

src/mcp/client/streamable_http.py

@@ -367,7 +367,13 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
                         else:
                             error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated")
                     else:
-                        error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response")
+                        # The status rides on `data`: INTERNAL_ERROR alone can't tell a
+                        # caller whether to re-authenticate (401/403) or retry (5xx).
+                        error_data = ErrorData(
+                            code=INTERNAL_ERROR,
+                            message="Server returned an error response",
+                            data={"http_status": response.status_code},
+                        )
                     session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
                     await ctx.read_stream_writer.send(session_message)
                 return

tests/client/test_streamable_http.py

@@ -748,3 +748,28 @@ async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contain
                 _abandoned_request_context(http, send), "evt-7", None, MAX_RECONNECTION_ATTEMPTS
             )
     send.close()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("status_code", [401, 403, 500, 502])
+async def test_non_2xx_post_preserves_http_status(status_code: int) -> None:
+    """The HTTP status survives on `ErrorData.data` so 401/403 stay distinguishable from 5xx.
+
+    The JSON-RPC code cannot carry it: every non-404 failure maps to INTERNAL_ERROR,
+    which on its own says nothing about whether the caller should re-authenticate.
+    """
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(status_code)
+
+    with anyio.fail_after(5):
+        async with (
+            httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http,
+            streamable_http_client("http://test/mcp", http_client=http) as (read, write),
+        ):
+            await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={})))
+            reply = await read.receive()
+
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.error.data == {"http_status": status_code}

tests/shared/test_sse.py

@@ -480,3 +480,42 @@ async def test_sse_session_cleanup_on_disconnect() -> None:
             headers={"Content-Type": "application/json"},
         )
         assert response.status_code == 404
+
+
+def make_post_rejecting_app(status_code: int) -> Starlette:
+    """SSE transport whose message POST endpoint always fails with `status_code`."""
+    sse = SseServerTransport(
+        "/messages/", security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False)
+    )
+    server = Server(SERVER_NAME, on_read_resource=_handle_read_resource)
+
+    async def handle_sse(request: Request) -> Response:
+        async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream):
+            await server.run(read_stream, write_stream, server.create_initialization_options())
+        return Response()
+
+    async def reject_post(request: Request) -> Response:
+        return Response(status_code=status_code)
+
+    return Starlette(
+        routes=[
+            Route("/sse", endpoint=handle_sse),
+            Route("/messages/", endpoint=reject_post, methods=["POST"]),
+        ]
+    )
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("status_code", [401, 403, 500])
+async def test_sse_post_error_reaches_caller(status_code: int) -> None:
+    """A non-2xx POST surfaces to the awaiting caller instead of hanging it forever."""
+    factory = in_process_client_factory(make_post_rejecting_app(status_code))
+
+    with anyio.fail_after(5):
+        async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams:
+            async with ClientSession(*streams) as session:
+                with pytest.raises(MCPError) as exc_info:
+                    await session.initialize()
+
+    # The HTTP status survives on `data` — 401/403 must stay distinguishable from 5xx.
+    assert exc_info.value.error.data == {"http_status": status_code}