From beeb2544d8e92fc3b1cad4b702a84f259a660db5 Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Sat, 30 May 2026 23:56:49 +0200 Subject: [PATCH 01/14] test(streamable-http): ensure resumption failure does not hang --- .../transports/test_hosting_resume.py | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/tests/interaction/transports/test_hosting_resume.py b/tests/interaction/transports/test_hosting_resume.py index 78492ccb7e..92e58a1991 100644 --- a/tests/interaction/transports/test_hosting_resume.py +++ b/tests/interaction/transports/test_hosting_resume.py @@ -373,6 +373,82 @@ async def call() -> None: assert received == snapshot(["before close", "after close"]) +@requirement("hosting:resume:close-stream") +@requirement("transport:streamable-http:resumability") +@requirement("client-transport:http:reconnect-post-priming") +@requirement("client-transport:http:reconnect-retry-value") +async def test_a_call_whose_stream_closes_and_cannot_be_resumed_fails_instead_of_hanging() -> None: + """If a resumable response stream disconnects and the server session is gone, the client fails + the request instead of hanging forever. + + The server closes the call's SSE stream after emitting one related notification. The test then + deletes the active server-side session to force the client's reconnect GET to return 404. + Without a terminal response/error on the read stream, ClientSession.send_request waits forever + (read timeout defaults to None). The transport must surface a request-scoped error when it + gives up reconnecting. + """ + reconnect_attempted = anyio.Event() + allow_exit = anyio.Event() + done = anyio.Event() + raised: list[BaseException] = [] + manager_ref = None + deleted_session = False + + mcp = MCPServer("resumable") + + @mcp.tool() + async def interrupt(ctx: Context) -> str: + await ctx.info("before close") + await ctx.close_sse_stream() + await allow_exit.wait() + return "unreachable" + + async def record_request(request: httpx.Request) -> None: + nonlocal deleted_session + if request.method != "GET": + return + if request.headers.get("last-event-id") is None: + return + reconnect_attempted.set() + if deleted_session or manager_ref is None: + return + session_ids = list(manager_ref._server_instances.keys()) + if session_ids: # pragma: no branch + del manager_ref._server_instances[session_ids[0]] + deleted_session = True + + async with mounted_app(mcp, event_store=SequencedEventStore(), retry_interval=0, on_request=record_request) as ( + http, + manager, + ): + manager_ref = manager + with anyio.fail_after(5): # pragma: no branch + async with ( + streamable_http_client(f"{BASE_URL}/mcp", http_client=http, terminate_on_close=False) as (r, w), + ClientSession(r, w) as session, + anyio.create_task_group() as tg, + ): + await session.initialize() + + async def call() -> None: + try: + await session.call_tool("interrupt", {}) + except BaseException as exc: + raised.append(exc) + finally: + done.set() + + tg.start_soon(call) + await reconnect_attempted.wait() + await done.wait() + allow_exit.set() + tg.cancel_scope.cancel() + + assert len(raised) == 1 + assert isinstance(raised[0], Exception) + assert "disconnected" in str(raised[0]).lower() + + @requirement("client-transport:http:resume-stream-api") async def test_a_captured_resumption_token_replays_missed_messages_on_a_new_connection() -> None: """A resumption token captured via on_resumption_token_update on one connection lets a fresh From 67b93e5050164181406db4d5c509153731873a22 Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Sun, 31 May 2026 00:06:41 +0200 Subject: [PATCH 02/14] test(streamable-http): cover disconnect without resumption anchor --- tests/client/test_streamable_http.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index da17a71c1e..fff838ece7 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -81,6 +81,30 @@ def test_mcp_name_header_values_are_base64_wrapped_when_unsafe_for_an_http_field assert encoded == raw +@pytest.mark.anyio +async def test_sse_response_disconnect_before_any_event_id_fails_request() -> None: + transport = StreamableHTTPTransport("http://example.com/mcp") + async with httpx.AsyncClient() as client: + read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](1) + request = JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={"name": "noop", "arguments": {}}) + ctx = RequestContext( + client=client, + session_id=None, + session_message=SessionMessage(request), + metadata=None, + read_stream_writer=read_stream_writer, + ) + response = httpx.Response(200, headers={"content-type": "text/event-stream"}, content=b"") + + async with read_stream_writer, read_stream: + await transport._handle_sse_response(response, ctx) + message = await read_stream.receive() + + assert isinstance(message.message, JSONRPCError) + assert message.message.id == 1 + assert message.message.error.code == CONNECTION_CLOSED + + @pytest.mark.anyio async def test_post_request_merges_per_message_metadata_headers() -> None: """`ClientMessageMetadata.headers` on a `SessionMessage` are merged into the outgoing POST headers From c4cff468e2f67958245d2f3213e5da25d6fdc027 Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Sun, 31 May 2026 00:18:08 +0200 Subject: [PATCH 03/14] fix: make streamable-http tests pass CI --- src/mcp/client/streamable_http.py | 2 +- tests/client/test_streamable_http.py | 1 + tests/interaction/transports/test_hosting_resume.py | 2 +- 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 5c3501d8c4..8c6dd0867b 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -528,7 +528,7 @@ async def _handle_reconnection( # Stream ended again without response - reconnect again (reset attempt counter) logger.info("SSE stream disconnected, reconnecting...") await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0) - except Exception as e: # pragma: no cover + except Exception as e: logger.debug(f"Reconnection failed: {e}") # Try to reconnect again if we still have an event ID await self._handle_reconnection(ctx, last_event_id, retry_interval_ms, attempt + 1) diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index fff838ece7..a096557023 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -100,6 +100,7 @@ async def test_sse_response_disconnect_before_any_event_id_fails_request() -> No await transport._handle_sse_response(response, ctx) message = await read_stream.receive() + assert isinstance(message, SessionMessage) assert isinstance(message.message, JSONRPCError) assert message.message.id == 1 assert message.message.error.code == CONNECTION_CLOSED diff --git a/tests/interaction/transports/test_hosting_resume.py b/tests/interaction/transports/test_hosting_resume.py index 92e58a1991..4a1b5703f5 100644 --- a/tests/interaction/transports/test_hosting_resume.py +++ b/tests/interaction/transports/test_hosting_resume.py @@ -423,7 +423,7 @@ async def record_request(request: httpx.Request) -> None: ): manager_ref = manager with anyio.fail_after(5): # pragma: no branch - async with ( + async with ( # pragma: no branch streamable_http_client(f"{BASE_URL}/mcp", http_client=http, terminate_on_close=False) as (r, w), ClientSession(r, w) as session, anyio.create_task_group() as tg, From 0b325f2fc1c3d0d4beaad3f3ea741d0a7c006633 Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Sat, 20 Jun 2026 21:36:16 +0200 Subject: [PATCH 04/14] test(streamable-http): suppress deprecated logging warning --- tests/interaction/transports/test_hosting_resume.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/interaction/transports/test_hosting_resume.py b/tests/interaction/transports/test_hosting_resume.py index 4a1b5703f5..1fe5d713bb 100644 --- a/tests/interaction/transports/test_hosting_resume.py +++ b/tests/interaction/transports/test_hosting_resume.py @@ -398,7 +398,7 @@ async def test_a_call_whose_stream_closes_and_cannot_be_resumed_fails_instead_of @mcp.tool() async def interrupt(ctx: Context) -> str: - await ctx.info("before close") + await ctx.info("before close") # pyright: ignore[reportDeprecated] await ctx.close_sse_stream() await allow_exit.wait() return "unreachable" From eef7986a8b1551a77b43e3cf23afc7a3260e582f Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Sun, 28 Jun 2026 23:52:02 +0200 Subject: [PATCH 05/14] fix(streamable-http): tolerate closed read streams during SSE cleanup --- src/mcp/client/streamable_http.py | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 8c6dd0867b..79f7d8e2b9 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -46,6 +46,16 @@ StreamWriter = ContextSendStream[SessionMessageOrError] StreamReader = ContextReceiveStream[SessionMessage] + +async def _send_or_ignore_closed(read_stream_writer: StreamWriter, message: SessionMessageOrError) -> bool: + try: + await read_stream_writer.send(message) + except (anyio.BrokenResourceError, anyio.ClosedResourceError): + logger.debug("Read stream closed before Streamable HTTP message could be delivered", exc_info=True) + return False + return True + + MCP_SESSION_ID = "mcp-session-id" LAST_EVENT_ID = "last-event-id" @@ -179,17 +189,17 @@ async def _handle_sse_event( # Otherwise, return False to continue listening return isinstance(message, JSONRPCResponse | JSONRPCError) - # Forwarding to a closed read stream lands here when the caller cancels mid-SSE - # (BrokenResourceError, not a parse failure); coverage is timing-dependent in the - # streaming story's modern HTTP cancellation leg. + except (anyio.BrokenResourceError, anyio.ClosedResourceError): + logger.debug("Read stream closed while forwarding SSE message", exc_info=True) + return True except Exception as exc: # pragma: lax no cover logger.exception("Error parsing SSE message") if original_request_id is not None: error_data = ErrorData(code=PARSE_ERROR, message=f"Failed to parse SSE message: {exc}") error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data)) - await read_stream_writer.send(error_msg) + await _send_or_ignore_closed(read_stream_writer, error_msg) return True - await read_stream_writer.send(exc) + await _send_or_ignore_closed(read_stream_writer, exc) return False else: # pragma: no cover logger.warning(f"Unknown SSE event: {sse.event}") From 2abc3c1837b1a5c41c7d55b3cb22eb719e71a19b Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Mon, 29 Jun 2026 00:00:40 +0200 Subject: [PATCH 06/14] test(streamable-http): cover closed read stream cleanup --- tests/client/test_streamable_http.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index a096557023..5def33d128 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -106,6 +106,26 @@ async def test_sse_response_disconnect_before_any_event_id_fails_request() -> No assert message.message.error.code == CONNECTION_CLOSED +@pytest.mark.anyio +async def test_sse_response_disconnect_ignores_closed_read_stream() -> None: + transport = StreamableHTTPTransport("http://example.com/mcp") + async with httpx.AsyncClient() as client: + read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](1) + request = JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={"name": "noop", "arguments": {}}) + ctx = RequestContext( + client=client, + session_id=None, + session_message=SessionMessage(request), + metadata=None, + read_stream_writer=read_stream_writer, + ) + response = httpx.Response(200, headers={"content-type": "text/event-stream"}, content=b"") + + async with read_stream_writer, read_stream: + await read_stream.aclose() + await transport._handle_sse_response(response, ctx) + + @pytest.mark.anyio async def test_post_request_merges_per_message_metadata_headers() -> None: """`ClientMessageMetadata.headers` on a `SessionMessage` are merged into the outgoing POST headers From 9e33664f30e9077cb320899f6f0df36970df31ab Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Mon, 29 Jun 2026 00:52:32 +0200 Subject: [PATCH 07/14] fix(streamable-http): bound empty SSE reconnect loops --- src/mcp/client/streamable_http.py | 9 ++++- tests/client/test_streamable_http.py | 57 +++++++++++++++++++++++++++- 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 79f7d8e2b9..a8b314fc0e 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -518,12 +518,15 @@ async def _handle_reconnection( # Track for potential further reconnection reconnect_last_event_id: str = last_event_id reconnect_retry_ms = retry_interval_ms + made_progress = False async for sse in event_source.aiter_sse(): if sse.id: # pragma: no branch reconnect_last_event_id = sse.id if sse.retry is not None: reconnect_retry_ms = sse.retry + if sse.event == "message" and bool(sse.data): + made_progress = True is_complete = await self._handle_sse_event( sse, @@ -535,9 +538,11 @@ async def _handle_reconnection( await event_source.response.aclose() return - # Stream ended again without response - reconnect again (reset attempt counter) + # Stream ended again without response - reconnect again. Only reset + # the retry counter when the resumed stream delivered real data. logger.info("SSE stream disconnected, reconnecting...") - await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, 0) + next_attempt = 0 if made_progress else attempt + 1 + await self._handle_reconnection(ctx, reconnect_last_event_id, reconnect_retry_ms, next_attempt) except Exception as e: logger.debug(f"Reconnection failed: {e}") # Try to reconnect again if we still have an event ID diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index 5def33d128..09943cf1d4 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -98,7 +98,8 @@ async def test_sse_response_disconnect_before_any_event_id_fails_request() -> No async with read_stream_writer, read_stream: await transport._handle_sse_response(response, ctx) - message = await read_stream.receive() + with anyio.fail_after(5): + message = await read_stream.receive() assert isinstance(message, SessionMessage) assert isinstance(message.message, JSONRPCError) @@ -106,6 +107,60 @@ async def test_sse_response_disconnect_before_any_event_id_fails_request() -> No assert message.message.error.code == CONNECTION_CLOSED +@pytest.mark.anyio +async def test_reconnection_empty_streams_count_toward_max_attempts(monkeypatch: pytest.MonkeyPatch) -> None: + class PrimingOnlyEventSource: + def __init__(self) -> None: + self.response = httpx.Response(200) + + async def __aenter__(self) -> "PrimingOnlyEventSource": + nonlocal reconnect_attempts + reconnect_attempts += 1 + return self + + async def __aexit__(self, *args: object) -> None: + return None + + async def aiter_sse(self) -> object: + yield type( + "SSE", + (), + {"event": "message", "data": "", "id": f"event-{reconnect_attempts}", "retry": 0}, + )() + + def connect_sse(*args: object, **kwargs: object) -> PrimingOnlyEventSource: + return PrimingOnlyEventSource() + + reconnect_attempts = 0 + monkeypatch.setattr( + "mcp.client.streamable_http.aconnect_sse", + connect_sse, + ) + + transport = StreamableHTTPTransport("http://example.com/mcp") + async with httpx.AsyncClient() as client: + read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](1) + request = JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/call", params={"name": "noop", "arguments": {}}) + ctx = RequestContext( + client=client, + session_id=None, + session_message=SessionMessage(request), + metadata=None, + read_stream_writer=read_stream_writer, + ) + + async with read_stream_writer, read_stream: + with anyio.fail_after(5): + await transport._handle_reconnection(ctx, "event-1", retry_interval_ms=0) + message = await read_stream.receive() + + assert reconnect_attempts == 2 + assert isinstance(message, SessionMessage) + assert isinstance(message.message, JSONRPCError) + assert message.message.id == 1 + assert message.message.error.code == CONNECTION_CLOSED + + @pytest.mark.anyio async def test_sse_response_disconnect_ignores_closed_read_stream() -> None: transport = StreamableHTTPTransport("http://example.com/mcp") From e39f069015853a6aa414a2defdc102b2e7851950 Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Mon, 29 Jun 2026 00:57:35 +0200 Subject: [PATCH 08/14] test(streamable-http): keep reconnect fake covered --- tests/client/test_streamable_http.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index 09943cf1d4..c25a0ada42 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -10,6 +10,7 @@ import json from collections.abc import AsyncIterator, Callable, Mapping from typing import Any +from types import SimpleNamespace import anyio import httpx @@ -122,11 +123,7 @@ async def __aexit__(self, *args: object) -> None: return None async def aiter_sse(self) -> object: - yield type( - "SSE", - (), - {"event": "message", "data": "", "id": f"event-{reconnect_attempts}", "retry": 0}, - )() + yield SimpleNamespace(event="message", data="", id=f"event-{reconnect_attempts}", retry=0) def connect_sse(*args: object, **kwargs: object) -> PrimingOnlyEventSource: return PrimingOnlyEventSource() From 4c0174adeb00515104473a07103517b888298f91 Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Tue, 30 Jun 2026 19:46:25 +0200 Subject: [PATCH 09/14] test(streamable-http): exempt dropped async generator yield --- tests/client/test_streamable_http.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index c25a0ada42..31d6adc8b6 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -123,7 +123,12 @@ async def __aexit__(self, *args: object) -> None: return None async def aiter_sse(self) -> object: - yield SimpleNamespace(event="message", data="", id=f"event-{reconnect_attempts}", retry=0) + yield SimpleNamespace( # pragma: lax no cover - coverage.py drops this nested async-generator yield + event="message", + data="", + id=f"event-{reconnect_attempts}", + retry=0, + ) def connect_sse(*args: object, **kwargs: object) -> PrimingOnlyEventSource: return PrimingOnlyEventSource() From bb31f9b37eeaf5a29f8956d7bc6a77683a52a531 Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Tue, 30 Jun 2026 19:52:50 +0200 Subject: [PATCH 10/14] test(streamable-http): mark traced timeout guards no-branch --- tests/client/test_streamable_http.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index 31d6adc8b6..8d68d1947b 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -99,7 +99,7 @@ async def test_sse_response_disconnect_before_any_event_id_fails_request() -> No async with read_stream_writer, read_stream: await transport._handle_sse_response(response, ctx) - with anyio.fail_after(5): + with anyio.fail_after(5): # pragma: no branch message = await read_stream.receive() assert isinstance(message, SessionMessage) @@ -152,7 +152,7 @@ def connect_sse(*args: object, **kwargs: object) -> PrimingOnlyEventSource: ) async with read_stream_writer, read_stream: - with anyio.fail_after(5): + with anyio.fail_after(5): # pragma: no branch await transport._handle_reconnection(ctx, "event-1", retry_interval_ms=0) message = await read_stream.receive() From 9aad16dfa3159e775e66cd071958d7d4ccaac86c Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Thu, 2 Jul 2026 22:03:57 +0200 Subject: [PATCH 11/14] test(streamable-http): apply ruff import order --- tests/client/test_streamable_http.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index 8d68d1947b..5d18d367ee 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -9,8 +9,8 @@ import base64 import json from collections.abc import AsyncIterator, Callable, Mapping -from typing import Any from types import SimpleNamespace +from typing import Any import anyio import httpx From 95730a710f2e9d29d817583a5f65b060beb78965 Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Thu, 2 Jul 2026 22:12:50 +0200 Subject: [PATCH 12/14] test(streamable-http): cover closed SSE reader --- tests/client/test_streamable_http.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index 5d18d367ee..d6ccff29ac 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -15,6 +15,7 @@ import anyio import httpx import pytest +from httpx_sse import ServerSentEvent from inline_snapshot import snapshot from mcp_types import ( CLIENT_CAPABILITIES_META_KEY, @@ -183,6 +184,20 @@ async def test_sse_response_disconnect_ignores_closed_read_stream() -> None: await transport._handle_sse_response(response, ctx) +@pytest.mark.anyio +async def test_sse_message_ignores_closed_read_stream() -> None: + transport = StreamableHTTPTransport("http://example.com/mcp") + read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](1) + response = JSONRPCResponse(jsonrpc="2.0", id=1, result={}) + sse = ServerSentEvent(event="message", data=response.model_dump_json(by_alias=True)) + + async with read_stream_writer, read_stream: + await read_stream.aclose() + complete = await transport._handle_sse_event(sse, read_stream_writer, original_request_id=1) + + assert complete is True + + @pytest.mark.anyio async def test_post_request_merges_per_message_metadata_headers() -> None: """`ClientMessageMetadata.headers` on a `SessionMessage` are merged into the outgoing POST headers From e271dc176170e83d8af6b562e980766dbe6cde18 Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Tue, 7 Jul 2026 16:41:48 +0200 Subject: [PATCH 13/14] test(streamable-http): align reconnect exhaustion assertion --- tests/interaction/transports/test_hosting_resume.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/interaction/transports/test_hosting_resume.py b/tests/interaction/transports/test_hosting_resume.py index 1fe5d713bb..406e23149a 100644 --- a/tests/interaction/transports/test_hosting_resume.py +++ b/tests/interaction/transports/test_hosting_resume.py @@ -446,7 +446,7 @@ async def call() -> None: assert len(raised) == 1 assert isinstance(raised[0], Exception) - assert "disconnected" in str(raised[0]).lower() + assert "reconnection attempts were exhausted" in str(raised[0]).lower() @requirement("client-transport:http:resume-stream-api") From f7f12ad753a26743155a11f655379205b3d06a3c Mon Sep 17 00:00:00 2001 From: Jianke LIN Date: Tue, 7 Jul 2026 16:52:26 +0200 Subject: [PATCH 14/14] test(streamable-http): cover closed-send helper --- tests/client/test_streamable_http.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index d6ccff29ac..fbf3a7caa2 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -36,6 +36,7 @@ MAX_RECONNECTION_ATTEMPTS, RequestContext, StreamableHTTPTransport, + _send_or_ignore_closed, streamable_http_client, ) from mcp.server import Server @@ -198,6 +199,18 @@ async def test_sse_message_ignores_closed_read_stream() -> None: assert complete is True +@pytest.mark.anyio +async def test_send_or_ignore_closed_reports_delivery_status() -> None: + read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](1) + message = SessionMessage(JSONRPCResponse(jsonrpc="2.0", id=1, result={})) + + async with read_stream_writer, read_stream: + assert await _send_or_ignore_closed(read_stream_writer, message) is True # pyright: ignore[reportPrivateUsage] + assert await read_stream.receive() == message + await read_stream.aclose() + assert await _send_or_ignore_closed(read_stream_writer, message) is False # pyright: ignore[reportPrivateUsage] + + @pytest.mark.anyio async def test_post_request_merges_per_message_metadata_headers() -> None: """`ClientMessageMetadata.headers` on a `SessionMessage` are merged into the outgoing POST headers