fix: stop the standalone GET stream reconnecting forever on empty connections

src/mcp/client/session.py

@@ -166,6 +166,8 @@ async def __call__(
 async def _default_message_handler(
     message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
 ) -> None:
+    if isinstance(message, Exception):
+        logger.warning("Unhandled exception in message handler: %s", message)
     await anyio.lowlevel.checkpoint()
 
 

src/mcp/client/streamable_http.py

@@ -214,7 +214,9 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer:
                     event_source.response.raise_for_status()
                     logger.debug("GET SSE connection established")
 
+                    received_events = False
                     async for sse in event_source.aiter_sse():
+                        received_events = True
                         # Track last event ID for reconnection
                         if sse.id:
                             last_event_id = sse.id
@@ -224,14 +226,18 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer:
 
                         await self._handle_sse_event(sse, read_stream_writer)
 
-                    # Stream ended normally (server closed) - reset attempt counter
-                    attempt = 0
+                    # Only reset attempts if we actually received events;
+                    # empty connections count toward MAX_RECONNECTION_ATTEMPTS
+                    if received_events:
+                        attempt = 0
+                    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_session.py

@@ -38,7 +38,7 @@
 from mcp import MCPError
 from mcp.client import ClientRequestContext
 from mcp.client.client import Client
-from mcp.client.session import DEFAULT_CLIENT_INFO, ClientSession
+from mcp.client.session import DEFAULT_CLIENT_INFO, ClientSession, _default_message_handler
 from mcp.client.subscriptions import ToolsListChanged, listen
 from mcp.server import Server, ServerRequestContext
 from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
@@ -1030,6 +1030,15 @@ async def handler(msg: object) -> None:
     assert "message_handler raised on transport exception" in caplog.text
 
 
+@pytest.mark.anyio
+async def test_default_message_handler_logs_unhandled_transport_exceptions(caplog: pytest.LogCaptureFixture):
+    """The default handler has no per-request waiter to resolve, so a transport-level `Exception`
+    item (e.g. a GET stream failure) would vanish silently; it logs a warning as a safety net."""
+    with caplog.at_level("WARNING", logger="client"):
+        await _default_message_handler(RuntimeError("boom"))
+    assert "Unhandled exception in message handler" in caplog.text
+
+
 @pytest.mark.anyio
 async def test_message_handler_awaiting_session_traffic_on_transport_exception_completes():
     """A `message_handler` that awaits session traffic on a transport `Exception` item completes:

tests/client/test_streamable_http.py

@@ -748,3 +748,26 @@ async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contain
                 _abandoned_request_context(http, send), "evt-7", None, MAX_RECONNECTION_ATTEMPTS
             )
     send.close()
+
+
+@pytest.mark.anyio
+async def test_get_stream_gives_up_after_repeated_empty_connections(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A GET stream that keeps opening but closing with no events counts each empty connection toward
+    the reconnection budget, so the loop terminates instead of reconnecting forever."""
+    monkeypatch.setattr("mcp.client.streamable_http.DEFAULT_RECONNECTION_DELAY_MS", 0)
+    get_requests = 0
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        nonlocal get_requests
+        get_requests += 1
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=b"")
+
+    transport = StreamableHTTPTransport("http://test/mcp")
+    transport.session_id = "sess-1"
+    send, receive = create_context_streams[SessionMessage | Exception](1)
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
+        with anyio.fail_after(5):
+            await transport.handle_get_stream(http, send)
+    assert get_requests == MAX_RECONNECTION_ATTEMPTS
+    send.close()
+    receive.close()