Check every alias form when reserving core surface wire keys

A claim model could route a reserved key through validation_alias,
serialization_alias, or an AliasChoices entry and still hit the core
pre-validation dead end the reservation exists to prevent. _wire_keys
collects every top-level key a field can read from or write to, and
the reservation checks the full set.

Author: maxisbey

src/mcp/client/extension.py

@@ -13,7 +13,8 @@
 
 from mcp_types import CORE_RESULT_TYPES, CallToolResult, InputRequiredResult, Result
 from mcp_types.version import MODERN_PROTOCOL_VERSIONS
-from pydantic import BaseModel
+from pydantic import AliasChoices, AliasPath, BaseModel
+from pydantic.fields import FieldInfo
 
 from mcp.shared.extension import validate_extension_identifier
 
@@ -35,6 +36,22 @@
 _RESERVED_WIRE_ALIASES: Final[frozenset[str]] = frozenset({"requestState", "inputRequests"})
 """Typed optional fields of the core result surface that pre-validates every inbound result."""
 
+
+def _wire_keys(name: str, field: FieldInfo) -> frozenset[str]:
+    """Every top-level wire key this field can read from or write to."""
+    keys = {field.alias or name}
+    if field.serialization_alias:
+        keys.add(field.serialization_alias)
+    validation_alias = field.validation_alias
+    choices = validation_alias.choices if isinstance(validation_alias, AliasChoices) else [validation_alias]
+    for choice in choices:
+        if isinstance(choice, AliasPath):
+            choice = choice.path[0]
+        if isinstance(choice, str):
+            keys.add(choice)
+    return frozenset(keys)
+
+
 ClaimedT = TypeVar("ClaimedT", bound=Result)
 NotifyParamsT = TypeVar("NotifyParamsT", bound=BaseModel)
 
@@ -74,11 +91,11 @@ def __post_init__(self) -> None:
         if issubclass(self.model, CallToolResult | InputRequiredResult):
             raise ValueError("claim models must not subclass core result types")
         for name, model_field in self.model.model_fields.items():
-            if (model_field.alias or name) in _RESERVED_WIRE_ALIASES:
+            for clash in sorted(_wire_keys(name, model_field) & _RESERVED_WIRE_ALIASES):
                 raise ValueError(
-                    f"{self.model.__name__}.{name} aliases {model_field.alias or name!r}, a typed field "
-                    "of the core result surface; a colliding value would fail core validation before "
-                    "the claim adapter runs"
+                    f"{self.model.__name__}.{name} aliases {clash!r}, a typed field of the core "
+                    "result surface; a colliding value would fail core validation before the "
+                    "claim adapter runs"
                 )
         field = self.model.model_fields.get("result_type")
         if field is None or get_args(field.annotation) != (self.result_type,):

tests/client/test_extension.py

@@ -7,13 +7,15 @@
 from inline_snapshot import snapshot
 from mcp_types import CallToolResult, InputRequiredResult, Result
 from mcp_types.version import MODERN_PROTOCOL_VERSIONS
-from pydantic import BaseModel
+from pydantic import AliasChoices, AliasPath, BaseModel, Field
+from pydantic.fields import FieldInfo
 
 from mcp.client.extension import (
     ClaimContext,
     ClientExtension,
     NotificationBinding,
     ResultClaim,
+    _wire_keys,
     advertise,
 )
 
@@ -148,6 +150,60 @@ def test_claim_rejects_model_aliasing_core_surface_fields() -> None:
     )
 
 
+class _ValidationAliasResult(Result):
+    result_type: Literal["va"] = "va"
+    vendor_state: dict[str, Any] | None = Field(default=None, validation_alias="requestState")
+
+
+class _SerializationAliasResult(Result):
+    result_type: Literal["sa"] = "sa"
+    vendor_state: dict[str, Any] | None = Field(default=None, serialization_alias="inputRequests")
+
+
+class _AliasChoicesResult(Result):
+    result_type: Literal["ac"] = "ac"
+    vendor_state: dict[str, Any] | None = Field(
+        default=None, validation_alias=AliasChoices("vendorKey", "requestState")
+    )
+
+
+class _AliasPathResult(Result):
+    result_type: Literal["ap"] = "ap"
+    vendor_state: dict[str, Any] | None = Field(
+        default=None, validation_alias=AliasChoices(AliasPath("requestState", "nested"))
+    )
+
+
+def test_wire_keys_for_a_bare_field_is_just_its_name() -> None:
+    """SDK-defined: a field with no aliases reads and writes only its own name."""
+    assert _wire_keys("plain", FieldInfo(annotation=str)) == frozenset({"plain"})
+
+
+def test_claim_rejects_reserved_aliases_in_every_alias_form() -> None:
+    """SDK-defined: validation_alias, serialization_alias, and AliasChoices routes to a reserved key are all caught."""
+    messages: dict[str, str] = {}
+    for model in (_ValidationAliasResult, _SerializationAliasResult, _AliasChoicesResult, _AliasPathResult):
+        with pytest.raises(ValueError) as exc_info:
+            ResultClaim(result_type=model.model_fields["result_type"].default, model=model, resolve=_resolve)
+        messages[model.__name__] = str(exc_info.value)
+
+    assert messages == snapshot(
+        {
+            "_ValidationAliasResult": "_ValidationAliasResult.vendor_state aliases "
+            "'requestState', a typed field of the core result surface; a colliding value would fail "
+            "core validation before the claim adapter runs",
+            "_SerializationAliasResult": "_SerializationAliasResult.vendor_state aliases "
+            "'inputRequests', a typed field of the core result surface; a colliding value would fail "
+            "core validation before the claim adapter runs",
+            "_AliasChoicesResult": "_AliasChoicesResult.vendor_state aliases 'requestState', a typed field of the core "
+            "result surface; a colliding value would fail core validation before the claim adapter runs",
+            "_AliasPathResult": "_AliasPathResult.vendor_state aliases "
+            "'requestState', a typed field of the core result surface; a colliding value would fail "
+            "core validation before the claim adapter runs",
+        }
+    )
+
+
 def test_claim_rejects_method_outside_the_closed_verb_set() -> None:
     """SDK-defined: claims attach to `tools/call` only, even for values that dodge the static Literal gate."""
     with pytest.raises(ValueError) as exc_info: