fix(client): harden the resumption GET and SSE POST error paths per review

Address the review findings on the previous revision:

- Dispatch resumption on message type as well as metadata: a notification
  stamped with a resumption token is POSTed as usual instead of tripping
  the resumption path's request-only assertion and killing the write loop.
- Treat any non-2xx as a failure (response.is_success), restoring the
  raise_for_status() semantics the checks replaced: an unfollowed redirect
  resolves the caller instead of being logged as success.
- Map a 404 on the resumption GET while a session id is held to
  INVALID_REQUEST / "Session terminated", the POST path's session-expiry
  signal, so reconnect logic keyed on it works across both.
- Contain the resumption read loop like _handle_sse_response: a stream
  dying mid-read or ending cleanly without a response resolves the waiter
  (CONNECTION_CLOSED) instead of tearing down the transport or hanging.
- Resolve the resumption GET's status errors via _resolve_abandoned_request
  for its closed-stream containment instead of hand-building the error.
- Surface network-level errors (httpx.HTTPError) on the SSE message POST
  through the same correlated path: on this transport nothing escapes
  loudly, so the caller previously hung forever.
- Deduplicate the SSE test app wiring behind make_app(wrap_post=...).

Seven new regression tests pin the above; each fails against the previous
revision (hang into fail_after, transport teardown, or wrong error).

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

Author: claude

src/mcp/client/sse.py

@@ -120,31 +120,39 @@ async def post_writer(endpoint_url: str):
                     async with write_stream_reader, write_stream:
 
                         async def _send_message(session_message: SessionMessage) -> None:
+                            # A POST failure must not raise: the post_writer handler below
+                            # would swallow it, hanging the waiting caller forever and killing
+                            # the write loop (#2110). Mirror the streamable-HTTP transport
+                            # instead: resolve the waiter with an error correlated to its
+                            # request id, keeping the session usable.
                             logger.debug(f"Sending client message: {session_message}")
                             message = session_message.message
-                            response = await client.post(
-                                endpoint_url,
-                                json=message.model_dump(
-                                    by_alias=True,
-                                    mode="json",
-                                    exclude_unset=True,
-                                ),
-                            )
-                            if response.status_code >= 400:
-                                # Resolve the waiting caller with an error correlated to its
-                                # request id, mirroring the streamable-HTTP transport: raising
-                                # here would be swallowed by the post_writer handler below and
-                                # the caller would hang forever (#2110). A notification has no
-                                # waiter to resolve, so the failure is only logged.
+                            try:
+                                response = await client.post(
+                                    endpoint_url,
+                                    json=message.model_dump(
+                                        by_alias=True,
+                                        mode="json",
+                                        exclude_unset=True,
+                                    ),
+                                )
+                            except httpx2.HTTPError as exc:
+                                logger.exception("Error POSTing message")
+                                error = types.ErrorData(
+                                    code=types.CONNECTION_CLOSED, message=f"Failed to send message: {exc}"
+                                )
+                            else:
+                                if response.is_success:
+                                    logger.debug(f"Client message sent successfully: {response.status_code}")
+                                    return
                                 logger.error(f"Message POST returned HTTP status {response.status_code}")
-                                if isinstance(message, types.JSONRPCRequest):
-                                    error_data = types.ErrorData(
-                                        code=types.INTERNAL_ERROR, message="Server returned an error response"
-                                    )
-                                    reply = types.JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)
-                                    await read_stream_writer.send(SessionMessage(reply))
-                                return
-                            logger.debug(f"Client message sent successfully: {response.status_code}")
+                                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)
+                                await read_stream_writer.send(SessionMessage(reply))
 
                         async for session_message in write_stream_reader:
                             sender_ctx = write_stream_reader.last_context

src/mcp/client/streamable_http.py

@@ -248,32 +248,51 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
         else:
             raise ResumptionError("Resumption request requires a resumption token")  # pragma: no cover
 
-        # Only requests resume: a resumption token is only ever attached by a
-        # request's metadata, so the original id is always available to map responses.
+        # Only requests resume: post_writer dispatches here on message type as well as
+        # metadata, so the original id is always available to map responses.
         assert isinstance(ctx.session_message.message, JSONRPCRequest)
         original_request_id = ctx.session_message.message.id
 
-        async with ctx.client.sse(self.url, headers=headers) as event_source:
-            if event_source.response.status_code >= 400:
-                # 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).
-                error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response")
-                reply = JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data)
-                await ctx.read_stream_writer.send(SessionMessage(reply))
-                return
-            logger.debug("Resumption GET SSE connection established")
+        try:
+            async with ctx.client.sse(self.url, headers=headers) as event_source:
+                if not event_source.response.is_success:
+                    # 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
+                    await self._resolve_abandoned_request(
+                        ctx.read_stream_writer,
+                        original_request_id,
+                        "Server returned an error response",
+                        code=INTERNAL_ERROR,
+                    )
+                    return
+                logger.debug("Resumption GET SSE connection established")
 
-            async for sse in event_source:  # pragma: no branch
-                is_complete = await self._handle_sse_event(
-                    sse,
-                    ctx.read_stream_writer,
-                    original_request_id,
-                    ctx.metadata.on_resumption_token_update if ctx.metadata else None,
-                )
-                if is_complete:
-                    await event_source.response.aclose()
-                    break
+                async for sse in event_source:
+                    is_complete = await self._handle_sse_event(
+                        sse,
+                        ctx.read_stream_writer,
+                        original_request_id,
+                        ctx.metadata.on_resumption_token_update if ctx.metadata else None,
+                    )
+                    if is_complete:
+                        await event_source.response.aclose()
+                        return
+        except Exception:
+            logger.debug("Resumption stream ended", exc_info=True)
+
+        # Stream ended without a response, cleanly or mid-read: resolve the waiter,
+        # mirroring `_handle_sse_response`, else the caller would hang forever.
+        await self._resolve_abandoned_request(
+            ctx.read_stream_writer, original_request_id, "resumption stream ended without a response"
+        )
 
     def _consume_modern_cancellation(self, session_message: SessionMessage) -> bool:
         """Translate an outbound `notifications/cancelled` at 2026; True means "do not POST".
@@ -563,8 +582,10 @@ async def _handle_message(session_message: SessionMessage) -> None:
                         else None
                     )
 
-                    # Check if this is a resumption request
-                    is_resumption = bool(metadata and metadata.resumption_token)
+                    # Only a request resumes: the token names an interrupted request's
+                    # stream, and `_handle_resumption_request` needs the id to correlate
+                    # its outcome. A notification stamped with one is POSTed as usual.
+                    is_resumption = bool(metadata and metadata.resumption_token) and isinstance(message, JSONRPCRequest)
 
                     logger.debug(f"Sending client message: {message}")
 

tests/client/test_streamable_http.py

@@ -34,6 +34,7 @@
 from mcp.client.streamable_http import (
     LAST_EVENT_ID,
     MAX_RECONNECTION_ATTEMPTS,
+    MCP_SESSION_ID,
     RequestContext,
     StreamableHTTPTransport,
     streamable_http_client,
@@ -135,11 +136,12 @@ def handler(request: httpx2.Request) -> httpx2.Response:
 
 
 @pytest.mark.anyio
-@pytest.mark.parametrize("status", [401, 403, 500])
+@pytest.mark.parametrize("status", [302, 401, 403, 500])
 async def test_resumption_get_http_error_resolves_caller_and_transport_survives(status: int) -> None:
     """A non-2xx on the resumption GET resolves the waiting request with a JSON-RPC error
     correlated to its id, and the transport stays usable for follow-up requests (SDK-defined;
     #2110 — the status error used to escape into the task group and tear down every stream).
+    An unfollowed redirect counts: its body is no event stream, so no response can arrive.
     """
 
     def handler(request: httpx2.Request) -> httpx2.Response:
@@ -174,6 +176,79 @@ def handler(request: httpx2.Request) -> httpx2.Response:
     assert follow_up.message.id == 2
 
 
+@pytest.mark.anyio
+async def test_resumption_get_404_with_session_reports_session_terminated() -> None:
+    """A 404 on the resumption GET while a session id is held reports "Session terminated"
+    (INVALID_REQUEST) to the waiter, the same session-expiry mapping as the POST path, so
+    reconnect logic keyed on that error works across both (SDK-defined)."""
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if request.method == "GET" and LAST_EVENT_ID in request.headers:
+            return httpx2.Response(404)
+        if request.method == "DELETE":  # session termination on close
+            return httpx2.Response(200)
+        body = json.loads(request.content)
+        return httpx2.Response(
+            200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}}, headers={MCP_SESSION_ID: "sess-1"}
+        )
+
+    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),
+        ):
+            # An initialize round-trip stores the session id the server stamps on its response.
+            await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="initialize", params={})))
+            assert isinstance(await read.receive(), SessionMessage)
+
+            await write.send(
+                SessionMessage(
+                    message=JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/call", params={}),
+                    metadata=ClientMessageMetadata(resumption_token="token-1"),
+                )
+            )
+            reply = await read.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.id == 2
+    assert reply.message.error.code == INVALID_REQUEST
+    assert reply.message.error.message == snapshot("Session terminated")
+
+
+@pytest.mark.anyio
+async def test_notification_with_resumption_token_is_posted_not_resumed() -> None:
+    """A notification stamped with a resumption token is POSTed like any notification, and the
+    write loop survives to serve the next request (SDK-defined: the token names an interrupted
+    request's stream, so resumption applies to requests only)."""
+    recorded: list[httpx2.Request] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        recorded.append(request)
+        body = json.loads(request.content)
+        if "id" not in body:
+            return httpx2.Response(202)
+        return httpx2.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}})
+
+    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(
+                    message=JSONRPCNotification(jsonrpc="2.0", method="notifications/foo", params={}),
+                    metadata=ClientMessageMetadata(resumption_token="token-1"),
+                )
+            )
+            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, JSONRPCResponse)
+    assert reply.message.id == 1
+    # The stamped notification went out as a plain POST, not a resumption GET.
+    assert [r.method for r in recorded] == ["POST", "POST"]
+
+
 @pytest.mark.anyio
 async def test_initialize_post_clears_cached_pv_header_and_unstamped_posts_read_it() -> None:
     """``initialize`` discards the cached protocol-version header; every other POST reads it.
@@ -674,6 +749,74 @@ def handler(request: httpx2.Request) -> httpx2.Response:
     assert reply.message.error.code == CONNECTION_CLOSED
 
 
+@pytest.mark.anyio
+async def test_resumption_stream_dying_mid_read_resolves_caller_and_transport_survives() -> None:
+    """A resumption GET stream that dies mid-read resolves the waiter with CONNECTION_CLOSED
+    and the transport stays usable for follow-up requests (SDK-defined; #2110 — the read error
+    used to escape into the task group and tear down every stream)."""
+    dying = _DyingSSEStream()
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        if request.method == "GET" and LAST_EVENT_ID in request.headers:
+            return httpx2.Response(200, headers={"content-type": "text/event-stream"}, stream=dying)
+        body = json.loads(request.content)
+        return httpx2.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}})
+
+    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(
+                    message=JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={}),
+                    metadata=ClientMessageMetadata(resumption_token="token-1"),
+                )
+            )
+            reply = await read.receive()
+            assert isinstance(reply, SessionMessage)
+            assert isinstance(reply.message, JSONRPCError)
+            assert reply.message.id == 1
+            assert reply.message.error.code == CONNECTION_CLOSED
+            assert reply.message.error.message == snapshot("resumption stream ended without a response")
+
+            # The transport survived: a plain follow-up request still round-trips.
+            await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/list", params={})))
+            follow_up = await read.receive()
+    assert isinstance(follow_up, SessionMessage)
+    assert isinstance(follow_up.message, JSONRPCResponse)
+    assert follow_up.message.id == 2
+
+
+@pytest.mark.anyio
+async def test_resumption_stream_clean_end_without_response_resolves_caller() -> None:
+    """A resumption GET stream that closes cleanly without delivering a response (e.g. the
+    server no longer holds the resumed request's events) resolves the waiter with an error
+    instead of hanging it forever (SDK-defined; #2110)."""
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        assert request.method == "GET" and LAST_EVENT_ID in request.headers
+        return httpx2.Response(200, headers={"content-type": "text/event-stream"}, content=b": nothing to replay\n\n")
+
+    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(
+                    message=JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={}),
+                    metadata=ClientMessageMetadata(resumption_token="token-1"),
+                )
+            )
+            reply = await read.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.id == 1
+    assert reply.message.error.code == CONNECTION_CLOSED
+    assert reply.message.error.message == snapshot("resumption stream ended without a response")
+
+
 class _DeliverOnCommandSSEStream(httpx2.AsyncByteStream):
     """Parks after opening, then delivers one JSON-RPC response when told."""