modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Drive resolver elicitation over the 2026-07-28 input_required flow (#2986)

Browse source 
Co-authored-by: Max Isbey <224885523+maxisbey@users.noreply.github.com>
This commit is contained in:
Marcelo Trylesinski 2026-06-29 15:39:43 +02:00 • committed by GitHub
commit c85836a081
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 1521 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

39
src/mcp/server/elicitation.py
View file

@ -87,6 +87,18 @@ def _validate_rendered_properties(json_schema: dict[str, Any]) -> None:
) from None
def render_elicitation_schema(schema: type[BaseModel]) -> dict[str, Any]:
"""Render a model as the spec-valid `requested_schema` for an elicitation.
Raises:
TypeError: If a field renders as something the spec's
`PrimitiveSchemaDefinition` does not accept.
"""
json_schema = schema.model_json_schema(schema_generator=_ElicitationJsonSchema)
_validate_rendered_properties(json_schema)
return json_schema
async def elicit_with_validation(
session: ServerSession,
message: str,
@ -102,9 +114,12 @@ async def elicit_with_validation(
the user or automatically generating a response.
For sensitive data like credentials or OAuth flows, use elicit_url() instead.
Raises:
ValueError: If the client accepted the elicitation without supplying
content, or with content that does not match the requested schema.
"""
json_schema = schema.model_json_schema(schema_generator=_ElicitationJsonSchema)
_validate_rendered_properties(json_schema)
json_schema = render_elicitation_schema(schema)
result = await session.elicit_form(
message=message,
@ -112,17 +127,19 @@ async def elicit_with_validation(
related_request_id=related_request_id,
)
if result.action == "accept" and result.content is not None:
# Validate and parse the content using the schema
validated_data = schema.model_validate(result.content)
if result.action == "accept":
if result.content is None:
raise ValueError("Received an accepted elicitation with no 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)
elif result.action == "decline":
if result.action == "decline":
return DeclinedElicitation()
elif result.action == "cancel":
return CancelledElicitation()
else: # pragma: no cover
# This should never happen, but handle it just in case
raise ValueError(f"Unexpected elicitation action: {result.action}")
return CancelledElicitation()
async def elicit_url(

5
src/mcp/server/mcpserver/context.py
View file

@ -232,6 +232,11 @@ class Context(BaseModel, Generic[LifespanContextT, RequestT]):
"""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`.

410
src/mcp/server/mcpserver/resolve.py
View file

@ -7,25 +7,47 @@ take tool arguments by name, and take the `Context`. A resolver may return
`Elicit[T]` to ask the client; the framework runs the elicitation and injects the
answer.
The framework picks the elicitation transport from the negotiated protocol. At
>= 2026-07-28 it returns an `InputRequiredResult` carrying the batched questions
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. 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:
- `Annotated[T, Resolve(fn)]` -> unwrapped `T`; decline/cancel aborts the call.
- `Annotated[ElicitationResult[T], Resolve(fn)]` (or a specific member) -> the
full outcome; the consumer branches on accept/decline/cancel.
Each resolver runs at most once per `tools/call` (memoized by function identity).
"""
from __future__ import annotations
import inspect
import types
import typing
from collections.abc import Callable, Hashable, Mapping
from typing import Annotated, Any, Generic, cast, get_args, get_origin
from typing import Annotated, Any, Generic, Literal, TypeGuard, get_args, get_origin
import anyio.to_thread
from pydantic import BaseModel
from mcp_types import (
MISSING_REQUIRED_CLIENT_CAPABILITY,
ClientCapabilities,
ElicitationCapability,
ElicitRequest,
ElicitRequestFormParams,
ElicitResult,
FormElicitationCapability,
InputRequests,
InputRequiredResult,
InputResponses,
MissingRequiredClientCapabilityErrorData,
)
from mcp_types.version import is_version_at_least
from pydantic import BaseModel, ValidationError
from typing_extensions import TypeVar
from mcp.server.elicitation import (
@ -33,16 +55,24 @@ from mcp.server.elicitation import (
CancelledElicitation,
DeclinedElicitation,
ElicitationResult,
render_elicitation_schema,
)
from mcp.server.mcpserver.context import Context
from mcp.server.mcpserver.exceptions import InvalidSignature, ToolError
from mcp.shared._callable_inspection import is_async_callable
from mcp.shared.exceptions import MCPError
T = TypeVar("T", bound=BaseModel)
# The union members the framework injects when a consumer opts into the outcome.
_ELICITATION_RESULT_MEMBERS = (AcceptedElicitation, DeclinedElicitation, CancelledElicitation)
# First protocol revision whose `tools/call` carries elicitation inside
# `InputRequiredResult` rather than as a standalone server-to-client request.
# Pinned (not `LATEST_MODERN_VERSION`, which moves when newer revisions are added).
_INPUT_REQUIRED_VERSION = "2026-07-28"
_STATE_VERSION = 1
class Resolve:
"""Marker for `Annotated[T, Resolve(fn)]`: fill the parameter by running `fn`."""
@ -79,10 +109,24 @@ class _ParamPlan:
class _ResolverPlan:
"""A resolver's parameters and whether it is async, analyzed once."""
def __init__(self, fn: Callable[..., Any], params: dict[str, _ParamPlan], is_async: bool) -> None:
def __init__(
self,
fn: Callable[..., Any],
params: dict[str, _ParamPlan],
is_async: bool,
elicit_schema: type[BaseModel] | None,
wire_key: str,
) -> None:
self.fn = fn
self.params = params
self.is_async = is_async
# The `T` from the resolver's `Elicit[T]` return arm, if annotated. Used to
# re-validate an outcome restored from `request_state` into a model.
self.elicit_schema = elicit_schema
# Deterministic, collision-free key for this resolver's elicitation on the
# wire (`input_requests`/`request_state`). Assigned at registration so it is
# stable across rounds even when `module:qualname` collides (closures).
self.wire_key = wire_key
def _type_hints(fn: Callable[..., Any]) -> dict[str, Any]:
@ -139,6 +183,36 @@ def _contains_resolve(annotation: Any) -> bool:
return any(_contains_resolve(arg) for arg in get_args(annotation))
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,)
# 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:
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).
@ -187,6 +261,9 @@ def build_resolver_plans(
or a tool argument by name).
"""
plans: dict[Hashable, _ResolverPlan] = {}
# Count how many distinct resolvers share each `module:qualname` base so closures
# from one factory get distinct, deterministic wire keys (`base`, `base#1`, ...).
base_counts: dict[str, int] = {}
def analyze(fn: Callable[..., Any], stack: tuple[Hashable, ...]) -> None:
key = _resolver_key(fn)
@ -195,6 +272,11 @@ def build_resolver_plans(
if key in plans:
return
base = _state_key(fn)
seen = base_counts.get(base, 0)
base_counts[base] = seen + 1
wire_key = base if seen == 0 else f"{base}#{seen}"
hints = _type_hints(fn)
sig = inspect.signature(fn)
params: dict[str, _ParamPlan] = {}
@ -217,7 +299,8 @@ def build_resolver_plans(
"expected a Context, an Annotated[_, Resolve(...)], or a tool argument by name"
)
plans[key] = _ResolverPlan(fn, params, is_async_callable(fn))
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,))
@ -241,76 +324,337 @@ def _is_context_annotation(annotation: Any) -> bool:
return any(isinstance(c, type) and issubclass(c, Context) for c in candidates)
class _Pending(Exception):
"""Internal: a resolver needs client input not yet available this round."""
class _Resolution:
"""Per-`tools/call` resolution state, shared across the DAG walk.
`input_required` selects the transport: at >= 2026-07-28 elicitations are
batched into `pending` and surfaced as an `InputRequiredResult`; at older
revisions each `Elicit` is answered synchronously via `ctx.elicit`.
"""
def __init__(
self,
plans: Mapping[Hashable, _ResolverPlan],
tool_args: Mapping[str, Any],
context: Context[Any, Any],
input_required: bool,
) -> None:
self.plans = plans
self.tool_args = tool_args
self.context = context
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 {}
# In-call dedup keyed by resolver identity (distinguishes two instances of
# 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.persist: dict[str, _StateEntry] = {}
self.pending: InputRequests = {}
def _state_key(fn: Callable[..., Any]) -> str:
"""Worker-stable base wire key for a resolver, derived only from registration data.
`input_requests`/`request_state` must round-trip through the client and resume on
any worker (stateless HTTP), so the key carries no `id(...)`: it is the resolver's
`module:qualname` (a callable object uses its type's). Distinct resolvers that
share this base - two instances of one method, two closures from one factory - are
disambiguated deterministically by `build_resolver_plans` (`base`, `base#1`, ...).
"""
qualname = getattr(fn, "__qualname__", None) or type(fn).__qualname__
module = getattr(fn, "__module__", None) or type(fn).__module__
return f"{module}:{qualname}"
async def resolve_arguments(
resolved_params: Mapping[str, tuple[Resolve, bool]],
plans: Mapping[Hashable, _ResolverPlan],
tool_args: Mapping[str, Any],
context: Context[Any, Any],
) -> dict[str, Any]:
) -> dict[str, Any] | InputRequiredResult:
"""Resolve every `Resolve`-marked tool parameter into a concrete value.
Each resolver runs at most once (memoized by function identity). Returns a
mapping of tool parameter name to the value to inject.
Returns the mapping of tool parameter name to injected value when every
resolver is satisfied. When a resolver still needs client input (and the
negotiated protocol is >= 2026-07-28), returns an `InputRequiredResult`
carrying the batched questions instead; the tool body is not run.
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).
"""
cache: dict[Hashable, ElicitationResult[Any]] = {}
# `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():
outcome = await _resolve(marker.fn, plans, tool_args, context, cache)
try:
outcome = await _resolve(marker.fn, res)
except _Pending:
continue
injected[name] = outcome if wants_union else _unwrap(outcome, name)
if res.pending:
return InputRequiredResult(input_requests=res.pending, request_state=_encode_state(res.persist))
return injected
async def _resolve(
fn: Callable[..., Any],
plans: Mapping[Hashable, _ResolverPlan],
tool_args: Mapping[str, Any],
context: Context[Any, Any],
cache: dict[Hashable, ElicitationResult[Any]],
) -> ElicitationResult[Any]:
key = _resolver_key(fn)
if key in cache:
return cache[key]
async def _resolve(fn: Callable[..., Any], res: _Resolution) -> ElicitationResult[Any]:
"""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.
"""
cache_key = _resolver_key(fn)
if cache_key in res.cache:
return res.cache[cache_key]
plan = res.plans[cache_key]
wire_key = plan.wire_key
if wire_key in res.pending:
# Already asked this round by another consumer; don't run the resolver again.
raise _Pending
# Restore a prior round's outcome directly only when its model is known from the
# `Elicit[T]` return arm. Without that (a resolver that elicits but isn't annotated
# `-> ... Elicit[T]`), fall through and re-run the resolver so `_elicit` can
# re-validate the stored answer against the live `Elicit.schema`.
if wire_key in res.state and (plan.elicit_schema is not None or res.state[wire_key].action != "accept"):
outcome = _restore_outcome(res, wire_key, plan.elicit_schema)
if outcome is not None:
res.cache[cache_key] = outcome
return outcome
plan = plans[key]
kwargs: dict[str, Any] = {}
dep_pending = False
for param_name, param_plan in plan.params.items():
if param_plan.kind == "context":
kwargs[param_name] = context
kwargs[param_name] = res.context
elif param_plan.kind == "by_name":
kwargs[param_name] = tool_args[param_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, plans, tool_args, context, cache)
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:
result = await fn(**kwargs)
else:
result = await anyio.to_thread.run_sync(lambda: fn(**kwargs))
outcome: ElicitationResult[Any]
if isinstance(result, Elicit):
elicit = cast("Elicit[BaseModel]", result)
outcome = await context.elicit(elicit.message, elicit.schema)
if _is_elicit(result):
outcome = await _elicit(result, wire_key, res)
else:
# A resolver may return any type (not just `BaseModel`); `model_construct`
# wraps it as an accepted result without validating against the schema bound.
outcome = cast("AcceptedElicitation[Any]", AcceptedElicitation.model_construct(data=result))
# 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
# persisted in `request_state`; the resolver re-runs next round instead.
outcome = _accepted(result)
cache[key] = outcome
res.cache[cache_key] = outcome
return outcome
async def _elicit(elicit: Elicit[Any], key: str, res: _Resolution) -> ElicitationResult[Any]:
"""Turn a resolver's `Elicit` into an outcome via the negotiated transport."""
if not res.input_required:
return await res.context.elicit(elicit.message, elicit.schema)
# Answered in a prior round (restored without a known schema, e.g. an unannotated
# resolver): re-validate the stored entry against the live `Elicit.schema`. A
# recorded outcome wins over a re-sent answer; an invalid entry self-deletes and
# falls through to the fresh answer (or to re-asking).
outcome = _restore_outcome(res, key, elicit.schema)
if outcome is not None:
return outcome
answer = res.answers.get(key)
if answer is None:
_require_form_elicitation(res.context, key)
res.pending[key] = _elicit_request(elicit)
raise _Pending
if not isinstance(answer, ElicitResult):
raise ToolError(f"Resolver {key!r} received a non-elicitation response")
if answer.action == "accept":
if answer.content is None:
raise ToolError(f"Resolver {key!r} received an accepted elicitation with no content")
try:
data = elicit.schema.model_validate(answer.content)
except ValidationError as e:
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()
def _unwrap(outcome: ElicitationResult[Any], name: str) -> Any:
if isinstance(outcome, AcceptedElicitation):
return outcome.data
raise ToolError(f"Resolver for parameter {name!r} could not resolve: elicitation was {outcome.action}")
def _is_elicit(value: Any) -> TypeGuard[Elicit[Any]]:
"""Runtime narrow of a resolver's return value to a (parameter-erased) `Elicit`."""
return isinstance(value, Elicit)
def _accepted(data: Any) -> AcceptedElicitation[Any]:
"""Wrap a resolved value as an accepted outcome without schema validation.
A resolver may return any type (the schema bound only constrains `Elicit[T]`),
and a value restored from `request_state` is already validated.
"""
return AcceptedElicitation[Any].model_construct(data=data)
def _uses_input_required(protocol_version: str | None) -> bool:
"""True when this request must elicit via `InputRequiredResult` (>= 2026-07-28).
Older revisions still carry a standalone `elicitation/create` server-to-client
request, so the framework keeps the synchronous `ctx.elicit()` path for them.
"""
return protocol_version is not None and is_version_at_least(protocol_version, _INPUT_REQUIRED_VERSION)
def _require_form_elicitation(context: Context[Any, Any], key: str) -> None:
"""Assert the client declared form elicitation before queueing a question for it.
The spec forbids sending an `input_requests` entry the client has not declared a
capability for. A bare `elicitation: {}` declaration (the only shape before modes
existed) counts as form support; an explicit url-only declaration does not.
Raises:
MCPError: With code `MISSING_REQUIRED_CLIENT_CAPABILITY` and a
`requiredCapabilities` payload when form elicitation is not declared.
"""
capabilities = context.client_capabilities
elicitation = capabilities.elicitation if capabilities is not None else None
if elicitation is not None and (elicitation.form is not None or elicitation.url is None):
return
data = MissingRequiredClientCapabilityErrorData(
required_capabilities=ClientCapabilities(elicitation=ElicitationCapability(form=FormElicitationCapability()))
)
raise MCPError(
code=MISSING_REQUIRED_CLIENT_CAPABILITY,
message=f"Client did not declare the form elicitation capability required by resolver {key!r}",
data=data.model_dump(by_alias=True, mode="json", exclude_none=True),
)
def _elicit_request(elicit: Elicit[Any]) -> ElicitRequest:
"""Render an `Elicit[T]` as the embedded `elicitation/create` request for `input_requests`."""
json_schema = render_elicitation_schema(elicit.schema)
return ElicitRequest(params=ElicitRequestFormParams(message=elicit.message, requested_schema=json_schema))
class _StateEntry(BaseModel):
"""One resolver's recorded outcome inside `request_state`."""
action: Literal["accept", "decline", "cancel"]
data: Any = None
class _State(BaseModel):
"""The decoded `request_state`: resolver outcomes from earlier rounds."""
v: int
outcomes: dict[str, _StateEntry] = {}
def _decode_state(request_state: str | None) -> dict[str, _StateEntry]:
"""Decode the per-call resolution progress from `request_state`.
`request_state` is client-trusted (integrity sealing is a follow-up); validate
it through `_State` and treat anything malformed as "no progress yet".
"""
if not request_state:
return {}
try:
state = _State.model_validate_json(request_state)
except ValidationError:
return {}
return state.outcomes if state.v == _STATE_VERSION else {}
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]:
"""Rebuild an `ElicitationResult` from a decoded `request_state` entry.
Raises:
ValidationError: If `schema` is known and the entry's data does not
validate against it.
"""
if entry.action == "decline":
return DeclinedElicitation()
if entry.action == "cancel":
return CancelledElicitation()
data = entry.data
if schema is not None:
data = schema.model_validate(data)
return _accepted(data)
def _restore_outcome(res: _Resolution, key: str, schema: type[BaseModel] | None) -> ElicitationResult[Any] | None:
"""Restore `key`'s recorded outcome from a prior round, or `None` when absent.
`request_state` is client-trusted, so an entry whose data fails validation gets
the `_decode_state` treatment - dropped as if no progress was recorded, so the
question is asked again - rather than surfacing a validation error.
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:
return None
try:
outcome = _outcome_from_state(entry, schema)
except ValidationError:
del res.state[key]
return None
res.persist[key] = entry
return outcome
__all__ = [
"Resolve",
"Elicit",

11
src/mcp/server/mcpserver/tools/base.py
View file

@ -4,7 +4,7 @@ from collections.abc import Callable, Hashable
from functools import cached_property
from typing import TYPE_CHECKING, Any
from mcp_types import Icon, ToolAnnotations
from mcp_types import Icon, InputRequiredResult, ToolAnnotations
from pydantic import BaseModel, Field
from mcp.server.mcpserver.exceptions import ToolError
@ -135,9 +135,12 @@ class Tool(BaseModel):
pre_validated: dict[str, Any] | None = None
if self.resolved_params:
pre_validated = self.fn_metadata.validate_arguments(arguments)
pass_directly |= await resolve_arguments(
self.resolved_params, self.resolver_plans, pre_validated, context
)
resolved = await resolve_arguments(self.resolved_params, self.resolver_plans, pre_validated, context)
if isinstance(resolved, InputRequiredResult):
# A resolver still needs client input (>= 2026-07-28): surface the
# batched questions instead of running the tool body this round.
return self.fn_metadata.convert_result(resolved) if convert_result else resolved
pass_directly |= resolved
result = await self.fn_metadata.call_fn_with_arg_validation(
self.fn,