fix(client): contain arbitrary POST failures and dedupe the status-error mapping

Address the third review round:

- Broaden the SSE message POST's failure catch to a terminal containment
  boundary (except Exception): user-supplied auth flows and hooks raise
  arbitrary types from inside client.post(), so an enumerated catch cannot
  keep the caller from hanging.
- Extract the status -> JSON-RPC error mapping into
  mcp.client._transport.status_error_data and use it from the message POST
  handler, the resumption GET, and the SSE POST; the message POST keeps its
  pre-session 404 -> METHOD_NOT_FOUND case locally. Wire-identical.
- Contain the SSE error-resolution send against a concurrently closed read
  stream (BrokenResourceError/ClosedResourceError -> debug log), mirroring
  _resolve_abandoned_request, so the teardown race cannot kill the write
  loop.

Tests: the auth-failure test is parametrized over OAuthTokenError and
RuntimeError, and a raw-stream teardown-race test pins that a failing
POST whose error is undeliverable leaves the write loop serving later
messages. Both fail against the previous revision.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AuJi8kEB3bhikW2pzbmhUL

Author: claude

src/mcp/client/_transport.py

@@ -5,6 +5,8 @@
 from contextlib import AbstractAsyncContextManager
 from typing import Protocol
 
+from mcp_types import INTERNAL_ERROR, INVALID_REQUEST, ErrorData
+
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.message import SessionMessage
 
@@ -13,6 +15,18 @@
 TransportStreams = tuple[ReadStream[SessionMessage | Exception], WriteStream[SessionMessage]]
 
 
