fix(client): propagate SSE POST errors to caller instead of hanging

In `sse_client`, the message-POST coroutine `_send_message` called
`response.raise_for_status()` with no handler. When the server returned a
non-2xx (401/403/404/5xx) or the POST hit a network error, the exception
propagated into the `post_writer` task group and was swallowed by its
`except Exception: logger.exception("Error in post_writer")`. The failure
was never delivered through the read stream, so a caller blocked on
`read_stream.receive()` (e.g. `ClientSession.initialize()`) hung forever.

Catch `httpx.HTTPError` inside `_send_message` and forward it to
`read_stream_writer`, the same pattern stdio.py and websocket.py already
use, and that `streamable_http.py` uses for its >= 400 responses. The
caller now receives the error promptly instead of deadlocking.

Adds a regression test driving a real SSE handshake whose message POST
returns 503; it asserts the caller receives the `HTTPStatusError` via the
read stream within a bounded timeout. The test fails (times out) against
the unpatched client.

Refs #2110

Author: fede-kamel

src/mcp/client/sse.py

@@ -120,15 +120,24 @@ async def post_writer(endpoint_url: str):
 
                         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 httpx.HTTPError as exc:
+                                # Forward the failure to the caller via the read stream instead of
+                                # letting it surface as a swallowed task-group error, which would
+                                # leave read_stream.receive() blocked forever (#2110). Mirrors the
+                                # stream-error handling in stdio.py and streamable_http.py.
+                                logger.exception("Error sending client message")
+                                await read_stream_writer.send(exc)
+                                return
                             logger.debug(f"Client message sent successfully: {response.status_code}")
 
                         async for session_message in write_stream_reader:

tests/shared/test_sse.py

@@ -14,7 +14,7 @@
 from inline_snapshot import snapshot
 from starlette.applications import Starlette
 from starlette.requests import Request
-from starlette.responses import Response
+from starlette.responses import Response, StreamingResponse
 from starlette.routing import Mount, Route
 
 import mcp.client.sse
@@ -25,6 +25,7 @@
 from mcp.server.sse import SseServerTransport
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
+from mcp.shared.message import SessionMessage
 from mcp.types import (
     CallToolRequestParams,
     CallToolResult,
@@ -161,6 +162,66 @@ async def http_client(server: None, server_url: str) -> AsyncGenerator[httpx.Asy
         yield client
 
 
+def make_failing_post_app() -> Starlette:  # pragma: no cover
+    """SSE app that completes the handshake but fails every message POST.
+
+    The `/sse` stream emits a valid `endpoint` event (so the client reaches the
+    POST path) and then stays open; `/messages/` always returns 503. Used to
+    drive the client's POST-error propagation path (#2110).
+    """
+
+    async def handle_sse(request: Request) -> Response:
+        async def event_stream() -> AsyncGenerator[bytes, None]:
+            yield b"event: endpoint\r\ndata: /messages/\r\n\r\n"
+            await anyio.sleep(30)  # keep the SSE connection open for the test's duration
+
+        return StreamingResponse(event_stream(), media_type="text/event-stream")
+
+    async def handle_message(request: Request) -> Response:
+        return Response("upstream exploded", status_code=503)
+
+    return Starlette(
+        routes=[
+            Route("/sse", endpoint=handle_sse),
+            Route("/messages/", endpoint=handle_message, methods=["POST"]),
+        ]
+    )
+
+
+def run_failing_post_server(server_port: int) -> None:  # pragma: no cover
+    app = make_failing_post_app()
+    server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=server_port, log_level="error"))
+    server.run()
+
+
+@pytest.fixture()
+def failing_post_server(server_port: int) -> Generator[None, None, None]:
+    proc = multiprocessing.Process(target=run_failing_post_server, kwargs={"server_port": server_port}, daemon=True)
+    proc.start()
+    wait_for_server(server_port)
+    yield
+    proc.kill()
+    proc.join(timeout=2)
+    if proc.is_alive():  # pragma: no cover
+        print("failing-post server process failed to terminate")
+
+
+@pytest.mark.anyio
+async def test_sse_client_post_error_propagates_to_caller(failing_post_server: None, server_url: str) -> None:
+    """A non-2xx on the message POST must surface to the caller via the read stream.
+
+    Regression test for #2110: previously the error was swallowed by the post_writer
+    task group and `read_stream.receive()` blocked forever.
+    """
+    async with sse_client(server_url + "/sse") as (read_stream, write_stream):
+        await write_stream.send(SessionMessage(types.JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")))
+        with anyio.fail_after(10):
+            item = await read_stream.receive()
+
+    assert isinstance(item, httpx.HTTPStatusError)
+    assert item.response.status_code == 503
+
+
 # Tests
 @pytest.mark.anyio
 async def test_raw_sse_connection(http_client: httpx.AsyncClient) -> None: