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/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,
@@ -43,6 +43,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

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

@@ -4,7 +4,7 @@
 import json
 from collections.abc import Callable
 from pathlib import Path
-from typing import Any
+from typing import Any, Optional
 
 import anyio
 import anyio.to_thread
@@ -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
 from mcp.types import 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
 
@@ -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
 
@@ -51,24 +52,30 @@ class FunctionResource(Resource):
     """
 
     fn: Callable[[], Any] = Field(exclude=True)
+    context_kwarg: Optional[str] = Field(None, exclude=True)
+
+    async def read(self, context: Any | None = None) -> str | bytes:
+        """Read the resource content by calling the function."""
+        args = {}
+        if self.context_kwarg:
+            args[self.context_kwarg] = context
 
-    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.iscoroutine(result):
-                result = await result
-
-            if isinstance(result, Resource):
-                return await result.read()
-            elif isinstance(result, bytes):
-                return result
-            elif isinstance(result, str):
-                return result
+            if inspect.iscoroutinefunction(self.fn):
+                result = await self.fn(**args)
             else:
-                return pydantic_core.to_json(result, fallback=str, indent=2).decode()
+                result = self.fn(**args)
+
+            if isinstance(result, (str, bytes)):
+                return result
+            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}")
 
@@ -88,6 +95,8 @@ def from_function(
         if func_name == "<lambda>":
             raise ValueError("You must provide a name for lambda functions")
 
+        context_kwarg = find_context_parameter(fn)
+
         # ensure the arguments are properly cast
         fn = validate_call(fn)
 
@@ -99,6 +108,7 @@ def from_function(
             mime_type=mime_type or "text/plain",
             fn=fn,
             icons=icons,
+            context_kwarg=context_kwarg,
         )
 
 
@@ -135,7 +145,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:
@@ -151,7 +161,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:
             response = await client.get(self.url)
@@ -189,7 +199,7 @@ def list_files(self) -> list[Path]:
         except Exception as e:
             raise ValueError(f"Error listing directory {self.path}: {e}")
 
-    async def read(self) -> str:  # Always returns JSON string
+    async def read(self, context: Any | None = None) -> str:  # Always returns JSON string
         """Read the directory listing."""
         try:
             files = await anyio.to_thread.run_sync(self.list_files)

src/mcp/server/fastmcp/server.py

@@ -348,7 +348,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:
             logger.exception(f"Error reading resource {uri}")
@@ -531,27 +531,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

@@ -31,17 +31,16 @@ def find_context_parameter(fn: Callable[..., Any]) -> str | None:
 
     # Check each parameter's type hint
     for param_name, annotation in hints.items():
-        # Handle direct Context type
+        # Handle direct Context type and generic aliases of Context
+        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
 
     return None
 

tests/server/fastmcp/test_integration.py

@@ -32,6 +32,7 @@
     sampling,
     structured_output,
     tool_progress,
+    context_resource,
 )
 from mcp.client.session import ClientSession
 from mcp.client.sse import sse_client
@@ -124,6 +125,8 @@ def run_server_with_transport(module_name: str, port: int, transport: str) -> No
         mcp = fastmcp_quickstart.mcp
     elif module_name == "structured_output":
         mcp = structured_output.mcp
+    elif module_name == "context_resource":
+        mcp = context_resource.mcp
     else:
         raise ImportError(f"Unknown module: {module_name}")
 
@@ -697,3 +700,42 @@ async def test_structured_output(server_transport: str, server_url: str) -> None
             assert "sunny" in result_text  # condition
             assert "45" in result_text  # humidity
             assert "5.2" in result_text  # wind_speed
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "server_transport",
+    [
+        ("context_resource", "sse"),
+        ("context_resource", "streamable-http"),
+    ],
+    indirect=True,
+)
+async def test_context_only_resource(server_transport: str, server_url: str) -> None:
+    """Test that a resource with only a context argument is registered as a regular resource."""
+    transport = server_transport
+    client_cm = create_client_for_transport(transport, server_url)
+
+    async with client_cm as client_streams:
+        read_stream, write_stream = unpack_streams(client_streams)
+        async with ClientSession(read_stream, write_stream) as session:
+            # Test initialization
+            result = await session.initialize()
+            assert isinstance(result, InitializeResult)
+            assert result.serverInfo.name == "Context Resource Example"
+
+            # Check that it is not in templates
+            templates = await session.list_resource_templates()
+            assert len(templates.resourceTemplates) == 0
+
+            # Check that it is in resources
+            resources = await session.list_resources()
+            assert len(resources.resources) == 1
+            resource = resources.resources[0]
+            assert resource.uri == AnyUrl("resource://only_context")
+
+            # Check that we can read it
+            read_result = await session.read_resource(AnyUrl("resource://only_context"))
+            assert len(read_result.contents) == 1
+            assert isinstance(read_result.contents[0], TextResourceContents)
+            assert read_result.contents[0].text == "Resource with only context injected"

tests/server/fastmcp/test_server.py

@@ -1046,6 +1046,33 @@ def resource_custom_ctx(id: str, my_ctx: Context[ServerSession, None]) -> str:
             assert isinstance(content, TextResourceContents)
             assert "Resource 123 with context" in content.text
 
+    @pytest.mark.anyio
+    async def test_resource_only_context(self):
+        """Test that resources without template args can receive context."""
+        mcp = FastMCP()
+
+        @mcp.resource("resource://only_context", name="resource_with_context_no_args")
+        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"
+
+        # Test via client
+        async with client_session(mcp._mcp_server) as client:
+            # Verify resource is registered via client
+            resources = await client.list_resources()
+            assert len(resources.resources) == 1
+            resource = resources.resources[0]
+            assert resource.uri == AnyUrl("resource://only_context")
+            assert resource.name == "resource_with_context_no_args"
+
+            # Test reading the resource
+            result = await client.read_resource(AnyUrl("resource://only_context"))
+            assert len(result.contents) == 1
+            content = result.contents[0]
+            assert isinstance(content, TextResourceContents)
+            assert content.text == "Resource with only context injected"
+
     @pytest.mark.anyio
     async def test_prompt_with_context(self):
         """Test that prompts can receive context parameter."""