Interaction matrix: era-aware unstamped fixture

The shared strip helper was if-present, so a modern cell whose result
lost its serverInfo stamp would still pass. The helper is now strict
(stamp must be present and well-formed before it is stripped), and the
interaction matrix resolves 'unstamped' as a fixture keyed on the
cell's spec_version: modern cells assert-and-strip, handshake-era
cells assert the result was never stamped. Every existing comparison
site now enforces the stamp contract in both directions.

The connect fixture returns a CellConnect callable that names its
cell's spec_version, so sibling fixtures can key on the era without
re-deriving it.

No-Verification-Needed: test-only change

Author: maxisbey

tests/_stamp.py

@@ -1,32 +1,43 @@
-"""Shared helper: drop the 2026-era serverInfo `_meta` stamp from a result.
+"""Shared helper: strip the 2026-era serverInfo `_meta` stamp from a result.
 
 Servers stamp `io.modelcontextprotocol/serverInfo` into every 2026-era
 result's `_meta` and never into handshake-era ones, and the stamp's `version`
-defaults to the installed package version. Suites that share one expected
-payload across eras (the interaction matrix) or that construct servers
-without a pinned version strip the stamp before exact comparison. Stamp
-semantics themselves have dedicated coverage in tests/server/test_runner.py.
+defaults to the installed package version. Suites whose expected payloads
+should stay identity-free strip the stamp before exact comparison - and the
+strip is strict, so a modern result that lost its stamp fails the test
+instead of passing silently.
+
+The interaction matrix does not use this function directly: its `unstamped`
+fixture (tests/interaction/conftest.py) resolves per cell to this strict
+strip on modern cells and to a must-not-be-stamped assertion on
+handshake-era cells, so one comparison line enforces both eras.
 """
 
-from typing import Any, TypeVar
+from typing import Any, Protocol, TypeVar
 
 from mcp_types import SERVER_INFO_META_KEY, Result
 
 R = TypeVar("R", bound=Result)
 
 
+class Unstamp(Protocol):
+    """An era-appropriate stamp normalizer: strips or forbids the stamp."""
+
+    def __call__(self, result: R) -> R: ...
+
+
 def unstamped(result: R) -> R:
-    """Remove the serverInfo stamp in place if present, asserting its shape.
+    """Assert the result carries a well-formed serverInfo stamp, then remove it.
 
-    Present-versus-absent is era-dependent and belongs to the runner tests;
-    this only guarantees that when a stamp exists it is a well-formed
-    identity object, then returns the result for inline use in comparisons.
+    Returns the result for inline use in comparisons. Use only where a stamp
+    is required (a 2026-era result); the interaction matrix's `unstamped`
+    fixture handles the era split.
     """
     meta = result.meta
-    if meta is not None and SERVER_INFO_META_KEY in meta:
-        stamp: Any = meta.pop(SERVER_INFO_META_KEY)
-        assert isinstance(stamp, dict)
-        assert "name" in stamp and "version" in stamp
-        if not meta:
-            result.meta = None
+    assert meta is not None and SERVER_INFO_META_KEY in meta, "expected a serverInfo stamp on this result"
+    stamp: Any = meta.pop(SERVER_INFO_META_KEY)
+    assert isinstance(stamp, dict)
+    assert "name" in stamp and "version" in stamp
+    if not meta:
+        result.meta = None
     return result

tests/interaction/conftest.py

@@ -10,10 +10,14 @@
 from typing import Any
 
 import pytest
+from mcp_types import SERVER_INFO_META_KEY
+from mcp_types.version import MODERN_PROTOCOL_VERSIONS
 
 from mcp.client.client import Client
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
+from tests._stamp import R, Unstamp
+from tests._stamp import unstamped as _strip_required_stamp
 from tests.interaction._connect import (
     Connect,
     connect_in_memory,
@@ -39,26 +43,56 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
     metafunc.parametrize("connect", compute_cells(requirements), indirect=True)
 
 
-@pytest.fixture
-def connect(request: pytest.FixtureRequest) -> Connect:
-    """The transport-parametrized connection factory: a test using it runs once per matrix cell.
+class CellConnect:
+    """The cell's connection factory, also naming the cell's `spec_version`.
 
-    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.
+    Callable exactly like the `Connect` factories it wraps; the attribute lets
+    sibling fixtures (`unstamped`) key on the cell's era without re-deriving it.
     """
-    transport, spec_version = request.param
-    assert isinstance(transport, str)
-    assert isinstance(spec_version, str)
-    factory = _FACTORIES[transport]
 
-    def _connect(server: Server | MCPServer, **kwargs: Any) -> AbstractAsyncContextManager[Client]:
+    def __init__(self, factory: Connect, spec_version: str) -> None:
+        self._factory = factory
+        self.spec_version = spec_version
+
+    def __call__(self, server: Server | MCPServer, **kwargs: Any) -> AbstractAsyncContextManager[Client]:
         # The matrix compares exact result payloads, and the 2026-era serverInfo
         # `_meta` stamp carries the server version, which defaults to the
         # commit-dependent installed package version. Pin it so expected
         # payloads stay deterministic across commits.
         lowlevel = server._lowlevel_server if isinstance(server, MCPServer) else server
         if lowlevel.version is None:
             lowlevel.version = "1.0.0"
-        return factory(server, spec_version=spec_version, **kwargs)
+        return self._factory(server, spec_version=self.spec_version, **kwargs)
+
+
+@pytest.fixture
+def connect(request: pytest.FixtureRequest) -> CellConnect:
+    """The transport-parametrized connection factory: a test using it runs once per matrix cell.
+
+    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, spec_version = request.param
+    assert isinstance(transport, str)
+    assert isinstance(spec_version, str)
+    return CellConnect(_FACTORIES[transport], spec_version)
+
+
+@pytest.fixture
+def unstamped(connect: CellConnect) -> Unstamp:
+    """The cell's era-aware serverInfo-stamp normalizer, for full-result comparisons.
+
+    On a modern cell the stamp MUST be present (asserted) and is stripped so
+    one expected payload stays valid across eras; on a handshake-era cell the
+    result must not be stamped at all. Either direction failing is a runner
+    regression, so the same comparison line enforces both.
+    """
+    if connect.spec_version in MODERN_PROTOCOL_VERSIONS:
+        return _strip_required_stamp
+
+    def _assert_never_stamped(result: R) -> R:
+        meta = result.meta
+        assert meta is None or SERVER_INFO_META_KEY not in meta, "handshake-era results are never stamped"
+        return result
 
-    return _connect
+    return _assert_never_stamped

tests/interaction/lowlevel/test_cancellation.py

@@ -33,7 +33,7 @@
 from mcp.server import Server, ServerRequestContext
 from mcp.shared.memory import MessageStream, create_client_server_memory_streams
 from mcp.shared.message import SessionMessage
-from tests._stamp import unstamped
+from tests._stamp import Unstamp
 from tests.interaction._connect import Connect
 from tests.interaction._helpers import IncomingMessage
 from tests.interaction._requirements import requirement
@@ -138,7 +138,7 @@ async def call_and_swallow_cancellation_error() -> None:
 
 
 @requirement("protocol:cancel:unknown-id-ignored")
-async def test_cancellation_for_unknown_request_is_ignored(connect: Connect) -> None:
+async def test_cancellation_for_unknown_request_is_ignored(connect: Connect, unstamped: Unstamp) -> None:
     """A cancellation referencing a request id that is not in flight is ignored without error."""
 
     async def list_tools(
@@ -351,7 +351,7 @@ async def scripted_server(streams: MessageStream) -> None:
 
 
 @requirement("protocol:cancel:abort-signal")
-async def test_abandoning_a_call_stops_the_server_handler(connect: Connect) -> None:
+async def test_abandoning_a_call_stops_the_server_handler(connect: Connect, unstamped: Unstamp) -> None:
     """Cancelling the task that awaits a call cancels the request itself, not just the local wait:
     the server-side handler is interrupted, and the session serves later requests normally.
 
@@ -402,7 +402,7 @@ async def call_and_abandon() -> None:
 
 
 @requirement("protocol:cancel:abort-scoped")
-async def test_abandoning_one_call_leaves_a_concurrent_call_running(connect: Connect) -> None:
+async def test_abandoning_one_call_leaves_a_concurrent_call_running(connect: Connect, unstamped: Unstamp) -> None:
     """Cancellation is scoped to the request it names: with two calls genuinely in flight,
     abandoning the first interrupts only its handler and the second returns its result.
 

tests/interaction/lowlevel/test_completion.py

@@ -15,7 +15,7 @@
 
 from mcp import MCPError
 from mcp.server import Server, ServerRequestContext
-from tests._stamp import unstamped
+from tests._stamp import Unstamp
 from tests.interaction._connect import Connect
 from tests.interaction._requirements import requirement
 
@@ -24,7 +24,7 @@
 
 @requirement("completion:prompt-arg")
 @requirement("completion:result-shape")
-async def test_complete_prompt_argument(connect: Connect) -> None:
+async def test_complete_prompt_argument(connect: Connect, unstamped: Unstamp) -> 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.
@@ -51,7 +51,7 @@ async def completion(ctx: ServerRequestContext, params: types.CompleteRequestPar
 
 
 @requirement("completion:resource-template-arg")
-async def test_complete_resource_template_variable(connect: Connect) -> None:
+async def test_complete_resource_template_variable(connect: Connect, unstamped: Unstamp) -> 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:
@@ -72,7 +72,7 @@ async def completion(ctx: ServerRequestContext, params: types.CompleteRequestPar
 
 
 @requirement("completion:context-arguments")
-async def test_complete_receives_context_arguments(connect: Connect) -> None:
+async def test_complete_receives_context_arguments(connect: Connect, unstamped: Unstamp) -> None:
     """Previously-resolved arguments passed as completion context reach the handler.
 
     The returned value is derived from the context, proving it arrived.

tests/interaction/lowlevel/test_flows.py

@@ -32,7 +32,7 @@
 from mcp.client import ClientRequestContext
 from mcp.server import Server, ServerRequestContext
 from mcp.server.session import ServerSession
-from tests._stamp import unstamped
+from tests._stamp import Unstamp
 from tests.interaction._connect import Connect
 from tests.interaction._helpers import IncomingMessage
 from tests.interaction._requirements import requirement
@@ -54,7 +54,9 @@ async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestPa
 
 
 @requirement("flow:tool-result:resource-link-follow")
-async def test_a_resource_link_returned_by_a_tool_can_be_followed_with_read(connect: Connect) -> None:
+async def test_a_resource_link_returned_by_a_tool_can_be_followed_with_read(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """A tool returns a resource_link; reading that link's URI returns the referenced contents.
 
     Steps: (1) call the tool, (2) extract the link from its content, (3) read_resource on the

tests/interaction/lowlevel/test_logging.py

@@ -11,7 +11,7 @@
 from mcp_types import CallToolResult, EmptyResult, LoggingMessageNotificationParams, TextContent
 
 from mcp.server import Server, ServerRequestContext
-from tests._stamp import unstamped
+from tests._stamp import Unstamp
 from tests.interaction._connect import Connect
 from tests.interaction._requirements import requirement
 
@@ -47,7 +47,7 @@ async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelReq
 
 @requirement("logging:message:fields")
 @requirement("tools:call:logging-mid-execution")
-async def test_log_messages_reach_logging_callback_in_order(connect: Connect) -> None:
+async def test_log_messages_reach_logging_callback_in_order(connect: Connect, unstamped: Unstamp) -> None:
     """Log messages sent during a tool call arrive at the logging callback, in order, before the call returns.
 
     The two messages pin the full notification shape: severity, optional logger name, and both

tests/interaction/lowlevel/test_meta.py

@@ -11,7 +11,7 @@
 from mcp_types import CallToolResult, RequestParamsMeta, TextContent
 
 from mcp.server import Server, ServerRequestContext
-from tests._stamp import unstamped
+from tests._stamp import Unstamp
 from tests.interaction._connect import Connect
 from tests.interaction._requirements import requirement
 
@@ -44,7 +44,7 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
 
 
 @requirement("meta:result-to-client")
-async def test_result_meta_reaches_client(connect: Connect) -> None:
+async def test_result_meta_reaches_client(connect: Connect, unstamped: Unstamp) -> None:
     """The _meta object a handler attaches to its result is delivered to the client unchanged."""
     result_meta = {"example.com/cost": 3}
 

tests/interaction/lowlevel/test_pagination.py

@@ -22,15 +22,15 @@
 
 from mcp import MCPError
 from mcp.server import Server, ServerRequestContext
-from tests._stamp import unstamped
+from tests._stamp import Unstamp
 from tests.interaction._connect import Connect
 from tests.interaction._requirements import requirement
 
 pytestmark = pytest.mark.anyio
 
 
 @requirement("tools:list:pagination")
-async def test_next_cursor_round_trips_through_the_client(connect: Connect) -> None:
+async def test_next_cursor_round_trips_through_the_client(connect: Connect, unstamped: Unstamp) -> None:
     """The next_cursor a list handler returns reaches the client, and the cursor the client sends
     back on the following call reaches the handler verbatim.
     """

tests/interaction/lowlevel/test_progress.py

@@ -18,7 +18,7 @@
 from mcp.server import Server, ServerRequestContext
 from mcp.server.session import ServerSession
 from mcp.shared.session import ProgressFnT
-from tests._stamp import unstamped
+from tests._stamp import Unstamp
 from tests.interaction._connect import Connect
 from tests.interaction._helpers import IncomingMessage
 from tests.interaction._requirements import requirement
@@ -28,7 +28,7 @@
 
 @requirement("protocol:progress:callback")
 @requirement("tools:call:progress")
-async def test_progress_during_tool_call_reaches_callback_in_order(connect: Connect) -> None:
+async def test_progress_during_tool_call_reaches_callback_in_order(connect: Connect, unstamped: Unstamp) -> None:
     """Progress notifications emitted by a tool handler reach the caller's progress callback in order."""
     received: list[tuple[float, float | None, str | None]] = []
 
@@ -84,7 +84,7 @@ async def ignore(progress: float, total: float | None, message: str | None) -> N
 
 
 @requirement("protocol:progress:no-token")
-async def test_no_progress_callback_means_no_token(connect: Connect) -> None:
+async def test_no_progress_callback_means_no_token(connect: Connect, unstamped: Unstamp) -> None:
     """Without a progress callback the request carries no progress token.
 
     The low-level API has no way to report request-scoped progress without a token, so a handler

tests/interaction/lowlevel/test_prompts.py

@@ -21,15 +21,15 @@
 
 from mcp import MCPError
 from mcp.server import Server, ServerRequestContext
-from tests._stamp import unstamped
+from tests._stamp import Unstamp
 from tests.interaction._connect import Connect
 from tests.interaction._requirements import requirement
 
 pytestmark = pytest.mark.anyio
 
 
 @requirement("prompts:list:basic")
-async def test_list_prompts_returns_registered_prompts(connect: Connect) -> None:
+async def test_list_prompts_returns_registered_prompts(connect: Connect, unstamped: Unstamp) -> None:
     """The prompts returned by the handler reach the client with their argument declarations intact."""
 
     async def list_prompts(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListPromptsResult:
@@ -72,7 +72,7 @@ async def list_prompts(ctx: ServerRequestContext, params: types.PaginatedRequest
 
 
 @requirement("prompts:get:with-args")
-async def test_get_prompt_substitutes_arguments(connect: Connect) -> None:
+async def test_get_prompt_substitutes_arguments(connect: Connect, unstamped: Unstamp) -> None:
     """Arguments supplied by the client reach the prompt handler; the templated message comes back."""
 
     async def get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestParams) -> GetPromptResult:
@@ -97,7 +97,7 @@ async def get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestPa
 
 
 @requirement("prompts:get:multi-message")
-async def test_get_prompt_multiple_messages_preserve_roles_and_order(connect: Connect) -> None:
+async def test_get_prompt_multiple_messages_preserve_roles_and_order(connect: Connect, unstamped: Unstamp) -> None:
     """A prompt returning a user/assistant conversation reaches the client with roles and order intact."""
 
     async def get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestParams) -> GetPromptResult:
@@ -127,7 +127,7 @@ async def get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestPa
 
 
 @requirement("prompts:get:no-args")
-async def test_get_prompt_without_arguments_returns_the_messages(connect: Connect) -> None:
+async def test_get_prompt_without_arguments_returns_the_messages(connect: Connect, unstamped: Unstamp) -> None:
     """A prompt fetched with no arguments delivers None as the handler's arguments and returns its messages."""
 
     async def get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestParams) -> GetPromptResult:
@@ -148,7 +148,7 @@ async def get_prompt(ctx: ServerRequestContext, params: types.GetPromptRequestPa
 @requirement("prompts:get:content:image")
 @requirement("prompts:get:content:audio")
 @requirement("prompts:get:content:embedded-resource")
-async def test_get_prompt_with_non_text_content_round_trips(connect: Connect) -> None:
+async def test_get_prompt_with_non_text_content_round_trips(connect: Connect, unstamped: Unstamp) -> None:
     """Prompt messages can carry image, audio, and embedded-resource content; all reach the client.
 
     A single full-result snapshot proves all three content types round-trip: each block in the result