diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index ed28fcc275..b3adf20ad8 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -28,8 +28,10 @@ ) from mcp.shared.message import ClientMessageMetadata, SessionMessage from mcp.types import ( + CONNECTION_CLOSED, ErrorData, InitializeResult, + INVALID_REQUEST, JSONRPCError, JSONRPCMessage, JSONRPCNotification, @@ -345,6 +347,15 @@ async def _handle_post_request(self, ctx: RequestContext) -> None: ) as response: if response.status_code == 202: logger.debug("Received 202 Accepted") + if isinstance(message.root, JSONRPCRequest): + # A request's response arrives on this POST's body; 202 says + # none will follow. Resolve rather than park the caller forever. + await self._resolve_abandoned_request( + ctx.read_stream_writer, + message.root.id, + "server answered a request with 202 Accepted", + code=INVALID_REQUEST, + ) return if response.status_code == 404: # pragma: no branch @@ -404,6 +415,10 @@ async def _handle_sse_response( last_event_id: str | None = None retry_interval_ms: int | None = None + original_request_id = None + if isinstance(ctx.session_message.message.root, JSONRPCRequest): + original_request_id = ctx.session_message.message.root.id + try: event_source = EventSource(response) async for sse in event_source.aiter_sse(): # pragma: no branch @@ -430,9 +445,34 @@ async def _handle_sse_response( logger.debug(f"SSE stream ended: {e}") # Stream ended without response - reconnect if we received an event with ID - if last_event_id is not None: # pragma: no branch + if last_event_id is not None: logger.info("SSE stream disconnected, reconnecting...") await self._handle_reconnection(ctx, last_event_id, retry_interval_ms) + else: + # Not resumable: resolve the waiter, else the request would hang + # forever instead of learning the connection is lost. + await self._resolve_abandoned_request( + ctx.read_stream_writer, original_request_id, "SSE stream ended without a response" + ) + + async def _resolve_abandoned_request( + self, + read_stream_writer: StreamWriter, + request_id: RequestId, + message: str, + *, + code: int = CONNECTION_CLOSED, + ) -> None: + """Resolve a request whose response can never arrive with a synthesized error. + + Best-effort: a closed read stream means the session is tearing down. + """ + error_data = ErrorData(code=code, message=message) + session_message = SessionMessage(JSONRPCMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data))) + try: + await read_stream_writer.send(session_message) + except (anyio.BrokenResourceError, anyio.ClosedResourceError): + logger.debug("read stream closed before request %r could be resolved", request_id) async def _handle_reconnection( self, @@ -442,9 +482,22 @@ async def _handle_reconnection( attempt: int = 0, ) -> None: """Reconnect with Last-Event-ID to resume stream after server disconnect.""" + # Extract original request ID to map responses + original_request_id = None + if isinstance(ctx.session_message.message.root, JSONRPCRequest): + original_request_id = ctx.session_message.message.root.id + # Bail if max retries exceeded - if attempt >= MAX_RECONNECTION_ATTEMPTS: # pragma: no cover + if attempt >= MAX_RECONNECTION_ATTEMPTS: logger.debug(f"Max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded") + # Resolve on give-up: a request with no read timeout would otherwise + # hang its caller forever. + if original_request_id is not None: + await self._resolve_abandoned_request( + ctx.read_stream_writer, + original_request_id, + "SSE stream ended and reconnection attempts were exhausted", + ) return # Always wait - use server value or default @@ -454,11 +507,6 @@ async def _handle_reconnection( headers = self._prepare_headers() headers[LAST_EVENT_ID] = last_event_id - # Extract original request ID to map responses - original_request_id = None - if isinstance(ctx.session_message.message.root, JSONRPCRequest): # pragma: no branch - original_request_id = ctx.session_message.message.root.id - try: async with aconnect_sse( ctx.client, diff --git a/tests/client/test_streamable_http_abandoned_request.py b/tests/client/test_streamable_http_abandoned_request.py new file mode 100644 index 0000000000..ccc1fa3894 --- /dev/null +++ b/tests/client/test_streamable_http_abandoned_request.py @@ -0,0 +1,171 @@ +"""Deterministic regression tests for abandoned-request resolution (#3441). + +When a request's response can never arrive - the server answers the POST +with 202 Accepted, the per-request SSE stream ends without a response +event, or reconnection attempts are exhausted - the v1.x transport must +resolve the pending request with a synthesized JSONRPCError instead of +parking the caller until its own timeout fires. +""" + +from collections.abc import AsyncIterator + +import anyio +import httpx +import pytest +from mcp.shared.message import SessionMessage +from mcp.types import ( + CONNECTION_CLOSED, + INVALID_REQUEST, + JSONRPCError, + JSONRPCMessage, + JSONRPCNotification, + JSONRPCRequest, +) + +from mcp.client.streamable_http import ( + MAX_RECONNECTION_ATTEMPTS, + RequestContext, + StreamableHTTPTransport, +) + + +def _make_request_context( + client: httpx.AsyncClient, + message: JSONRPCMessage, + read_stream_writer, +) -> RequestContext: + session_message = SessionMessage(message) + return RequestContext( + client=client, + session_id=None, + session_message=session_message, + metadata=None, + read_stream_writer=read_stream_writer, + ) + + +def _request(id_: str) -> JSONRPCMessage: + return JSONRPCMessage(JSONRPCRequest(jsonrpc="2.0", id=id_, method="tools/call", params={})) + + +class _DyingSSEStream(httpx.AsyncByteStream): + """Emits one id-less comment then breaks - a non-resumable stream dropping.""" + + def __init__(self) -> None: + self.opened = anyio.Event() + + async def __aiter__(self) -> AsyncIterator[bytes]: + self.opened.set() + yield b": hello\n\n" + raise httpx.ReadError("connection reset") + + async def aclose(self) -> None: + pass + + +@pytest.mark.anyio +async def test_non_resumable_sse_drop_resolves_request_with_error() -> None: + """A per-request SSE stream that dies having carried no event ids can never + deliver its response; the transport resolves the waiter with CONNECTION_CLOSED + instead of hanging forever.""" + transport = StreamableHTTPTransport("http://test/mcp") + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=_DyingSSEStream()) + + streams = anyio.create_memory_object_stream(4) + try: + write_stream, read_stream = streams + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + ctx = _make_request_context(client, _request("req-sse"), write_stream) + with anyio.fail_after(5): + await transport._handle_post_request(ctx) + reply = await read_stream.receive() + finally: + write_stream.close() + read_stream.close() + + assert isinstance(reply.message.root, JSONRPCError) + assert reply.message.root.id == "req-sse" + assert reply.message.root.error.code == CONNECTION_CLOSED + + +@pytest.mark.anyio +async def test_post_answered_with_202_resolves_request_with_error() -> None: + """A request answered with 202 Accepted will never receive a response body; + the transport resolves the waiter with INVALID_REQUEST instead of hanging.""" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(202) + + transport = StreamableHTTPTransport("http://test/mcp") + streams = anyio.create_memory_object_stream(4) + try: + write_stream, read_stream = streams + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + ctx = _make_request_context(client, _request("req-202"), write_stream) + with anyio.fail_after(5): + await transport._handle_post_request(ctx) + reply = await read_stream.receive() + finally: + write_stream.close() + read_stream.close() + + assert isinstance(reply.message.root, JSONRPCError) + assert reply.message.root.id == "req-202" + assert reply.message.root.error.code == INVALID_REQUEST + + +@pytest.mark.anyio +async def test_post_answered_with_202_does_not_resolve_notifications() -> None: + """Notifications have no waiter to resolve; a 202 answer must not inject an + error into the read stream for them.""" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(202) + + transport = StreamableHTTPTransport("http://test/mcp") + notification = JSONRPCMessage(JSONRPCNotification(jsonrpc="2.0", method="notifications/initialized")) + streams = anyio.create_memory_object_stream(4) + try: + write_stream, read_stream = streams + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + ctx = _make_request_context(client, notification, write_stream) + with anyio.move_on_after(1): + await transport._handle_post_request(ctx) + await read_stream.receive() + pytest.fail("a notification must not be resolved with an error") + finally: + write_stream.close() + read_stream.close() + + +@pytest.mark.anyio +async def test_exhausted_reconnection_attempts_resolve_request_with_error() -> None: + """When reconnection attempts are exhausted for a request whose SSE stream + keeps dying, the transport resolves the waiter with CONNECTION_CLOSED.""" + transport = StreamableHTTPTransport("http://test/mcp") + + def handler(request: httpx.Request) -> httpx.Response: # pragma: no cover + pytest.fail("exhausted reconnection must resolve without further HTTP calls") + + streams = anyio.create_memory_object_stream(4) + try: + write_stream, read_stream = streams + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + ctx = _make_request_context(client, _request("req-ex"), write_stream) + with anyio.fail_after(5): + await transport._handle_reconnection( + ctx, + last_event_id="1", + retry_interval_ms=1, + attempt=MAX_RECONNECTION_ATTEMPTS, + ) + reply = await read_stream.receive() + finally: + write_stream.close() + read_stream.close() + + assert isinstance(reply.message.root, JSONRPCError) + assert reply.message.root.id == "req-ex" + assert reply.message.root.error.code == CONNECTION_CLOSED