Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions src/mcp/client/streamable_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -224,8 +226,9 @@ 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)
Expand Down
25 changes: 25 additions & 0 deletions tests/client/test_streamable_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -736,6 +736,31 @@ 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:
await transport.handle_get_stream(http, send)
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
Resolved

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."""
Expand Down
Loading