resolve

Resolver dependency injection for MCPServer tools.

A tool parameter annotated Annotated[T, Resolve(fn)] is filled by running the resolver fn before the tool body, instead of from the LLM-supplied arguments. Resolvers form a DAG: a resolver may declare its own Resolve(...) dependencies, 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). Resolver bodies may re-run on every round; a recorded outcome is consulted only when the body asks its question again, so a resolver's own computation always wins over anything the client echoes back in request_state.

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.

AcceptedElicitation

Bases: BaseModel, Generic[ElicitSchemaModelT]

Result when user accepts the elicitation.

Source code in src/mcp/server/elicitation.py

class AcceptedElicitation(BaseModel, Generic[ElicitSchemaModelT]):
    """Result when user accepts the elicitation."""

    action: Literal["accept"] = "accept"
    data: ElicitSchemaModelT

CancelledElicitation

Bases: BaseModel

Result when user cancels the elicitation.

Source code in src/mcp/server/elicitation.py

class CancelledElicitation(BaseModel):
    """Result when user cancels the elicitation."""

    action: Literal["cancel"] = "cancel"

DeclinedElicitation

Bases: BaseModel

Result when user declines the elicitation.

Source code in src/mcp/server/elicitation.py

class DeclinedElicitation(BaseModel):
    """Result when user declines the elicitation."""

    action: Literal["decline"] = "decline"

Resolve

Marker for Annotated[T, Resolve(fn)]: fill the parameter by running fn.

Source code in src/mcp/server/mcpserver/resolve.py

class Resolve:
    """Marker for `Annotated[T, Resolve(fn)]`: fill the parameter by running `fn`."""

    def __init__(self, fn: Callable[..., Any]) -> None:
        self.fn = fn

Elicit

Bases: Generic[T]

A resolver's request to ask the client.

Returned from a resolver to signal that the value must be elicited. The framework runs ctx.elicit(message, schema) and injects the outcome.

Source code in src/mcp/server/mcpserver/resolve.py

class Elicit(Generic[T]):
    """A resolver's request to ask the client.

    Returned from a resolver to signal that the value must be elicited. The
    framework runs `ctx.elicit(message, schema)` and injects the outcome.
    """

    def __init__(self, message: str, schema: type[T]) -> None:
        self.message = message
        self.schema = schema

find_resolved_parameters

find_resolved_parameters(
    fn: Callable[..., Any],
) -> dict[str, tuple[Resolve, bool]]

Find parameters of fn annotated Annotated[_, Resolve(...)].

Returns a mapping of parameter name to (Resolve, wants_union), where wants_union is True when the annotated type is an ElicitationResult member (the consumer wants the full outcome rather than the unwrapped model).

Source code in src/mcp/server/mcpserver/resolve.py

def find_resolved_parameters(fn: Callable[..., Any]) -> dict[str, tuple[Resolve, bool]]:
    """Find parameters of `fn` annotated `Annotated[_, Resolve(...)]`.

    Returns a mapping of parameter name to `(Resolve, wants_union)`, where
    `wants_union` is True when the annotated type is an `ElicitationResult` member
    (the consumer wants the full outcome rather than the unwrapped model).
    """
    hints = _type_hints(fn)
    resolved: dict[str, tuple[Resolve, bool]] = {}
    for name in inspect.signature(fn).parameters:
        annotation = hints.get(name)
        if get_origin(annotation) is not Annotated:
            # A `Resolve` marker is only honored at the top level; flag (rather than
            # silently drop) one buried in a union, e.g. `Annotated[T, Resolve(f)] | None`.
            if _contains_resolve(annotation):
                raise InvalidSignature(
                    f"Parameter {name!r} of {_resolver_name(fn)!r} wraps `Resolve(...)` in a "
                    "union; annotate the parameter directly as `Annotated[T, Resolve(...)]`"
                )
            continue
        type_arg, *metadata = get_args(annotation)
        marker = next((m for m in metadata if isinstance(m, Resolve)), None)
        if marker is not None:
            resolved[name] = (marker, _wants_union(type_arg))
    return resolved

returns_input_required

returns_input_required(fn: Callable[..., Any]) -> bool

True when fn's return annotation carries an InputRequiredResult arm.

