Harden claim and binding construction invariants

Reject an empty claim sequence at the session constructor: the ad-filter
would otherwise treat the identifier as claim-bearing and silently drop
it from the capability ad at every version, leaving the invariant to
caller discipline. Validate ResultClaim.method against the closed verb
set at construction so an unchecked runtime value cannot fold into
tools/call parsing. Create notification binding queues before the
dispatcher starts so the enqueue path indexes a complete dict by
construction rather than by scheduling order. Copy the extensions ad
dict at the constructor boundary. Pins added: modern re-adoption after
legacy reactivates claims; a legacy-version discover probe drops
claim-bearing identifiers from its ad.

Author: maxisbey

src/mcp/client/extension.py

@@ -12,7 +12,7 @@
 
 from collections.abc import Awaitable, Callable, Sequence
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar, get_args
+from typing import TYPE_CHECKING, Any, Final, Generic, Literal, TypeVar, get_args
 
 from mcp_types import CORE_RESULT_TYPES, CallToolResult, InputRequiredResult, Result
 from mcp_types.version import MODERN_PROTOCOL_VERSIONS
@@ -32,6 +32,9 @@
     "advertise",
 ]
 
+_CLAIM_METHODS: Final[frozenset[str]] = frozenset({"tools/call"})
+"""The closed set of verbs a claim may attach to (widened with the `method` Literal)."""
+
 ClaimedT = TypeVar("ClaimedT", bound=Result)
 NotifyParamsT = TypeVar("NotifyParamsT", bound=BaseModel)
 
@@ -82,6 +85,8 @@ class ResultClaim(Generic[ClaimedT]):
     protocol_versions: frozenset[str] | None = None
 
     def __post_init__(self) -> None:
+        if self.method not in _CLAIM_METHODS:
+            raise ValueError(f"claims attach to {sorted(_CLAIM_METHODS)} only; got method {self.method!r}")
         if self.result_type in CORE_RESULT_TYPES:
             raise ValueError(f"resultType {self.result_type!r} is core protocol vocabulary")
         if issubclass(self.model, CallToolResult | InputRequiredResult):

src/mcp/client/session.py

@@ -277,6 +277,11 @@ def _index_claims(
                 f"result_claims key {identifier!r} has no extensions entry; a claim is only "
                 "advertised through its extension's capability ad"
             )
+        if not claims:
+            raise ValueError(
+                f"result_claims[{identifier!r}] is empty; an empty claim set would drop the "
+                "extension from the capability ad at every version — omit the key instead"
+            )
         for claim in claims:
             key = (claim.method, claim.result_type)
             if key in seen:
@@ -344,7 +349,7 @@ def __init__(
         self._client_info = client_info or DEFAULT_CLIENT_INFO
         self._sampling_callback = sampling_callback or _default_sampling_callback
         self._sampling_capabilities = sampling_capabilities
-        self._extensions = extensions
+        self._extensions = dict(extensions) if extensions is not None else None
         self._result_claims = _index_claims(result_claims, extensions)
         self._notification_bindings = _index_bindings(notification_bindings)
         self._active_claims: dict[str, ResultClaim[Any]] = {}
@@ -387,10 +392,14 @@ async def __aenter__(self) -> Self:
         self._task_group = anyio.create_task_group()
         await self._task_group.__aenter__()
         try:
-            await self._task_group.start(self._dispatcher.run, self._on_request, self._on_notify)
+            # Queues exist before the dispatcher can deliver: _on_notify may run as
+            # soon as the dispatcher starts, and its enqueue indexes this dict.
             for binding in self._notification_bindings.values():
                 send, receive = anyio.create_memory_object_stream[BaseModel](_NOTIFICATION_QUEUE_SIZE)
                 self._binding_queues[binding.method] = (send, receive)
+            await self._task_group.start(self._dispatcher.run, self._on_request, self._on_notify)
+            for binding in self._notification_bindings.values():
+                _, receive = self._binding_queues[binding.method]
                 self._task_group.start_soon(self._deliver_bound_notifications, binding, receive)
         except BaseException:
             # Unwind the entered task group before propagating: a cancellation

tests/client/test_extension.py

@@ -7,7 +7,7 @@
 """
 
 from dataclasses import FrozenInstanceError
-from typing import Any, Literal
+from typing import Any, Literal, cast
 
 import pytest
 from inline_snapshot import snapshot
@@ -142,6 +142,15 @@ def test_claim_rejects_mismatched_result_type_literal() -> None:
     assert str(exc_info.value) == snapshot("_OtherTagResult.result_type must be Literal['task']")
 
 
+def test_claim_rejects_method_outside_the_closed_verb_set() -> None:
+    """SDK-defined: claims attach to `tools/call` only (the Literal is the static gate);
+    an unchecked runtime value must not fold into tools/call parsing silently."""
+    with pytest.raises(ValueError) as exc_info:
+        _claim(method=cast("Literal['tools/call']", "prompts/get"))
+
+    assert str(exc_info.value) == snapshot("claims attach to ['tools/call'] only; got method 'prompts/get'")
+
+
 def test_claim_rejects_empty_protocol_versions() -> None:
     """SDK-defined: an empty version set could never activate; `None` is the
     spelling for "every modern version"."""

tests/client/test_session_claims.py

@@ -166,6 +166,19 @@ def test_claims_keyed_to_unadvertised_extension_rejected() -> None:
     )
 
 
+def test_empty_claim_sequence_rejected() -> None:
+    """SDK-defined: an empty claim set would make the ad-filter treat the identifier as
+    claim-bearing and drop it from the capability ad at every version; "claim-less" and
+    "advertises everywhere" stay the same thing by rejecting the empty spelling."""
+    with pytest.raises(ValueError) as exc_info:
+        ClientSession(dispatcher=_RecordingDispatcher(), extensions={_TASKS_EXT: {}}, result_claims={_TASKS_EXT: []})
+
+    assert str(exc_info.value) == snapshot(
+        "result_claims['com.example/tasks'] is empty; an empty claim set would drop the "
+        "extension from the capability ad at every version — omit the key instead"
+    )
+
+
 def test_empty_settings_count_as_an_advertised_extension() -> None:
     """SDK-defined: an extension advertised with empty settings ({}) is still an ad —
     claims keyed to it construct fine."""
@@ -228,6 +241,25 @@ async def test_legacy_adopt_clears_active_claims() -> None:
             assert dispatcher.calls[-1][0] == "tools/call"
 
 
+@pytest.mark.anyio
+async def test_modern_readopt_after_legacy_reactivates_claims() -> None:
+    """SDK-defined: adoption is re-entrant in both directions — after modern→legacy→
+    modern the claims are active again and the adopt-built adapter routes claimed raws."""
+    dispatcher = _RecordingDispatcher(tool_result=_CLAIMED_TASK_RESULT)
+    session = _claims_session(dispatcher, _task_claim())
+    with anyio.fail_after(5):
+        async with session:
+            _adopt_modern(session)
+            _adopt_handshake(session)
+            assert session._call_tool_adapter is _CallToolResultAdapter
+
+            _adopt_modern(session)
+            result = await session.call_tool("t", {}, allow_claimed=True)
+
+    assert isinstance(result, _TaskResult)
+    assert session._call_tool_adapter is not _CallToolResultAdapter
+
+
 # ── The version-aware capability ad ─────────────────────────────────────────
 
 
@@ -303,6 +335,22 @@ async def test_discover_probe_ad_includes_claim_identifiers_at_the_probe_version
     assert capabilities["extensions"] == {_TASKS_EXT: {}}
 
 
+@pytest.mark.anyio
+async def test_discover_probe_ad_drops_claim_identifiers_at_a_legacy_probe_version() -> None:
+    """SDK-defined: a lowlevel `send_discover` at a non-modern version string builds an
+    ad where no claim can be active, so the claim-bearing identifier drops coherently."""
+    dispatcher = _RecordingDispatcher()
+    session = _claims_session(dispatcher, _task_claim())
+    with anyio.fail_after(5):
+        async with session:
+            await session.send_discover(LATEST_HANDSHAKE_VERSION)
+
+    [(_, params, _)] = dispatcher.calls
+    assert params is not None
+    capabilities = params["_meta"][CLIENT_CAPABILITIES_META_KEY]
+    assert "extensions" not in capabilities
+
+
 # ── Routing through the adopt-built adapter ─────────────────────────────────