fix(client): cover OAuth failures and 404 session expiry on the SSE POST; document the new error contract

Address the second review round:

- Widen the SSE message POST's failure catch to (httpx.HTTPError,
  OAuthFlowError): an OAuthClientProvider re-auth failing inside
  client.post() previously took the same swallowed path and hung the
  waiting caller forever.
- Map a 404 on the SSE message POST to INVALID_REQUEST / "Session
  terminated" when the endpoint URL carries a session id (the SSE
  analogue of the streamable transport's session check); keep the
  generic error when it does not.
- Document the changed error behavior in docs/migration.md: resumption
  GET and SSE message POST outcome tables, and scope the "connect-level
  failures still escape" sentence to the streamable message POST, the
  one place it still holds.

Three new regression tests; the OAuth and 404-session ones fail against
the previous revision.

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

Author: claude

docs/migration.md

@@ -2188,6 +2188,25 @@ In v1, a non-2xx response to a message POST (other than 404) raised `httpx.HTTPS
 | 404, no session yet | `McpError` with positive code `32600` | `MCPError(-32601, 'Not Found')` |
 | Any other 4xx/5xx | `httpx.HTTPStatusError` escapes as `ExceptionGroup` | `MCPError(-32603, 'Server returned an error response')` |
 
+The same contract covers the resumption GET — a request re-attached with a resumption token (`Last-Event-ID`). In v1 a failure there escaped the context as an `ExceptionGroup` that failed every pending request, or hung the resumed call forever:
+
+| Resumption GET outcome | v1 | v2 |
+| --- | --- | --- |
+| 404, session established | `httpx.HTTPStatusError` escapes as `ExceptionGroup` | `MCPError(-32600, 'Session terminated')` |
+| Any other non-2xx | `httpx.HTTPStatusError` escapes as `ExceptionGroup` | `MCPError(-32603, 'Server returned an error response')` |
+| Stream drops mid-read | error escapes as `ExceptionGroup` | `MCPError(-32000, 'resumption stream ended without a response')` |
+| Stream ends cleanly with no response | resumed call hangs forever | `MCPError(-32000, 'resumption stream ended without a response')` |
+
+The SSE transport (`sse_client`) applies the same rule to its message POST. In v1 *any* POST failure — a non-2xx status, a network error, or an OAuth re-auth failure raised by the configured `auth` — was caught and logged inside the transport's writer task: the waiting caller hung forever and the write loop died, so every later send was silently dropped. In v2 the failing request resolves promptly and the session stays usable:
+
+| Message POST outcome | v1 | v2 |
+| --- | --- | --- |
+| 404, endpoint URL carries a session id | caller hangs forever; write loop dies | `MCPError(-32600, 'Session terminated')` |
+| Any other non-2xx | caller hangs forever; write loop dies | `MCPError(-32603, 'Server returned an error response')` |
+| Network-level failure (`httpx2.ConnectError`, timeouts) or OAuth flow failure | caller hangs forever; write loop dies | `MCPError(-32000, 'Failed to send message: ...')` |
+
+A failed POST of a *notification* has no caller to resolve; v2 logs and drops it, keeping the write loop (and every later send) alive.
+
 Both common v1 patterns silently stop working: an `except* httpx.HTTPStatusError` around the transport context becomes dead code because status errors no longer escape the context, and a session-expiry check on `error.code == 32600` never matches again because the code is now the standard negative `-32600`.
 
 **Before (v1):**
@@ -2230,7 +2249,7 @@ async with streamable_http_client(url) as (read, write):
                 raise
 ```
 
