Enables context injection into resources

Author: jonvet

src/mcp/server/fastmcp/resources/resource_manager.py

@@ -1,14 +1,17 @@
 """Resource manager functionality."""
 
 from collections.abc import Callable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from pydantic import AnyUrl
 
 from mcp.server.fastmcp.resources.base import Resource
 from mcp.server.fastmcp.resources.templates import ResourceTemplate
 from mcp.server.fastmcp.utilities.logging import get_logger
 
+if TYPE_CHECKING:
+    from mcp.server.fastmcp.server import Context
+
 logger = get_logger(__name__)
 
 
@@ -65,7 +68,9 @@ def add_template(
         self._templates[template.uri_template] = template
         return template
 
-    async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
+    async def get_resource(
+        self, uri: AnyUrl | str, context: "Context | None" = None
+    ) -> Resource | None:
         """Get resource by URI, checking concrete resources first, then templates."""
         uri_str = str(uri)
         logger.debug("Getting resource", extra={"uri": uri_str})
@@ -78,7 +83,9 @@ async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
         for template in self._templates.values():
             if params := template.matches(uri_str):
                 try:
-                    return await template.create_resource(uri_str, params)
+                    return await template.create_resource(
+                        uri_str, params, context=context
+                    )
                 except Exception as e:
                     raise ValueError(f"Error creating resource from template: {e}")
 

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

@@ -6,10 +6,15 @@
 import re
 from collections.abc import Callable
 from typing import Any
+from typing import TYPE_CHECKING, Any, Callable
 
-from pydantic import BaseModel, Field, TypeAdapter, validate_call
+from pydantic import BaseModel, Field, validate_call
 
 from mcp.server.fastmcp.resources.types import FunctionResource, Resource
+from mcp.server.fastmcp.utilities.func_metadata import func_metadata
+
+if TYPE_CHECKING:
+    from mcp.server.fastmcp.server import Context
 
 
 class ResourceTemplate(BaseModel):
@@ -27,6 +32,9 @@ class ResourceTemplate(BaseModel):
     parameters: dict[str, Any] = Field(
         description="JSON schema for function parameters"
     )
+    context_kwarg: str | None = Field(
+        None, description="Name of the kwarg that should receive context"
+    )
 
     @classmethod
     def from_function(
@@ -42,8 +50,24 @@ def from_function(
         if func_name == "<lambda>":
             raise ValueError("You must provide a name for lambda functions")
 
-        # Get schema from TypeAdapter - will fail if function isn't properly typed
-        parameters = TypeAdapter(fn).json_schema()
+        # Find context parameter if it exists
+        context_kwarg = None
+        sig = inspect.signature(fn)
+        for param_name, param in sig.parameters.items():
+            if (
+                param.annotation.__name__ == "Context"
+                if hasattr(param.annotation, "__name__")
+                else False
+            ):
+                context_kwarg = param_name
+                break
+
+        # Get schema from func_metadata, excluding context parameter
+        func_arg_metadata = func_metadata(
+            fn,
+            skip_names=[context_kwarg] if context_kwarg is not None else [],
+        )
+        parameters = func_arg_metadata.arg_model.model_json_schema()
 
         # ensure the arguments are properly cast
         fn = validate_call(fn)
@@ -55,6 +79,7 @@ def from_function(
             mime_type=mime_type or "text/plain",
             fn=fn,
             parameters=parameters,
+            context_kwarg=context_kwarg,
         )
 
     def matches(self, uri: str) -> dict[str, Any] | None:
@@ -66,9 +91,15 @@ def matches(self, uri: str) -> dict[str, Any] | None:
             return match.groupdict()
         return None
 
-    async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
+    async def create_resource(
+        self, uri: str, params: dict[str, Any], context: "Context | None" = None
+    ) -> Resource:
         """Create a resource from the template with the given parameters."""
         try:
+            # Add context to params if needed
+            if self.context_kwarg is not None and context is not None:
+                params = {**params, self.context_kwarg: context}
+
             # Call function and check if result is a coroutine
             result = self.fn(**params)
             if inspect.iscoroutine(result):

src/mcp/server/fastmcp/server.py

@@ -230,7 +230,8 @@ async def list_resource_templates(self) -> list[MCPResourceTemplate]:
     async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:
         """Read a resource by URI."""
 
-        resource = await self._resource_manager.get_resource(uri)
+        context = self.get_context()
+        resource = await self._resource_manager.get_resource(uri, context=context)
         if not resource:
             raise ResourceError(f"Unknown resource: {uri}")
 
@@ -327,6 +328,10 @@ def resource(
         If the URI contains parameters (e.g. "resource://{param}") or the function
         has parameters, it will be registered as a template resource.
 
+        Resources can optionally request a Context object by adding a parameter with the
+        Context type annotation. The context provides access to MCP capabilities like
+        logging, progress reporting, and resource access.
+
         Args:
             uri: URI for the resource (e.g. "resource://my-resource" or "resource://{param}")
             name: Optional name for the resource
@@ -351,6 +356,12 @@ def get_weather(city: str) -> str:
             async def get_weather(city: str) -> str:
                 data = await fetch_weather(city)
                 return f"Weather for {city}: {data}"
+
+            @server.resource("resource://{city}/weather")
+            async def get_weather(city: str, ctx: Context) -> str:
+                await ctx.info(f"Getting weather for {city}")
+                data = await fetch_weather(city)
+                return f"Weather for {city}: {data}"
         """
         # Check if user passed function directly instead of calling decorator
         if callable(uri):
@@ -367,7 +378,18 @@ def decorator(fn: AnyFunction) -> AnyFunction:
             if has_uri_params or has_func_params:
                 # Validate that URI params match function params
                 uri_params = set(re.findall(r"{(\w+)}", uri))
-                func_params = set(inspect.signature(fn).parameters.keys())
+
+                # Get all function params except 'ctx' or any parameter of type Context
+                sig = inspect.signature(fn)
+                func_params = set()
+                for param_name, param in sig.parameters.items():
+                    # Skip context parameters
+                    if param_name == "ctx" or (
+                        hasattr(param.annotation, "__name__")
+                        and param.annotation.__name__ == "Context"
+                    ):
+                        continue
+                    func_params.add(param_name)
 
                 if uri_params != func_params:
                     raise ValueError(

tests/server/fastmcp/resources/test_resource_manager.py

@@ -139,3 +139,104 @@ def test_list_resources(self, temp_file: Path):
         resources = manager.list_resources()
         assert len(resources) == 2
         assert resources == [resource1, resource2]
+
+    @pytest.mark.anyio
+    async def test_context_injection_in_template(self):
+        """Test that context is injected when getting a resource from a template."""
+        from mcp.server.fastmcp import Context, FastMCP
+
+        manager = ResourceManager()
+
+        def resource_with_context(name: str, ctx: Context) -> str:
+            assert isinstance(ctx, Context)
+            return f"Hello, {name}!"
+
+        template = ResourceTemplate.from_function(
+            fn=resource_with_context,
+            uri_template="greet://{name}",
+            name="greeter",
+        )
+        manager._templates[template.uri_template] = template
+
+        mcp = FastMCP()
+        ctx = mcp.get_context()
+
+        resource = await manager.get_resource(AnyUrl("greet://world"), context=ctx)
+        assert isinstance(resource, FunctionResource)
+        content = await resource.read()
+        assert content == "Hello, world!"
+
+    @pytest.mark.anyio
+    async def test_context_injection_in_async_template(self):
+        """Test that context is properly injected in async template functions."""
+        from mcp.server.fastmcp import Context, FastMCP
+
+        manager = ResourceManager()
+
+        async def async_resource(name: str, ctx: Context) -> str:
+            assert isinstance(ctx, Context)
+            return f"Async Hello, {name}!"
+
+        template = ResourceTemplate.from_function(
+            fn=async_resource,
+            uri_template="async-greet://{name}",
+            name="async-greeter",
+        )
+        manager._templates[template.uri_template] = template
+
+        mcp = FastMCP()
+        ctx = mcp.get_context()
+
+        resource = await manager.get_resource(
+            AnyUrl("async-greet://world"), context=ctx
+        )
+        assert isinstance(resource, FunctionResource)
+        content = await resource.read()
+        assert content == "Async Hello, world!"
+
+    @pytest.mark.anyio
+    async def test_optional_context_in_template(self):
+        """Test that context is optional when getting a resource from a template."""
+        from mcp.server.fastmcp import Context
+
+        manager = ResourceManager()
+
+        def resource_with_optional_context(
+            name: str, ctx: Context | None = None
+        ) -> str:
+            return f"Hello, {name}!"
+
+        template = ResourceTemplate.from_function(
+            fn=resource_with_optional_context,
+            uri_template="greet://{name}",
+            name="greeter",
+        )
+        manager._templates[template.uri_template] = template
+
+        resource = await manager.get_resource(AnyUrl("greet://world"))
+        assert isinstance(resource, FunctionResource)
+        content = await resource.read()
+        assert content == "Hello, world!"
+
+    @pytest.mark.anyio
+    async def test_context_error_handling_in_template(self):
+        """Test error handling when context injection fails in a template."""
+        from mcp.server.fastmcp import Context, FastMCP
+
+        manager = ResourceManager()
+
+        def failing_resource(name: str, ctx: Context) -> str:
+            raise ValueError("Test error")
+
+        template = ResourceTemplate.from_function(
+            fn=failing_resource,
+            uri_template="greet://{name}",
+            name="greeter",
+        )
+        manager._templates[template.uri_template] = template
+
+        mcp = FastMCP()
+        ctx = mcp.get_context()
+
+        with pytest.raises(ValueError, match="Error creating resource from template"):
+            await manager.get_resource(AnyUrl("greet://world"), context=ctx)