Keep the live Context when a prompt or template annotates Context[T] (#3624)

Author: maxisbey

docs/handlers/lifespan.md

@@ -41,7 +41,7 @@ Nothing new. `ctx` is a **Context** parameter, so the SDK injects it and it neve
 
 `genre` is the only argument the model can pass. The lifespan is your server's business.
 
-`@mcp.resource()` and `@mcp.prompt()` functions can take a `ctx` parameter too, written as a bare `Context` for a reason the next section gets to. Everything `ctx` carries is in **[The Context](context.md)**.
+`@mcp.resource()` and `@mcp.prompt()` functions can take a `ctx` parameter too. Everything `ctx` carries is in **[The Context](context.md)**.
 
 ### It really is typed
 
@@ -51,19 +51,6 @@ That one type parameter is why `ctx.request_context.lifespan_context` **is** an
 
 Write a bare `Context` instead and `lifespan_context` is typed as `dict[str, Any]`: the type checker has no way to know what your lifespan yielded. The object is still there at runtime; you've lost the help.
 
-!!! warning
-    `Context[AppContext]` is a **tool-only** spelling. Put it on an `@mcp.resource()` or
-    `@mcp.prompt()` function and every call to that handler fails. The client gets an error back,
-    and the server log shows why:
-
-    ```text
-    Context is not available outside of a request
-    ```
-
-    In resources and prompts, write the bare `ctx: Context`. The object your lifespan yielded is
-    still `ctx.request_context.lifespan_context` at runtime; you give up the type parameter, not
-    the object.
-
 !!! tip
     There is always a lifespan. If you don't pass one, the SDK's default yields an empty `dict`,
     so `ctx.request_context.lifespan_context` is `{}`, never `None`. That default is also why a
@@ -96,7 +83,7 @@ Strip the server down to the lifecycle: give `Database` a `connected` flag, flip
 * Code before the `yield` is startup. The `finally` after it is shutdown.
 * It runs once, around the whole life of the server, not per request.
 * Whatever you `yield` is `ctx.request_context.lifespan_context` in every tool, resource, and prompt.
-* `ctx: Context[AppContext]` makes that access fully typed in tools. Resources and prompts take the bare `Context`.
+* `ctx: Context[AppContext]` makes that access fully typed.
 * No `lifespan=` means an empty `dict`, never `None`.
 
 A handler that stops mid-call to ask the user for something only they know is **[Elicitation](elicitation.md)**.

docs/migration.md

@@ -1714,8 +1714,6 @@ async def my_tool(ctx: Context) -> str: ...
 async def my_tool(ctx: Context[MyLifespanState]) -> str: ...
 ```
 
-The parametrized `Context[MyLifespanState]` annotation currently works only on `@mcp.tool()` handlers. On `@mcp.prompt()` and templated `@mcp.resource("scheme://{param}")` handlers, annotate the parameter as bare `Context` for now: these handlers are wrapped in `pydantic.validate_call`, which re-validates the injected `Context` into a fresh `Context[MyLifespanState]` detached from the request, so the first access to `ctx.request_id`, `ctx.session`, or `ctx.request_context` raises `ValueError: Context is not available outside of a request` (the client sees an internal server error, or `Error creating resource from template ...`). Bare `Context` still exposes `ctx.request_context.lifespan_context`; only its static type is lost.
-
 ### `ServerSession` is now a thin proxy (no longer a `BaseSession`)
 
 `ServerSession` no longer subclasses `BaseSession`. It is now a small per-request proxy that exposes `send_request`, `send_notification`, the typed convenience helpers — `create_message`, `elicit` / `elicit_form` / `elicit_url`, `send_elicit_complete`, `list_roots`, `send_log_message`, `send_resource_updated`, `send_resource_list_changed` / `send_tool_list_changed` / `send_prompt_list_changed`, `send_ping`, `send_progress_notification`, and the new `report_progress` — plus `check_client_capability` and the read-only `client_params`, `client_capabilities`, `protocol_version`, and `can_send_request` properties. The receive loop, `initialize` handling, and per-request task isolation that previously lived in `ServerSession` have moved to `JSONRPCDispatcher` and `ServerRunner`.

src/mcp/server/mcpserver/context.py

@@ -4,8 +4,8 @@
 from typing import TYPE_CHECKING, Any, Generic, cast
 
 from mcp_types import ClientCapabilities, InputRequiredResult, InputResponseRequestParams, InputResponses, LoggingLevel
-from pydantic import AnyUrl, BaseModel
-from typing_extensions import deprecated
+from pydantic import AnyUrl, BaseModel, ModelWrapValidatorHandler, model_validator
+from typing_extensions import Self, deprecated
 
 from mcp.server.context import LifespanContextT, RequestT, ServerRequestContext
 from mcp.server.elicitation import (
@@ -84,6 +84,13 @@ def __init__(
         self._input_params = input_params
         self._subscriptions = subscriptions
 
+    @model_validator(mode="wrap")
+    @classmethod
+    def _keep_instance(cls, value: Any, handler: ModelWrapValidatorHandler[Self]) -> Self:
+        """Validate an existing `Context` to itself. `Context[T]` is a separate class at runtime, so
+        pydantic would otherwise rebuild an instance of plain `Context` without its request state."""
+        return cast(Self, value) if isinstance(value, Context) else handler(value)
+
     @property
     def mcp_server(self) -> MCPServer:
         """Access to the MCPServer instance."""

tests/docs_src/test_lifespan.py

@@ -5,7 +5,7 @@
 from mcp_types import TextContent, TextResourceContents
 
 from docs_src.lifespan import tutorial001, tutorial002
-from mcp import Client, MCPError
+from mcp import Client
 from mcp.server import MCPServer
 from mcp.server.mcpserver import Context
 
@@ -74,8 +74,8 @@ def stock_report(ctx: Context) -> str:
         assert message.content == TextContent(type="text", text="Summarise a shelf of 3 books.")
 
 
-async def test_parameterized_context_is_tool_only(caplog: pytest.LogCaptureFixture) -> None:
-    """`Context[AppContext]` on a resource or prompt fails every call; the server logs the `ValueError`."""
+async def test_parameterized_context_reaches_the_lifespan_object_in_resources_and_prompts() -> None:
+    """`ctx: Context[AppContext]` gives a resource or prompt the same typed lifespan object a tool gets."""
     mcp = MCPServer("Bookshop", lifespan=tutorial001.app_lifespan)
 
     @mcp.resource("books://{genre}/count")
@@ -89,14 +89,13 @@ def stock_report(ctx: Context[tutorial001.AppContext]) -> str:
         return f"Summarise a shelf of {ctx.request_context.lifespan_context.db.query()} books."
 
     async with Client(mcp) as client:
-        with pytest.raises(MCPError, match="Error creating resource from template"):
-            await client.read_resource("books://poetry/count")
-        assert "ValueError: Context is not available outside of a request" in caplog.text
-
-        caplog.clear()
-        with pytest.raises(MCPError):
-            await client.get_prompt("stock_report")
-        assert "ValueError: Context is not available outside of a request" in caplog.text
+        resource = await client.read_resource("books://poetry/count")
+        assert resource.contents == [
+            TextResourceContents(uri="books://poetry/count", mime_type="text/plain", text="3 books in 'poetry'.")
+        ]
+        prompt = await client.get_prompt("stock_report")
+        (message,) = prompt.messages
+        assert message.content == TextContent(type="text", text="Summarise a shelf of 3 books.")
 
 
 async def test_default_lifespan_yields_an_empty_dict() -> None:

tests/server/mcpserver/test_server.py

@@ -1,5 +1,8 @@
 import base64
 import logging
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Annotated, Any
@@ -1337,6 +1340,45 @@ def prompt_no_context(text: str) -> str:
             assert content.text == "Prompt 'test' works"
 
 
+async def test_parameterized_context_carries_the_request_into_templates_and_prompts():
+    """`ctx: Context[AppState]` on a resource template or a sync or async prompt is the request's own
+    context, as it is on a tool: lifespan state and the negotiated protocol version are readable."""
+
+    @dataclass
+    class AppState:
+        greeting: str
+
+    @asynccontextmanager
+    async def lifespan(_: MCPServer[AppState]) -> AsyncIterator[AppState]:
+        yield AppState(greeting="Hello")
+
+    mcp = MCPServer(lifespan=lifespan)
+
+    @mcp.resource("greeting://{name}")
+    def greeting(name: str, ctx: Context[AppState]) -> str:
+        return f"{ctx.request_context.lifespan_context.greeting}, {name} ({ctx.protocol_version})"
+
+    @mcp.prompt()
+    def greet_sync(name: str, ctx: Context[AppState]) -> str:
+        return f"{ctx.request_context.lifespan_context.greeting}, {name} ({ctx.protocol_version})"
+
+    @mcp.prompt()
+    async def greet_async(name: str, ctx: Context[AppState]) -> str:
+        return f"{ctx.request_context.lifespan_context.greeting}, {name} ({ctx.protocol_version})"
+
+    async with Client(mcp, mode="2026-07-28") as client:
+        resource = await client.read_resource("greeting://Alice")
+        sync_prompt = await client.get_prompt("greet_sync", {"name": "Alice"})
+        async_prompt = await client.get_prompt("greet_async", {"name": "Alice"})
+
+    assert resource.contents == [
+        TextResourceContents(uri="greeting://Alice", mime_type="text/plain", text="Hello, Alice (2026-07-28)")
+    ]
+    expected = [PromptMessage(role="user", content=TextContent(type="text", text="Hello, Alice (2026-07-28)"))]
+    assert sync_prompt.messages == expected
+    assert async_prompt.messages == expected
+
+
 class TestServerPrompts:
     """Test prompt functionality in MCPServer server."""
 
@@ -2151,6 +2193,35 @@ async def briefing(ctx: Context) -> list[UserMessage] | InputRequiredResult:
     assert block.text == "Brief Alice (state=r1)"
 
 
+async def test_prompt_with_parameterized_context_reads_input_responses_on_retry():
+    """A prompt annotated `ctx: Context[T]` sees the retry's input_responses and request_state, so the
+    multi-round-trip flow completes instead of asking the same question again."""
+    mcp = MCPServer()
+
+    @mcp.prompt()
+    async def briefing(ctx: Context[dict[str, Any]]) -> list[UserMessage] | InputRequiredResult:
+        responses = ctx.input_responses
+        if responses and "who" in responses:
+            who = responses["who"]
+            assert isinstance(who, ElicitResult) and who.content is not None
+            return [UserMessage(content=f"Brief {who.content['name']} (state={ctx.request_state})")]
+        return InputRequiredResult(input_requests={"who": _ask_who()}, request_state="r1")
+
+    with anyio.fail_after(5):
+        async with Client(mcp, mode="2026-07-28") as client:
+            r1 = await client.session.get_prompt("briefing", allow_input_required=True)
+            assert isinstance(r1, InputRequiredResult)
+
+            r2 = await client.session.get_prompt(
+                "briefing",
+                input_responses={"who": ElicitResult(action="accept", content={"name": "Alice"})},
+                request_state=r1.request_state,
+                allow_input_required=True,
+            )
+    assert isinstance(r2, GetPromptResult)
+    assert r2.messages == [PromptMessage(role="user", content=TextContent(type="text", text="Brief Alice (state=r1)"))]
+
+
 async def test_prompt_input_required_result_on_legacy_session_is_a_serialization_error():
     """Pins the shared era gate: a pre-2026 session has no input_required vocabulary, so
     the runner rejects the frame with -32603 — the same posture the tools path has."""
@@ -3114,6 +3185,11 @@ def test_context_mcp_server_outside_request_raises() -> None:
         _ = Context().mcp_server
 
 
+def test_context_request_context_outside_request_raises() -> None:
+    with pytest.raises(ValueError, match="outside of a request"):
+        _ = Context().request_context
+
+
 async def test_context_notify_outside_a_request_raises() -> None:
     with pytest.raises(ValueError, match="outside of a request"):
         await Context().notify_tools_changed()