Adopt the serverInfo _meta stamp fixture across the interaction suite

Main now stamps io.modelcontextprotocol/serverInfo into the _meta of
every 2026-era result and ships an era-aware strip helper. Route the
branch's own result comparisons through it: era-parametrized tests thread
the unstamped fixture; tests pinned to a single modern connection call
the strict module helper directly, aliased strip_stamp wherever a file
also uses the fixture so the name never carries two meanings; raw-wire
snapshots pin the stamp on the wire, with ad-hoc test servers given an
explicit version so no snapshot pins an empty one. Drop the
DiscoverResult(server_info=...) kwargs the reshaped model silently
ignored, and correct the tests/_stamp.py docstring to match the new
direct usage.

No-Verification-Needed: test-only change

Author: maxisbey

tests/_stamp.py

@@ -6,10 +6,13 @@
 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
+Era-parametrized interaction tests go through the `unstamped` fixture
+(tests/interaction/conftest.py), which 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.
+handshake-era cells, so one comparison line enforces both eras. Interaction
+tests pinned to a single modern connection call this function directly
+(imported as `strip_stamp` where a file also uses the fixture, so the two
+never share a name).
 """
 
 from typing import Any, Protocol, TypeVar
@@ -29,8 +32,8 @@ def unstamped(result: R) -> R:
     """Assert the result carries a well-formed serverInfo stamp, then remove it.
 
     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.
+    is required (a 2026-era result); tests that also run on handshake-era
+    cells use the `unstamped` fixture instead, which handles the era split.
     """
     meta = result.meta
     assert meta is not None and SERVER_INFO_META_KEY in meta, "expected a serverInfo stamp on this result"

tests/interaction/lowlevel/test_caching.py

@@ -9,6 +9,7 @@
 import mcp_types as types
 import pytest
 from mcp_types import (
+    SERVER_INFO_META_KEY,
     DiscoverResult,
     ElicitRequest,
     ElicitRequestFormParams,
@@ -216,7 +217,7 @@ async def read_resource(
             cache_scope="public",
         )
 
-    server = Server("cached", on_read_resource=read_resource)
+    server = Server("cached", version="1.0.0", on_read_resource=read_resource)
 
     async def answer_who(context: ClientRequestContext, params: types.ElicitRequestParams) -> ElicitResult:
         assert isinstance(params, ElicitRequestFormParams)
@@ -245,7 +246,9 @@ async def answer_who(context: ClientRequestContext, params: types.ElicitRequestP
     interim = responses[reads[0].id]
     complete = responses[reads[1].id]
     # Exact key vocabulary, stronger than `not in` checks: any field added to interim frames fails loudly.
-    assert sorted(interim) == ["inputRequests", "resultType"]
+    # The interim frame is a 2026 result, so it carries the serverInfo stamp but still no caching hints.
+    assert sorted(interim) == ["_meta", "inputRequests", "resultType"]
+    assert interim["_meta"] == {SERVER_INFO_META_KEY: {"name": "cached", "version": "1.0.0"}}
     assert interim["resultType"] == "input_required"
     assert complete["ttlMs"] == 60_000
     assert complete["cacheScope"] == "public"
@@ -289,13 +292,7 @@ async def scripted_server() -> None:
             ClientSession(client_read, client_write, client_info=Implementation(name="cli", version="0")) as session,
         ):
             task_group.start_soon(scripted_server)
-            session.adopt(
-                DiscoverResult(
-                    supported_versions=[LATEST_MODERN_VERSION],
-                    capabilities=ServerCapabilities(),
-                    server_info=Implementation(name="srv", version="0"),
-                )
-            )
+            session.adopt(DiscoverResult(supported_versions=[LATEST_MODERN_VERSION], capabilities=ServerCapabilities()))
             with anyio.fail_after(5):
                 result = await session.list_tools()
 

tests/interaction/lowlevel/test_elicitation.py

@@ -38,6 +38,8 @@
 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 Unstamp
+from tests._stamp import unstamped as strip_stamp
 from tests.interaction._connect import BASE_URL, Connect, mounted_app
 from tests.interaction._helpers import RecordingTransport
 from tests.interaction._requirements import requirement
@@ -672,7 +674,9 @@ async def scripted_server(streams: MessageStream) -> None:
 
 
 @requirement("elicitation:mrtr:form:basic")
-async def test_embedded_form_elicitation_accepted_content_returns_to_retried_handler(connect: Connect) -> None:
+async def test_embedded_form_elicitation_accepted_content_returns_to_retried_handler(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """An embedded form elicitation reaches the callback as sent and its accepted content reaches the handler.
 
     Spec-mandated: at 2026-07-28 elicitation/create rides the MRTR flow, not a server-initiated request.
