test: add wire-level invariant tests via a recording transport

A RecordingTransport wrapper tees every message crossing the client's
transport boundary so the suite can assert properties that are invisible to
API callers: request ids are unique and never null, notifications are never
answered, and exactly one initialized notification is sent between the
initialize response and the first feature request.

Author: maxisbey

tests/interaction/_helpers.py

@@ -1,17 +1,107 @@
-"""Shared type aliases for the interaction suite.
+"""Shared helpers for the interaction suite.
 
-Keep this module small: it exists only for types that every test would otherwise have to
-assemble from the SDK's internals to annotate a client callback. Server fixtures and assertion
-helpers belong in the test that uses them.
+Keep this module small: it exists only for (a) types that every test would otherwise have to
+assemble from the SDK's internals to annotate a client callback, and (b) the recording transport
+used by the wire-level tests. Server fixtures and assertion helpers belong in the test that uses
+them.
 """
 
+from types import TracebackType
+
+import anyio
+from typing_extensions import Self
+
+from mcp.client._transport import ReadStream, Transport, TransportStreams, WriteStream
+from mcp.shared.message import SessionMessage
 from mcp.shared.session import RequestResponder
 from mcp.types import ClientResult, ServerNotification, ServerRequest
 
 # TODO: this union is the parameter type of every client message handler (MessageHandlerFnT),
 # but the SDK does not export a name for it -- writing a correctly-typed handler requires
 # importing RequestResponder from mcp.shared.session and assembling the union by hand. It
 # should be a named, exported alias next to MessageHandlerFnT (like ClientRequestContext is
-# for the request callbacks), at which point this module can be deleted.
+# for the request callbacks), at which point this alias can be deleted.
 IncomingMessage = RequestResponder[ServerRequest, ClientResult] | ServerNotification | Exception
 """Everything a client message handler can receive."""
+
+
+class _RecordingReadStream:
+    """Delegates to a read stream, appending every received message to a log."""
+
+    def __init__(self, inner: ReadStream[SessionMessage | Exception], log: list[SessionMessage | Exception]) -> None:
+        self._inner = inner
+        self._log = log
+
+    async def receive(self) -> SessionMessage | Exception:
+        item = await self._inner.receive()
+        self._log.append(item)
+        return item
+
+    async def aclose(self) -> None:
+        await self._inner.aclose()
+
+    def __aiter__(self) -> Self:
+        return self
+
+    async def __anext__(self) -> SessionMessage | Exception:
+        try:
+            return await self.receive()
+        except anyio.EndOfStream:
+            raise StopAsyncIteration from None
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> bool | None:
+        await self.aclose()
+        return None
+
+
+class _RecordingWriteStream:
+    """Delegates to a write stream, appending every sent message to a log."""
+
+    def __init__(self, inner: WriteStream[SessionMessage], log: list[SessionMessage]) -> None:
+        self._inner = inner
+        self._log = log
+
+    async def send(self, item: SessionMessage, /) -> None:
+        self._log.append(item)
+        await self._inner.send(item)
+
+    async def aclose(self) -> None:
+        await self._inner.aclose()
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> bool | None:
+        await self.aclose()
+        return None
+
+
+class RecordingTransport:
+    """Wraps a Transport and records every message crossing the client's transport boundary.
+
+    `sent` holds everything the client wrote towards the server; `received` holds everything the
+    server delivered to the client. The recording sits at the transport seam -- the exact payloads
+    a real transport would serialise -- and never touches the session, so wire-level assertions
+    written against it survive changes to the receive path.
+    """
+
+    def __init__(self, inner: Transport) -> None:
+        self.inner = inner
+        self.sent: list[SessionMessage] = []
+        self.received: list[SessionMessage | Exception] = []
+
+    async def __aenter__(self) -> TransportStreams:
+        read_stream, write_stream = await self.inner.__aenter__()
+        return _RecordingReadStream(read_stream, self.received), _RecordingWriteStream(write_stream, self.sent)
+
+    async def __aexit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> bool | None:
+        return await self.inner.__aexit__(exc_type, exc_val, exc_tb)

tests/interaction/_requirements.py

@@ -49,6 +49,20 @@ class Requirement:
     # ═══════════════════════════════════════════════════════════════════════════
     # Protocol primitives
     # ═══════════════════════════════════════════════════════════════════════════
+    "protocol:request-id:unique": Requirement(
+        source=f"{SPEC_BASE_URL}/basic#requests",
+        behavior=(
+            "Every request sent on a session carries a unique, non-null integer id; ids are never reused "
+            "within the session."
+        ),
+    ),
+    "protocol:notifications:no-response": Requirement(
+        source=f"{SPEC_BASE_URL}/basic#notifications",
+        behavior=(
+            "Notifications are never answered: every message the server delivers is either the response "
+            "to a request the client sent or a notification carrying no id."
+        ),
+    ),
     "protocol:error:internal-error": Requirement(
         source=f"{SPEC_BASE_URL}/basic#responses",
         behavior="An unhandled exception in a request handler is returned to the caller as a JSON-RPC error.",
@@ -106,6 +120,13 @@ class Requirement:
         source=f"{SPEC_BASE_URL}/basic/lifecycle#initialization",
         behavior="A request sent before the initialization handshake completes is rejected with an error.",
     ),
+    "lifecycle:initialized-notification": Requirement(
+        source=f"{SPEC_BASE_URL}/basic/lifecycle#initialization",
+        behavior=(
+            "The client sends exactly one initialized notification, after the initialize response and "
+            "before its first feature request."
+        ),
+    ),
     # ═══════════════════════════════════════════════════════════════════════════
     # Cancellation
     # ═══════════════════════════════════════════════════════════════════════════

tests/interaction/lowlevel/test_timeouts.py

@@ -86,12 +86,7 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
 
 @requirement("timeouts:session-default")
 async def test_session_level_timeout_applies_to_every_request() -> None:
-    """A read timeout configured on the client applies to requests that do not set their own.
-
-    The session default also governs the initialize handshake, so this is the one test in the
-    suite that needs a real (50ms) timeout: it must be long enough for the in-process handshake
-    to complete and is then waited out in full by the blocked tool call.
-    """
+    """A read timeout configured on the client applies to requests that do not set their own."""
 
     async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
         assert params.name == "block"
@@ -100,6 +95,12 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
 
     server = Server("blocker", on_call_tool=call_tool)
 
+    # The one real wall-clock wait in the suite, and it cannot be made effectively zero like the
+    # per-request timeouts: a session-level timeout also governs the initialize handshake, so the
+    # value must be long enough for the in-process handshake to complete before the blocked tool
+    # call waits it out in full. 50ms buys a ~50x safety margin over the handshake's actual
+    # latency; lowering it only erodes the margin against CI scheduler jitter without saving
+    # anything perceptible.
     async with Client(server, read_timeout_seconds=0.05) as client:
         with pytest.raises(MCPError) as exc_info:
             await client.call_tool("block", {})

tests/interaction/lowlevel/test_wire.py

@@ -0,0 +1,122 @@
+"""Wire-level invariants observed at the client's transport boundary.
+
+These behaviours are invisible to API callers -- they are properties of the raw JSON-RPC frames.
+The tests wrap the in-memory transport in a RecordingTransport, which tees every message crossing
+the transport seam into a list without touching the session, so the assertions hold for whatever
+the session implementation sends rather than for what its API returns.
+"""
+
+import anyio
+import pytest
+from inline_snapshot import snapshot
+
+from mcp import types
+from mcp.client._memory import InMemoryTransport
+from mcp.client.client import Client
+from mcp.server import Server, ServerRequestContext
+from mcp.shared.message import SessionMessage
+from mcp.types import CallToolResult, JSONRPCNotification, JSONRPCRequest, JSONRPCResponse, TextContent
+from tests.interaction._helpers import RecordingTransport, _RecordingReadStream
+from tests.interaction._requirements import requirement
+
+pytestmark = pytest.mark.anyio
+
+
+def _echo_server() -> Server:
+    """A server with one echo tool, used by every test in this module."""
+
+    async def list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(tools=[types.Tool(name="echo", input_schema={"type": "object"})])
+
+    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
+        assert params.name == "echo"
+        return CallToolResult(content=[TextContent(text="ok")])
+
+    return Server("wire", on_list_tools=list_tools, on_call_tool=call_tool)
+
+
+@requirement("protocol:request-id:unique")
+async def test_request_ids_are_unique_and_never_null() -> None:
+    """Every request the client sends carries a distinct, non-null id.
+
+    The id sequence is pinned: sequential integers from zero, in send order, including the
+    schema-cache refresh the client performs after the first successful tool call.
+    """
+    recording = RecordingTransport(InMemoryTransport(_echo_server()))
+
+    async with Client(recording) as client:
+        await client.list_tools()
+        await client.call_tool("echo", {})
+        await client.call_tool("echo", {})
+        await client.send_ping()
+
+    sent = [message.message for message in recording.sent]
+    request_ids = [message.id for message in sent if isinstance(message, JSONRPCRequest)]
+    assert all(request_id is not None for request_id in request_ids)
+    assert len(request_ids) == len(set(request_ids))
+    # initialize, tools/list, tools/call, tools/call, ping -- the client does not issue a
+    # schema-cache refresh here because the explicit tools/list already populated the cache.
+    assert request_ids == snapshot([0, 1, 2, 3, 4])
+
+
+@requirement("protocol:notifications:no-response")
+async def test_notifications_are_never_answered() -> None:
+    """A notification produces no response: everything the server sends back answers a request.
+
+    The client sends two notifications (initialized and roots/list_changed) and several requests;
+    the messages received from the server must be exactly one response per request, each carrying
+    the id of the request it answers, and nothing else.
+    """
+    recording = RecordingTransport(InMemoryTransport(_echo_server()))
+
+    async with Client(recording) as client:
+        await client.send_roots_list_changed()
+        await client.send_ping()
+
+    sent = [message.message for message in recording.sent]
+    sent_request_ids = [message.id for message in sent if isinstance(message, JSONRPCRequest)]
+    sent_notifications = [message for message in sent if isinstance(message, JSONRPCNotification)]
+    received = [message.message for message in recording.received if isinstance(message, SessionMessage)]
+    received_responses = [message for message in received if isinstance(message, JSONRPCResponse)]
+
+    assert len(sent_notifications) == 2  # notifications/initialized and notifications/roots/list_changed
+    assert len(received_responses) == len(received)  # nothing the server sent was anything but a response
+    assert [message.id for message in received_responses] == sent_request_ids
+
+
+async def test_recording_read_stream_ends_iteration_when_the_sender_closes() -> None:
+    """The recording wrapper preserves the end-of-stream behaviour of the stream it wraps.
+
+    This exercises the helper itself rather than an interaction-model behaviour: a transport whose
+    far end closes must end the client's receive loop cleanly, and the wrapper must not swallow or
+    mistranslate that.
+    """
+    send_stream, receive_stream = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+    log: list[SessionMessage | Exception] = []
+    async with send_stream, _RecordingReadStream(receive_stream, log) as wrapped:
+        await send_stream.aclose()
+        items = [item async for item in wrapped]
+    assert items == []
+    assert log == []
+
+
+@requirement("lifecycle:initialized-notification")
+async def test_exactly_one_initialized_notification_is_sent_after_the_handshake() -> None:
+    """The client sends initialized exactly once, between the initialize response and its first request.
+
+    The full method sequence the client puts on the wire is pinned in send order.
+    """
+    recording = RecordingTransport(InMemoryTransport(_echo_server()))
+
+    async with Client(recording) as client:
+        await client.list_tools()
+
+    sent_methods = [
+        message.message.method
+        for message in recording.sent
+        if isinstance(message.message, JSONRPCRequest | JSONRPCNotification)
+    ]
+    assert sent_methods.count("notifications/initialized") == 1
+    assert sent_methods == snapshot(["initialize", "notifications/initialized", "tools/list"])

tests/interaction/mcpserver/test_tools.py

@@ -173,9 +173,6 @@ def primes() -> list[int]:
 async def test_call_tool_invalid_arguments_become_error_result() -> None:
     """Arguments that fail validation against the tool's signature are reported as an is_error
     result describing the failure, not as a protocol error.
-
-    The description is raw pydantic output (version-dependent and leaking the internal argument
-    model name), so only the stable prefix is asserted rather than the full text.
     """
     mcp = MCPServer("adder")
 
@@ -187,6 +184,9 @@ def add(a: int, b: int) -> str:
     async with Client(mcp) as client:
         result = await client.call_tool("add", {"b": 3})
 
+    # The description is raw pydantic output -- it embeds a pydantic-version-specific
+    # errors.pydantic.dev URL and the internal `addArguments` model name -- so only the stable
+    # prefix is asserted; a full snapshot would break on every pydantic upgrade.
     assert result.is_error is True
     assert isinstance(result.content[0], TextContent)
     assert result.content[0].text.startswith("Error executing tool add: 1 validation error")