test: fix interaction suite for 3.10/3.11/3.14 and lowest-direct CI legs

- raise pytest floor to 8.4.0 (RaisesGroup, used in the auth tests)
- collect twice in connect_over_sse so the unclosed sse_stream_reader is
  finalized on 3.10 (PEP 442 cycle needs a second pass)
- collapse stacked async-with statements into comma form where it reads
  cleanly, and apply # pragma: no branch on the remaining sync-with +
  async-with shapes coverage.py mis-traces on 3.11/3.14

Author: maxisbey

pyproject.toml

@@ -78,7 +78,7 @@ dev = [
     # We add mcp[cli,ws] so `uv sync` considers the extras.
     "mcp[cli,ws]",
     "pyright>=1.1.400",
-    "pytest>=8.3.4",
+    "pytest>=8.4.0",
     "ruff>=0.8.5",
     "trio>=0.26.2",
     "pytest-flakefinder>=1.1.0",

tests/interaction/README.md

@@ -208,5 +208,8 @@ assert after the call, with no synchronisation. The exceptions:
 
 CI requires 100% line and branch coverage, including `tests/`, and `strict-no-cover` fails the
 build if a line marked `# pragma: no cover` is ever executed. When a new test starts covering a
-pragma'd line in `src/`, delete the pragma in the same change. Do not add new `# pragma`,
-`# type: ignore`, or `# noqa` comments; restructure instead.
+pragma'd line in `src/`, delete the pragma in the same change. Do not add new `# type: ignore` or
+`# noqa` comments; restructure instead. The one sanctioned pragma is `# pragma: no branch` on a
+`with`/`async with` line whose only fault is coverage.py mis-tracing the exit arc of a nested
+async context — restructure first, and reserve the pragma for shapes that cannot collapse (a sync
+`with` adjacent to an `async with`).

tests/interaction/_connect.py

@@ -130,20 +130,21 @@ async def connect_over_streamable_http(
         retry_interval=retry_interval,
         transport_security=NO_DNS_REBINDING_PROTECTION,
     )
-    async with server.session_manager.run():
-        async with httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http_client:
-            transport = streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client)
-            async with Client(
-                transport,
-                read_timeout_seconds=read_timeout_seconds,
-                sampling_callback=sampling_callback,
-                list_roots_callback=list_roots_callback,
-                logging_callback=logging_callback,
-                message_handler=message_handler,
-                client_info=client_info,
-                elicitation_callback=elicitation_callback,
-            ) as client:
-                yield client
+    async with (
+        server.session_manager.run(),
+        httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http_client,
+        Client(
+            streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client),
+            read_timeout_seconds=read_timeout_seconds,
+            sampling_callback=sampling_callback,
+            list_roots_callback=list_roots_callback,
+            logging_callback=logging_callback,
+            message_handler=message_handler,
+            client_info=client_info,
+            elicitation_callback=elicitation_callback,
+        ) as client,
+    ):
+        yield client
 
 
 @asynccontextmanager
@@ -183,11 +184,13 @@ async def mounted_app(
         auth_server_provider=auth_server_provider,
     )
     event_hooks = {"request": [on_request]} if on_request is not None else None
-    async with server.session_manager.run():
-        async with httpx.AsyncClient(
+    async with (
+        server.session_manager.run(),
+        httpx.AsyncClient(
             transport=StreamingASGITransport(app), base_url=BASE_URL, event_hooks=event_hooks, headers=headers
-        ) as http_client:
-            yield http_client, server.session_manager
+        ) as http_client,
+    ):
+        yield http_client, server.session_manager
 
 
 @asynccontextmanager
@@ -357,11 +360,13 @@ def httpx_client_factory(
         ) as client:
             yield client
     finally:
-        # SseServerTransport.connect_sse hands its internal SSE-chunk receive stream to
-        # sse_starlette's EventSourceResponse, which never closes it when its task group is
-        # cancelled on disconnect (see notes/findings.md). Collect the orphan here so its
-        # ResourceWarning fires deterministically inside this fixture instead of at an
-        # arbitrary later GC.
+        # SseServerTransport.connect_sse never closes its sse_stream_reader (handed to
+        # sse_starlette.EventSourceResponse, which does not aclose() its content on cancel).
+        # After teardown that reader is held only by a reference cycle through the connect_sse
+        # frame and its task objects; collecting twice runs the cycle's finalizers and then
+        # frees the reader while ResourceWarning is suppressed, instead of at an arbitrary
+        # later GC under pytest's error filter. One pass suffices on 3.11+; 3.10 needs both.
         with warnings.catch_warnings():
             warnings.simplefilter("ignore", ResourceWarning)
             gc.collect()
