FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

fix: stop the standalone GET stream reconnecting forever on empty connections by heyhayes · Pull Request #3087 · modelcontextprotocol/python-sdk · GitHub

Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension .py  (4) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
2 changes: 2 additions & 0 deletions src/mcp/client/session.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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()


Expand Down
12 changes: 9 additions & 3 deletions src/mcp/client/streamable_http.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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

Expand Down
11 changes: 10 additions & 1 deletion tests/client/test_session.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
23 changes: 23 additions & 0 deletions tests/client/test_streamable_http.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Loading

Back | FazBrowse Home | New Git URL