+def status_error_data(status_code: int, *, has_session: bool) -> ErrorData:
+    """Map a non-2xx HTTP status on a client transport request to the error its waiting caller receives.
+
+    A 404 while a session is held is the session-expiry signal (`INVALID_REQUEST`,
+    "Session terminated"); anything else gets the generic stand-in. A call site with
+    an extra status mapping (e.g. the message POST's pre-session 404) branches first.
+    """
+    if status_code == 404 and has_session:
+        return ErrorData(code=INVALID_REQUEST, message="Session terminated")
+    return ErrorData(code=INTERNAL_ERROR, message="Server returned an error response")
+
+
 class Transport(AbstractAsyncContextManager[TransportStreams], Protocol):
     """Protocol for MCP transports.
 

src/mcp/client/sse.py

@@ -10,7 +10,7 @@
 from anyio.abc import TaskStatus
 from httpx2 import SSEError
 
-from mcp.client.auth.exceptions import OAuthFlowError
+from mcp.client._transport import status_error_data
 from mcp.shared._compat import resync_tracer
 from mcp.shared._context_streams import create_context_streams
 from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client
@@ -137,9 +137,11 @@ async def _send_message(session_message: SessionMessage) -> None:
                                         exclude_unset=True,
                                     ),
                                 )
-                            except (httpx2.HTTPError, OAuthFlowError) as exc:
-                                # OAuthFlowError: OAuthClientProvider re-auth failing inside
-                                # client.post() must resolve the waiter like any network error.
+                            except Exception as exc:
+                                # Terminal containment boundary: beyond httpx's own errors,
+                                # user-supplied auth flows and hooks can raise arbitrary types
+                                # from inside `client.post()`, so an enumerated catch cannot
+                                # keep the caller from hanging.
                                 logger.exception("Error POSTing message")
                                 error = types.ErrorData(
                                     code=types.CONNECTION_CLOSED, message=f"Failed to send message: {exc}"
@@ -149,21 +151,22 @@ async def _send_message(session_message: SessionMessage) -> None:
                                     logger.debug(f"Client message sent successfully: {response.status_code}")
                                     return
                                 logger.error(f"Message POST returned HTTP status {response.status_code}")
-                                if (
-                                    response.status_code == 404
-                                    and _extract_session_id_from_endpoint(endpoint_url) is not None
-                                ):
-                                    # The endpoint URL carries the session id, so a 404 is the
-                                    # session-expiry signal - same mapping as streamable HTTP.
-                                    error = types.ErrorData(code=types.INVALID_REQUEST, message="Session terminated")
-                                else:
-                                    error = types.ErrorData(
-                                        code=types.INTERNAL_ERROR, message="Server returned an error response"
-                                    )
+                                # The endpoint URL carrying a session id is this transport's
+                                # "session established" signal, as `self.session_id` is for
+                                # streamable HTTP.
+                                error = status_error_data(
+                                    response.status_code,
+                                    has_session=_extract_session_id_from_endpoint(endpoint_url) is not None,
+                                )
                             # A notification has no waiter to resolve, so its failure is only logged.
                             if isinstance(message, types.JSONRPCRequest):
                                 reply = types.JSONRPCError(jsonrpc="2.0", id=message.id, error=error)
-                                await read_stream_writer.send(SessionMessage(reply))
+                                try:
+                                    await read_stream_writer.send(SessionMessage(reply))
+                                except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+                                    # Teardown race: the reader is gone, so there is nobody
+                                    # left to resolve - contain it, keeping the write loop up.
+                                    logger.debug("read stream closed before request %r could be resolved", message.id)
 
                         async for session_message in write_stream_reader:
                             sender_ctx = write_stream_reader.last_context

src/mcp/client/streamable_http.py

@@ -14,7 +14,6 @@
 from httpx2 import EventSource, ServerSentEvent
 from mcp_types import (
     CONNECTION_CLOSED,
-    INTERNAL_ERROR,
     INVALID_REQUEST,
     METHOD_NOT_FOUND,
     PARSE_ERROR,
@@ -30,7 +29,7 @@
 from mcp_types.version import MODERN_PROTOCOL_VERSIONS
 from pydantic import ValidationError
 
-from mcp.client._transport import TransportStreams
+from mcp.client._transport import TransportStreams, status_error_data
 from mcp.shared._compat import resync_tracer
 from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream, create_context_streams
 from mcp.shared._httpx_utils import create_mcp_http_client
@@ -259,18 +258,11 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
                     # Resolve the waiting caller with an error correlated to its request,
                     # mirroring `_handle_post_request`: an escaping `HTTPStatusError` would
                     # tear down the transport's task group and every stream with it (#2110).
-                    if event_source.response.status_code == 404 and self.session_id is not None:
-                        # The GET carried our Mcp-Session-Id, so a 404 is the session-expiry
-                        # signal reconnect logic keys on - same mapping as the POST path.
-                        await self._resolve_abandoned_request(
-                            ctx.read_stream_writer, original_request_id, "Session terminated", code=INVALID_REQUEST
-                        )
-                        return
+                    error_data = status_error_data(
+                        event_source.response.status_code, has_session=self.session_id is not None
+                    )
                     await self._resolve_abandoned_request(
-                        ctx.read_stream_writer,
-                        original_request_id,
-                        "Server returned an error response",
-                        code=INTERNAL_ERROR,
+                        ctx.read_stream_writer, original_request_id, error_data.message, code=error_data.code
                     )
                     return
                 logger.debug("Resumption GET SSE connection established")
@@ -384,16 +376,13 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
                         except (httpx2.StreamError, ValidationError):
                             pass
                         logger.debug("Non-2xx body was not a JSON-RPC error; using fallback")
-                    if response.status_code == 404:
-                        if self.session_id is None:
-                            # No session yet → 404 is the HTTP-level spelling of
-                            # METHOD_NOT_FOUND (gateway / legacy server doesn't know
-                            # this method); "Session terminated" would be a lie here.
-                            error_data = ErrorData(code=METHOD_NOT_FOUND, message="Not Found")
-                        else:
-                            error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated")
+                    if response.status_code == 404 and self.session_id is None:
+                        # No session yet → 404 is the HTTP-level spelling of
+                        # METHOD_NOT_FOUND (gateway / legacy server doesn't know
+                        # this method); "Session terminated" would be a lie here.
+                        error_data = ErrorData(code=METHOD_NOT_FOUND, message="Not Found")
                     else:
-                        error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response")
+                        error_data = status_error_data(response.status_code, has_session=self.session_id is not None)
                     session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data))
                     await ctx.read_stream_writer.send(session_message)
                 return

tests/shared/test_sse.py

@@ -19,6 +19,7 @@
     EmptyResult,
     Implementation,
     InitializeResult,
+    JSONRPCRequest,
     JSONRPCResponse,
     ListToolsResult,
     PaginatedRequestParams,
@@ -44,6 +45,7 @@
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared._httpx_utils import McpHttpClientFactory
 from mcp.shared.exceptions import MCPError
+from mcp.shared.message import SessionMessage
 from tests.interaction.transports import StreamingASGITransport
 
 SERVER_NAME = "test_server_for_SSE"
@@ -392,17 +394,21 @@ async def handle_post(request: Request) -> Response:
 
 
 @pytest.mark.anyio
-async def test_sse_client_oauth_failure_on_post_reaches_caller_and_session_survives() -> None:
-    """An SDK OAuth flow failure raised from inside a request's message POST reaches the waiting
-    caller promptly as a JSON-RPC error correlated to the request, and the session stays usable
+@pytest.mark.parametrize("exc_type", [OAuthTokenError, RuntimeError])
+async def test_sse_client_auth_failure_on_post_reaches_caller_and_session_survives(
+    exc_type: type[Exception],
+) -> None:
+    """A failure raised from inside a request's message POST by a user-supplied hook — an SDK
+    OAuth flow error, or any exception from a custom auth flow — reaches the waiting caller
+    promptly as a JSON-RPC error correlated to the request, and the session stays usable
     (SDK-defined; #2110 — like any network error, it used to be swallowed inside post_writer)."""
 
     class _RefusingAuth(httpx2.Auth):
