Backport abandoned-request resolution from 2.x to the 1.x Streamable HTTP client

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,

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