@@ -723,13 +727,15 @@ async def answer_form(context: ClientRequestContext, params: types.ElicitRequest
             )
         ]
     )
-    assert result == snapshot(
+    assert unstamped(result) == snapshot(
         CallToolResult(content=[TextContent(text="accept")], structured_content={"username": "ada", "newsletter": True})
     )
 
 
 @requirement("elicitation:mrtr:form:action:decline")
-async def test_embedded_form_elicitation_decline_reaches_retried_handler_with_no_content(connect: Connect) -> None:
+async def test_embedded_form_elicitation_decline_reaches_retried_handler_with_no_content(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """An embedded form elicitation declined by the callback reaches the retried handler with no content."""
 
     async def list_tools(
@@ -765,11 +771,13 @@ async def answer_form(context: ClientRequestContext, params: types.ElicitRequest
     async with connect(server, elicitation_callback=answer_form) as client:
         result = await client.call_tool("confirm", {})
 
-    assert result == snapshot(CallToolResult(content=[TextContent(text="decline content=None")]))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="decline content=None")]))
 
 
 @requirement("elicitation:mrtr:form:action:cancel")
-async def test_embedded_form_elicitation_cancel_reaches_retried_handler_with_no_content(connect: Connect) -> None:
+async def test_embedded_form_elicitation_cancel_reaches_retried_handler_with_no_content(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """An embedded form elicitation cancelled by the callback reaches the retried handler with no content."""
 
     async def list_tools(
@@ -805,11 +813,13 @@ async def answer_form(context: ClientRequestContext, params: types.ElicitRequest
     async with connect(server, elicitation_callback=answer_form) as client:
         result = await client.call_tool("confirm", {})
 
-    assert result == snapshot(CallToolResult(content=[TextContent(text="cancel content=None")]))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="cancel content=None")]))
 
 
 @requirement("elicitation:mrtr:form:schema:primitives")
-async def test_embedded_form_elicitation_schema_primitives_reach_the_callback_as_sent(connect: Connect) -> None:
+async def test_embedded_form_elicitation_schema_primitives_reach_the_callback_as_sent(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """Primitive requested-schema fields on an embedded form elicitation reach the callback intact.
 
     Spec-mandated. One representative constraint per type; the exhaustive sweep lives with the 2025 push-path sibling.
@@ -877,7 +887,7 @@ async def answer_form(context: ClientRequestContext, params: types.ElicitRequest
             )
         ]
     )
-    assert result == snapshot(
+    assert unstamped(result) == snapshot(
         CallToolResult(
             content=[TextContent(text="accept")],
             structured_content={"email": "ada@example.com", "age": 36, "score": 9.5, "subscribed": True},
@@ -887,7 +897,7 @@ async def answer_form(context: ClientRequestContext, params: types.ElicitRequest
 
 @requirement("elicitation:mrtr:capability:not-declared")
 async def test_server_embeds_elicitation_for_a_client_that_declared_no_elicitation_capability(
-    connect: Connect,
+    connect: Connect, unstamped: Unstamp
 ) -> None:
     """Pins a known gap: the SDK embeds an elicitation for a client that declared no elicitation capability.
 
@@ -916,7 +926,7 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
         raw = await client.session.call_tool("ask", {}, allow_input_required=True)
 
     assert isinstance(raw, InputRequiredResult)
-    assert raw == snapshot(
+    assert unstamped(raw) == snapshot(
         InputRequiredResult(
             input_requests={
                 "ask": ElicitRequest(
@@ -983,7 +993,7 @@ async def answer_url(context: ClientRequestContext, params: types.ElicitRequestP
     assert received == snapshot(
         [ElicitRequestURLParams(message="Sign in to continue.", url="https://example.com/auth")]
     )
-    assert result == snapshot(CallToolResult(content=[TextContent(text="accept content=None")]))
+    assert strip_stamp(result) == snapshot(CallToolResult(content=[TextContent(text="accept content=None")]))
     # Positive control: the interim input_required leg was captured, so the scan below is not vacuous.
     interim = [
         message.message

tests/interaction/lowlevel/test_mrtr.py

@@ -54,6 +54,8 @@
 from mcp.shared.exceptions import NoBackChannelError
 from mcp.shared.memory import MessageStream, create_client_server_memory_streams
 from mcp.shared.message import SessionMessage
+from tests._stamp import Unstamp
+from tests._stamp import unstamped as strip_stamp
 from tests.interaction._connect import BASE_URL, Connect, base_headers, mounted_app
 from tests.interaction._helpers import RecordingTransport
 from tests.interaction._requirements import requirement
@@ -106,7 +108,9 @@ async def call_tool(
 
 
 @requirement("mrtr:tools-call:write-once-roundtrip")
-async def test_input_required_tool_call_is_auto_fulfilled_and_retried_to_completion(connect: Connect) -> None:
+async def test_input_required_tool_call_is_auto_fulfilled_and_retried_to_completion(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """An input_required tools/call is auto-fulfilled by the client driver and retried to completion.
 
     The byte-exact requestState echo (spec MUST) is the only observable proxy for the MUST NOT
@@ -125,13 +129,15 @@ async def answer_login(context: ClientRequestContext, params: types.ElicitReques
     async with connect(server, elicitation_callback=answer_login) as client:
         result = await client.call_tool("login", {})
 
-    assert result == snapshot(CallToolResult(content=[TextContent(text="hello octocat")]))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="hello octocat")]))
     assert prompts == ["Provide your GitHub username"]
     assert request_states == [None, OPAQUE_STATE]
 
 
 @requirement("mrtr:request-state-only:retry")
-async def test_state_only_input_required_is_retried_with_no_responses_and_echoed_state(connect: Connect) -> None:
+async def test_state_only_input_required_is_retried_with_no_responses_and_echoed_state(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """A state-only input_required result is retried with no inputResponses and the state echoed.
 
     No callbacks are registered: a driver that wrongly dispatched here would error the call.
@@ -161,12 +167,14 @@ async def call_tool(
     async with connect(server) as client:
         result = await client.call_tool("resume", {})
 
-    assert result == snapshot(CallToolResult(content=[TextContent(text="done")]))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="done")]))
     assert request_states == [None, resume_token]
 
 
 @requirement("mrtr:multi-round:complete")
-async def test_server_reprompts_across_two_productive_rounds_then_completes(connect: Connect) -> None:
+async def test_server_reprompts_across_two_productive_rounds_then_completes(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """A server re-prompting with input_required across two productive rounds completes normally.
 
     Round 1's answer rides forward inside request_state (the spec's stateless-server pattern). Each
@@ -216,7 +224,7 @@ async def answer_by_prompt(context: ClientRequestContext, params: types.ElicitRe
     async with connect(server, elicitation_callback=answer_by_prompt) as client:
         result = await client.call_tool("enroll", {})
 
-    assert result == snapshot(CallToolResult(content=[TextContent(text="one+two")]))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="one+two")]))
     assert prompts == ["first question", "second question"]
     assert request_states == [None, "s1", "s2:one"]
 
@@ -314,7 +322,9 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
 
 
 @requirement("mrtr:input-responses:key-correspondence")
-async def test_multi_request_input_responses_are_keyed_by_the_input_request_keys(connect: Connect) -> None:
+async def test_multi_request_input_responses_are_keyed_by_the_input_request_keys(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """inputResponses on the retry are keyed by the inputRequests keys, each value that key's typed result.
 
     ElicitResult and ListRootsResult prove the map contract; sampling fidelity belongs to the sampling entries.
@@ -354,11 +364,11 @@ async def answer_roots(context: ClientRequestContext) -> ListRootsResult:
     async with connect(server, elicitation_callback=answer_login, list_roots_callback=answer_roots) as client:
         result = await client.call_tool("profile", {})
 
-    assert result == snapshot(CallToolResult(content=[TextContent(text="octocat@file:///workspace")]))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="octocat@file:///workspace")]))
 
 
 @requirement("mrtr:input-responses:missing-reprompted")
-async def test_retry_missing_a_requested_key_is_reprompted_not_errored(connect: Connect) -> None:
+async def test_retry_missing_a_requested_key_is_reprompted_not_errored(connect: Connect, unstamped: Unstamp) -> None:
     """A retry omitting a requested inputResponses key is re-prompted, not errored (spec SHOULD).
 
     The re-prompt decision belongs to the test's handler; the SDK obligation pinned is that the partial
@@ -423,13 +433,15 @@ async def call_tool(
             allow_input_required=True,
         )
 
-    assert result == snapshot(CallToolResult(content=[TextContent(text="one+two")]))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="one+two")]))
     # The partial map reached the handler as sent, not filtered or rejected.
     assert seen == [None, {"first"}, {"second"}]
 
 
 @requirement("mrtr:input-responses:unknown-ignored")
-async def test_retry_with_an_unrequested_extra_key_is_tolerated_and_the_call_completes(connect: Connect) -> None:
+async def test_retry_with_an_unrequested_extra_key_is_tolerated_and_the_call_completes(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """A retry carrying an unrequested inputResponses key completes normally (spec SHOULD: ignore).
 
     The ignoring happens in the test's handler; the SDK half pinned is that the stray entry is
@@ -472,12 +484,14 @@ async def call_tool(
             allow_input_required=True,
         )
 
-    assert result == snapshot(CallToolResult(content=[TextContent(text="hello ada")]))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="hello ada")]))
     assert seen == [None, {"name", "stray"}]
 
 
 @requirement("mrtr:push-api:loud-fail-2026")