-        """Stands in for OAuthClientProvider whose re-auth fails mid-session."""
+        """Stands in for OAuthClientProvider (or any user auth hook) failing mid-session."""
 
         async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
             if request.method == "POST" and json.loads(request.content).get("method") == "resources/read":
-                raise OAuthTokenError("re-authentication failed")
+                raise exc_type("re-authentication failed")
             yield request
 
     factory = in_process_client_factory(make_server_app())
@@ -423,6 +429,44 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
             assert isinstance(await session.send_ping(), EmptyResult)
 
 
+@pytest.mark.anyio
+async def test_sse_client_post_error_after_reader_closed_is_contained() -> None:
+    """A failing POST whose error can no longer be delivered — the read stream already closed
+    with the server's SSE stream — is contained: the write loop survives and later messages
+    still reach the server (SDK-defined teardown-race guard). Raw streams, because the race
+    needs the read side closed while the write side keeps sending."""
+    posted: list[str] = []
+    second_post = anyio.Event()
+
+    async def handle_sse(request: Request) -> StreamingResponse:
+        async def stream() -> AsyncGenerator[str, None]:
+            # The stream ends right after the endpoint event: the client's reader
+            # observes EOF and closes the read stream.
+            yield "event: endpoint\ndata: /messages/\n\n"
+
+        return StreamingResponse(stream(), media_type="text/event-stream")
+
+    async def handle_post(request: Request) -> Response:
+        posted.append(json.loads(await request.body())["method"])
+        if len(posted) == 2:
+            second_post.set()
+        return Response(status_code=500)
+
+    app = Starlette(routes=[Route("/sse", handle_sse), Route("/messages/", handle_post, methods=["POST"])])
+    factory = in_process_client_factory(app)
+    with anyio.fail_after(5):
+        async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as (read, write):
+            # Wait for the reader to observe the server's EOF and close the read stream.
+            with pytest.raises(anyio.EndOfStream):
+                await read.receive()
+
+            await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="first/call", params={})))
+            await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=2, method="second/call", params={})))
+            await second_post.wait()
+    # The first POST's undeliverable error was contained; the second still went out.
+    assert posted == ["first/call", "second/call"]
+
+
 @pytest.mark.anyio
 async def test_sse_client_notification_post_http_error_leaves_session_usable() -> None:
     """A non-2xx on a notification's message POST resolves no caller (a notification has no