Rewrite the unrecognized-resultType pin onto the extension seam; simplify four tests

The server extension API can now author an arbitrary resultType through
public surface, so the scripted memory-stream peer is replaced with an
MCPServer extension driven through the connect fixture (one node becomes
the entry's two cells). The parallel-MRTR isolation test drops its wire
recording for handler capture (the rendezvous still forces interleaving),
the GET-stream step-up shim moves into the auth harness beside its POST
twin, a redundant metadata shim argument goes away, and the claimed-shape
extension test pins its ValidationError by structured fields instead of
accepting any validation failure.

Author: maxisbey

tests/interaction/_requirements.py

@@ -3799,13 +3799,14 @@ def __post_init__(self) -> None:
         added_in="2026-07-28",
         divergence=Divergence(
             note=(
-                "The client never rejects an unrecognized resultType: ResultType is a "
-                "deliberately open Literal-or-str union (src/mcp-types/mcp_types/_types.py), "
-                "the 2026-07-28 wire surface types resultType as a bare str, and the client's "
-                "only result-kind dispatch is isinstance(result, InputRequiredResult) "
-                "(src/mcp/client/session.py), so an unrecognized value round-trips and is "
-                "surfaced on the returned result unchanged -- on both eras (the in-code TODO "
-                "in src/mcp/server/runner.py records the missing rejection)."
+                "The client accepts an unrecognized resultType whenever the body also parses "
+                "as a complete core result: ResultType is a deliberately open Literal-or-str "
+                "union (src/mcp-types/mcp_types/_types.py), and the client's discriminated "
+                "claim adapter (src/mcp/client/session.py) routes unknown tags to the core "
+                "arm, so such a value is surfaced on the returned result unchanged. A body "
+                "that does not parse as a core result fails result validation -- that reject "
+                "arm is pinned by extensions:client:claimed-result-undeclared-invalid. The "
+                "in-code TODO in src/mcp/server/runner.py records the missing rejection."
             ),
             issue="L117",
         ),
@@ -4917,9 +4918,9 @@ def __post_init__(self) -> None:
         ),
         added_in="2026-07-28",
         note=(
-            "Known leniency: the monolith result surface still accepts an unknown tag when the payload "
-            "also parses as a complete core result (open result_type, extras ignored). Rejecting tags "
-            "outside core plus active claims is a tracked follow-up ruling."
+            "The lenient accept arm -- an unknown tag whose body still parses as a complete core result "
+            "surfaces unchanged -- is a recorded divergence owned by "
+            "protocol:result-type:unrecognized-invalid; this entry pins only the reject arm."
         ),
     ),
     "extensions:client:capability-ad:gates-server-behaviour": Requirement(

tests/interaction/auth/_harness.py

@@ -16,9 +16,10 @@
 from typing import Any
 from urllib.parse import parse_qs, parse_qsl, urlsplit
 
+import anyio
 import httpx
 from pydantic import AnyHttpUrl, AnyUrl, BaseModel
-from starlette.types import ASGIApp, Receive, Scope, Send
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 from mcp.client.auth import OAuthClientProvider
 from mcp.client.client import Client
@@ -371,6 +372,55 @@ async def wrapped(scope: Scope, receive: Receive, send: Send) -> None:
     return factory
 
 
+def get_stream_step_up_shim(www_authenticate: str) -> tuple[list[int], anyio.Event, AppShim]:
+    """Build an `app_shim` that 403s the first authenticated GET to `/mcp` with the given challenge.
+
+    Returns:
+        The statuses of every authenticated GET response (live-updated), an event set when one
+        of those responses starts with status 200 (the reopened stream), and the shim factory.
+    """
+    statuses: list[int] = []
+    reopened = anyio.Event()
+    fired = False
+
+    def factory(app: ASGIApp) -> ASGIApp:
+        async def wrapped(scope: Scope, receive: Receive, send: Send) -> None:
+            nonlocal fired
+            if not (
+                scope["type"] == "http"
+                and scope["path"] == "/mcp"
+                and scope["method"] == "GET"
+                and b"authorization" in dict(scope["headers"])
+            ):
+                await app(scope, receive, send)
+                return
+
+            async def recording_send(message: Message) -> None:
+                if message["type"] == "http.response.start":
+                    statuses.append(message["status"])
+                    if message["status"] == 200:
+                        reopened.set()
+                await send(message)
+
+            if not fired:
+                fired = True
+                await recording_send(
+                    {
+                        "type": "http.response.start",
+                        "status": 403,
+                        "headers": [(b"www-authenticate", www_authenticate.encode())],
+                    }
+                )
+                await recording_send({"type": "http.response.body", "body": b""})
+                return
+            # The reopened SSE stream stays open until the test's exit cancels it; nothing may follow this await.
+            await app(scope, receive, recording_send)
+
+        return wrapped
+
+    return statuses, reopened, factory
+
+
 def m2m_token_shim(provider: InMemoryAuthorizationServerProvider, *, scopes: list[str]) -> AppShim:
     """Build an `app_shim` that handles `grant_type=client_credentials` at `/token`.
 

tests/interaction/auth/test_authorize_token.py

@@ -509,12 +509,7 @@ async def test_an_iss_differing_only_by_a_trailing_slash_is_rejected_without_nor
 
     with anyio.fail_after(5):
         with pytest.RaisesGroup(pytest.RaisesExc(OAuthFlowError, match=f"^{mismatch}$"), flatten_subgroups=True):
-            await connect_with_oauth(
-                server,
-                provider=provider,
-                app_shim=lambda app: shimmed_app(app, serve=canned_asm(iss_advertised=None)),
-                on_request=on_request,
-            ).__aenter__()
+            await connect_with_oauth(server, provider=provider, on_request=on_request).__aenter__()
 
     # The recorded unauthenticated trigger POST guards the negative below against an unwired hook.
     assert find(recorded, "POST", "/mcp") != []

tests/interaction/auth/test_lifecycle.py

@@ -16,7 +16,6 @@
 from inline_snapshot import snapshot
 from mcp_types import INTERNAL_ERROR, ErrorData, ListToolsResult, Tool
 from pydantic import AnyHttpUrl, AnyUrl
-from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 from mcp import MCPError
 from mcp.client.auth.extensions.client_credentials import ClientCredentialsOAuthProvider, PrivateKeyJWTOAuthProvider
@@ -26,11 +25,11 @@
 from tests.interaction._requirements import requirement
 from tests.interaction.auth._harness import (
     REDIRECT_URI,
-    AppShim,
     InMemoryTokenStorage,
     RecordedRequest,
     auth_settings,
     connect_with_oauth,
+    get_stream_step_up_shim,
     m2m_token_shim,
     metadata_body,
     record_requests,
@@ -738,61 +737,12 @@ async def test_a_second_insufficient_scope_403_after_a_step_up_surfaces_without_
     assert counts[("POST", "/token")] == 2
 
 
-def get_stream_step_up_shim(www_authenticate: str) -> tuple[list[int], anyio.Event, AppShim]:
-    """Build an `app_shim` that 403s the first authenticated GET to `/mcp` with the given challenge.
-
-    Returns:
-        The statuses of every authenticated GET response (live-updated), an event set when one
-        of those responses starts with status 200 (the reopened stream), and the shim factory.
-    """
-    statuses: list[int] = []
-    reopened = anyio.Event()
-    fired = False
-
-    def factory(app: ASGIApp) -> ASGIApp:
-        async def wrapped(scope: Scope, receive: Receive, send: Send) -> None:
-            nonlocal fired
-            if not (
-                scope["type"] == "http"
-                and scope["path"] == "/mcp"
-                and scope["method"] == "GET"
-                and b"authorization" in dict(scope["headers"])
-            ):
-                await app(scope, receive, send)
-                return
-
-            async def recording_send(message: Message) -> None:
-                if message["type"] == "http.response.start":
-                    statuses.append(message["status"])
-                    if message["status"] == 200:
-                        reopened.set()
-                await send(message)
-
-            if not fired:
-                fired = True
-                await recording_send(
-                    {
-                        "type": "http.response.start",
-                        "status": 403,
-                        "headers": [(b"www-authenticate", www_authenticate.encode())],
-                    }
-                )
-                await recording_send({"type": "http.response.body", "body": b""})
-                return
-            # The reopened SSE stream stays open until the test's exit cancels it; nothing may follow this await.
-            await app(scope, receive, recording_send)
-
-        return wrapped
-
-    return statuses, reopened, factory
-
-
 @requirement("client-auth:stepup:get-stream-403")
 async def test_a_403_on_the_get_stream_open_steps_up_and_reopens_the_stream_with_the_upgraded_token() -> None:
     """A 403 `insufficient_scope` on the standalone GET stream open steps up and reopens the stream.
 
     The standalone GET (a 2025-11-25 mechanism, removed at 2026-07-28) is opened by the SDK in the
-    background and is invisible to `Client`, so the file-local shim records each authenticated
+    background and is invisible to `Client`, so the harness shim records each authenticated
     GET's response status and the test waits on the reopened stream's 200 before acting. The
     failure arm stays unpinned: the transport swallows GET failures into a timed reconnect loop
     this suite cannot observe without sleeps.

tests/interaction/lowlevel/test_mrtr.py

@@ -22,7 +22,6 @@
     ClientCapabilities,
     CreateMessageRequest,
     CreateMessageRequestParams,
-    DiscoverResult,
     ElicitRequest,
     ElicitRequestFormParams,
     ElicitResult,
@@ -50,7 +49,9 @@
 from mcp.client import ClientRequestContext, ClientSession
 from mcp.client.client import Client
 from mcp.client.streamable_http import streamable_http_client
-from mcp.server import Server, ServerRequestContext
+from mcp.server import MCPServer, Server, ServerRequestContext
+from mcp.server.context import CallNext, HandlerResult
+from mcp.server.extension import Extension
 from mcp.shared.exceptions import NoBackChannelError
 from mcp.shared.memory import MessageStream, create_client_server_memory_streams
 from mcp.shared.message import SessionMessage
@@ -754,8 +755,10 @@ async def test_parallel_mrtr_calls_keep_request_state_and_responses_isolated() -
     """Parallel MRTR calls keep requestState and inputResponses scoped to their originating request.
 
     A symmetric rendezvous in the elicitation callback forces both loops mid-flight before either
-    retry leaves; the exhaustive scan over every recorded tools/call frame proves no leak (spec MUST NOT).
+    retry leaves (spec MUST NOT). Handler capture suffices: every tools/call the client sends is
+    delivered to the handler, so the captured rounds are 1:1 with the sent frames.
     """
+    rounds: list[tuple[str, str | None, set[str] | None]] = []
 
     async def list_tools(
         ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
@@ -772,12 +775,14 @@ async def call_tool(
     ) -> CallToolResult | InputRequiredResult:
         assert params.name in ("alpha", "beta")
         name = params.name
+        rounds.append(
+            (name, params.request_state, None if params.input_responses is None else set(params.input_responses))
+        )
         if params.input_responses is None:
             return InputRequiredResult(
                 input_requests={f"q-{name}": _form_request(f"for {name}")},
                 request_state=f"state-{name}",
             )
-        assert params.request_state == f"state-{name}"
         return CallToolResult(content=[TextContent(text=name)])
 
     server = Server("parallel", on_list_tools=list_tools, on_call_tool=call_tool)
@@ -799,12 +804,7 @@ async def answer(context: ClientRequestContext, params: types.ElicitRequestParam
 
     with anyio.fail_after(5):
         async with (
-            mounted_app(server) as (http, _),
-            Client(
-                recording := RecordingTransport(streamable_http_client(f"{BASE_URL}/mcp", http_client=http)),
-                mode=LATEST_MODERN_VERSION,
-                elicitation_callback=answer,
-            ) as client,
+            Client(server, mode=LATEST_MODERN_VERSION, elicitation_callback=answer) as client,
             # Last item so it exits first: both calls complete while the client is still open.
             anyio.create_task_group() as task_group,
         ):
@@ -815,28 +815,10 @@ async def call(name: str) -> None:
             task_group.start_soon(call, "alpha")
             task_group.start_soon(call, "beta")
 
-    frames = [
-        message.message
-        for message in recording.sent
-        if isinstance(message.message, JSONRPCRequest) and message.message.method == "tools/call"
-    ]
-    by_name: dict[str, list[dict[str, Any]]] = {"alpha": [], "beta": []}
-    for frame in frames:
-        assert frame.params is not None
-        by_name[frame.params["name"]].append(frame.params)
-    for name, sent_params in by_name.items():
-        assert len(sent_params) == 2
-        initial, retry = sent_params
-        assert "requestState" not in initial
-        assert "inputResponses" not in initial
-        assert retry["requestState"] == f"state-{name}"
-        assert set(retry["inputResponses"]) == {f"q-{name}"}
-    # The exhaustive negative: no frame anywhere carries the other call's state or responses.
-    for params in (frame.params for frame in frames):
-        assert params is not None
-        other = "beta" if params["name"] == "alpha" else "alpha"
-        assert params.get("requestState") in (None, f"state-{params['name']}")
-        assert f"q-{other}" not in params.get("inputResponses", {})
+    # The rendezvous guarantees both initial rounds land before either retry; order within a phase is free.
+    assert sorted(rounds[:2]) == [("alpha", None, None), ("beta", None, None)]
+    # Each retry carries exactly its own call's state and response key -- nothing crossed over.
+    assert sorted(rounds[2:]) == [("alpha", "state-alpha", {"q-alpha"}), ("beta", "state-beta", {"q-beta"})]
     assert results == {
         "alpha": CallToolResult(content=[TextContent(text="alpha")]),
         "beta": CallToolResult(content=[TextContent(text="beta")]),
@@ -966,7 +948,7 @@ async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestPara
     assert error.error == snapshot(ErrorData(code=INVALID_PARAMS, message="Invalid request parameters", data=""))
 
 
-# --- scripted server peer: result bodies a real Server cannot emit ---
+# --- scripted server peer: byte-controlled absence of the resultType key ---
 
 
 @requirement("protocol:result-type:absent-is-complete")
@@ -1034,63 +1016,40 @@ def respond(request_id: types.RequestId, result: dict[str, object]) -> SessionMe
         assert result == snapshot(CallToolResult(content=[TextContent(text="plain")]))
 
 
+# --- unrecognized resultType: a server extension puts an arbitrary tag on the wire ---
+
+
 @requirement("protocol:result-type:unrecognized-invalid")
-async def test_an_unrecognized_result_type_value_is_surfaced_unchanged_instead_of_treated_as_invalid() -> None:
+async def test_an_unrecognized_result_type_value_is_surfaced_unchanged_instead_of_treated_as_invalid(
+    connect: Connect,
+) -> None:
     """PINS A KNOWN GAP: an unrecognized resultType round-trips instead of being treated as invalid (spec MUST).
 
-    The client's open ResultType union accepts any string. Scripted peer over memory streams
-    because the typed Server cannot author an arbitrary resultType. When the client starts
-    rejecting unrecognized resultType values: re-pin to the typed rejection and delete the Divergence.
+    The leniency is narrow: the unknown tag survives only because the body also parses as a
+    complete core result. When the client starts rejecting unrecognized resultType values:
+    re-pin to the typed rejection and delete the Divergence.
     """
 
-    async def scripted_server(streams: MessageStream) -> None:
-        server_read, server_write = streams
+    class BogusIssuer(Extension):
+        identifier = "com.example/bogus"
 
-        def respond(request_id: types.RequestId, result: dict[str, object]) -> SessionMessage:
-            return SessionMessage(JSONRPCResponse(jsonrpc="2.0", id=request_id, result=result))
+        async def intercept_tool_call(
+            self, params: types.CallToolRequestParams, ctx: ServerRequestContext[Any, Any], call_next: CallNext
+        ) -> HandlerResult:
+            assert params.name == "probe"
+            # "bogus" is in no core or extension vocabulary -- exactly the value the MUST addresses.
+            return {"resultType": "bogus", "content": [{"type": "text", "text": "still here"}]}
 
-        call = await server_read.receive()
-        assert isinstance(call, SessionMessage)
-        assert isinstance(call.message, JSONRPCRequest)
-        assert call.message.method == "tools/call"
-        # "bogus" is in no core or extension vocabulary -- exactly the value the MUST addresses.
-        await server_write.send(
-            respond(call.message.id, {"resultType": "bogus", "content": [{"type": "text", "text": "still here"}]})
-        )
+    server = MCPServer("bogus-issuer", extensions=[BogusIssuer()])
 
-        # The client's output-schema cache refresh follows the call result; stopping here hangs the test.
-        refresh = await server_read.receive()
-        assert isinstance(refresh, SessionMessage)
-        assert isinstance(refresh.message, JSONRPCRequest)
-        assert refresh.message.method == "tools/list"
-        await server_write.send(
-            respond(
-                refresh.message.id,
-                {
-                    "tools": [{"name": "x", "inputSchema": {"type": "object"}}],
-                    "resultType": "complete",
-                    "ttlMs": 0,
-                    "cacheScope": "private",
-                },
-            )
-        )
+    @server.tool()
+    def probe() -> CallToolResult:
+        """Probe the unrecognized-tag path."""
+        raise NotImplementedError  # the server extension answers before the tool runs
 
-    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, client_info=Implementation(name="cli", version="0")) as session,
-    ):
-        task_group.start_soon(scripted_server, server_streams)
-        session.adopt(
-            DiscoverResult(
-                supported_versions=[LATEST_MODERN_VERSION],
-                capabilities=ServerCapabilities(),
-                server_info=Implementation(name="srv", version="0"),
-            )
-        )
-        with anyio.fail_after(5):
-            result = await session.call_tool("x", {})
+    async with connect(server) as client:
+        result = await client.call_tool("probe", {})
 
-        # 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"))
+    # 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"))