Add resolver dependency injection for MCPServer tools

docs/migration.md

@@ -1473,6 +1473,65 @@ app = server.streamable_http_app(
 
 The lowlevel `Server` also now exposes a `session_manager` property to access the `StreamableHTTPSessionManager` after calling `streamable_http_app()`.
 
+### Resolver dependency injection for tools (`Resolve` / `Elicit`)
+
+A tool parameter annotated `Annotated[T, Resolve(fn)]` is filled by running the resolver `fn` before the tool body, instead of by the calling LLM. Resolvers form a dependency graph: a resolver may declare its own `Resolve(...)` dependencies, read the `Context` (including `ctx.headers`), and receive the tool's own arguments by name. A resolver may return `Elicit[T]` to ask the client; the SDK runs the elicitation and injects the answer. A resolver only elicits when it needs to - it can also resolve a value directly and skip the question. Each resolver runs at most once per `tools/call`.
+
+The injected type follows the consumer's annotation. Annotating the unwrapped model (`Annotated[Confirm, Resolve(confirm)]`) injects the model on accept and aborts the call with an error result on decline or cancel. To branch on the outcome instead - so the tool can react to decline and cancel - annotate `ElicitationResult[Confirm]` (or an explicit `AcceptedElicitation[Confirm] | DeclinedElicitation | CancelledElicitation` union):
+
+```python
+from typing import Annotated
+
+from pydantic import BaseModel
+
+from mcp.server.mcpserver import (
+    AcceptedElicitation,
+    CancelledElicitation,
+    DeclinedElicitation,
+    Elicit,
+    ElicitationResult,
+    MCPServer,
+    Resolve,
+)
+
+mcp = MCPServer(name="files")
+
+
+class Confirm(BaseModel):
+    ok: bool
+
+
+async def confirm_delete(path: str) -> Confirm | Elicit[Confirm]:
+    file_count = len(list_files(path))
+    if file_count == 0:
+        return Confirm(ok=True)  # empty folder: nothing to confirm, no question
+    return Elicit(f"{path} has {file_count} file(s). Delete anyway?", Confirm)
+
+
+@mcp.tool()
+async def delete_folder(
+    path: str,
+    confirm: Annotated[ElicitationResult[Confirm], Resolve(confirm_delete)],
+) -> str:
+    """Delete a folder, asking for confirmation when it is not empty."""
+    match confirm:
+        case AcceptedElicitation(data=Confirm(ok=True)):
+            delete(path)
+            return f"deleted {path}"
+        case AcceptedElicitation():
+            return "kept the folder"
+        case DeclinedElicitation():
+            return "declined: folder not deleted"
+        case CancelledElicitation():
+            return "cancelled: folder not deleted"
+```
+
+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.
+
+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.
+
+`ElicitationResult` is now a subscriptable generic alias (so `ElicitationResult[T]` works in annotations) instead of a plain union. A runtime `isinstance(result, ElicitationResult)` therefore raises `TypeError`; check against the member classes directly - `isinstance(result, AcceptedElicitation)` (or `DeclinedElicitation` / `CancelledElicitation`).
+
 ## Need Help?
 
 If you encounter issues during migration:

docs/tutorial/context.md

@@ -63,6 +63,7 @@ The injected object is small. Besides `request_id`:
 * `await ctx.report_progress(progress, total, message)`: stream progress back to the caller during a long call. The whole story is in **Progress**.
 * `await ctx.elicit(message, schema)` and `await ctx.elicit_url(...)`: pause the tool and ask the user a question. That's **Elicitation**.
 * `ctx.session`: the server's side of the conversation with this client. Notifications you send to the client live here; the last section uses it.
+* `ctx.headers`: the request headers the transport carried, or `None` on stdio. Read a custom header with `(ctx.headers or {}).get("x-...")`.
 * `ctx.request_context`: the raw per-request record. The field you'll reach for is `lifespan_context`, the object your startup code yielded (see **Lifespan**).
 
 Logging is deliberately not on that list. A server logs with Python's `logging` module, like any other Python program. **Logging** is the short chapter on why.

docs/tutorial/elicitation.md

@@ -79,6 +79,22 @@ A refusal is not an error. The tool decides what declining means (here, no booki
     `"maybe"` for a `bool` doesn't corrupt your booking: the call fails with the
     `ValidationError`, your `if` never runs.
 
+## Ask before the tool runs
+
+The booking tool above weaves the question into its own body. When the question is really a *precondition* - confirm before deleting, authenticate before acting - you can lift it out of the tool into a **resolver** and let the framework ask for you.
+
+A parameter annotated `Annotated[T, Resolve(fn)]` is filled by running `fn` before the tool body. The resolver returns the value directly when it already knows it, or returns `Elicit(...)` to have the framework ask:
+
+```python title="server.py" hl_lines="24-30 35-36"
+--8<-- "docs_src/elicitation/tutorial004.py"
+```
+
+* `confirm_delete` reads the tool's own `path` argument by name, lists the folder, and **only elicits when it must** - an empty folder resolves to `Confirm(ok=True)` with no round-trip to the client.
+* `delete_folder` annotates `ElicitationResult[Confirm]`, so the framework injects the whole outcome and the tool `match`es every case: accept-and-confirm, accept-but-keep (`ok=False`), decline, cancel.
+* The `confirm` parameter never appears in the tool's input schema - the client supplies `path`, the resolver supplies `confirm`.
+
+Annotate the unwrapped model (`Annotated[Confirm, Resolve(confirm_delete)]`) instead when the tool doesn't need to branch: it receives the model on accept and the call aborts with an error on decline or cancel.
+
 ## Send the user to a URL
 
 Some things must not go through the model or the client: credentials, card numbers, OAuth consent. For those you don't ask for data; you ask the user to go somewhere:

docs_src/elicitation/tutorial004.py

@@ -0,0 +1,47 @@
+from typing import Annotated
+
+from pydantic import BaseModel
+
+from mcp.server import MCPServer
+from mcp.server.mcpserver import (
+    AcceptedElicitation,
+    CancelledElicitation,
+    DeclinedElicitation,
+    Elicit,
+    ElicitationResult,
+    Resolve,
+)
+
+mcp = MCPServer("Files")
+
+_FOLDERS: dict[str, list[str]] = {"/tmp/empty": [], "/tmp/project": ["main.py", "README.md"]}
+
+
+class Confirm(BaseModel):
+    ok: bool
+
+
+async def confirm_delete(path: str) -> Confirm | Elicit[Confirm]:
+    """Resolver: ask for confirmation only when the folder is not empty."""
+    file_count = len(_FOLDERS.get(path, []))
+    if file_count == 0:
+        return Confirm(ok=True)  # nothing to confirm, no round-trip to the client
+    return Elicit(f"{path} has {file_count} file(s). Delete anyway?", Confirm)
+
+
+@mcp.tool()
+async def delete_folder(
+    path: str,
+    confirm: Annotated[ElicitationResult[Confirm], Resolve(confirm_delete)],
+) -> str:
+    """Delete a folder, asking for confirmation when it is not empty."""
+    match confirm:
+        case AcceptedElicitation(data=Confirm(ok=True)):
+            _FOLDERS.pop(path, None)
+            return f"deleted {path}"
+        case AcceptedElicitation():
+            return "kept the folder"
+        case DeclinedElicitation():
+            return "declined: folder not deleted"
+        case CancelledElicitation():
+            return "cancelled: folder not deleted"

src/mcp/server/elicitation.py

@@ -11,6 +11,7 @@
 from pydantic import BaseModel, ValidationError
 from pydantic.json_schema import GenerateJsonSchema, JsonSchemaValue
 from pydantic_core import core_schema
+from typing_extensions import TypeAliasType
 
 from mcp.server.session import ServerSession
 
@@ -36,7 +37,11 @@ class CancelledElicitation(BaseModel):
     action: Literal["cancel"] = "cancel"
 
 
-ElicitationResult = AcceptedElicitation[ElicitSchemaModelT] | DeclinedElicitation | CancelledElicitation
+ElicitationResult = TypeAliasType(
+    "ElicitationResult",
+    AcceptedElicitation[ElicitSchemaModelT] | DeclinedElicitation | CancelledElicitation,
+    type_params=(ElicitSchemaModelT,),
+)
 
 
 class AcceptedUrlElicitation(BaseModel):

src/mcp/server/mcpserver/__init__.py

@@ -3,7 +3,27 @@
 from mcp_types import Icon
 
 from .context import Context
+from .resolve import (
+    AcceptedElicitation,
+    CancelledElicitation,
+    DeclinedElicitation,
+    Elicit,
+    ElicitationResult,
+    Resolve,
+)
 from .server import MCPServer
 from .utilities.types import Audio, Image
 
-__all__ = ["MCPServer", "Context", "Image", "Audio", "Icon"]
+__all__ = [
+    "MCPServer",
+    "Context",
+    "Image",
+    "Audio",
+    "Icon",
+    "Resolve",
+    "Elicit",
+    "ElicitationResult",
+    "AcceptedElicitation",
+    "DeclinedElicitation",
+    "CancelledElicitation",
+]

src/mcp/server/mcpserver/context.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-from collections.abc import Iterable
-from typing import TYPE_CHECKING, Any, Generic
+from collections.abc import Iterable, Mapping
+from typing import TYPE_CHECKING, Any, Generic, Protocol, cast
 
 from mcp_types import ClientCapabilities, InputResponseRequestParams, InputResponses, LoggingLevel
 from pydantic import AnyUrl, BaseModel
@@ -22,6 +22,11 @@
     from mcp.server.mcpserver.server import MCPServer
 
 
+class _HasHeaders(Protocol):
+    @property
+    def headers(self) -> Mapping[str, str]: ...
+
+
 class Context(BaseModel, Generic[LifespanContextT, RequestT]):
     """Context object providing access to MCP capabilities.
 
@@ -217,6 +222,17 @@ def client_id(self) -> str | None:
         """
         return self.request_context.meta.get("client_id") if self.request_context.meta else None  # pragma: no cover
 
+    @property
+    def headers(self) -> Mapping[str, str] | None:
+        """Request headers carried by this message, when the transport has them.
+
+        Populated by HTTP-based transports; `None` on stdio.
+        """
+        request = self.request_context.request
+        if request is None:
+            return None
+        return cast("_HasHeaders", request).headers
+
     @property
     def request_id(self) -> str:
         """Get the unique ID for this request."""