Fix typed Context in resources and prompts

Author: morluto

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 the same typed `ctx` parameter. 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 in tools, resources, and prompts.
 * 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

@@ -1712,7 +1712,7 @@ 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.
+The parametrized `Context[MyLifespanState]` annotation works on `@mcp.tool()`, `@mcp.prompt()`, and templated `@mcp.resource("scheme://{param}")` handlers. The SDK validates client-supplied arguments separately and passes the server-created context directly, preserving its request state and the static type of `ctx.request_context.lifespan_context`.
 
 ### `ServerSession` is now a thin proxy (no longer a `BaseSession`)
 

src/mcp/server/mcpserver/prompts/base.py

@@ -9,9 +9,13 @@
 import anyio.to_thread
 import pydantic_core
 from mcp_types import ContentBlock, Icon, InputRequiredResult, TextContent
-from pydantic import BaseModel, Field, TypeAdapter, validate_call
+from pydantic import BaseModel, Field, TypeAdapter
 
-from mcp.server.mcpserver.utilities.context_injection import find_context_parameter, inject_context
+from mcp.server.mcpserver.utilities.context_injection import (
+    find_context_parameter,
+    inject_context,
+    validate_call_with_injected_context,
+)
 from mcp.server.mcpserver.utilities.func_metadata import func_metadata
 from mcp.shared._callable_inspection import is_async_callable
 from mcp.shared.exceptions import MCPError
@@ -126,7 +130,7 @@ def from_function(
                 )
 
         # ensure the arguments are properly cast
-        fn = validate_call(fn)
+        fn = validate_call_with_injected_context(fn, context_kwarg)
 
         return cls(
             name=func_name,

src/mcp/server/mcpserver/resources/templates.py

@@ -9,11 +9,15 @@
 
 import anyio.to_thread
 from mcp_types import Annotations, Icon, InputRequiredResult
-from pydantic import BaseModel, Field, validate_call
+from pydantic import BaseModel, Field
 
 from mcp.server.mcpserver.exceptions import ResourceError
 from mcp.server.mcpserver.resources.types import FunctionResource, Resource
-from mcp.server.mcpserver.utilities.context_injection import find_context_parameter, inject_context
+from mcp.server.mcpserver.utilities.context_injection import (
+    find_context_parameter,
+    inject_context,
+    validate_call_with_injected_context,
+)
 from mcp.server.mcpserver.utilities.func_metadata import func_metadata
 from mcp.server.mcpserver.utilities.logging import get_logger
 from mcp.shared._callable_inspection import is_async_callable
@@ -160,7 +164,7 @@ def from_function(
         parameters = func_arg_metadata.arg_model.model_json_schema()
 
         # ensure the arguments are properly cast
-        fn = validate_call(fn)
+        fn = validate_call_with_injected_context(fn, context_kwarg)
 
         return cls(
             uri_template=uri_template,

src/mcp/server/mcpserver/utilities/context_injection.py

@@ -2,12 +2,16 @@
 
 from __future__ import annotations
 
+import functools
 import inspect
 import typing
 from collections.abc import Callable
 from typing import Any
 
+from pydantic import SkipValidation, validate_call
+
 from mcp.server.mcpserver.context import Context
+from mcp.shared._callable_inspection import is_async_callable
 
 
 def find_context_parameter(fn: Callable[..., Any]) -> str | None:
@@ -66,3 +70,59 @@ def inject_context(
     if context_kwarg is not None and context is not None:
         return {**kwargs, context_kwarg: context}
     return kwargs
+
+
+def validate_call_with_injected_context(
+    fn: Callable[..., Any],
+    context_kwarg: str | None,
+) -> Callable[..., Any]:
+    """Validate a handler call without re-validating its injected context."""
+    if context_kwarg is None:
+        return validate_call(fn)
+
+    signature = inspect.signature(fn, eval_str=True)
+    context_parameter = signature.parameters.get(context_kwarg)
+    if context_parameter is None:
+        return validate_call(fn)
+
+    context_annotation = context_parameter.annotation
+    if context_annotation is inspect.Parameter.empty:
+        context_annotation = Any
+    skipped_context_annotation = SkipValidation[context_annotation]
+    validation_signature = signature.replace(
+        parameters=[
+            parameter.replace(annotation=skipped_context_annotation) if parameter.name == context_kwarg else parameter
+            for parameter in signature.parameters.values()
+        ]
+    )
+    validation_annotations = {
+        parameter.name: parameter.annotation
+        for parameter in validation_signature.parameters.values()
+        if parameter.annotation is not inspect.Parameter.empty
+    }
+    if validation_signature.return_annotation is not inspect.Signature.empty:
+        validation_annotations["return"] = validation_signature.return_annotation
+
+    if is_async_callable(fn):
+
+        @functools.wraps(fn)
+        async def async_forwarding(*args: Any, **kwargs: Any) -> Any:
+            return await fn(*args, **kwargs)
+
+        forwarding: Callable[..., Any] = async_forwarding
+    else:
+
+        @functools.wraps(fn)
+        def sync_forwarding(*args: Any, **kwargs: Any) -> Any:
+            return fn(*args, **kwargs)
+
+        forwarding = sync_forwarding
+
+    # Pydantic builds its validator from these temporary annotations. Restore the
+    # handler's public signature after validation is compiled.
+    setattr(forwarding, "__signature__", validation_signature)
+    forwarding.__annotations__ = validation_annotations
+    validated = validate_call(forwarding)
+    setattr(validated, "__signature__", signature)
+    validated.__annotations__ = fn.__annotations__
+    return validated

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_resources_and_prompts() -> None:
+    """`Context[AppContext]` preserves typed lifespan state in resources and prompts."""
     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/resources/test_resource_template.py

@@ -221,6 +221,21 @@ def my_func(key: str, value: int) -> dict[str, Any]:
         data = json.loads(content)
         assert data == {"key": "foo", "value": 123}
 
+    @pytest.mark.anyio
+    async def test_create_resource_preserves_context_with_variadic_kwargs(self):
+        """Injected context bypasses validation without changing **kwargs handling."""
+        context = Context()
+
+        def my_func(ctx: Context[None], **kwargs: str) -> str:
+            assert ctx is context
+            return kwargs["key"]
+
+        template = ResourceTemplate.from_function(fn=my_func, uri_template="test://{key}")
+        resource = await template.create_resource("test://value", {"key": "value"}, context)
+
+        assert isinstance(resource, FunctionResource)
+        assert await resource.read() == "value"
+
     @pytest.mark.anyio
     async def test_template_error(self):
         """Test error handling in template resource creation."""

tests/server/mcpserver/test_server.py

@@ -1197,9 +1197,9 @@ async def test_resource_with_context(self):
         mcp = MCPServer()
 
         @mcp.resource("resource://context/{name}")
-        def resource_with_context(name: str, ctx: Context) -> str:
+        async def resource_with_context(name: str, ctx: Context[None]) -> str:
             """Resource that receives context."""
-            assert ctx is not None
+            assert ctx.request_context is not None
             return f"Resource {name} - context injected"
 
         # Verify template has context_kwarg set
@@ -1280,9 +1280,9 @@ async def test_prompt_with_context(self):
         mcp = MCPServer()
 
         @mcp.prompt("prompt_with_ctx")
-        def prompt_with_context(text: str, ctx: Context) -> str:
+        def prompt_with_context(text: str, ctx: Context[None]) -> str:
             """Prompt that expects context."""
-            assert ctx is not None
+            assert ctx.request_context is not None
             return f"Prompt '{text}' - context injected"
 
         # Test via client