Fix core-arm sentinel collision and close binding queues on every exit

A claim tagged 'core' could collide with the adapter's internal routing
sentinel and hijack ordinary tools/call parsing; the sentinel is now
derived to never equal a claimed tag. Binding queues close in finally
blocks so a raising task-group exit cannot leak them. The mapping-form
migration error now also covers an empty dict; claim dedup keys by
resultType alone (the verb is single-valued at construction, and the
activation map already keys by tag); the one-time private-spelling
tree-grep test is gone; docstring backticks and a stray __future__
import brought in line with repo conventions.

Author: maxisbey

src/mcp/client/client.py

@@ -207,22 +207,22 @@ def _fold_extensions(extensions: Sequence[ClientExtension] | None) -> _FoldedExt
     Mirrors the server's consumption-time posture (`MCPServer._apply_extension`): a
     per-instance identifier is validated here because no class attribute existed to
     validate at definition time. `settings()` is read exactly once per extension and
-    the returned dict is held by reference. Duplicate `(method, resultType)` claims
-    and duplicate notification methods are rejected here, where both owning extensions
+    the returned dict is held by reference. Duplicate resultType claims and
+    duplicate notification methods are rejected here, where both owning extensions
     can be named — the session's own duplicate checks know only methods and tags.
     """
-    if not extensions:
-        return _FoldedExtensions(ad=None, claims=None, bindings=None, by_model={})
     if isinstance(extensions, Mapping):
         raise TypeError(
             "extensions= takes a sequence of ClientExtension instances; the mapping form was "
             "replaced — use advertise(identifier, settings) for advertise-only entries"
         )
+    if not extensions:
+        return _FoldedExtensions(ad=None, claims=None, bindings=None, by_model={})
     ad: dict[str, dict[str, Any]] = {}
     claims: dict[str, tuple[ResultClaim[Any], ...]] = {}
     bindings: list[NotificationBinding[Any]] = []
     by_model: dict[type[Result], ResultClaim[Any]] = {}
-    claim_owners: dict[tuple[str, str], str] = {}
+    claim_owners: dict[str, str] = {}
     binding_owners: dict[str, str] = {}
     for extension in extensions:
         identifier = getattr(extension, "identifier", None)
@@ -237,20 +237,18 @@ def _fold_extensions(extensions: Sequence[ClientExtension] | None) -> _FoldedExt
         ad[identifier] = extension.settings()
         extension_claims = tuple(extension.claims())
         for claim in extension_claims:
-            key = (claim.method, claim.result_type)
-            if key in claim_owners:
-                owner = claim_owners[key]
+            tag = claim.result_type
+            if tag in claim_owners:
+                owner = claim_owners[tag]
                 both = (
                     f"extension {identifier!r} claims"
                     if owner == identifier
                     else (f"extensions {owner!r} and {identifier!r} both claim")
                 )
-                raise ValueError(
-                    f"{both} {claim.method!r} resultType {claim.result_type!r}; a wire tag can have only one resolver"
-                )
-            claim_owners[key] = identifier
-            # Collision-free by construction: a model's `result_type` Literal pins it to
-            # exactly one tag, and each (method, tag) pair has exactly one owner.
+                raise ValueError(f"{both} resultType {tag!r}; a wire tag can have only one resolver")
+            claim_owners[tag] = identifier
+            # One model, one tag: the model's result_type Literal is pinned to exactly
+            # this tag at claim construction, so the type-keyed index cannot collide.
             by_model[claim.model] = claim
         if extension_claims:
             claims[identifier] = extension_claims

src/mcp/client/session.py

@@ -241,6 +241,9 @@ def _build_call_tool_adapter(
     if not active:
         return _CallToolResultAdapter
     tags = frozenset(active)
+    core_arm = "core"
+    while core_arm in tags:  # the routing sentinel must never collide with a claimed tag
+        core_arm += "-"
 
     def _route(value: Any) -> str:
         # pydantic hands the discriminator either the raw inbound dict or an
@@ -251,9 +254,9 @@ def _route(value: Any) -> str:
             tag = cast("dict[str, Any]", value).get("resultType")
         else:
             tag = getattr(value, "result_type", None)
-        return tag if isinstance(tag, str) and tag in tags else "core"
+        return tag if isinstance(tag, str) and tag in tags else core_arm
 
-    arms: list[Any] = [Annotated[types.CallToolResult | types.InputRequiredResult, Tag("core")]]
+    arms: list[Any] = [Annotated[types.CallToolResult | types.InputRequiredResult, Tag(core_arm)]]
     arms += [Annotated[claim.model, Tag(tag)] for tag, claim in active.items()]
     # reduce(or_, ...) builds the Union dynamically; PEP-646 star-unpack needs py3.11+.
     return TypeAdapter(Annotated[reduce(or_, arms), Discriminator(_route)])
@@ -270,7 +273,7 @@ def _index_claims(
     advertised extension and no wire tag may be claimed twice.
     """
     indexed: dict[str, tuple[ResultClaim[Any], ...]] = {}
-    seen: set[tuple[str, str]] = set()
+    seen: set[str] = set()
     for identifier, claims in (result_claims or {}).items():
         if extensions is None or identifier not in extensions:
             raise ValueError(
@@ -283,10 +286,9 @@ def _index_claims(
                 "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:
-                raise ValueError(f"duplicate result claim for {claim.method!r} resultType {claim.result_type!r}")
-            seen.add(key)
+            if claim.result_type in seen:
+                raise ValueError(f"duplicate result claim for resultType {claim.result_type!r}")
+            seen.add(claim.result_type)
         indexed[identifier] = tuple(claims)
     return indexed
 
@@ -411,8 +413,10 @@ async def __aenter__(self) -> Self:
             # Shield the group's own scope (a new one would break LIFO exit)
             # so a pending outer cancellation cannot re-fire inside __aexit__.
             task_group.cancel_scope.shield = True
-            await task_group.__aexit__(None, None, None)
-            self._close_binding_queues()
+            try:
+                await task_group.__aexit__(None, None, None)
+            finally:
+                self._close_binding_queues()
             raise
         return self
 
@@ -425,8 +429,10 @@ async def __aexit__(
         # Exit must not block: cancel the dispatcher, binding consumers, and in-flight callbacks.
         assert self._task_group is not None
         self._task_group.cancel_scope.cancel()
-        result = await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
-        self._close_binding_queues()
+        try:
+            result = await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
+        finally:
+            self._close_binding_queues()
         await resync_tracer()
         return result
 
@@ -975,15 +981,15 @@ async def call_tool(
             allow_input_required: When ``False`` (default), an `InputRequiredResult`
                 from the server raises `RuntimeError`; when ``True``, it is returned
                 so the caller can resolve the requests and retry.
-            allow_claimed: When ``False`` (default), a claimed (extension) result
-                shape raises `UnexpectedClaimedResult`; when ``True``, the parsed
+            allow_claimed: When `False` (default), a claimed (extension) result
+                shape raises `UnexpectedClaimedResult`; when `True`, the parsed
                 claim model is returned for the caller to handle.
 
         Raises:
             RuntimeError: If the server returns an `InputRequiredResult` and
                 ``allow_input_required`` is ``False``.
             UnexpectedClaimedResult: If a claimed result shape parses and
-                ``allow_claimed`` is ``False``; carries the parsed value.
+                `allow_claimed` is `False`; carries the parsed value.
         """
         result = await self.send_request(
             types.CallToolRequest(

tests/client/test_client_extensions.py

@@ -185,7 +185,7 @@ def test_one_extension_claiming_a_tag_twice_reads_as_one_owner() -> None:
         Client(_add_server(), extensions=[_SelfConflictingClaims()])
 
     assert str(exc_info.value) == snapshot(
-        "extension 'com.example/twice' claims 'tools/call' resultType 'twice'; a wire tag can have only one resolver"
+        "extension 'com.example/twice' claims resultType 'twice'; a wire tag can have only one resolver"
     )
 
 
@@ -252,8 +252,8 @@ def test_conflicting_claims_across_extensions_name_both_owners() -> None:
         Client(_add_server(), extensions=[_VoucherExtension(_unreachable_resolve), _RivalVoucherExtension()])
 
     assert str(exc_info.value) == snapshot(
-        "extensions 'com.example/voucher' and 'com.example/rival' both claim 'tools/call' "
-        "resultType 'voucher'; a wire tag can have only one resolver"
+        "extensions 'com.example/voucher' and 'com.example/rival' both claim resultType "
+        "'voucher'; a wire tag can have only one resolver"
     )
 
 

tests/client/test_send_request_mcp_name.py

@@ -7,7 +7,6 @@
 `send_request` typing: a `Request[...]` subclass passes without a cast.
 """
 
-from __future__ import annotations
 
 from collections.abc import Mapping
 from typing import Any, Literal

tests/client/test_session_claims.py

@@ -139,7 +139,7 @@ def test_duplicate_claim_tag_across_extensions_rejected() -> None:
             result_claims={_TASKS_EXT: [_task_claim()], _AD_ONLY_EXT: [_task_claim()]},
         )
 
-    assert str(exc_info.value) == snapshot("duplicate result claim for 'tools/call' resultType 'task'")
+    assert str(exc_info.value) == snapshot("duplicate result claim for resultType 'task'")
 
 
 def test_claims_keyed_to_unadvertised_extension_rejected() -> None:
@@ -351,6 +351,35 @@ async def test_discover_probe_ad_drops_claim_identifiers_at_a_legacy_probe_versi
     assert "extensions" not in capabilities
 
 
+class _CoreTaggedResult(Result):
+    """A claim whose wire tag collides with the adapter's internal routing sentinel."""
+
+    result_type: Literal["core"] = "core"
+    payload: str = ""
+
+
+async def _resolve_core_tagged(result: _CoreTaggedResult, ctx: ClaimContext) -> CallToolResult:
+    raise NotImplementedError
+
+
+@pytest.mark.anyio
+async def test_claim_tagged_core_cannot_hijack_core_parsing() -> None:
+    """SDK-defined: "core" is not protocol vocabulary, so a claim may use it as a wire
+    tag — and the adapter's internal routing sentinel must not collide: ordinary tool
+    results still parse as core results, and a claimed `core` raw routes to the model."""
+    claim = ResultClaim(result_type="core", model=_CoreTaggedResult, resolve=_resolve_core_tagged)
+    dispatcher = _RecordingDispatcher(tool_result={"resultType": "core", "payload": "p-1"})
+    session = ClientSession(dispatcher=dispatcher, extensions={_TASKS_EXT: {}}, result_claims={_TASKS_EXT: [claim]})
+    with anyio.fail_after(5):
+        async with session:
+            _adopt_modern(session)
+            ordinary = session._call_tool_adapter.validate_python(_COMPLETE_TOOL_RESULT)
+            claimed = await session.call_tool("t", {}, allow_claimed=True)
+
+    assert isinstance(ordinary, CallToolResult)
+    assert isinstance(claimed, _CoreTaggedResult)
+
+
 # ── Routing through the adopt-built adapter ─────────────────────────────────
 
 

tests/client/test_session_promotions.py

@@ -1,7 +1,5 @@
 """`dispatch_input_request` and `validate_tool_result` are public `ClientSession` API."""
 
-import re
-from pathlib import Path
 
 import mcp_types as types
 import pytest
@@ -68,17 +66,3 @@ async def test_validate_tool_result_raises_on_schema_mismatch() -> None:
         # Stable SDK prefix only: the message tail is jsonschema text that shifts with the dependency.
         with pytest.raises(RuntimeError, match="Invalid structured content returned by tool t"):
             await client.session.validate_tool_result("t", CallToolResult(content=[], structured_content={"x": "no"}))
-
-
-def _spell_private(name: str) -> str:
-    return f"_{name}"
-
-
-def test_no_private_spelling_references_remain() -> None:
-    """The promotions are renames, not aliases — the old private names are gone from `src/`."""
-    pattern = re.compile(f"{_spell_private('dispatch_input_request')}|{_spell_private('validate_tool_result')}")
-    src = Path(__file__).resolve().parents[2] / "src"
-    offenders = [
-        (path.name, match) for path in sorted(src.rglob("*.py")) for match in pattern.findall(path.read_text())
-    ]
-    assert not offenders