-async def test_push_elicit_on_2026_raises_typed_local_error_and_call_still_completes(connect: Connect) -> None:
+async def test_push_elicit_on_2026_raises_typed_local_error_and_call_still_completes(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """A push API call on a 2026 connection raises a typed local error and the call still completes.
 
     Spec-mandated outcome, era-routed enforcement: every modern dispatch path installs a
@@ -510,7 +524,7 @@ async def never_deliverable(context: ClientRequestContext, params: types.ElicitR
         result = await client.call_tool("ask", {})
 
     # The failed push did not poison the request: the call completes with the handler's fallback.
-    assert result == snapshot(CallToolResult(content=[TextContent(text="fallback")]))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="fallback")]))
     assert len(caught) == 1
     assert caught[0].method == "elicitation/create"
     assert caught[0].error == snapshot(
@@ -562,7 +576,7 @@ async def never_deliverable(context: ClientRequestContext, params: types.ElicitR
         result = await client.call_tool("ask", {})
 
     # The failed push did not poison the request: the call completes with the handler's fallback.
-    assert result == snapshot(CallToolResult(content=[TextContent(text="fallback")]))
+    assert strip_stamp(result) == snapshot(CallToolResult(content=[TextContent(text="fallback")]))
     assert len(caught) == 1
     assert caught[0].method == "elicitation/create"
     assert caught[0].error == snapshot(
@@ -819,7 +833,7 @@ async def answer(context: ClientRequestContext, params: types.ElicitRequestParam
         ):
 
             async def call(name: str) -> None:
-                results[name] = await client.call_tool(name, {})
+                results[name] = strip_stamp(await client.call_tool(name, {}))
 
             task_group.start_soon(call, "alpha")
             task_group.start_soon(call, "beta")
@@ -879,7 +893,7 @@ async def answer(context: ClientRequestContext, params: types.ElicitRequestParam
             result = await client.call_tool("ask", {})
 
     # Non-vacuity: the elicitation genuinely happened and the round trip completed through it.
-    assert result == snapshot(CallToolResult(content=[TextContent(text="done:Berlin:s1")]))
+    assert strip_stamp(result) == snapshot(CallToolResult(content=[TextContent(text="done:Berlin:s1")]))
     assert elicited == ["Need a name"]
     # Prove the received log holds only messages before narrowing: a filtered-out transport exception would fake it.
     received_messages = [message for message in recording.received if isinstance(message, SessionMessage)]
@@ -1030,7 +1044,7 @@ def respond(request_id: types.RequestId, result: dict[str, object]) -> SessionMe
 
 @requirement("protocol:result-type:unrecognized-invalid")
 async def test_an_unrecognized_result_type_value_is_surfaced_unchanged_instead_of_treated_as_invalid(
-    connect: Connect,
+    connect: Connect, unstamped: Unstamp
 ) -> None:
     """PINS A KNOWN GAP: an unrecognized resultType round-trips instead of being treated as invalid (spec MUST).
 
@@ -1061,4 +1075,4 @@ def probe() -> CallToolResult:
 
     # The divergent observable: the unrecognized discriminator survives unchanged, never a rejection.
     assert result.result_type == "bogus"
-    assert result == snapshot(CallToolResult(content=[TextContent(text="still here")], result_type="bogus"))
+    assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="still here")], result_type="bogus"))

tests/interaction/lowlevel/test_pagination.py

@@ -62,7 +62,9 @@ async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestPa
 
 
 @requirement("protocol:pagination:empty-cursor-valid")
-async def test_an_empty_string_next_cursor_round_trips_as_a_cursor_not_end_of_results(connect: Connect) -> None:
+async def test_an_empty_string_next_cursor_round_trips_as_a_cursor_not_end_of_results(
+    connect: Connect, unstamped: Unstamp
+) -> None:
     """An empty-string next_cursor round-trips verbatim, distinct from absent.
 
     Spec-mandated: an empty string is a valid cursor and MUST NOT be treated as the end of results.
@@ -84,7 +86,9 @@ async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestPa
 
     assert first_page.next_cursor == ""
     assert seen_cursors == [None, ""]
-    assert second_page == snapshot(ListToolsResult(tools=[Tool(name="beta", input_schema={"type": "object"})]))
+    assert unstamped(second_page) == snapshot(
+        ListToolsResult(tools=[Tool(name="beta", input_schema={"type": "object"})])
+    )
 
 
 @requirement("pagination:exhaustion")