fix(client): bound empty GET stream retries

src/mcp/client/streamable_http.py

@@ -214,7 +214,9 @@ async def handle_get_stream(self, client: httpx2.AsyncClient, read_stream_writer
                     event_source.response.raise_for_status()
                     logger.debug("GET SSE connection established")
 
+                    saw_event = False
                     async for sse in event_source:
+                        saw_event = True
                         # Track last event ID for reconnection
                         if sse.id:
                             last_event_id = sse.id
@@ -224,14 +226,15 @@ async def handle_get_stream(self, client: httpx2.AsyncClient, read_stream_writer
 
                         await self._handle_sse_event(sse, read_stream_writer)
 
-                    # Stream ended normally (server closed) - reset attempt counter
-                    attempt = 0
+                    # A clean stream close is retryable only when the server sent an event.
+                    # An empty stream is a terminated endpoint, so count it against the retry budget.
+                    attempt = 0 if saw_event else attempt + 1
 
             except Exception:
                 logger.debug("GET stream error", exc_info=True)
                 attempt += 1
 
-            if attempt >= MAX_RECONNECTION_ATTEMPTS:  # pragma: no cover
+            if attempt >= MAX_RECONNECTION_ATTEMPTS:
                 logger.debug(f"GET stream max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
                 return
 

tests/client/test_streamable_http.py

@@ -736,6 +736,32 @@ async def test_exhausted_reconnection_attempts_resolve_the_request_with_an_error
     receive.close()
 
 
+@pytest.mark.anyio
+async def test_empty_get_stream_exhausts_reconnection_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
+    """An empty, cleanly closed GET stream must not reset the reconnection budget."""
+    requests: list[httpx2.Request] = []
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        requests.append(request)
+        return httpx2.Response(200, headers={"content-type": "text/event-stream"}, content=b"")
+
+    async def no_sleep(_delay: float) -> None:
+        pass
+
+    monkeypatch.setattr("mcp.client.streamable_http.anyio.sleep", no_sleep)
+    transport = StreamableHTTPTransport("http://test/mcp")
+    transport.session_id = "session-1"
+    send, receive = create_context_streams[SessionMessage | Exception](0)
+
+    async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http:
+        with anyio.fail_after(5):
+            await transport.handle_get_stream(http, send)
+
+    assert len(requests) == MAX_RECONNECTION_ATTEMPTS
+    send.close()
+    receive.close()
+
+
 @pytest.mark.anyio
 async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contained() -> None:
     """Teardown race: a stream dying after the reader closed resolves best-effort and must not crash."""