|
| 1 | +"""Deterministic regression tests for abandoned-request resolution (#3441). |
| 2 | +
|
| 3 | +When a request's response can never arrive - the server answers the POST |
| 4 | +with 202 Accepted, the per-request SSE stream ends without a response |
| 5 | +event, or reconnection attempts are exhausted - the v1.x transport must |
| 6 | +resolve the pending request with a synthesized JSONRPCError instead of |
| 7 | +parking the caller until its own timeout fires. |
| 8 | +""" |
| 9 | + |
| 10 | +from collections.abc import AsyncIterator |
| 11 | + |
| 12 | +import anyio |
| 13 | +import httpx |
| 14 | +import pytest |
| 15 | +from mcp.shared.message import SessionMessage |
| 16 | +from mcp.types import ( |
| 17 | + CONNECTION_CLOSED, |
| 18 | + INVALID_REQUEST, |
| 19 | + JSONRPCError, |
| 20 | + JSONRPCMessage, |
| 21 | + JSONRPCNotification, |
| 22 | + JSONRPCRequest, |
| 23 | +) |
| 24 | + |
| 25 | +from mcp.client.streamable_http import ( |
| 26 | + MAX_RECONNECTION_ATTEMPTS, |
| 27 | + RequestContext, |
| 28 | + StreamableHTTPTransport, |
| 29 | +) |
| 30 | + |
| 31 | + |
| 32 | +def _make_request_context( |
| 33 | + client: httpx.AsyncClient, |
| 34 | + message: JSONRPCMessage, |
| 35 | + read_stream_writer, |
| 36 | +) -> RequestContext: |
| 37 | + session_message = SessionMessage(message) |
| 38 | + return RequestContext( |
| 39 | + client=client, |
| 40 | + session_id=None, |
| 41 | + session_message=session_message, |
| 42 | + metadata=None, |
| 43 | + read_stream_writer=read_stream_writer, |
| 44 | + ) |
| 45 | + |
| 46 | + |
| 47 | +def _request(id_: str) -> JSONRPCMessage: |
| 48 | + return JSONRPCMessage(JSONRPCRequest(jsonrpc="2.0", id=id_, method="tools/call", params={})) |
| 49 | + |
| 50 | + |
| 51 | +class _DyingSSEStream(httpx.AsyncByteStream): |
| 52 | + """Emits one id-less comment then breaks - a non-resumable stream dropping.""" |
| 53 | + |
| 54 | + def __init__(self) -> None: |
| 55 | + self.opened = anyio.Event() |
| 56 | + |
| 57 | + async def __aiter__(self) -> AsyncIterator[bytes]: |
| 58 | + self.opened.set() |
| 59 | + yield b": hello\n\n" |
| 60 | + raise httpx.ReadError("connection reset") |
| 61 | + |
| 62 | + async def aclose(self) -> None: |
| 63 | + pass |
| 64 | + |
| 65 | + |
| 66 | +@pytest.mark.anyio |
| 67 | +async def test_non_resumable_sse_drop_resolves_request_with_error() -> None: |
| 68 | + """A per-request SSE stream that dies having carried no event ids can never |
| 69 | + deliver its response; the transport resolves the waiter with CONNECTION_CLOSED |
| 70 | + instead of hanging forever.""" |
| 71 | + transport = StreamableHTTPTransport("http://test/mcp") |
| 72 | + |
| 73 | + def handler(request: httpx.Request) -> httpx.Response: |
| 74 | + return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=_DyingSSEStream()) |
| 75 | + |
| 76 | + streams = anyio.create_memory_object_stream(4) |
| 77 | + try: |
| 78 | + write_stream, read_stream = streams |
| 79 | + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: |
| 80 | + ctx = _make_request_context(client, _request("req-sse"), write_stream) |
| 81 | + with anyio.fail_after(5): |
| 82 | + await transport._handle_post_request(ctx) |
| 83 | + reply = await read_stream.receive() |
| 84 | + finally: |
| 85 | + write_stream.close() |
| 86 | + read_stream.close() |
| 87 | + |
| 88 | + assert isinstance(reply.message.root, JSONRPCError) |
| 89 | + assert reply.message.root.id == "req-sse" |
| 90 | + assert reply.message.root.error.code == CONNECTION_CLOSED |
| 91 | + |
| 92 | + |
| 93 | +@pytest.mark.anyio |
| 94 | +async def test_post_answered_with_202_resolves_request_with_error() -> None: |
| 95 | + """A request answered with 202 Accepted will never receive a response body; |
| 96 | + the transport resolves the waiter with INVALID_REQUEST instead of hanging.""" |
| 97 | + |
| 98 | + def handler(request: httpx.Request) -> httpx.Response: |
| 99 | + return httpx.Response(202) |
| 100 | + |
| 101 | + transport = StreamableHTTPTransport("http://test/mcp") |
| 102 | + streams = anyio.create_memory_object_stream(4) |
| 103 | + try: |
| 104 | + write_stream, read_stream = streams |
| 105 | + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: |
| 106 | + ctx = _make_request_context(client, _request("req-202"), write_stream) |
| 107 | + with anyio.fail_after(5): |
| 108 | + await transport._handle_post_request(ctx) |
| 109 | + reply = await read_stream.receive() |
| 110 | + finally: |
| 111 | + write_stream.close() |
| 112 | + read_stream.close() |
| 113 | + |
| 114 | + assert isinstance(reply.message.root, JSONRPCError) |
| 115 | + assert reply.message.root.id == "req-202" |
| 116 | + assert reply.message.root.error.code == INVALID_REQUEST |
| 117 | + |
| 118 | + |
| 119 | +@pytest.mark.anyio |
| 120 | +async def test_post_answered_with_202_does_not_resolve_notifications() -> None: |
| 121 | + """Notifications have no waiter to resolve; a 202 answer must not inject an |
| 122 | + error into the read stream for them.""" |
| 123 | + |
| 124 | + def handler(request: httpx.Request) -> httpx.Response: |
| 125 | + return httpx.Response(202) |
| 126 | + |
| 127 | + transport = StreamableHTTPTransport("http://test/mcp") |
| 128 | + notification = JSONRPCMessage(JSONRPCNotification(jsonrpc="2.0", method="notifications/initialized")) |
| 129 | + streams = anyio.create_memory_object_stream(4) |
| 130 | + try: |
| 131 | + write_stream, read_stream = streams |
| 132 | + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: |
| 133 | + ctx = _make_request_context(client, notification, write_stream) |
| 134 | + with anyio.move_on_after(1): |
| 135 | + await transport._handle_post_request(ctx) |
| 136 | + await read_stream.receive() |
| 137 | + pytest.fail("a notification must not be resolved with an error") |
| 138 | + finally: |
| 139 | + write_stream.close() |
| 140 | + read_stream.close() |
| 141 | + |
| 142 | + |
| 143 | +@pytest.mark.anyio |
| 144 | +async def test_exhausted_reconnection_attempts_resolve_request_with_error() -> None: |
| 145 | + """When reconnection attempts are exhausted for a request whose SSE stream |
| 146 | + keeps dying, the transport resolves the waiter with CONNECTION_CLOSED.""" |
| 147 | + transport = StreamableHTTPTransport("http://test/mcp") |
| 148 | + |
| 149 | + def handler(request: httpx.Request) -> httpx.Response: # pragma: no cover |
| 150 | + pytest.fail("exhausted reconnection must resolve without further HTTP calls") |
| 151 | + |
| 152 | + streams = anyio.create_memory_object_stream(4) |
| 153 | + try: |
| 154 | + write_stream, read_stream = streams |
| 155 | + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: |
| 156 | + ctx = _make_request_context(client, _request("req-ex"), write_stream) |
| 157 | + with anyio.fail_after(5): |
| 158 | + await transport._handle_reconnection( |
| 159 | + ctx, |
| 160 | + last_event_id="1", |
| 161 | + retry_interval_ms=1, |
| 162 | + attempt=MAX_RECONNECTION_ATTEMPTS, |
| 163 | + ) |
| 164 | + reply = await read_stream.receive() |
| 165 | + finally: |
| 166 | + write_stream.close() |
| 167 | + read_stream.close() |
| 168 | + |
| 169 | + assert isinstance(reply.message.root, JSONRPCError) |
| 170 | + assert reply.message.root.id == "req-ex" |
| 171 | + assert reply.message.root.error.code == CONNECTION_CLOSED |
0 commit comments