Used at tool registration to reject combining Resolve(...) parameters with a hand-rolled InputRequiredResult flow: a call has a single input_responses/request_state channel, so the two flows would overwrite each other's state and the call could never converge.

Source code in src/mcp/server/mcpserver/resolve.py

def returns_input_required(fn: Callable[..., Any]) -> bool:
    """True when `fn`'s return annotation carries an `InputRequiredResult` arm.

    Used at tool registration to reject combining `Resolve(...)` parameters with a
    hand-rolled `InputRequiredResult` flow: a call has a single
    `input_responses`/`request_state` channel, so the two flows would overwrite
    each other's state and the call could never converge.
    """
    return _has_input_required_arm(_type_hints(fn).get("return"))

build_resolver_plans

build_resolver_plans(
    resolved_params: Mapping[str, tuple[Resolve, bool]],
    tool_arg_names: set[str],
) -> dict[Hashable, _ResolverPlan]

Statically analyze the resolver DAG rooted at a tool's resolved parameters.

Raises:

Type	Description
InvalidSignature	If a resolver has a cyclic dependency, or a resolver parameter cannot be classified (not a Context, a nested Resolve, or a tool argument by name).

Source code in src/mcp/server/mcpserver/resolve.py

def build_resolver_plans(
    resolved_params: Mapping[str, tuple[Resolve, bool]],
    tool_arg_names: set[str],
) -> dict[Hashable, _ResolverPlan]:
    """Statically analyze the resolver DAG rooted at a tool's resolved parameters.

    Raises:
        InvalidSignature: If a resolver has a cyclic dependency, or a resolver
            parameter cannot be classified (not a `Context`, a nested `Resolve`,
            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)
        if key in stack:
            raise InvalidSignature(f"Resolver {_resolver_name(fn)!r} has a cyclic dependency")
        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] = {}
        nested: list[Callable[..., Any]] = []
        for param_name in sig.parameters:
            annotation = hints.get(param_name)
            if annotation is not None and _is_context_annotation(annotation):
                params[param_name] = _ParamPlan("context")
                continue
            marker, wants_union = _resolve_marker(annotation)
            if marker is not None:
                params[param_name] = _ParamPlan("resolve", marker, wants_union)
                nested.append(marker.fn)
                continue
            if param_name in tool_arg_names:
                params[param_name] = _ParamPlan("by_name")
                continue
            raise InvalidSignature(
                f"Resolver {_resolver_name(fn)!r} parameter {param_name!r} cannot be resolved: "
                "expected a Context, an Annotated[_, Resolve(...)], or a tool argument by name"
            )

        _check_elicit_return(hints.get("return"), _resolver_name(fn))
        plans[key] = _ResolverPlan(fn, params, is_async_callable(fn), wire_key)
        for dep in nested:
            analyze(dep, stack + (key,))

    for marker, _ in resolved_params.values():
        analyze(marker.fn, ())
    return plans

resolve_arguments async

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] | InputRequiredResult

Resolve every Resolve-marked tool parameter into a concrete value.

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.

Each question is asked once - its answer is carried in request_state across rounds and satisfies the question when the resolver asks it again. Resolver bodies themselves may re-run on each round; a recorded answer is consulted only when the body asks, never in place of running it.

Raises:

Type	Description
ToolError	If an elicited value is declined or cancelled and the consumer asked for the unwrapped model (rather than the result union).

Source code in src/mcp/server/mcpserver/resolve.py

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] | InputRequiredResult:
    """Resolve every `Resolve`-marked tool parameter into a concrete value.

    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.

    Each question is asked once - its answer is carried in `request_state` across
    rounds and satisfies the question when the resolver asks it again. Resolver
    bodies themselves may re-run on each round; a recorded answer is consulted
    only when the body asks, never in place of running it.

    Raises:
        ToolError: If an elicited value is declined or cancelled and the consumer
            asked for the unwrapped model (rather than the result union).
    """
    # `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:
            outcome = await _resolve(marker.fn, res)
        except _Pending:
            continue
        injected[name] = outcome if wants_union else _unwrap(outcome, name)

    if res.pending:
        asked = {key: _request_digest(request) for key, request in res.pending.items()}
        return InputRequiredResult(input_requests=res.pending, request_state=_encode_state(res.persist, asked))
    return injected