-Move HTTP-status failure handling from around the transport context to around the individual calls, catching `MCPError` (see [`McpError` renamed to `MCPError`](#mcperror-renamed-to-mcperror)). Connect-level failures such as `httpx2.ConnectError` still escape the transport context as before; keep context-level handling for those only.
+Move HTTP-status failure handling from around the transport context to around the individual calls, catching `MCPError` (see [`McpError` renamed to `MCPError`](#mcperror-renamed-to-mcperror)). On the streamable HTTP transport, connect-level failures such as `httpx2.ConnectError` on a message POST still escape the transport context as before — keep context-level handling for those; on the resumption GET and on the SSE transport's message POST they resolve the failing request instead, as above.
 
 ### `terminate_windows_process` removed
 

src/mcp/client/sse.py

@@ -10,6 +10,7 @@
 from anyio.abc import TaskStatus
 from httpx2 import SSEError
 
+from mcp.client.auth.exceptions import OAuthFlowError
 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
@@ -136,7 +137,9 @@ async def _send_message(session_message: SessionMessage) -> None:
                                         exclude_unset=True,
                                     ),
                                 )
-                            except httpx2.HTTPError as exc:
+                            except (httpx2.HTTPError, OAuthFlowError) as exc:
+                                # OAuthFlowError: OAuthClientProvider re-auth failing inside
+                                # client.post() must resolve the waiter like any network error.
                                 logger.exception("Error POSTing message")
                                 error = types.ErrorData(
                                     code=types.CONNECTION_CLOSED, message=f"Failed to send message: {exc}"
@@ -146,9 +149,17 @@ 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}")
-                                error = types.ErrorData(
-                                    code=types.INTERNAL_ERROR, message="Server returned an error response"
-                                )
+                                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"
+                                    )
                             # 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)

tests/shared/test_sse.py

@@ -31,11 +31,12 @@
 )
 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
 from starlette.types import ASGIApp, Receive, Scope, Send
 
 import mcp.client.sse
+from mcp.client.auth.exceptions import OAuthTokenError
 from mcp.client.session import ClientSession
 from mcp.client.sse import _extract_session_id_from_endpoint, sse_client
 from mcp.server import Server, ServerRequestContext
@@ -340,6 +341,88 @@ def factory(
             assert isinstance(await session.send_ping(), EmptyResult)
 
 
+@pytest.mark.anyio
+async def test_sse_client_post_404_with_session_endpoint_reports_session_terminated() -> None:
+    """A 404 on a request's message POST while the endpoint URL carries a session id reports
+    "Session terminated" (INVALID_REQUEST) to the caller, the same session-expiry mapping as
+    the streamable HTTP transport (SDK-defined)."""
+    factory = in_process_client_factory(make_app_rejecting_posts({"resources/read": 404}))
+    with anyio.fail_after(5):
+        # One parenthesized async-with: separately nested ones trip a phantom
+        # branch arc under coverage on Python 3.14 (see the note in mcp.client.sse).
+        async with (
+            sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams,
+            ClientSession(*streams) as session,
+        ):
+            await session.initialize()
+
+            with pytest.raises(MCPError) as exc_info:
+                await session.read_resource(uri="foobar://should-work")
+            assert exc_info.value.error.code == types.INVALID_REQUEST
+            assert exc_info.value.error.message == snapshot("Session terminated")
+
+
+@pytest.mark.anyio
+async def test_sse_client_post_404_without_session_endpoint_keeps_generic_error() -> None:
+    """A 404 on a request's message POST when the endpoint URL carries no session id keeps the
+    generic error: with no session to expire, "Session terminated" would be a lie (SDK-defined).
+    The raw endpoint is scripted because `SseServerTransport` always issues a session id."""
+
+    async def handle_sse(request: Request) -> StreamingResponse:
+        async def stream() -> AsyncGenerator[str, None]:
+            yield "event: endpoint\ndata: /messages/\n\n"
+            await anyio.Event().wait()  # park until the client disconnects
+
+        return StreamingResponse(stream(), media_type="text/event-stream")
+
+    async def handle_post(request: Request) -> Response:
+        return Response(status_code=404)
+
+    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 streams,
+            ClientSession(*streams) as session,
+        ):
+            with pytest.raises(MCPError) as exc_info:
+                await session.initialize()
+            assert exc_info.value.error.code == types.INTERNAL_ERROR
+            assert exc_info.value.error.message == snapshot("Server returned an error 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
+    (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."""
+
+        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")
+            yield request
+
+    factory = in_process_client_factory(make_server_app())
+    with anyio.fail_after(5):
+        async with (
+            sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory, auth=_RefusingAuth()) as streams,
+            ClientSession(*streams) as session,
+        ):
+            await session.initialize()
+
+            with pytest.raises(MCPError) as exc_info:
+                await session.read_resource(uri="foobar://should-work")
+            assert exc_info.value.error.code == types.CONNECTION_CLOSED
+            # The message embeds the auth exception's text; pin only the SDK-authored prefix.
+            assert exc_info.value.error.message.startswith("Failed to send message:")
+
+            # The session survived the failed POST: the next request round-trips.
+            assert isinstance(await session.send_ping(), EmptyResult)
+
+
 @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