+            gc.collect()

tests/interaction/auth/_harness.py

@@ -11,7 +11,7 @@
 
 import json
 from collections.abc import AsyncIterator, Callable, Mapping, Sequence
-from contextlib import asynccontextmanager
+from contextlib import AsyncExitStack, asynccontextmanager
 from dataclasses import dataclass, field
 from typing import Any
 from urllib.parse import parse_qs, parse_qsl, urlsplit
@@ -451,11 +451,15 @@ async def hook(request: httpx.Request) -> None:
 
         event_hooks = {"request": [hook]}
 
-    async with server.session_manager.run():
-        async with httpx.AsyncClient(
-            transport=StreamingASGITransport(app), base_url=BASE_URL, auth=oauth, event_hooks=event_hooks
-        ) as http_client:
-            headless.bind(http_client)
-            transport = streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client)
-            async with Client(transport) as client:
-                yield client, headless
+    async with AsyncExitStack() as stack:
+        await stack.enter_async_context(server.session_manager.run())
+        http_client = await stack.enter_async_context(
+            httpx.AsyncClient(
+                transport=StreamingASGITransport(app), base_url=BASE_URL, auth=oauth, event_hooks=event_hooks
+            )
+        )
+        headless.bind(http_client)
+        client = await stack.enter_async_context(
+            Client(streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client))
+        )
+        yield client, headless

tests/interaction/lowlevel/test_cancellation.py

@@ -13,7 +13,7 @@
 from mcp import MCPError, types
 from mcp.client import ClientSession
 from mcp.server import Server, ServerRequestContext
-from mcp.shared.memory import create_client_server_memory_streams
+from mcp.shared.memory import MessageStream, create_client_server_memory_streams
 from mcp.shared.message import SessionMessage
 from mcp.types import (
     CallToolResult,
@@ -170,63 +170,65 @@ async def test_a_response_for_an_unknown_request_id_surfaces_to_the_message_hand
     scripted-peer mechanism is the in-memory stream pair, not because the behaviour is
     transport-specific.
     """
-    async with create_client_server_memory_streams() as (client_streams, server_streams):
-        client_read, client_write = client_streams
-        server_read, server_write = server_streams
-
-        async def scripted_server() -> None:
-            def respond(request_id: types.RequestId, result: types.Result) -> SessionMessage:
-                return SessionMessage(
-                    JSONRPCResponse(
-                        jsonrpc="2.0",
-                        id=request_id,
-                        # Serialized exactly as a real server serializes results onto the wire.
-                        result=result.model_dump(by_alias=True, mode="json", exclude_none=True),
-                    )
-                )
 
-            init = await server_read.receive()
-            assert isinstance(init, SessionMessage)
-            assert isinstance(init.message, JSONRPCRequest)
-            assert init.message.method == "initialize"
-            await server_write.send(
-                respond(
-                    init.message.id,
-                    InitializeResult(
-                        protocol_version="2025-11-25",
-                        capabilities=ServerCapabilities(),
-                        server_info=Implementation(name="scripted", version="0.0.1"),
-                    ),
+    async def scripted_server(streams: MessageStream) -> None:
+        server_read, server_write = streams
+
+        def respond(request_id: types.RequestId, result: types.Result) -> SessionMessage:
+            return SessionMessage(
+                JSONRPCResponse(
+                    jsonrpc="2.0",
+                    id=request_id,
+                    # Serialized exactly as a real server serializes results onto the wire.
+                    result=result.model_dump(by_alias=True, mode="json", exclude_none=True),
                 )
             )
 
-            initialized = await server_read.receive()
-            assert isinstance(initialized, SessionMessage)
-            assert isinstance(initialized.message, JSONRPCNotification)
-            assert initialized.message.method == "notifications/initialized"
-
-            ping = await server_read.receive()
-            assert isinstance(ping, SessionMessage)
-            assert isinstance(ping.message, JSONRPCRequest)
-            assert ping.message.method == "ping"
-            # First answer with a fabricated id that matches nothing in flight, then the real id.
-            await server_write.send(respond(9999, EmptyResult()))
-            await server_write.send(respond(ping.message.id, EmptyResult()))
-
-        incoming: list[IncomingMessage] = []
-
-        async def message_handler(message: IncomingMessage) -> None:
-            incoming.append(message)
-
-        async with anyio.create_task_group() as task_group:
-            task_group.start_soon(scripted_server)
-            async with ClientSession(client_read, client_write, message_handler=message_handler) as session:
-                with anyio.fail_after(5):
-                    await session.initialize()
-                    pong = await session.send_request(PingRequest(), EmptyResult)
-
-            assert pong == snapshot(EmptyResult())
-            assert len(incoming) == 1
-            assert isinstance(incoming[0], RuntimeError)
-            # The full message embeds the response object's repr; only the prefix is stable.
-            assert str(incoming[0]).startswith("Received response with an unknown request ID:")
+        init = await server_read.receive()
+        assert isinstance(init, SessionMessage)
+        assert isinstance(init.message, JSONRPCRequest)
+        assert init.message.method == "initialize"
+        await server_write.send(
+            respond(
+                init.message.id,
+                InitializeResult(
+                    protocol_version="2025-11-25",
+                    capabilities=ServerCapabilities(),
+                    server_info=Implementation(name="scripted", version="0.0.1"),
+                ),
+            )
+        )
+
+        initialized = await server_read.receive()
+        assert isinstance(initialized, SessionMessage)
+        assert isinstance(initialized.message, JSONRPCNotification)
+        assert initialized.message.method == "notifications/initialized"
+
+        ping = await server_read.receive()
+        assert isinstance(ping, SessionMessage)
+        assert isinstance(ping.message, JSONRPCRequest)
+        assert ping.message.method == "ping"
+        # First answer with a fabricated id that matches nothing in flight, then the real id.
+        await server_write.send(respond(9999, EmptyResult()))
+        await server_write.send(respond(ping.message.id, EmptyResult()))
+
+    incoming: list[IncomingMessage] = []
+
+    async def message_handler(message: IncomingMessage) -> None:
+        incoming.append(message)
+
+    async with (
+        create_client_server_memory_streams() as ((client_read, client_write), server_streams),
+        anyio.create_task_group() as task_group,
+        ClientSession(client_read, client_write, message_handler=message_handler) as session,
+    ):
+        task_group.start_soon(scripted_server, server_streams)
+        with anyio.fail_after(5):
+            await session.initialize()
+            pong = await session.send_request(PingRequest(), EmptyResult)
+
+        assert pong == snapshot(EmptyResult())
+        assert len(incoming) == 1
+        assert isinstance(incoming[0], RuntimeError)
+        # The full message embeds the response object's repr; only the prefix is stable.
+        assert str(incoming[0]).startswith("Received response with an unknown request ID:")

tests/interaction/lowlevel/test_elicitation.py

@@ -11,7 +11,7 @@
 from mcp import MCPError, UrlElicitationRequiredError, types
 from mcp.client import ClientRequestContext, ClientSession
 from mcp.server import Server, ServerRequestContext
-from mcp.shared.memory import create_client_server_memory_streams
+from mcp.shared.memory import MessageStream, create_client_server_memory_streams
 from mcp.shared.message import SessionMessage
 from mcp.types import (
     CallToolResult,
@@ -594,68 +594,68 @@ async def answer_form(context: ClientRequestContext, params: types.ElicitRequest
         received.append(params)
         return ElicitResult(action="accept", content={})
 
-    async with create_client_server_memory_streams() as (client_streams, server_streams):
-        client_read, client_write = client_streams
-        server_read, server_write = server_streams
-
-        async def scripted_server() -> None:
-            initialize = await server_read.receive()
-            assert isinstance(initialize, SessionMessage)
-            request = initialize.message
-            assert isinstance(request, JSONRPCRequest)
-            assert request.method == "initialize"
-            result = InitializeResult(
-                protocol_version="2025-11-25",
-                capabilities=ServerCapabilities(),
-                server_info=Implementation(name="legacy", version="0.0.1"),
-            )
-            await server_write.send(
-                SessionMessage(
-                    JSONRPCResponse(
-                        jsonrpc="2.0",
-                        id=request.id,
-                        result=result.model_dump(by_alias=True, mode="json", exclude_none=True),
-                    )
+    async def scripted_server(streams: MessageStream) -> None:
+        server_read, server_write = streams
+        initialize = await server_read.receive()
+        assert isinstance(initialize, SessionMessage)
+        request = initialize.message
+        assert isinstance(request, JSONRPCRequest)
+        assert request.method == "initialize"
+        result = InitializeResult(
+            protocol_version="2025-11-25",
+            capabilities=ServerCapabilities(),
+            server_info=Implementation(name="legacy", version="0.0.1"),
+        )
+        await server_write.send(
+            SessionMessage(
+                JSONRPCResponse(
+                    jsonrpc="2.0",
+                    id=request.id,
+                    result=result.model_dump(by_alias=True, mode="json", exclude_none=True),
                 )
             )
-            initialized = await server_read.receive()
-            assert isinstance(initialized, SessionMessage)
-            assert isinstance(initialized.message, JSONRPCNotification)
-            assert initialized.message.method == "notifications/initialized"
-            # No mode key: a server speaking a pre-mode revision of the spec sends only message + schema.
-            await server_write.send(
-                SessionMessage(
-                    JSONRPCRequest(
-                        jsonrpc="2.0",
-                        id=2,
-                        method="elicitation/create",
-                        params={"message": "Legacy ask.", "requestedSchema": {"type": "object", "properties": {}}},
-                    )
+        )
+        initialized = await server_read.receive()
+        assert isinstance(initialized, SessionMessage)
+        assert isinstance(initialized.message, JSONRPCNotification)
+        assert initialized.message.method == "notifications/initialized"
+        # No mode key: a server speaking a pre-mode revision of the spec sends only message + schema.
+        await server_write.send(
+            SessionMessage(
+                JSONRPCRequest(
+                    jsonrpc="2.0",
+                    id=2,
+                    method="elicitation/create",
+                    params={"message": "Legacy ask.", "requestedSchema": {"type": "object", "properties": {}}},
                 )
             )
-            response = await server_read.receive()
-            assert isinstance(response, SessionMessage)
-            server_received.append(response.message)
-            answered.set()
-
-        async with anyio.create_task_group() as tg:
-            tg.start_soon(scripted_server)
-            async with ClientSession(client_read, client_write, elicitation_callback=answer_form) as session:
-                with anyio.fail_after(5):
-                    await session.initialize()
-                    await answered.wait()
-
-            assert received == snapshot(
-                [
-                    ElicitRequestFormParams(
-                        _meta=None,
-                        message="Legacy ask.",
-                        requested_schema={"type": "object", "properties": {}},
-                    )
-                ]
+        )
+        response = await server_read.receive()
+        assert isinstance(response, SessionMessage)
+        server_received.append(response.message)
+        answered.set()
+
+    async with (
+        create_client_server_memory_streams() as ((client_read, client_write), server_streams),
+        anyio.create_task_group() as tg,
+        ClientSession(client_read, client_write, elicitation_callback=answer_form) as session,
+    ):
+        tg.start_soon(scripted_server, server_streams)
+        with anyio.fail_after(5):
+            await session.initialize()
+            await answered.wait()
+
+    assert received == snapshot(
+        [
+            ElicitRequestFormParams(
+                _meta=None,
+                message="Legacy ask.",
+                requested_schema={"type": "object", "properties": {}},
             )
-            assert isinstance(received[0], ElicitRequestFormParams)
-            assert received[0].mode == "form"
-            assert len(server_received) == 1
-            assert isinstance(server_received[0], JSONRPCResponse)
-            assert server_received[0].id == 2
+        ]
+    )
+    assert isinstance(received[0], ElicitRequestFormParams)
+    assert received[0].mode == "form"
+    assert len(server_received) == 1
+    assert isinstance(server_received[0], JSONRPCResponse)
+    assert server_received[0].id == 2