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.

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).

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

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] = {}

    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

        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"
            )

        plans[key] = _ResolverPlan(fn, params, is_async_callable(fn))
        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]

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.

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]:
    """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.

    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]] = {}
    injected: dict[str, Any] = {}
    for name, (marker, wants_union) in resolved_params.items():
        outcome = await _resolve(marker.fn, plans, tool_args, context, cache)
        injected[name] = outcome if wants_union else _unwrap(outcome, name)
    return injected