test: run the interaction suite over both in-memory and streamable HTTP transports

Author: maxisbey

src/mcp/server/streamable_http.py

@@ -374,7 +374,7 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No
             await error_response(scope, receive, send)
             return
 
-        if self._terminated:  # pragma: no cover
+        if self._terminated:  # pragma: lax no cover
             # If the session has been terminated, return 404 Not Found
             response = self._create_error_response(
                 "Not Found: Session has been terminated",
@@ -635,7 +635,7 @@ async def sse_writer():  # pragma: lax no cover
                 finally:
                     await sse_stream_reader.aclose()
 
-        except Exception as err:  # pragma: no cover
+        except Exception as err:  # pragma: lax no cover
             logger.exception("Error handling POST request")
             response = self._create_error_response(
                 f"Error handling POST request: {err}",

tests/interaction/README.md

@@ -37,17 +37,31 @@ The whole suite is in-memory and event-driven; it runs in about a second.
 tests/interaction/
   _requirements.py      the requirements manifest (see below)
   _helpers.py           shared type aliases + the wire-recording transport
+  _connect.py           the transport-parametrized connection factories
+  conftest.py           the connect fixture (the transport matrix)
   test_coverage.py      enforces the manifest ↔ test contract
   lowlevel/             one file per feature area, against the low-level Server
   mcpserver/            the same feature areas in MCPServer's natural idiom
-  transports/           a smoke subset over the streamable HTTP framing
+  transports/           behaviour specific to one transport (modes, streams, framing)
 ```
 
 The two server APIs produce genuinely different wire output for the same conceptual feature
 (`MCPServer` generates schemas, converts exceptions to `isError` results, attaches structured
 content), so they get parallel directories with mirrored file names rather than one parametrized
 test body — each directory pins its flavour's true output exactly.
 
+### The transport matrix
+
+Transport-agnostic tests take the `connect` fixture instead of constructing `Client(server)`
+directly, and therefore run once per transport: over the in-memory transport and over the
+server's real streamable HTTP app driven in process through the streaming bridge. A test connects
+the same way in either case — `async with connect(server, ...) as client:` — and asserts the same
+output, because the transport is not supposed to change observable behaviour. Tests that are tied
+to one transport do not use the fixture: the wire-recording tests (their seam is the in-memory
+stream pair), the bare-`ClientSession` lifecycle tests, the real-clock timeout tests (the timeout
+machinery is transport-independent and must not race transport latency), and everything under
+`transports/`, which pins behaviour only observable on that transport.
+
 ## The requirements manifest
 
 `_requirements.py` maps every behaviour the suite covers to the reason it must hold:

tests/interaction/_connect.py

@@ -0,0 +1,117 @@
+"""Transport-parametrized connection factories for the interaction suite.
+
+The `connect` fixture (see conftest.py) hands tests one of these factories so the same test body
+runs over the in-memory transport and over streamable HTTP without naming either: the factory is a
+drop-in replacement for constructing `Client(server, ...)` and yields the connected client. The
+streamable HTTP factory drives the server's real Starlette app through the in-process streaming
+bridge, so the full HTTP framing layer (session ids, SSE encoding, session management) runs with
+no sockets, threads, or subprocesses.
+"""
+
+from collections.abc import AsyncIterator
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from typing import Protocol
+
+import httpx
+
+from mcp.client.client import Client
+from mcp.client.session import ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
+from mcp.client.streamable_http import streamable_http_client
+from mcp.server import Server
+from mcp.server.mcpserver import MCPServer
+from mcp.server.transport_security import TransportSecuritySettings
+from mcp.types import Implementation
+from tests.interaction.transports._bridge import StreamingASGITransport
+
+# The in-process app is mounted at this origin purely so URLs are well-formed; nothing listens here.
+_BASE_URL = "http://127.0.0.1:8000"
+
+
+class Connect(Protocol):
+    """Connect a Client to a server over the transport selected by the `connect` fixture.
+
+    Accepts the same keyword arguments as `Client` and yields the connected client.
+    """
+
+    def __call__(
+        self,
+        server: Server | MCPServer,
+        *,
+        read_timeout_seconds: float | None = None,
+        sampling_callback: SamplingFnT | None = None,
+        list_roots_callback: ListRootsFnT | None = None,
+        logging_callback: LoggingFnT | None = None,
+        message_handler: MessageHandlerFnT | None = None,
+        client_info: Implementation | None = None,
+        elicitation_callback: ElicitationFnT | None = None,
+    ) -> AbstractAsyncContextManager[Client]: ...
+
+
+@asynccontextmanager
+async def connect_in_memory(
+    server: Server | MCPServer,
+    *,
+    read_timeout_seconds: float | None = None,
+    sampling_callback: SamplingFnT | None = None,
+    list_roots_callback: ListRootsFnT | None = None,
+    logging_callback: LoggingFnT | None = None,
+    message_handler: MessageHandlerFnT | None = None,
+    client_info: Implementation | None = None,
+    elicitation_callback: ElicitationFnT | None = None,
+) -> AsyncIterator[Client]:
+    """Yield a Client connected to the server over the in-memory transport."""
+    async with Client(
+        server,
+        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
+async def connect_over_streamable_http(
+    server: Server | MCPServer,
+    *,
+    stateless_http: bool = False,
+    json_response: bool = False,
+    read_timeout_seconds: float | None = None,
+    sampling_callback: SamplingFnT | None = None,
+    list_roots_callback: ListRootsFnT | None = None,
+    logging_callback: LoggingFnT | None = None,
+    message_handler: MessageHandlerFnT | None = None,
+    client_info: Implementation | None = None,
+    elicitation_callback: ElicitationFnT | None = None,
+) -> AsyncIterator[Client]:
+    """Yield a Client connected to the server's streamable HTTP app, entirely in process.
+
+    With the defaults this is the matrix leg (stateful sessions, SSE responses); the
+    transport-specific tests pass `stateless_http` or `json_response` to select the other
+    server modes.
+    """
+    # DNS-rebinding protection validates Host/Origin headers against a real network attack that
+    # cannot exist for an in-process ASGI app; leaving it on would also pull the origin-validation
+    # branch (deliberately uncovered in src) into coverage.
+    app = server.streamable_http_app(
+        stateless_http=stateless_http,
+        json_response=json_response,
+        transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
+    )
+    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

tests/interaction/conftest.py

@@ -0,0 +1,22 @@
+"""Shared fixtures for the interaction suite."""
+
+import pytest
+
+from tests.interaction._connect import Connect, connect_in_memory, connect_over_streamable_http
+
+_FACTORIES: dict[str, Connect] = {
+    "in-memory": connect_in_memory,
+    "streamable-http": connect_over_streamable_http,
+}
+
+
+@pytest.fixture(params=sorted(_FACTORIES))
+def connect(request: pytest.FixtureRequest) -> Connect:
+    """The transport-parametrized connection factory: a test using it runs once per transport.
+
+    Tests that are tied to one transport (the wire-recording tests, the bare-ClientSession tests,
+    the transport-specific tests under transports/) do not use this fixture and connect directly.
+    """
+    transport_name = request.param
+    assert isinstance(transport_name, str)
+    return _FACTORIES[transport_name]

tests/interaction/lowlevel/test_cancellation.py

@@ -11,17 +11,17 @@
 from inline_snapshot import snapshot
 
 from mcp import MCPError, types
-from mcp.client.client import Client
 from mcp.server import Server, ServerRequestContext
 from mcp.types import CallToolResult, ErrorData, TextContent
+from tests.interaction._connect import Connect
 from tests.interaction._requirements import requirement
 
 pytestmark = pytest.mark.anyio
 
 
 @requirement("protocol:cancel:in-flight")
 @requirement("protocol:cancel:handler-abort-propagates")
-async def test_cancellation_stops_in_flight_handler() -> None:
+async def test_cancellation_stops_in_flight_handler(connect: Connect) -> None:
     """Cancelling an in-flight request interrupts its handler and fails the pending call.
 
     The server answers the cancelled request with an error response (the spec says it should
@@ -47,7 +47,7 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
 
     server = Server("blocker", on_call_tool=call_tool)
 
-    async with Client(server) as client:
+    async with connect(server) as client:
         with anyio.fail_after(5):
             async with anyio.create_task_group() as task_group:
 
@@ -70,7 +70,7 @@ async def call_and_capture_error() -> None:
 
 
 @requirement("protocol:cancel:server-survives")
-async def test_session_serves_requests_after_cancellation() -> None:
+async def test_session_serves_requests_after_cancellation(connect: Connect) -> None:
     """A request cancelled mid-flight does not poison the session: the next request succeeds."""
     started = anyio.Event()
     request_ids: list[types.RequestId] = []
@@ -96,7 +96,7 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
 
     server = Server("blocker", on_list_tools=list_tools, on_call_tool=call_tool)
 
-    async with Client(server) as client:
+    async with connect(server) as client:
         with anyio.fail_after(5):
             async with anyio.create_task_group() as task_group:
 
@@ -116,7 +116,7 @@ async def call_and_swallow_cancellation_error() -> None:
 
 
 @requirement("protocol:cancel:unknown-id-ignored")
-async def test_cancellation_for_unknown_request_is_ignored() -> None:
+async def test_cancellation_for_unknown_request_is_ignored(connect: Connect) -> None:
     """A cancellation referencing a request id that is not in flight is ignored without error."""
 
     async def list_tools(
@@ -130,7 +130,7 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
 
     server = Server("calm", on_list_tools=list_tools, on_call_tool=call_tool)
 
-    async with Client(server) as client:
+    async with connect(server) as client:
         await client.session.send_notification(
             types.CancelledNotification(params=types.CancelledNotificationParams(request_id=9999))
         )

tests/interaction/lowlevel/test_completion.py

@@ -4,7 +4,6 @@
 from inline_snapshot import snapshot
 
 from mcp import MCPError, types
-from mcp.client.client import Client
 from mcp.server import Server, ServerRequestContext
 from mcp.types import (
     METHOD_NOT_FOUND,
@@ -14,14 +13,15 @@
     PromptReference,
     ResourceTemplateReference,
 )
+from tests.interaction._connect import Connect
 from tests.interaction._requirements import requirement
 
 pytestmark = pytest.mark.anyio
 
 
 @requirement("completion:prompt-arg")
 @requirement("completion:result-shape")
-async def test_complete_prompt_argument() -> None:
+async def test_complete_prompt_argument(connect: Connect) -> None:
     """Completing a prompt argument delivers the ref, argument name, and current value to the handler.
 
     The returned values are filtered by the argument's value, proving the value reached the handler.
@@ -37,7 +37,7 @@ async def completion(ctx: ServerRequestContext, params: types.CompleteRequestPar
 
     server = Server("completer", on_completion=completion)
 
-    async with Client(server) as client:
+    async with connect(server) as client:
         result = await client.complete(
             PromptReference(name="code_review"), argument={"name": "language", "value": "py"}
         )
@@ -48,7 +48,7 @@ async def completion(ctx: ServerRequestContext, params: types.CompleteRequestPar
 
 
 @requirement("completion:resource-template-arg")
-async def test_complete_resource_template_variable() -> None:
+async def test_complete_resource_template_variable(connect: Connect) -> None:
     """Completing a URI template variable delivers the template URI and variable name to the handler."""
 
     async def completion(ctx: ServerRequestContext, params: types.CompleteRequestParams) -> CompleteResult:
@@ -59,7 +59,7 @@ async def completion(ctx: ServerRequestContext, params: types.CompleteRequestPar
 
     server = Server("completer", on_completion=completion)
 
-    async with Client(server) as client:
+    async with connect(server) as client:
         result = await client.complete(
             ResourceTemplateReference(uri="github://repos/{owner}/{repo}"),
             argument={"name": "owner", "value": "model"},
@@ -69,7 +69,7 @@ async def completion(ctx: ServerRequestContext, params: types.CompleteRequestPar
 
 
 @requirement("completion:context-arguments")
-async def test_complete_receives_context_arguments() -> None:
+async def test_complete_receives_context_arguments(connect: Connect) -> None:
     """Previously-resolved arguments passed as completion context reach the handler.
 
     The returned value is derived from the context, proving it arrived.
@@ -83,7 +83,7 @@ async def completion(ctx: ServerRequestContext, params: types.CompleteRequestPar
 
     server = Server("completer", on_completion=completion)
 
-    async with Client(server) as client:
+    async with connect(server) as client:
         result = await client.complete(
             ResourceTemplateReference(uri="github://repos/{owner}/{repo}"),
             argument={"name": "repo", "value": ""},
@@ -95,11 +95,11 @@ async def completion(ctx: ServerRequestContext, params: types.CompleteRequestPar
 
 @requirement("completion:complete:not-supported")
 @requirement("protocol:error:method-not-found")
-async def test_complete_without_handler_is_method_not_found() -> None:
+async def test_complete_without_handler_is_method_not_found(connect: Connect) -> None:
     """A server with no completion handler advertises no completions capability and rejects the request."""
     server = Server("incomplete")
 
-    async with Client(server) as client:
+    async with connect(server) as client:
         assert client.initialize_result.capabilities.completions is None
 
         with pytest.raises(MCPError) as exc_info: