Fix resolver MRTR edge cases from review

- In-call cache keyed by _resolver_key (instance-distinct) again; _state_key adds
  id(__self__) for the wire key, so two instances of one bound method no longer
  collide and silently share an outcome.
- resolve_arguments reads ctx.protocol_version (new Context property, None outside
  a request) instead of dereferencing request_context, so direct
  MCPServer.call_tool() works for tools whose resolvers never elicit.
- request_state persists only elicited outcomes (always validated models); a
  resolver that resolves without eliciting is pure and re-runs each round. Fixes
  the json.dumps crash on non-serializable returns (datetime/set/...) and the
  dict-degradation of restored values.
- _elicit_return_schema handles a bare Elicit[T] return (not only unions).
- _INPUT_REQUIRED_VERSION pinned to '2026-07-28' instead of LATEST_MODERN_VERSION.
- accept with no content raises ToolError instead of silently reporting cancel.
- Independent nested resolver deps batch into one round (catch _Pending per dep).
- Test cleanup: drop dead helpers, hoist CreateMessageResult import.

Add regression tests for each; document the narrowed elicited-only persistence.

Author: Kludex

docs/migration.md

@@ -1515,7 +1515,7 @@ async def delete_folder(
 
 The `confirm_delete` resolver reads the tool's own `path` argument by name, lists the folder, and only elicits when the folder is non-empty - an empty folder resolves to `Confirm(ok=True)` with no round-trip to the client. Because `delete_folder` annotates the result union, it handles every outcome: the user accepting and confirming, accepting but declining to delete (`ok=False`), declining the elicitation, or cancelling it.
 
-The framework drives elicitation over whichever transport the negotiated protocol provides, so the resolver and tool code above is unchanged either way. At `2026-07-28` and later it returns an `InputRequiredResult` carrying the questions and resumes when the client retries `call_tool(..., input_responses=..., request_state=...)` (independent resolvers are batched into one round; a resolver that depends on another's answer is asked in a later round). At `2025-11-25` and earlier it issues a synchronous `elicitation/create` request mid-call. Resolved outcomes are carried in `request_state` across rounds so each resolver resolves once per call.
+The framework drives elicitation over whichever transport the negotiated protocol provides, so the resolver and tool code above is unchanged either way. At `2026-07-28` and later it returns an `InputRequiredResult` carrying the questions and resumes when the client retries `call_tool(..., input_responses=..., request_state=...)` (independent resolvers are batched into one round; a resolver that depends on another's answer is asked in a later round). At `2025-11-25` and earlier it issues a synchronous `elicitation/create` request mid-call. Elicited answers are carried in `request_state` across rounds, so each question is asked once; a resolver that resolves without eliciting is pure and may re-run each round.
 
 Resolved parameters are omitted from the tool's input schema, so the client never supplies them. Resolver parameters that cannot be classified, and cyclic resolver dependencies, raise at registration time.
 

src/mcp/server/mcpserver/context.py

@@ -238,6 +238,11 @@ def request_id(self) -> str:
         """Get the unique ID for this request."""
         return str(self.request_context.request_id)
 
+    @property
+    def protocol_version(self) -> str | None:
+        """The negotiated protocol version, or `None` outside of an active request."""
+        return self._request_context.protocol_version if self._request_context is not None else None
+
     @property
     def input_responses(self) -> InputResponses | None:
         """Client responses to a prior `InputRequiredResult.input_requests`.

src/mcp/server/mcpserver/resolve.py

@@ -12,8 +12,9 @@
 and resumes when the client retries with `input_responses`/`request_state`
 (independent resolvers are asked in one round; a resolver depending on another's
 answer is asked in a later round). At <= 2025-11-25 it issues a synchronous
-`elicitation/create` request mid-call. Resolved outcomes are carried in
-`request_state` across rounds, so each resolver resolves once per logical call.
+`elicitation/create` request mid-call. Only *elicited* outcomes are carried in
+`request_state` across rounds (so the user is asked each question once); a
+resolver that returns a value without eliciting is pure and may re-run each round.
 
 Whether the consumer receives the unwrapped model or the full
 `ElicitationResult` union is decided by the consumer's annotation:
@@ -26,6 +27,7 @@
 from __future__ import annotations
 
 import inspect
+import types
 import typing
 from collections.abc import Callable, Hashable, Mapping
 from typing import Annotated, Any, Generic, Literal, TypeGuard, get_args, get_origin
@@ -39,7 +41,7 @@
     InputRequiredResult,
     InputResponses,
 )
-from mcp_types.version import LATEST_MODERN_VERSION, is_version_at_least
+from mcp_types.version import is_version_at_least
 from pydantic import BaseModel, ValidationError
 from typing_extensions import TypeVar
 
@@ -61,7 +63,8 @@
 
 # First protocol revision whose `tools/call` carries elicitation inside
 # `InputRequiredResult` rather than as a standalone server-to-client request.
-_INPUT_REQUIRED_VERSION = LATEST_MODERN_VERSION  # "2026-07-28"
+# Pinned (not `LATEST_MODERN_VERSION`, which moves when newer revisions are added).
+_INPUT_REQUIRED_VERSION = "2026-07-28"
 _STATE_VERSION = 1
 
 
@@ -158,10 +161,12 @@ def find_resolved_parameters(fn: Callable[..., Any]) -> dict[str, tuple[Resolve,
 def _elicit_return_schema(return_annotation: Any) -> type[BaseModel] | None:
     """Extract `T` from a resolver return type's `Elicit[T]` arm, if present.
 
-    Lets an outcome restored from `request_state` (a plain dict) be re-validated
-    into its model so dependent resolvers and tools receive a typed value.
+    Handles a bare `-> Elicit[T]` and a `-> T | Elicit[T]` union. Lets an elicited
+    outcome restored from `request_state` (a plain dict) be re-validated into its
+    model so dependent resolvers and tools receive a typed value.
     """
-    candidates = get_args(return_annotation) if get_origin(return_annotation) is not None else (return_annotation,)
+    # A bare `Elicit[T]` is itself a candidate; a union contributes its members.
+    candidates = get_args(return_annotation) if _is_union(return_annotation) else (return_annotation,)
     for candidate in candidates:
         if get_origin(candidate) is Elicit:
             schema = get_args(candidate)[0]
@@ -170,6 +175,10 @@ def _elicit_return_schema(return_annotation: Any) -> type[BaseModel] | None:
     return None
 
 
+def _is_union(annotation: Any) -> bool:
+    return get_origin(annotation) in (typing.Union, types.UnionType)
+
+
 def _wants_union(type_arg: Any) -> bool:
     """True when `type_arg` is an `ElicitationResult` member (or a union of them).
 
@@ -296,19 +305,26 @@ def __init__(
         self.input_required = input_required
         self.answers: InputResponses = context.input_responses or {} if input_required else {}
         self.state = _decode_state(context.request_state) if input_required else {}
-        self.cache: dict[str, ElicitationResult[Any]] = {}
+        # In-call dedup keyed by resolver identity (distinguishes two instances of
+        # the same bound method); `elicited` holds only outcomes that came from an
+        # elicitation, keyed by their wire key - these are what `request_state`
+        # persists, since pure resolvers are cheap to re-run each round.
+        self.cache: dict[Hashable, ElicitationResult[Any]] = {}
+        self.elicited: dict[str, ElicitationResult[Any]] = {}
         self.pending: InputRequests = {}
 
 
 def _state_key(fn: Callable[..., Any]) -> str:
-    """Process-stable wire key for a resolver.
+    """Process-stable wire key for a resolver's elicitation.
 
-    `id`-based keys aren't stable across `input_required` rounds (a retry may land
-    on a different worker), so memoize and key `input_requests`/`request_state` by
-    the resolver's `module:qualname`. Two consumers of the same resolver therefore
-    share one cache entry, one question, and one stored outcome.
+    `id(fn)` isn't stable across `input_required` rounds, so key `input_requests` /
+    `request_state` by `module:qualname`. Bound methods add their `__self__` id so
+    two instances of the same method get distinct questions and stored outcomes
+    (the registered `Resolve(...)` holds the instance for the call's lifetime).
     """
-    return f"{getattr(fn, '__module__', '')}:{getattr(fn, '__qualname__', fn)}"
+    base = f"{getattr(fn, '__module__', '')}:{getattr(fn, '__qualname__', fn)!s}"
+    bound_self = getattr(fn, "__self__", None)
+    return f"{base}#{id(bound_self)}" if bound_self is not None else base
 
 
 async def resolve_arguments(
@@ -324,14 +340,19 @@ async def resolve_arguments(
     negotiated protocol is >= 2026-07-28), returns an `InputRequiredResult`
     carrying the batched questions instead; the tool body is not run.
 
-    Each resolver runs at most once per logical call - across multiple
-    `input_required` rounds, resolved outcomes are carried in `request_state`.
+    An eliciting resolver asks its question once - its answer is carried in
+    `request_state` across rounds - while a resolver that resolves without
+    eliciting is pure and may re-run on each round.
 
     Raises:
         ToolError: If an elicited value is declined or cancelled and the consumer
             asked for the unwrapped model (rather than the result union).
     """
-    res = _Resolution(plans, tool_args, context, uses_input_required(context.request_context.protocol_version))
+    # `ctx.protocol_version` is `None` outside an active request: `MCPServer.call_tool()`
+    # called directly builds such a `Context`, and a tool whose resolvers never elicit
+    # must still work there. A missing version means the synchronous (non-input_required)
+    # transport, which never reaches a server-to-client request anyway.
+    res = _Resolution(plans, tool_args, context, uses_input_required(context.protocol_version))
     injected: dict[str, Any] = {}
     for name, (marker, wants_union) in resolved_params.items():
         try:
@@ -341,39 +362,49 @@ async def resolve_arguments(
         injected[name] = outcome if wants_union else _unwrap(outcome, name)
 
     if res.pending:
-        return InputRequiredResult(input_requests=res.pending, request_state=_encode_state(res.cache))
+        return InputRequiredResult(input_requests=res.pending, request_state=_encode_state(res.elicited))
     return injected
 
 
 async def _resolve(fn: Callable[..., Any], res: _Resolution) -> ElicitationResult[Any]:
-    """Resolve one resolver, memoized by its process-stable state key.
+    """Resolve one resolver, deduped within the call by its resolver identity.
 
     Raises `_Pending` when the resolver (or one of its dependencies) needs client
     input that has not arrived yet.
     """
-    key = _state_key(fn)
-    if key in res.cache:
-        return res.cache[key]
-    if key in res.pending:
+    cache_key = _resolver_key(fn)
+    if cache_key in res.cache:
+        return res.cache[cache_key]
+
+    plan = res.plans[cache_key]
+    wire_key = _state_key(fn)
+    if wire_key in res.pending:
         # Already asked this round by another consumer; don't run the resolver again.
         raise _Pending
-
-    plan = res.plans[_resolver_key(fn)]
-    if key in res.state:
-        outcome = _outcome_from_state(res.state[key], plan.elicit_schema)
-        res.cache[key] = outcome
+    if wire_key in res.state:
+        outcome = _outcome_from_state(res.state[wire_key], plan.elicit_schema)
+        res.cache[cache_key] = outcome
         return outcome
 
     kwargs: dict[str, Any] = {}
+    dep_pending = False
     for param_name, param_plan in plan.params.items():
         if param_plan.kind == "context":
             kwargs[param_name] = res.context
         elif param_plan.kind == "by_name":
             kwargs[param_name] = res.tool_args[param_name]
         else:
             assert param_plan.resolve is not None
-            dep_outcome = await _resolve(param_plan.resolve.fn, res)
+            try:
+                # Visit every dependency so independent ones that need input are all
+                # collected into `res.pending` and batched into a single round.
+                dep_outcome = await _resolve(param_plan.resolve.fn, res)
+            except _Pending:
+                dep_pending = True
+                continue
             kwargs[param_name] = dep_outcome if param_plan.wants_union else _unwrap(dep_outcome, param_name)
+    if dep_pending:
+        raise _Pending
 
     result: Any
     if plan.is_async:
@@ -382,13 +413,15 @@ async def _resolve(fn: Callable[..., Any], res: _Resolution) -> ElicitationResul
         result = await anyio.to_thread.run_sync(lambda: fn(**kwargs))
 
     if _is_elicit(result):
-        outcome = await _elicit(result, key, res)
+        outcome = await _elicit(result, wire_key, res)
+        res.elicited[wire_key] = outcome
     else:
         # A resolver may return any type (not just `BaseModel`), so accept it as the
-        # outcome without validating against the schema bound.
+        # outcome without validating against the schema bound. Plain outcomes are not
+        # persisted in `request_state`; the resolver re-runs next round instead.
         outcome = _accepted(result)
 
-    res.cache[key] = outcome
+    res.cache[cache_key] = outcome
     return outcome
 
 
@@ -403,7 +436,9 @@ async def _elicit(elicit: Elicit[Any], key: str, res: _Resolution) -> Elicitatio
         raise _Pending
     if not isinstance(answer, ElicitResult):
         raise ToolError(f"Resolver {key!r} received a non-elicitation response")
-    if answer.action == "accept" and answer.content is not None:
+    if answer.action == "accept":
+        if answer.content is None:
+            raise ToolError(f"Resolver {key!r} received an accepted elicitation with no content")
         return AcceptedElicitation(data=elicit.schema.model_validate(answer.content))
     if answer.action == "decline":
         return DeclinedElicitation()