Persist elicited answers as raw wire content; reject ambiguous Elicit returns

request_state entries now store exactly the content the client sent (which
already passed validation) instead of re-deriving it from the validated model
with model_dump - persistence and restore previously used two codecs that
diverge whenever validation and serialization aliases differ, dropping a
legitimate stored answer on the third round and re-asking forever. Encode and
restore are now the identity on the client's bytes, so no serialization knob
can break the round-trip; the earlier by-alias dump is gone with the rest of
the model-dumping path.

A return annotation with more than one distinct Elicit arm now raises
InvalidSignature at registration: the restore path validates against the
single statically-known schema, so `-> Elicit[A] | Elicit[B]` either never
converges (the stored B answer fails A validation each round) or silently
injects a wrong-typed model when the shapes happen to overlap. Like cycles
and unclassifiable parameters, the ambiguous shape is rejected up front.

Legacy parity: elicit_with_validation wraps a schema-mismatched accepted
answer in the same stable error the 2026 path uses, instead of letting the
raw pydantic ValidationError text reach the client; noted in the migration
guide since callers catching ValidationError see ValueError now.

Author: maxisbey

docs/migration.md

@@ -786,6 +786,8 @@ Positional calls (`await ctx.info("hello")`) are unaffected.
 
 `Context.elicit()` (and `elicit_with_validation()`) now render the schema first and validate each property against the spec's `PrimitiveSchemaDefinition`, raising `TypeError` at the call site for anything outside it. `Optional[T]` fields render as `{"type": ...}` with the field omitted from `required` (previously the non-spec `anyOf` shape). A bare `list[str]` field is rejected because it renders without the required enum items; use `list[Literal[...]]` or `list[str]` with `json_schema_extra` supplying the items. Unions of multiple primitives (e.g. `int | str`) and nested models are rejected.
 
+A schema-mismatched *accepted* answer also fails differently: the call now raises `ValueError` with a stable message ("Received an accepted elicitation whose content does not match the requested schema") instead of letting pydantic's `ValidationError` escape with its internals. Code that caught `ValidationError` around `ctx.elicit()` should catch `ValueError` (or rely on the tool's error result).
+
 ### Replace `RootModel` by union types with `TypeAdapter` validation
 
 The following union types are no longer `RootModel` subclasses:

docs/tutorial/elicitation.md

@@ -76,8 +76,8 @@ A refusal is not an error. The tool decides what declining means (here, no booki
 
 !!! tip
     The answer is validated against your model before your code sees it. A client that sends
-    `"maybe"` for a `bool` doesn't corrupt your booking: the call fails with the
-    `ValidationError`, your `if` never runs.
+    `"maybe"` for a `bool` doesn't corrupt your booking: the call fails with a
+    schema-mismatch error, your `if` never runs.
 
 ## Ask before the tool runs
 

src/mcp/server/elicitation.py

@@ -116,7 +116,8 @@ async def elicit_with_validation(
     For sensitive data like credentials or OAuth flows, use elicit_url() instead.
 
     Raises:
-        ValueError: If the client accepted the elicitation without supplying content.
+        ValueError: If the client accepted the elicitation without supplying
+            content, or with content that does not match the requested schema.
     """
     json_schema = render_elicitation_schema(schema)
 
@@ -129,8 +130,12 @@ async def elicit_with_validation(
     if result.action == "accept":
         if result.content is None:
             raise ValueError("Received an accepted elicitation with no content")
-        # Validate and parse the content using the schema
-        validated_data = schema.model_validate(result.content)
+        try:
+            validated_data = schema.model_validate(result.content)
+        except ValidationError as e:
+            raise ValueError(
+                "Received an accepted elicitation whose content does not match the requested schema"
+            ) from e
         return AcceptedElicitation(data=validated_data)
     if result.action == "decline":
         return DeclinedElicitation()

src/mcp/server/mcpserver/resolve.py

@@ -183,21 +183,30 @@ def _contains_resolve(annotation: Any) -> bool:
     return any(_contains_resolve(arg) for arg in get_args(annotation))
 
 
-def _elicit_return_schema(return_annotation: Any) -> type[BaseModel] | None:
+def _elicit_return_schema(return_annotation: Any, name: str) -> type[BaseModel] | None:
     """Extract `T` from a resolver return type's `Elicit[T]` arm, if present.
 
     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.
+
+    Raises:
+        InvalidSignature: If the annotation has more than one `Elicit[...]` arm;
+            the runtime can honor only one static question schema per resolver.
     """
     # 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]
-            if isinstance(schema, type) and issubclass(schema, BaseModel):
-                return schema
-    return None
+    # Typing dedupes equal union members, so two arms here are genuinely distinct.
+    arms = [c for c in candidates if get_origin(c) is Elicit]
+    if len(arms) > 1:
+        raise InvalidSignature(
+            f"Resolver {name!r} return annotation has multiple Elicit arms; "
+            "a resolver asks one question - split it into separate resolvers"
+        )
+    if not arms:
+        return None
+    schema = get_args(arms[0])[0]
+    return schema if isinstance(schema, type) and issubclass(schema, BaseModel) else None
 
 
 def _is_union(annotation: Any) -> bool:
@@ -290,9 +299,8 @@ def analyze(fn: Callable[..., Any], stack: tuple[Hashable, ...]) -> None:
                 "expected a Context, an Annotated[_, Resolve(...)], or a tool argument by name"
             )
 
-        plans[key] = _ResolverPlan(
-            fn, params, is_async_callable(fn), _elicit_return_schema(hints.get("return")), wire_key
-        )
+        elicit_schema = _elicit_return_schema(hints.get("return"), _resolver_name(fn))
+        plans[key] = _ResolverPlan(fn, params, is_async_callable(fn), elicit_schema, wire_key)
         for dep in nested:
             analyze(dep, stack + (key,))
 
@@ -342,11 +350,13 @@ def __init__(
         self.answers: InputResponses = context.input_responses or {} if input_required else {}
         self.state = _decode_state(context.request_state) if input_required else {}
         # 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.
+        # the same bound method); `persist` holds the wire-shaped record of each
+        # elicited outcome, keyed by its wire key - exactly what the next round's
+        # `request_state` carries. Entries are the client's own (validated) wire
+        # data, never re-derived from a model, so encode-restore is the identity.
+        # Pure resolvers are cheap to re-run each round and are not persisted.
         self.cache: dict[Hashable, ElicitationResult[Any]] = {}
-        self.elicited: dict[str, ElicitationResult[Any]] = {}
+        self.persist: dict[str, _StateEntry] = {}
         self.pending: InputRequests = {}
 
 
@@ -399,7 +409,7 @@ 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.elicited))
+        return InputRequiredResult(input_requests=res.pending, request_state=_encode_state(res.persist))
     return injected
 
 
@@ -456,7 +466,6 @@ async def _resolve(fn: Callable[..., Any], res: _Resolution) -> ElicitationResul
 
     if _is_elicit(result):
         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. Plain outcomes are not
@@ -496,9 +505,14 @@ async def _elicit(elicit: Elicit[Any], key: str, res: _Resolution) -> Elicitatio
             raise ToolError(
                 f"Resolver {key!r} received an accepted elicitation whose content does not match the requested schema"
             ) from e
+        # Persist the exact wire content that just passed validation - never the
+        # model - so restoring next round revalidates the same bytes the client sent.
+        res.persist[key] = _StateEntry(action="accept", data=answer.content)
         return AcceptedElicitation(data=data)
     if answer.action == "decline":
+        res.persist[key] = _StateEntry(action="decline")
         return DeclinedElicitation()
+    res.persist[key] = _StateEntry(action="cancel")
     return CancelledElicitation()
 
 
@@ -591,18 +605,13 @@ def _decode_state(request_state: str | None) -> dict[str, _StateEntry]:
     return state.outcomes if state.v == _STATE_VERSION else {}
 
 
-def _encode_state(outcomes: Mapping[str, ElicitationResult[Any]]) -> str:
-    """Encode resolved outcomes (keyed by resolver path) for the next round."""
-    entries: dict[str, _StateEntry] = {}
-    for path, outcome in outcomes.items():
-        data = outcome.data if isinstance(outcome, AcceptedElicitation) else None
-        if isinstance(data, BaseModel):
-            # By alias: the stored shape must round-trip through
-            # `schema.model_validate` on restore, which expects the alias-keyed
-            # form the client answered with (the rendered schema is alias-keyed).
-            data = data.model_dump(mode="json", by_alias=True)
-        entries[path] = _StateEntry(action=outcome.action, data=data)
-    return _State(v=_STATE_VERSION, outcomes=entries).model_dump_json()
+def _encode_state(outcomes: Mapping[str, _StateEntry]) -> str:
+    """Encode recorded elicitation outcomes (keyed by wire key) for the next round.
+
+    Entries already hold the client's wire-shaped data exactly as it was sent (and
+    validated), so encoding is pure wrapping: encode-restore is the identity.
+    """
+    return _State(v=_STATE_VERSION, outcomes=dict(outcomes)).model_dump_json()
 
 
 def _outcome_from_state(entry: _StateEntry, schema: type[BaseModel] | None) -> ElicitationResult[Any]:
@@ -629,9 +638,10 @@ def _restore_outcome(res: _Resolution, key: str, schema: type[BaseModel] | None)
     the `_decode_state` treatment - dropped as if no progress was recorded, so the
     question is asked again - rather than surfacing a validation error.
 
-    Carries a restored outcome forward in `res.elicited`: if a later resolver is
-    still pending, the next round's `request_state` is built from `res.elicited`,
-    so an earlier answer must stay there or it would be dropped and re-asked.
+    Carries the original decoded entry forward unchanged in `res.persist`: if a
+    later resolver is still pending, the next round's `request_state` is built from
+    `res.persist`, so an earlier answer must stay there - byte-identical, never
+    re-derived - or it would be dropped and re-asked.
     """
     entry = res.state.get(key)
     if entry is None:
@@ -641,7 +651,7 @@ def _restore_outcome(res: _Resolution, key: str, schema: type[BaseModel] | None)
     except ValidationError:
         del res.state[key]
         return None
-    res.elicited[key] = outcome
+    res.persist[key] = entry
     return outcome
 
 

tests/docs_src/test_elicitation.py

@@ -124,7 +124,7 @@ async def on_elicit(context: ClientRequestContext, params: ElicitRequestParams)
         result = await client.call_tool("book_table", {"date": "2025-12-25", "party_size": 2})
     assert result.is_error
     assert isinstance(result.content[0], TextContent)
-    assert "Input should be a valid boolean" in result.content[0].text
+    assert "does not match the requested schema" in result.content[0].text
 
 
 class Address(BaseModel):