claude · Aug 11, 2026 · Aug 11, 2026 · Aug 11, 2026 · Aug 11, 2026 · Aug 11, 2026
diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py
@@ -120,17 +120,39 @@
                     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}")
-                            response = await client.post(
-                                endpoint_url,
-                                json=session_message.message.model_dump(
-                                    by_alias=True,
-                                    mode="json",
-                                    exclude_unset=True,
-                                ),
-                            )
-                            response.raise_for_status()
-                            logger.debug(f"Client message sent successfully: {response.status_code}")
+                            message = session_message.message
+                            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}")
+                                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

diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py
@@ -248,25 +248,51 @@
         else:
             raise ResumptionError("Resumption request requires a resumption token")  # pragma: no cover
 
-        # Extract original request ID to map responses
-        original_request_id = None
-        if isinstance(ctx.session_message.message, JSONRPCRequest):  # pragma: no branch
-            original_request_id = ctx.session_message.message.id
+        # 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:
-            event_source.response.raise_for_status()
-            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".
@@ -556,8 +582,10 @@
                         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}")
 

diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py
@@ -19,6 +19,7 @@
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
     CONNECTION_CLOSED,
+    INTERNAL_ERROR,
     INVALID_REQUEST,
     METHOD_NOT_FOUND,
     PROTOCOL_VERSION_META_KEY,
@@ -31,7 +32,9 @@
 from starlette.types import Receive, Scope, Send
 
 from mcp.client.streamable_http import (
+    LAST_EVENT_ID,
     MAX_RECONNECTION_ATTEMPTS,
+    MCP_SESSION_ID,
     RequestContext,
     StreamableHTTPTransport,
     streamable_http_client,
@@ -132,6 +135,120 @@ def handler(request: httpx2.Request) -> httpx2.Response:
     assert reply.message.error.code == METHOD_NOT_FOUND
 
 
+@pytest.mark.anyio
+@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:
+        if request.method == "GET" and LAST_EVENT_ID in request.headers:
+            return httpx2.Response(status)
+        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 == INTERNAL_ERROR
+            assert reply.message.error.message == snapshot("Server returned an error 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_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.
@@ -632,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."""