Add support for context-only resources (#1405)

Author: 0Delta

examples/snippets/servers/context_resource.py

@@ -0,0 +1,11 @@
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.server.session import ServerSession
+
+mcp = FastMCP(name="Context Resource Example")
+
+
+@mcp.resource("resource://only_context")
+def resource_only_context(ctx: Context[ServerSession, None]) -> str:
+    """Resource that only receives context."""
+    assert ctx is not None
+    return "Resource with only context injected"

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

@@ -99,12 +99,12 @@ def from_function(
 
         # Find context parameter if it exists
         if context_kwarg is None:  # pragma: no branch
-            context_kwarg = find_context_parameter(fn)
+            context_kwarg = find_context_parameter(fn) or ""
 
         # 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 [],
+            skip_names=[context_kwarg] if context_kwarg else [],
         )
         parameters = func_arg_metadata.arg_model.model_json_schema()
 

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

@@ -1,7 +1,7 @@
 """Base classes and interfaces for FastMCP resources."""
 
 import abc
-from typing import Annotated
+from typing import Annotated, Any
 
 from pydantic import (
     AnyUrl,
@@ -44,6 +44,6 @@ def set_default_name(cls, name: str | None, info: ValidationInfo) -> str:
         raise ValueError("Either name or uri must be provided")
 
     @abc.abstractmethod
-    async def read(self) -> str | bytes:
+    async def read(self, context: Any | None = None) -> str | bytes:
         """Read the resource content."""
         pass  # pragma: no cover

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

@@ -54,12 +54,12 @@ def from_function(
 
         # Find context parameter if it exists
         if context_kwarg is None:  # pragma: no branch
-            context_kwarg = find_context_parameter(fn)
+            context_kwarg = find_context_parameter(fn) or ""
 
         # 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 [],
+            skip_names=[context_kwarg] if context_kwarg else [],
         )
         parameters = func_arg_metadata.arg_model.model_json_schema()
 

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

@@ -14,6 +14,7 @@
 from pydantic import AnyUrl, Field, ValidationInfo, validate_call
 
 from mcp.server.fastmcp.resources.base import Resource
+from mcp.server.fastmcp.utilities.context_injection import find_context_parameter, inject_context
 from mcp.types import Annotations, Icon
 
 
@@ -22,7 +23,7 @@ class TextResource(Resource):
 
     text: str = Field(description="Text content of the resource")
 
-    async def read(self) -> str:
+    async def read(self, context: Any | None = None) -> str:
         """Read the text content."""
         return self.text  # pragma: no cover
 
@@ -32,7 +33,7 @@ class BinaryResource(Resource):
 
     data: bytes = Field(description="Binary content of the resource")
 
-    async def read(self) -> bytes:
+    async def read(self, context: Any | None = None) -> bytes:
         """Read the binary content."""
         return self.data  # pragma: no cover
 
@@ -51,24 +52,39 @@ class FunctionResource(Resource):
     """
 
     fn: Callable[[], Any] = Field(exclude=True)
+    context_kwarg: str | None = Field(None, exclude=True)
+
+    async def read(self, context: Any | None = None) -> str | bytes:
+        """Read the resource content by calling the function."""
+        # Inject context using utility which handles optimization
+        # If context_kwarg is set, it's used directly (fast)
+        # If not set (manual init), it falls back to inspection (safe)
+        args = inject_context(self.fn, {}, context, self.context_kwarg)
 
-    async def read(self) -> str | bytes:
-        """Read the resource by calling the wrapped function."""
         try:
-            # Call the function first to see if it returns a coroutine
-            result = self.fn()
-            # If it's a coroutine, await it
+            if inspect.iscoroutinefunction(self.fn):
+                result = await self.fn(**args)
+            else:
+                result = self.fn(**args)
+
+            # Support cases where a sync function returns a coroutine
             if inspect.iscoroutine(result):
-                result = await result
+                result = await result  # pragma: no cover
 
-            if isinstance(result, Resource):  # pragma: no cover
-                return await result.read()
-            elif isinstance(result, bytes):
-                return result
-            elif isinstance(result, str):
+            # Support returning a Resource instance (recursive read)
+            if isinstance(result, Resource):
+                return await result.read(context)  # pragma: no cover
+
+            if isinstance(result, str | bytes):
                 return result
-            else:
-                return pydantic_core.to_json(result, fallback=str, indent=2).decode()
+            if isinstance(result, pydantic.BaseModel):
+                return result.model_dump_json(indent=2)
+
+            # For other types, convert to a JSON string
+            try:
+                return json.dumps(pydantic_core.to_jsonable_python(result))
+            except pydantic_core.PydanticSerializationError:
+                return json.dumps(str(result))
         except Exception as e:
             raise ValueError(f"Error reading resource {self.uri}: {e}")
 
@@ -86,8 +102,10 @@ def from_function(
     ) -> "FunctionResource":
         """Create a FunctionResource from a function."""
         func_name = name or fn.__name__
-        if func_name == "<lambda>":  # pragma: no cover
-            raise ValueError("You must provide a name for lambda functions")
+        if func_name == "<lambda>":
+            raise ValueError("You must provide a name for lambda functions")  # pragma: no cover
+
+        context_kwarg = find_context_parameter(fn) or ""
 
         # ensure the arguments are properly cast
         fn = validate_call(fn)
@@ -100,6 +118,7 @@ def from_function(
             mime_type=mime_type or "text/plain",
             fn=fn,
             icons=icons,
+            context_kwarg=context_kwarg,
             annotations=annotations,
         )
 
@@ -125,7 +144,7 @@ class FileResource(Resource):
     def validate_absolute_path(cls, path: Path) -> Path:  # pragma: no cover
         """Ensure path is absolute."""
         if not path.is_absolute():
-            raise ValueError("Path must be absolute")
+            raise ValueError("Path must be absolute")  # pragma: no cover
         return path
 
     @pydantic.field_validator("is_binary")
@@ -137,7 +156,7 @@ def set_binary_from_mime_type(cls, is_binary: bool, info: ValidationInfo) -> boo
         mime_type = info.data.get("mime_type", "text/plain")
         return not mime_type.startswith("text/")
 
-    async def read(self) -> str | bytes:
+    async def read(self, context: Any | None = None) -> str | bytes:
         """Read the file content."""
         try:
             if self.is_binary:
@@ -153,7 +172,7 @@ class HttpResource(Resource):
     url: str = Field(description="URL to fetch content from")
     mime_type: str = Field(default="application/json", description="MIME type of the resource content")
 
-    async def read(self) -> str | bytes:
+    async def read(self, context: Any | None = None) -> str | bytes:
         """Read the HTTP content."""
         async with httpx.AsyncClient() as client:  # pragma: no cover
             response = await client.get(self.url)
@@ -191,7 +210,7 @@ def list_files(self) -> list[Path]:  # pragma: no cover
         except Exception as e:
             raise ValueError(f"Error listing directory {self.path}: {e}")
 
-    async def read(self) -> str:  # Always returns JSON string  # pragma: no cover
+    async def read(self, context: Any | None = None) -> str:  # Always returns JSON string  # pragma: no cover
         """Read the directory listing."""
         try:
             files = await anyio.to_thread.run_sync(self.list_files)

src/mcp/server/fastmcp/server.py

@@ -376,7 +376,7 @@ async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContent
             raise ResourceError(f"Unknown resource: {uri}")
 
         try:
-            content = await resource.read()
+            content = await resource.read(context=context)
             return [ReadResourceContents(content=content, mime_type=resource.mime_type)]
         except Exception as e:  # pragma: no cover
             logger.exception(f"Error reading resource {uri}")
@@ -575,27 +575,24 @@ async def get_weather(city: str) -> str:
             )
 
         def decorator(fn: AnyFunction) -> AnyFunction:
-            # Check if this should be a template
             sig = inspect.signature(fn)
-            has_uri_params = "{" in uri and "}" in uri
-            has_func_params = bool(sig.parameters)
+            context_param = find_context_parameter(fn)
+
+            # Determine effective parameters, excluding context
+            effective_func_params = {p for p in sig.parameters.keys() if p != context_param}
 
-            if has_uri_params or has_func_params:
-                # Check for Context parameter to exclude from validation
-                context_param = find_context_parameter(fn)
+            has_uri_params = "{" in uri and "}" in uri
+            has_effective_func_params = bool(effective_func_params)
 
-                # Validate that URI params match function params (excluding context)
+            if has_uri_params or has_effective_func_params:
+                # Register as template
                 uri_params = set(re.findall(r"{(\w+)}", uri))
-                # We need to remove the context_param from the resource function if
-                # there is any.
-                func_params = {p for p in sig.parameters.keys() if p != context_param}
 
-                if uri_params != func_params:
+                if uri_params != effective_func_params:
                     raise ValueError(
-                        f"Mismatch between URI parameters {uri_params} and function parameters {func_params}"
+                        f"Mismatch between URI parameters {uri_params} and function parameters {effective_func_params}"
                     )
 
-                # Register as template
                 self._resource_manager.add_template(
                     fn=fn,
                     uri_template=uri,

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

@@ -1,68 +1,57 @@
-"""Context injection utilities for FastMCP."""
-
-from __future__ import annotations
-
 import inspect
 import typing
 from collections.abc import Callable
-from typing import Any
-
-
-def find_context_parameter(fn: Callable[..., Any]) -> str | None:
-    """Find the parameter that should receive the Context object.
+from typing import TYPE_CHECKING, Any
 
-    Searches through the function's signature to find a parameter
-    with a Context type annotation.
+if TYPE_CHECKING:
+    from mcp.server.fastmcp import Context
 
-    Args:
-        fn: The function to inspect
 
-    Returns:
-        The name of the context parameter, or None if not found
+def find_context_parameter(fn: Callable[..., Any]) -> str | None:
     """
-    from mcp.server.fastmcp.server import Context
+    Inspect a function signature to find a parameter annotated with Context.
+    Returns the name of the parameter if found, otherwise None.
+    """
+    from mcp.server.fastmcp import Context
 
-    # Get type hints to properly resolve string annotations
     try:
-        hints = typing.get_type_hints(fn)
-    except Exception:
-        # If we can't resolve type hints, we can't find the context parameter
+        sig = inspect.signature(fn)
+    except ValueError:  # pragma: no cover
+        # Can't inspect signature (e.g. some builtins/wrappers)
         return None
 
-    # Check each parameter's type hint
-    for param_name, annotation in hints.items():
-        # Handle direct Context type
+    for param_name, param in sig.parameters.items():
+        annotation = param.annotation
+        if annotation is inspect.Parameter.empty:
+            continue
+
+        # Handle Optional[Context], Annotated[Context, ...], etc.
+        origin = typing.get_origin(annotation)
+
+        # Check if the annotation itself is Context or a subclass
         if inspect.isclass(annotation) and issubclass(annotation, Context):
             return param_name
 
-        # Handle generic types like Optional[Context]
-        origin = typing.get_origin(annotation)
-        if origin is not None:
-            args = typing.get_args(annotation)
-            for arg in args:
-                if inspect.isclass(arg) and issubclass(arg, Context):
-                    return param_name
+        # Check if it's a generic alias of Context (e.g., Context[...])
+        if origin is not None and inspect.isclass(origin) and issubclass(origin, Context):
+            return param_name  # pragma: no cover
 
     return None
 
 
 def inject_context(
     fn: Callable[..., Any],
     kwargs: dict[str, Any],
-    context: Any | None,
-    context_kwarg: str | None,
+    context: "Context[Any, Any, Any] | None",
+    context_kwarg: str | None = None,
 ) -> dict[str, Any]:
-    """Inject context into function kwargs if needed.
-
-    Args:
-        fn: The function that will be called
-        kwargs: The current keyword arguments
-        context: The context object to inject (if any)
-        context_kwarg: The name of the parameter to inject into
-
-    Returns:
-        Updated kwargs with context injected if applicable
     """
-    if context_kwarg is not None and context is not None:
-        return {**kwargs, context_kwarg: context}
+    Inject the Context object into kwargs if the function expects it.
+    Returns the updated kwargs.
+    """
+    if context_kwarg is None:
+        context_kwarg = find_context_parameter(fn)
+
+    if context_kwarg:
+        kwargs[context_kwarg] = context
     return kwargs