Harden the client's later-revision drop and widen its coverage

A `RootModel` wrapper row (the empty result served at pre-2026
versions) reported the pydantic-internal `root` key rather than the
keys of the model it wraps; unwrap it so a legal field a strict
surface merely omits, such as `_meta`, is never mistaken for a later
revision's field. This is latent against today's tables (none of the
wrapper rows has a newer-revision counterpart) but keeps the field-set
logic correct as revisions are added.

Extend the legacy-session test over every handshake protocol version,
cover the `resultType` tag on `tools/call` and `prompts/get` results,
and register the inbound tolerance in the interaction requirements
manifest alongside its existing outbound counterpart.

Author: maxisbey

src/mcp/client/session.py

@@ -80,14 +80,20 @@ def _clamp_inbound_ttl(raw: dict[str, Any]) -> None:
 
 @cache
 def _wire_fields(target: type[BaseModel] | UnionType) -> frozenset[str]:
-    """Top-level wire keys `target` declares (its members', for a union)."""
-    models: tuple[Any, ...] = get_args(target) if isinstance(target, UnionType) else (target,)
-    return frozenset(
-        field.alias or name
-        for model in models
-        if isinstance(model, type) and issubclass(model, BaseModel)
-        for name, field in model.model_fields.items()
-    )
+    """Top-level wire keys `target` declares (its members', for a union).
+
+    A `RootModel` row (e.g. an empty result carried as `RootModel[Result]`)
+    reports its wrapped type's keys, not the pydantic-internal `root`.
+    """
+    members: tuple[Any, ...] = get_args(target) if isinstance(target, UnionType) else (target,)
+    models = [m for m in members if isinstance(m, type) and issubclass(m, BaseModel)]
+    fields: set[str] = set()
+    for model in models:
+        if getattr(model, "__pydantic_root_model__", False):  # a RootModel wrapper row
+            fields |= _wire_fields(model.model_fields["root"].annotation)
+        else:
+            fields.update(field.alias or name for name, field in model.model_fields.items())
+    return frozenset(fields)
 
 
 @cache

tests/client/test_session.py

@@ -1767,29 +1767,67 @@ async def test_a_boolean_inbound_ttl_is_not_clamped_only_coerced_by_validation(w
     ("read_resource", {"contents": []}),
 ]
 
+_LEGACY_TAGGED_RESULTS: list[tuple[str, dict[str, Any]]] = [
+    ("call_tool", {"content": []}),
+    ("get_prompt", {"messages": []}),
+]
+
+
+def _legacy_init(version: str) -> dict[str, Any]:
+    return InitializeResult(
+        protocol_version=version,
+        capabilities=ServerCapabilities(),
+        server_info=Implementation(name="mock-server", version="0.1.0"),
+    ).model_dump(by_alias=True, mode="json", exclude_none=True)
+
+
+async def _call_legacy(session: ClientSession, verb: str) -> Any:
+    if verb == "read_resource":
+        return await session.read_resource("mem://x")
+    if verb == "call_tool":
+        return await session.call_tool("t", {})
+    if verb == "get_prompt":
+        return await session.get_prompt("p")
+    return await getattr(session, verb)()
+
 
 @pytest.mark.anyio
+@pytest.mark.parametrize("version", HANDSHAKE_PROTOCOL_VERSIONS)
 @pytest.mark.parametrize(("verb", "body"), _LEGACY_HINTED_RESULTS)
-async def test_cache_hints_from_a_legacy_server_never_reach_the_result(verb: str, body: dict[str, Any]) -> None:
-    """SDK-defined: on a pre-2026 session the caching fields are outside the negotiated
+async def test_cache_hints_from_a_legacy_server_never_reach_the_result(
+    version: str, verb: str, body: dict[str, Any]
+) -> None:
+    """SDK-defined: on any pre-2026 session the caching fields are outside the negotiated
     schema, so whatever a server puts in them - even values the 2026-07-28 enum
     rejects - is dropped and the model shows its conservative defaults."""
-    init_result = InitializeResult(
-        protocol_version=LATEST_HANDSHAKE_VERSION,
-        capabilities=ServerCapabilities(),
-        server_info=Implementation(name="mock-server", version="0.1.0"),
-    ).model_dump(by_alias=True, mode="json", exclude_none=True)
-    hinted = {**body, "ttlMs": -1, "cacheScope": "session"}
-    dispatcher = _ScriptedDispatcher(init_result, hinted)
+    dispatcher = _ScriptedDispatcher(_legacy_init(version), {**body, "ttlMs": -1, "cacheScope": "session"})
     with anyio.fail_after(5):
         async with ClientSession(dispatcher=dispatcher) as session:
             await session.initialize()
-            call = getattr(session, verb)
-            result = await (call("mem://x") if verb == "read_resource" else call())
+            result = await _call_legacy(session, verb)
     assert (result.ttl_ms, result.cache_scope) == (0, "private")
     assert not {"ttl_ms", "cache_scope"} & result.model_fields_set
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(("verb", "body"), _LEGACY_TAGGED_RESULTS)
+async def test_a_2026_result_type_tag_from_a_legacy_server_never_reaches_the_result(
+    verb: str, body: dict[str, Any]
+) -> None:
+    """SDK-defined: `resultType` is 2026-07-28 vocabulary that also feeds result-union
+    routing, so a tag on a pre-2026 wire (even one no union arm claims) is dropped and
+    the plain result is returned rather than mis-routing or failing."""
+    # `call_tool` re-lists tools to validate structured content; that answer trails harmlessly for `get_prompt`.
+    dispatcher = _ScriptedDispatcher(
+        _legacy_init(LATEST_HANDSHAKE_VERSION), {**body, "resultType": "task"}, {"tools": []}
+    )
+    with anyio.fail_after(5):
+        async with ClientSession(dispatcher=dispatcher) as session:
+            await session.initialize()
+            result = await _call_legacy(session, verb)
+    assert result.result_type == "complete"
+
+
 @pytest.mark.anyio
 async def test_session_call_tool_returns_input_required_result_when_opted_in() -> None:
     """`ClientSession.call_tool(..., allow_input_required=True)` surfaces the

tests/interaction/_requirements.py

@@ -3278,6 +3278,19 @@ def __post_init__(self) -> None:
             "at the transport seam."
         ),
     ),
+    "hosting:client:legacy-inbound-modern-vocabulary-dropped": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/basic/versioning",
+        behavior=(
+            "A client on a pre-2026 session that receives 2026-07-28 result vocabulary "
+            "(resultType, ttlMs, cacheScope) drops those fields before parsing, so a value "
+            "the 2026-07-28 types reject cannot fail the call and the result shows its defaults."
+        ),
+        deferred=(
+            "Client-inbound seam, covered by tests/client/test_session.py::"
+            "test_cache_hints_from_a_legacy_server_never_reach_the_result and "
+            "test_a_2026_result_type_tag_from_a_legacy_server_never_reaches_the_result"
+        ),
+    ),
     "hosting:http:modern:tools-call-stateless": Requirement(
         source=f"{SPEC_2026_BASE_URL}/basic/transports/streamable-http",
         behavior=(