Merge pull request #263 from jlowin/context

Support context injection in resources, templates, and prompts (like tools)

docs/servers/context.mdx

@@ -1,16 +1,16 @@
 ---
 title: MCP Context
 sidebarTitle: Context
-description: Access MCP capabilities like logging, progress, and resources within your tools.
+description: Access MCP capabilities like logging, progress, and resources within your MCP objects.
 icon: rectangle-code
 ---
 import { VersionBadge } from '/snippets/version-badge.mdx'
 
-When defining FastMCP [tools](/servers/tools), your functions might need to interact with the underlying MCP session or access server capabilities. FastMCP provides the `Context` object for this purpose.
+When defining FastMCP [tools](/servers/tools), [resources](/servers/resources), resource templates, or [prompts](/servers/prompts), your functions might need to interact with the underlying MCP session or access server capabilities. FastMCP provides the `Context` object for this purpose.
 
 ## What Is Context?
 
-The `Context` object provides a clean interface to access MCP features within your tool functions, including:
+The `Context` object provides a clean interface to access MCP features within your functions, including:
 
 - **Logging**: Send debug, info, warning, and error messages back to the client
 - **Progress Reporting**: Update the client on the progress of long-running operations
@@ -21,7 +21,7 @@ The `Context` object provides a clean interface to access MCP features within yo
 
 ## Accessing the Context
 
-To use the context object within your tool function, simply add a parameter to your function signature and type-hint it as `Context`. FastMCP will automatically inject the context instance when your tool is called.
+To use the context object within any of your functions, simply add a parameter to your function signature and type-hint it as `Context`. FastMCP will automatically inject the context instance when your function is called.
 
 ```python
 from fastmcp import FastMCP, Context
@@ -65,15 +65,15 @@ async def process_file(file_uri: str, ctx: Context) -> str:
 
 - The parameter name (e.g., `ctx`, `context`) doesn't matter, only the type hint `Context` is important.
 - The context parameter can be placed anywhere in your function's signature.
-- The context is optional - tools that don't need it can omit the parameter.
-- Context is only available within tool functions during a request; attempting to use context methods outside a request will raise errors.
-- Context methods are async, so your tool function usually needs to be async as well.
+- The context is optional - functions that don't need it can omit the parameter.
+- Context is only available during a request; attempting to use context methods outside a request will raise errors.
+- Context methods are async, so your function usually needs to be async as well.
 
 ## Context Capabilities
 
 ### Logging
 
-Send log messages back to the MCP client. This is useful for debugging and providing visibility into tool execution during a request.
+Send log messages back to the MCP client. This is useful for debugging and providing visibility into function execution during a request.
 
 ```python
 @mcp.tool()
@@ -97,14 +97,14 @@ async def analyze_data(data: list[float], ctx: Context) -> dict:
 **Available Logging Methods:**
 
 - **`ctx.debug(message: str)`**: Low-level details useful for debugging
-- **`ctx.info(message: str)`**: General information about tool execution
+- **`ctx.info(message: str)`**: General information about execution
 - **`ctx.warning(message: str)`**: Potential issues that didn't prevent execution
 - **`ctx.error(message: str)`**: Errors that occurred during execution
 - **`ctx.log(level: Literal["debug", "info", "warning", "error"], message: str, logger_name: str | None = None)`**: Generic log method supporting custom logger names
 
 ### Progress Reporting
 
-For long-running tools, notify the client about the progress of the operation. This allows clients to display progress indicators and provide a better user experience.
+For long-running operations, notify the client about the progress. This allows clients to display progress indicators and provide a better user experience.
 
 ```python
 @mcp.tool()
@@ -137,7 +137,7 @@ Progress reporting requires the client to have sent a `progressToken` in the ini
 
 ### Resource Access
 
-Read data from resources registered with your FastMCP server. This allows tools to access files, configuration, or dynamically generated content.
+Read data from resources registered with your FastMCP server. This allows functions to access files, configuration, or dynamically generated content.
 
 ```python
 @mcp.tool()
@@ -177,7 +177,7 @@ The returned content is typically accessed via `content_list[0].content` and can
 
 <VersionBadge version="2.0.0" />
 
-Request the client's LLM to generate text based on provided messages. This is useful when your tool needs to leverage the LLM's capabilities to process data or generate responses.
+Request the client's LLM to generate text based on provided messages. This is useful when your function needs to leverage the LLM's capabilities to process data or generate responses.
 
 ```python
 @mcp.tool()
@@ -279,6 +279,60 @@ async def advanced_tool(ctx: Context) -> str:
 Direct use of `session` or `request_context` requires understanding the low-level MCP Python SDK and may be less stable than using the methods provided directly on the `Context` object.
 </Warning>
 
-## Using Context in Other Components
+## Using Context in Different Components
 
-Currently, Context is primarily designed for use within tool functions. Support for Context in other components like resources and prompts is planned for future releases. 
+All FastMCP components (tools, resources, templates, and prompts) can use the Context object following the same pattern - simply add a parameter with the `Context` type annotation.
+
+### Context in Resources and Templates
+
+Resources and resource templates can access context to customize their behavior:
+
+```python
+@mcp.resource("resource://user-data")
+async def get_user_data(ctx: Context) -> dict:
+    """Fetch personalized user data based on the request context."""
+    user_id = ctx.client_id or "anonymous"
+    await ctx.info(f"Fetching data for user {user_id}")
+    
+    # Example of using context for dynamic resource generation
+    return {
+        "user_id": user_id,
+        "last_access": datetime.now().isoformat(),
+        "request_id": ctx.request_id
+    }
+
+@mcp.resource("resource://users/{user_id}/profile")
+async def get_user_profile(user_id: str, ctx: Context) -> dict:
+    """Fetch user profile from database with context-aware logging."""
+    await ctx.info(f"Fetching profile for user {user_id}")
+    
+    # Example of using context in a template resource
+    # In a real implementation, you might query a database
+    return {
+        "id": user_id,
+        "name": f"User {user_id}",
+        "request_id": ctx.request_id
+    }
+```
+
+### Context in Prompts
+
+Prompts can use context to generate more dynamic templates:
+
+```python
+@mcp.prompt()
+async def data_analysis_request(dataset: str, ctx: Context) -> str:
+    """Generate a request to analyze data with contextual information."""
+    await ctx.info(f"Generating data analysis prompt for {dataset}")
+    
+    # Could use context to read configuration or personalize the prompt
+    return f"""Please analyze the following dataset: {dataset}
+    
+Request initiated at: {datetime.now().isoformat()}
+Request ID: {ctx.request_id}
+"""
+```
+
+<VersionBadge version="2.3.0" />
+
+All FastMCP objects now support context injection using the same consistent pattern, making it easy to add session-aware capabilities to all aspects of your MCP server. 

docs/servers/prompts.mdx

@@ -171,33 +171,24 @@ async def data_based_prompt(data_id: str) -> str:
 
 Use `async def` when your prompt function performs I/O operations like network requests, database queries, file I/O, or external service calls.
 
-### The MCP Session
+### Accessing MCP Context
 
-Prompts can access the MCP features via the `Context` object, just like tools.
+<VersionBadge version="2.2.5" />
 
-```python
-from fastmcp import Context
+Prompts can access additional MCP information and features through the `Context` object. To access it, add a parameter to your prompt function with a type annotation of `Context`:
+
+```python {6}
+from fastmcp import FastMCP, Context
+
+mcp = FastMCP(name="PromptServer")
 
 @mcp.prompt()
 async def generate_report_request(report_type: str, ctx: Context) -> str:
-    """Generates a request for a report based on available data."""
-    # Log the request
-    await ctx.info(f"Generating prompt for report type: {report_type}")
-    
-    # Could potentially use ctx.read_resource to fetch data
-    # Or ctx.sample to get additional input from the LLM
-    
-    return f"Please create a {report_type} report based on the available data."
+    """Generates a request for a report."""
+    return f"Please create a {report_type} report. Request ID: {ctx.request_id}"
 ```
 
-Using the `ctx` parameter (based on its `Context` type hint), you can access:
-
--   **Logging:** `ctx.debug()`, `ctx.info()`, etc.
--   **Resource Access:** `ctx.read_resource(uri)`
--   **LLM Sampling:** `ctx.sample(...)`
--   **Request Info:** `ctx.request_id`, `ctx.client_id`
-
-Refer to the [Context documentation](/servers/context) for more details on these capabilities.
+For full documentation on the Context object and all its capabilities, see the [Context documentation](/servers/context).
 
 ## Server Behavior
 

docs/servers/resources.mdx

@@ -95,6 +95,36 @@ def get_application_status() -> dict:
 - **`mime_type`**: Specifies the content type (FastMCP often infers a default like `text/plain` or `application/json`, but explicit is better for non-text types).
 - **`tags`**: A set of strings for categorization, potentially used by clients for filtering.
 
+### Accessing MCP Context
+
+<VersionBadge version="2.2.5" />
+
+Resources and resource templates can access additional MCP information and features through the `Context` object. To access it, add a parameter to your resource function with a type annotation of `Context`:
+
+```python {6, 14}
+from fastmcp import FastMCP, Context
+
+mcp = FastMCP(name="DataServer")
+
+@mcp.resource("resource://system-status")
+async def get_system_status(ctx: Context) -> dict:
+    """Provides system status information."""
+    return {
+        "status": "operational",
+        "request_id": ctx.request_id
+    }
+
+@mcp.resource("resource://{name}/details")
+async def get_details(name: str, ctx: Context) -> dict:
+    """Get details for a specific name."""
+    return {
+        "name": name,
+        "accessed_at": ctx.request_id
+    }
+```
+
+For full documentation on the Context object and all its capabilities, see the [Context documentation](/servers/context).
+
 
 ### Asynchronous Resources
 
@@ -205,6 +235,8 @@ Note that this parameter is only available when using `add_resource()` directly
 
 Resource Templates allow clients to request resources whose content depends on parameters embedded in the URI. Define a template using the **same `@mcp.resource` decorator**, but include `{parameter_name}` placeholders in the URI string and add corresponding arguments to your function signature.
 
+Resource templates share most configuration options with regular resources (name, description, mime_type, tags), but add the ability to define URI parameters that map to function parameters.
+
 Resource templates generate a new resource for each unique set of parameters, which means that resources can be dynamically created on-demand. For example, if the resource template `"user://profile/{name}"` is registered, MCP clients could request `"user://profile/ford"` or `"user://profile/marvin"` to retrieve either of those two user profiles as resources, without having to register each resource individually.
 
 Here is a complete example that shows how to define two resource templates:
@@ -379,28 +411,6 @@ In this stacked decorator pattern:
 
 Templates provide a powerful way to expose parameterized data access points following REST-like principles.
 
-### Custom Template Keys
-
-<VersionBadge version="2.2.0" />
-
-Similar to resources, you can provide custom keys when directly adding templates:
-
-```python
-from fastmcp.resources import ResourceTemplate
-
-# Create a template with a function
-template = ResourceTemplate.from_function(
-    my_function,
-    uri_template="data://{id}/details",
-    name="Data Details"
-)
-
-# Register with a custom key
-mcp._resource_manager.add_template(template, key="custom://{id}/view")
-```
-
-This allows accessing the same template implementation through different URI patterns.
-
 ## Server Behavior
 
 ### Duplicate Resources

docs/servers/tools.mdx

@@ -263,7 +263,8 @@ FastMCP automatically catches exceptions raised within your tool function:
 
 Using informative exceptions helps the LLM understand failures and react appropriately.
 
-### Accessing MCP Context
+## MCP Context
+
 
 Tools can access MCP features like logging, reading resources, or reporting progress through the `Context` object. To use it, add a parameter to your tool function with the type hint `Context`.
 
@@ -304,39 +305,6 @@ The Context object provides access to:
 
 For full documentation on the Context object and all its capabilities, see the [Context documentation](/servers/context).
 
-## Server Behavior
-
-### Duplicate Tools
-
-<VersionBadge version="2.1.0" />
-
-You can control how the FastMCP server behaves if you try to register multiple tools with the same name. This is configured using the `on_duplicate_tools` argument when creating the `FastMCP` instance.
-
-```python
-from fastmcp import FastMCP
-
-mcp = FastMCP(
-    name="StrictServer",
-    # Configure behavior for duplicate tool names
-    on_duplicate_tools="error"
-)
-
-@mcp.tool()
-def my_tool(): return "Version 1"
-
-# This will now raise a ValueError because 'my_tool' already exists
-# and on_duplicate_tools is set to "error".
-# @mcp.tool()
-# def my_tool(): return "Version 2"
-```
-
-The duplicate behavior options are:
-
--   `"warn"` (default): Logs a warning and the new tool replaces the old one.
--   `"error"`: Raises a `ValueError`, preventing the duplicate registration.
--   `"replace"`: Silently replaces the existing tool with the new one.
--   `"ignore"`: Keeps the original tool and ignores the new registration attempt.
-
 ## Parameter Types
 
 FastMCP supports a wide variety of parameter types to give you flexibility when designing your tools.
@@ -663,3 +631,36 @@ Common validation options include:
 | `description` | Any | Human-readable description (appears in schema) |
 
 When a client sends invalid data, FastMCP will return a validation error explaining why the parameter failed validation.
+
+## Server Behavior
+
+### Duplicate Tools
+
+<VersionBadge version="2.1.0" />
+
+You can control how the FastMCP server behaves if you try to register multiple tools with the same name. This is configured using the `on_duplicate_tools` argument when creating the `FastMCP` instance.
+
+```python
+from fastmcp import FastMCP
+
+mcp = FastMCP(
+    name="StrictServer",
+    # Configure behavior for duplicate tool names
+    on_duplicate_tools="error"
+)
+
+@mcp.tool()
+def my_tool(): return "Version 1"
+
+# This will now raise a ValueError because 'my_tool' already exists
+# and on_duplicate_tools is set to "error".
+# @mcp.tool()
+# def my_tool(): return "Version 2"
+```
+
+The duplicate behavior options are:
+
+-   `"warn"` (default): Logs a warning and the new tool replaces the old one.
+-   `"error"`: Raises a `ValueError`, preventing the duplicate registration.
+-   `"replace"`: Silently replaces the existing tool with the new one.
+-   `"ignore"`: Keeps the original tool and ignores the new registration attempt.

src/fastmcp/prompts/prompt.py

@@ -1,9 +1,11 @@
 """Base classes for FastMCP prompts."""
 
+from __future__ import annotations as _annotations
+
 import inspect
 import json
 from collections.abc import Awaitable, Callable, Sequence
-from typing import Annotated, Any, Literal
+from typing import TYPE_CHECKING, Annotated, Any, Literal
 
 import pydantic_core
 from mcp.types import EmbeddedResource, ImageContent, TextContent
@@ -13,6 +15,12 @@ from pydantic import BaseModel, BeforeValidator, Field, TypeAdapter, validate_ca
 
 from fastmcp.utilities.types import _convert_set_defaults
 
+if TYPE_CHECKING:
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT
+
+    from fastmcp.server import Context
+
 CONTENT_TYPES = TextContent | ImageContent | EmbeddedResource
 
 
@@ -72,6 +80,9 @@ class Prompt(BaseModel):
         None, description="Arguments that can be passed to the prompt"
     )
     fn: Callable[..., PromptResult | Awaitable[PromptResult]]
+    context_kwarg: str | None = Field(
+        None, description="Name of the kwarg that should receive context"
+    )
 
     @classmethod
     def from_function(
@@ -80,7 +91,8 @@ class Prompt(BaseModel):
         name: str | None = None,
         description: str | None = None,
         tags: set[str] | None = None,
-    ) -> "Prompt":
+        context_kwarg: str | None = None,
+    ) -> Prompt:
         """Create a Prompt from a function.
 
         The function can return:
@@ -89,11 +101,24 @@ class Prompt(BaseModel):
         - A dict (converted to a message)
         - A sequence of any of the above
         """
+        from fastmcp import Context
+
         func_name = name or fn.__name__
 
         if func_name == "<lambda>":
             raise ValueError("You must provide a name for lambda functions")
 
+        # Auto-detect context parameter if not provided
+        if context_kwarg is None:
+            if inspect.ismethod(fn) and hasattr(fn, "__func__"):
+                sig = inspect.signature(fn.__func__)
+            else:
+                sig = inspect.signature(fn)
+            for param_name, param in sig.parameters.items():
+                if param.annotation is Context:
+                    context_kwarg = param_name
+                    break
+
         # Get schema from TypeAdapter - will fail if function isn't properly typed
         parameters = TypeAdapter(fn).json_schema()
 
@@ -101,6 +126,10 @@ class Prompt(BaseModel):
         arguments: list[PromptArgument] = []
         if "properties" in parameters:
             for param_name, param in parameters["properties"].items():
+                # Skip context parameter
+                if param_name == context_kwarg:
+                    continue
+
                 required = param_name in parameters.get("required", [])
                 arguments.append(
                     PromptArgument(
@@ -119,9 +148,14 @@ class Prompt(BaseModel):
             arguments=arguments,
             fn=fn,
             tags=tags or set(),
+            context_kwarg=context_kwarg,
         )
 
-    async def render(self, arguments: dict[str, Any] | None = None) -> list[Message]:
+    async def render(
+        self,
+        arguments: dict[str, Any] | None = None,
+        context: Context[ServerSessionT, LifespanContextT] | None = None,
+    ) -> list[Message]:
         """Render the prompt with arguments."""
         # Validate required arguments
         if self.arguments:
@@ -132,8 +166,13 @@ class Prompt(BaseModel):
                 raise ValueError(f"Missing required arguments: {missing}")
 
         try:
+            # Prepare arguments with context
+            kwargs = arguments.copy() if arguments else {}
+            if self.context_kwarg is not None and context is not None:
+                kwargs[self.context_kwarg] = context
+
             # Call function and check if result is a coroutine
-            result = self.fn(**(arguments or {}))
+            result = self.fn(**kwargs)
             if inspect.iscoroutine(result):
                 result = await result
 

src/fastmcp/prompts/prompt_manager.py

@@ -1,13 +1,21 @@
 """Prompt management functionality."""
 
+from __future__ import annotations as _annotations
+
 from collections.abc import Awaitable, Callable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from fastmcp.exceptions import NotFoundError
 from fastmcp.prompts.prompt import Message, Prompt, PromptResult
 from fastmcp.settings import DuplicateBehavior
 from fastmcp.utilities.logging import get_logger
 
+if TYPE_CHECKING:
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT
+
+    from fastmcp.server import Context
+
 logger = get_logger(__name__)
 
 
@@ -69,14 +77,17 @@ class PromptManager:
         return prompt
 
     async def render_prompt(
-        self, name: str, arguments: dict[str, Any] | None = None
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        context: Context[ServerSessionT, LifespanContextT] | None = None,
     ) -> list[Message]:
         """Render a prompt by name with arguments."""
         prompt = self.get_prompt(name)
         if not prompt:
             raise NotFoundError(f"Unknown prompt: {name}")
 
-        return await prompt.render(arguments)
+        return await prompt.render(arguments, context=context)
 
     def has_prompt(self, key: str) -> bool:
         """Check if a prompt exists."""

src/fastmcp/resources/resource.py

@@ -1,7 +1,9 @@
 """Base classes and interfaces for FastMCP resources."""
 
+from __future__ import annotations
+
 import abc
-from typing import Annotated, Any
+from typing import TYPE_CHECKING, Annotated, Any
 
 from mcp.types import Resource as MCPResource
 from pydantic import (
@@ -17,6 +19,12 @@ from pydantic import (
 
 from fastmcp.utilities.types import _convert_set_defaults
 
+if TYPE_CHECKING:
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT
+
+    from fastmcp.server import Context
+
 
 class Resource(BaseModel, abc.ABC):
     """Base class for all resources."""
@@ -58,7 +66,9 @@ class Resource(BaseModel, abc.ABC):
         raise ValueError("Either name or uri must be provided")
 
     @abc.abstractmethod
-    async def read(self) -> str | bytes:
+    async def read(
+        self, context: Context[ServerSessionT, LifespanContextT] | None = None
+    ) -> str | bytes:
         """Read the resource content."""
         pass
 

src/fastmcp/resources/resource_manager.py

@@ -61,9 +61,16 @@ class ResourceManager:
             The added resource or template. If a resource or template with the same URI already exists,
             returns the existing resource or template.
         """
+        from fastmcp.server.context import Context
+
         # Check if this should be a template
         has_uri_params = "{" in uri and "}" in uri
-        has_func_params = bool(inspect.signature(fn).parameters)
+        # check if the function has any parameters (other than injected context)
+        has_func_params = any(
+            p
+            for p in inspect.signature(fn).parameters.values()
+            if p.annotation is not Context
+        )
 
         if has_uri_params or has_func_params:
             return self.add_template_from_fn(
@@ -102,12 +109,12 @@ class ResourceManager:
             The added resource. If a resource with the same URI already exists,
             returns the existing resource.
         """
-        resource = FunctionResource(
+        resource = FunctionResource.from_function(
+            fn=fn,
             uri=AnyUrl(uri),
             name=name,
             description=description,
             mime_type=mime_type or "text/plain",
-            fn=fn,
             tags=tags or set(),
         )
         return self.add_resource(resource)
@@ -212,9 +219,13 @@ class ResourceManager:
                 return True
         return False
 
-    async def get_resource(self, uri: AnyUrl | str) -> Resource:
+    async def get_resource(self, uri: AnyUrl | str, context=None) -> Resource:
         """Get resource by URI, checking concrete resources first, then templates.
 
+        Args:
+            uri: The URI of the resource to get
+            context: Optional context object to pass to template resources
+
         Raises:
             NotFoundError: If no resource or template matching the URI is found.
         """
@@ -230,7 +241,11 @@ class ResourceManager:
             # Try to match against the storage key (which might be a custom key)
             if params := match_uri_template(uri_str, storage_key):
                 try:
-                    return await template.create_resource(uri_str, params)
+                    return await template.create_resource(
+                        uri_str,
+                        params=params,
+                        context=context,
+                    )
                 except Exception as e:
                     raise ValueError(f"Error creating resource from template: {e}")
 

src/fastmcp/resources/template.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import inspect
 import re
 from collections.abc import Callable
-from typing import Annotated, Any
+from typing import TYPE_CHECKING, Annotated, Any
 from urllib.parse import unquote
 
 from mcp.types import ResourceTemplate as MCPResourceTemplate
@@ -22,6 +22,12 @@ from pydantic import (
 from fastmcp.resources.types import FunctionResource, Resource
 from fastmcp.utilities.types import _convert_set_defaults
 
+if TYPE_CHECKING:
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT
+
+    from fastmcp.server import Context
+
 
 def build_regex(template: str) -> re.Pattern:
     parts = re.split(r"(\{[^}]+\})", template)
@@ -70,6 +76,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"
+    )
 
     @field_validator("mime_type", mode="before")
     @classmethod
@@ -88,18 +97,34 @@ class ResourceTemplate(BaseModel):
         description: str | None = None,
         mime_type: str | None = None,
         tags: set[str] | None = None,
+        context_kwarg: str | None = None,
     ) -> ResourceTemplate:
         """Create a template from a function."""
+        from fastmcp import Context
+
         func_name = name or fn.__name__
         if func_name == "<lambda>":
             raise ValueError("You must provide a name for lambda functions")
 
+        # Auto-detect context parameter if not provided
+        if context_kwarg is None:
+            if inspect.ismethod(fn) and hasattr(fn, "__func__"):
+                sig = inspect.signature(fn.__func__)
+            else:
+                sig = inspect.signature(fn)
+            for param_name, param in sig.parameters.items():
+                if param.annotation is Context:
+                    context_kwarg = param_name
+                    break
+
         # Validate that URI params match function params
         uri_params = set(re.findall(r"{(\w+)(?:\*)?}", uri_template))
         if not uri_params:
             raise ValueError("URI template must contain at least one parameter")
 
         func_params = set(inspect.signature(fn).parameters.keys())
+        if context_kwarg:
+            func_params.discard(context_kwarg)
 
         # get the parameters that are required
         required_params = {
@@ -107,6 +132,8 @@ class ResourceTemplate(BaseModel):
             for p in func_params
             if inspect.signature(fn).parameters[p].default is inspect.Parameter.empty
         }
+        if context_kwarg and context_kwarg in required_params:
+            required_params.discard(context_kwarg)
 
         if not required_params.issubset(uri_params):
             raise ValueError(
@@ -132,17 +159,28 @@ class ResourceTemplate(BaseModel):
             fn=fn,
             parameters=parameters,
             tags=tags or set(),
+            context_kwarg=context_kwarg,
         )
 
     def matches(self, uri: str) -> dict[str, Any] | None:
         """Check if URI matches template and extract parameters."""
         return match_uri_template(uri, self.uri_template)
 
-    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[ServerSessionT, LifespanContextT] | None = None,
+    ) -> Resource:
         """Create a resource from the template with the given parameters."""
         try:
+            # Add context to parameters if needed
+            kwargs = params.copy()
+            if self.context_kwarg is not None and context is not None:
+                kwargs[self.context_kwarg] = context
+
             # Call function and check if result is a coroutine
-            result = self.fn(**params)
+            result = self.fn(**kwargs)
             if inspect.iscoroutine(result):
                 result = await result
 
@@ -151,8 +189,9 @@ class ResourceTemplate(BaseModel):
                 name=self.name,
                 description=self.description,
                 mime_type=self.mime_type,
-                fn=lambda: result,  # Capture result in closure
+                fn=lambda **kwargs: result,  # Capture result in closure
                 tags=self.tags,
+                context_kwarg=self.context_kwarg,
             )
         except Exception as e:
             raise ValueError(f"Error creating resource from template: {e}")

src/fastmcp/resources/types.py

@@ -1,10 +1,12 @@
 """Concrete resource implementations."""
 
+from __future__ import annotations
+
 import inspect
 import json
 from collections.abc import Callable
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import anyio
 import anyio.to_thread
@@ -13,15 +15,24 @@ import pydantic.json
 import pydantic_core
 from pydantic import Field, ValidationInfo
 
+import fastmcp
 from fastmcp.resources.resource import Resource
 
+if TYPE_CHECKING:
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT
+
+    from fastmcp.server import Context
+
 
 class TextResource(Resource):
     """A resource that reads from a string."""
 
     text: str = Field(description="Text content of the resource")
 
-    async def read(self) -> str:
+    async def read(
+        self, context: Context[ServerSessionT, LifespanContextT] | None = None
+    ) -> str:
         """Read the text content."""
         return self.text
 
@@ -31,7 +42,9 @@ class BinaryResource(Resource):
 
     data: bytes = Field(description="Binary content of the resource")
 
-    async def read(self) -> bytes:
+    async def read(
+        self, context: Context[ServerSessionT, LifespanContextT] | None = None
+    ) -> bytes:
         """Read the binary content."""
         return self.data
 
@@ -50,15 +63,40 @@ class FunctionResource(Resource):
     """
 
     fn: Callable[[], Any]
+    context_kwarg: str | None = Field(
+        default=None, description="Name of the kwarg that should receive context"
+    )
 
-    async def read(self) -> str | bytes:
+    @classmethod
+    def from_function(
+        cls, fn: Callable[[], Any], context_kwarg: str | None = None, **kwargs
+    ) -> FunctionResource:
+        if context_kwarg is None:
+            parameters = inspect.signature(fn).parameters
+            context_param = next(
+                (p for p in parameters.values() if p.annotation is fastmcp.Context),
+                None,
+            )
+            if context_param is not None:
+                context_kwarg = context_param.name
+        return cls(fn=fn, context_kwarg=context_kwarg, **kwargs)
+
+    async def read(
+        self,
+        context: Context[ServerSessionT, LifespanContextT] | None = None,
+    ) -> str | bytes:
         """Read the resource by calling the wrapped function."""
         try:
-            result = (
-                await self.fn() if inspect.iscoroutinefunction(self.fn) else self.fn()
-            )
+            kwargs = {}
+            if self.context_kwarg is not None:
+                kwargs[self.context_kwarg] = context
+
+            result = self.fn(**kwargs)
+            if inspect.iscoroutinefunction(self.fn):
+                result = await result
+
             if isinstance(result, Resource):
-                return await result.read()
+                return await result.read(context=context)
             if isinstance(result, bytes):
                 return result
             if isinstance(result, str):
@@ -105,7 +143,9 @@ class FileResource(Resource):
         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: Context[ServerSessionT, LifespanContextT] | None = None
+    ) -> str | bytes:
         """Read the file content."""
         try:
             if self.is_binary:
@@ -123,7 +163,9 @@ class HttpResource(Resource):
         default="application/json", description="MIME type of the resource content"
     )
 
-    async def read(self) -> str | bytes:
+    async def read(
+        self, context: Context[ServerSessionT, LifespanContextT] | None = None
+    ) -> str | bytes:
         """Read the HTTP content."""
         async with httpx.AsyncClient() as client:
             response = await client.get(self.url)
@@ -175,7 +217,9 @@ class DirectoryResource(Resource):
         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: Context[ServerSessionT, LifespanContextT] | None = None
+    ) -> str:  # Always returns JSON string
         """Read the directory listing."""
         try:
             files = await anyio.to_thread.run_sync(self.list_files)

src/fastmcp/server/openapi.py

@@ -1,11 +1,13 @@
 """FastMCP server implementation for OpenAPI integration."""
 
+from __future__ import annotations
+
 import enum
 import json
 import re
 from dataclasses import dataclass
 from re import Pattern
-from typing import Any, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 import httpx
 from mcp.types import TextContent
@@ -22,6 +24,12 @@ from fastmcp.utilities.openapi import (
     format_description_with_responses,
 )
 
+if TYPE_CHECKING:
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT
+
+    from fastmcp.server import Context
+
 logger = get_logger(__name__)
 
 HttpMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"]
@@ -257,7 +265,9 @@ class OpenAPIResource(Resource):
         self._client = client
         self._route = route
 
-    async def read(self) -> str | bytes:
+    async def read(
+        self, context: Context[ServerSessionT, LifespanContextT] | None = None
+    ) -> str | bytes:
         """Fetch the resource data by making an HTTP request."""
         try:
             # Extract path parameters from the URI if present
@@ -347,11 +357,17 @@ class OpenAPIResourceTemplate(ResourceTemplate):
             fn=lambda **kwargs: None,
             parameters=parameters,
             tags=tags,
+            context_kwarg=None,
         )
         self._client = client
         self._route = route
 
-    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[ServerSessionT, LifespanContextT] | None = None,
+    ) -> Resource:
         """Create a resource with the given parameters."""
         # Generate a URI for this resource instance
         uri_parts = []

src/fastmcp/server/proxy.py

@@ -1,4 +1,6 @@
-from typing import Any, cast
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, cast
 from urllib.parse import quote
 
 import mcp.types
@@ -25,6 +27,12 @@ from fastmcp.tools.tool import Tool
 from fastmcp.utilities.func_metadata import func_metadata
 from fastmcp.utilities.logging import get_logger
 
+if TYPE_CHECKING:
+    from mcp.server.session import ServerSessionT
+    from mcp.shared.context import LifespanContextT
+
+    from fastmcp.server import Context
+
 logger = get_logger(__name__)
 
 
@@ -33,12 +41,12 @@ def _proxy_passthrough():
 
 
 class ProxyTool(Tool):
-    def __init__(self, client: "Client", **kwargs):
+    def __init__(self, client: Client, **kwargs):
         super().__init__(**kwargs)
         self._client = client
 
     @classmethod
-    async def from_client(cls, client: "Client", tool: mcp.types.Tool) -> "ProxyTool":
+    async def from_client(cls, client: Client, tool: mcp.types.Tool) -> ProxyTool:
         return cls(
             client=client,
             name=tool.name,
@@ -50,7 +58,9 @@ class ProxyTool(Tool):
         )
 
     async def run(
-        self, arguments: dict[str, Any], context: Context | None = None
+        self,
+        arguments: dict[str, Any],
+        context: Context[ServerSessionT, LifespanContextT] | None = None,
     ) -> Any:
         # the client context manager will swallow any exceptions inside a TaskGroup
         # so we return the raw result and raise an exception ourselves
@@ -64,17 +74,15 @@ class ProxyTool(Tool):
 
 
 class ProxyResource(Resource):
-    def __init__(
-        self, client: "Client", *, _value: str | bytes | None = None, **kwargs
-    ):
+    def __init__(self, client: Client, *, _value: str | bytes | None = None, **kwargs):
         super().__init__(**kwargs)
         self._client = client
         self._value = _value
 
     @classmethod
     async def from_client(
-        cls, client: "Client", resource: mcp.types.Resource
-    ) -> "ProxyResource":
+        cls, client: Client, resource: mcp.types.Resource
+    ) -> ProxyResource:
         return cls(
             client=client,
             uri=resource.uri,
@@ -83,7 +91,9 @@ class ProxyResource(Resource):
             mime_type=resource.mimeType,
         )
 
-    async def read(self) -> str | bytes:
+    async def read(
+        self, context: Context[ServerSessionT, LifespanContextT] | None = None
+    ) -> str | bytes:
         if self._value is not None:
             return self._value
 
@@ -98,14 +108,14 @@ class ProxyResource(Resource):
 
 
 class ProxyTemplate(ResourceTemplate):
-    def __init__(self, client: "Client", **kwargs):
+    def __init__(self, client: Client, **kwargs):
         super().__init__(**kwargs)
         self._client = client
 
     @classmethod
     async def from_client(
-        cls, client: "Client", template: mcp.types.ResourceTemplate
-    ) -> "ProxyTemplate":
+        cls, client: Client, template: mcp.types.ResourceTemplate
+    ) -> ProxyTemplate:
         return cls(
             client=client,
             uri_template=template.uriTemplate,
@@ -115,7 +125,12 @@ class ProxyTemplate(ResourceTemplate):
             parameters={},
         )
 
-    async def create_resource(self, uri: str, params: dict[str, Any]) -> ProxyResource:
+    async def create_resource(
+        self,
+        uri: str,
+        params: dict[str, Any],
+        context: Context[ServerSessionT, LifespanContextT] | None = None,
+    ) -> ProxyResource:
         # dont use the provided uri, because it may not be the same as the
         # uri_template on the remote server.
         # quote params to ensure they are valid for the uri_template
@@ -144,14 +159,12 @@ class ProxyTemplate(ResourceTemplate):
 
 
 class ProxyPrompt(Prompt):
-    def __init__(self, client: "Client", **kwargs):
+    def __init__(self, client: Client, **kwargs):
         super().__init__(**kwargs)
         self._client = client
 
     @classmethod
-    async def from_client(
-        cls, client: "Client", prompt: mcp.types.Prompt
-    ) -> "ProxyPrompt":
+    async def from_client(cls, client: Client, prompt: mcp.types.Prompt) -> ProxyPrompt:
         return cls(
             client=client,
             name=prompt.name,
@@ -160,14 +173,18 @@ class ProxyPrompt(Prompt):
             fn=_proxy_passthrough,
         )
 
-    async def render(self, arguments: dict[str, Any]) -> list[Message]:
+    async def render(
+        self,
+        arguments: dict[str, Any],
+        context: Context[ServerSessionT, LifespanContextT] | None = None,
+    ) -> list[Message]:
         async with self._client:
             result = await self._client.get_prompt(self.name, arguments)
         return [Message(role=m.role, content=m.content) for m in result]
 
 
 class FastMCPProxy(FastMCP):
-    def __init__(self, client: "Client", **kwargs):
+    def __init__(self, client: Client, **kwargs):
         super().__init__(**kwargs)
         self.client = client
 

src/fastmcp/server/server.py

@@ -398,9 +398,10 @@ class FastMCP(Generic[LifespanResultT]):
         server.
         """
         if self._resource_manager.has_resource(uri):
-            resource = await self._resource_manager.get_resource(uri)
+            context = self.get_context()
+            resource = await self._resource_manager.get_resource(uri, context=context)
             try:
-                content = await resource.read()
+                content = await resource.read(context=context)
                 return [
                     ReadResourceContents(content=content, mime_type=resource.mime_type)
                 ]
@@ -424,7 +425,10 @@ class FastMCP(Generic[LifespanResultT]):
 
         """
         if self._prompt_manager.has_prompt(name):
-            messages = await self._prompt_manager.render_prompt(name, arguments)
+            context = self.get_context()
+            messages = await self._prompt_manager.render_prompt(
+                name, arguments=arguments or {}, context=context
+            )
             return GetPromptResult(messages=pydantic_core.to_jsonable_python(messages))
         else:
             for server in self._mounted_servers.values():
@@ -562,6 +566,10 @@ class FastMCP(Generic[LifespanResultT]):
         - bytes for binary content
         - other types will be converted to JSON
 
+        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 session information.
+
         If the URI contains parameters (e.g. "resource://{param}") or the function
         has parameters, it will be registered as a template resource.
 
@@ -586,6 +594,11 @@ class FastMCP(Generic[LifespanResultT]):
             def get_weather(city: str) -> str:
                 return f"Weather for {city}"
 
+            @server.resource("resource://{city}/weather")
+            def get_weather_with_context(city: str, ctx: Context) -> str:
+                ctx.info(f"Fetching weather for {city}")
+                return f"Weather for {city}"
+
             @server.resource("resource://{city}/weather")
             async def get_weather(city: str) -> str:
                 data = await fetch_weather(city)
@@ -639,6 +652,10 @@ class FastMCP(Generic[LifespanResultT]):
     ) -> Callable[[AnyFunction], AnyFunction]:
         """Decorator to register a prompt.
 
+        Prompts 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 session information.
+
         Args:
             name: Optional name for the prompt (defaults to function name)
             description: Optional description of what the prompt does
@@ -655,6 +672,17 @@ class FastMCP(Generic[LifespanResultT]):
                     }
                 ]
 
+            @server.prompt()
+            def analyze_with_context(table_name: str, ctx: Context) -> list[Message]:
+                ctx.info(f"Analyzing table {table_name}")
+                schema = read_table_schema(table_name)
+                return [
+                    {
+                        "role": "user",
+                        "content": f"Analyze this schema:\n{schema}"
+                    }
+                ]
+
             @server.prompt()
             async def analyze_file(path: str) -> list[Message]:
                 content = await read_file(path)

src/fastmcp/tools/tool.py

@@ -101,13 +101,16 @@ class Tool(BaseModel):
     ) -> list[TextContent | ImageContent | EmbeddedResource]:
         """Run the tool with arguments."""
         try:
-            result = await self.fn_metadata.call_fn_with_arg_validation(
-                self.fn,
-                self.is_async,
-                arguments,
+            pass_args = (
                 {self.context_kwarg: context}
                 if self.context_kwarg is not None
-                else None,
+                else None
+            )
+            result = await self.fn_metadata.call_fn_with_arg_validation(
+                fn=self.fn,
+                fn_is_async=self.is_async,
+                arguments_to_validate=arguments,
+                arguments_to_pass_directly=pass_args,
             )
             return _convert_to_content(result)
         except Exception as e:

tests/server/test_server_interactions.py

@@ -682,7 +682,147 @@ class TestToolParameters:
             assert result[0].text == "0:16:40"
 
 
-class TestResources:
+class TestToolContextInjection:
+    """Test context injection in tools."""
+
+    async def test_context_detection(self):
+        """Test that context parameters are properly detected."""
+        mcp = FastMCP()
+
+        def tool_with_context(x: int, ctx: Context) -> str:
+            return f"Request {ctx.request_id}: {x}"
+
+        mcp.add_tool(tool_with_context)
+        async with Client(mcp) as client:
+            tools = await client.list_tools()
+            assert len(tools) == 1
+            assert tools[0].name == "tool_with_context"
+
+    async def test_context_injection(self):
+        """Test that context is properly injected into tool calls."""
+        mcp = FastMCP()
+
+        @mcp.tool()
+        def tool_with_context(x: int, ctx: Context) -> str:
+            assert isinstance(ctx, Context)
+            assert ctx.request_id is not None
+            return ctx.request_id
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("tool_with_context", {"x": 42})
+            assert len(result) == 1
+            content = result[0]
+            assert isinstance(content, TextContent)
+            assert content.text == "1"
+
+    async def test_async_context(self):
+        """Test that context works in async functions."""
+        mcp = FastMCP()
+
+        async def async_tool(x: int, ctx: Context) -> str:
+            assert ctx.request_id is not None
+            return f"Async request {ctx.request_id}: {x}"
+
+        mcp.add_tool(async_tool)
+        async with Client(mcp) as client:
+            result = await client.call_tool("async_tool", {"x": 42})
+            assert len(result) == 1
+            content = result[0]
+            assert isinstance(content, TextContent)
+            assert "Async request" in content.text
+            assert "42" in content.text
+
+    async def test_context_logging(self):
+        from unittest.mock import patch
+
+        import mcp.server.session
+
+        """Test that context logging methods work."""
+        mcp = FastMCP()
+
+        async def logging_tool(msg: str, ctx: Context) -> str:
+            await ctx.debug("Debug message")
+            await ctx.info("Info message")
+            await ctx.warning("Warning message")
+            await ctx.error("Error message")
+            return f"Logged messages for {msg}"
+
+        mcp.add_tool(logging_tool)
+
+        with patch("mcp.server.session.ServerSession.send_log_message") as mock_log:
+            async with Client(mcp) as client:
+                result = await client.call_tool("logging_tool", {"msg": "test"})
+                assert len(result) == 1
+                content = result[0]
+                assert isinstance(content, TextContent)
+                assert "Logged messages for test" in content.text
+
+                assert mock_log.call_count == 4
+                mock_log.assert_any_call(
+                    level="debug", data="Debug message", logger=None
+                )
+                mock_log.assert_any_call(level="info", data="Info message", logger=None)
+                mock_log.assert_any_call(
+                    level="warning", data="Warning message", logger=None
+                )
+                mock_log.assert_any_call(
+                    level="error", data="Error message", logger=None
+                )
+
+    async def test_optional_context(self):
+        """Test that context is optional."""
+        mcp = FastMCP()
+
+        def no_context(x: int) -> int:
+            return x * 2
+
+        mcp.add_tool(no_context)
+        async with Client(mcp) as client:
+            result = await client.call_tool("no_context", {"x": 21})
+            assert len(result) == 1
+            content = result[0]
+            assert isinstance(content, TextContent)
+            assert content.text == "42"
+
+    async def test_context_resource_access(self):
+        """Test that context can access resources."""
+        mcp = FastMCP()
+
+        @mcp.resource("test://data")
+        def test_resource() -> str:
+            return "resource data"
+
+        @mcp.tool()
+        async def tool_with_resource(ctx: Context) -> str:
+            r_iter = await ctx.read_resource("test://data")
+            r_list = list(r_iter)
+            assert len(r_list) == 1
+            r = r_list[0]
+            return f"Read resource: {r.content} with mime type {r.mime_type}"
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("tool_with_resource", {})
+            assert len(result) == 1
+            content = result[0]
+            assert isinstance(content, TextContent)
+            assert "Read resource: resource data" in content.text
+
+    async def test_tool_decorator_with_tags(self):
+        """Test that the tool decorator properly sets tags."""
+        mcp = FastMCP()
+
+        @mcp.tool(tags={"example", "test-tag"})
+        def sample_tool(x: int) -> int:
+            return x * 2
+
+        # Verify the tool exists
+        async with Client(mcp) as client:
+            tools = await client.list_tools()
+            assert len(tools) == 1
+            # Note: MCPTool from the client API doesn't expose tags
+
+
+class TestResource:
     async def test_text_resource(self):
         mcp = FastMCP()
 
@@ -756,6 +896,21 @@ class TestResources:
             assert result[0].blob == base64.b64encode(b"Binary file data").decode()
 
 
+class TestResourceContext:
+    async def test_resource_with_context_annotation_gets_context(self):
+        mcp = FastMCP()
+
+        @mcp.resource("resource://test")
+        def resource_with_context(ctx: Context) -> str:
+            assert isinstance(ctx, Context)
+            return ctx.request_id
+
+        async with Client(mcp) as client:
+            result = await client.read_resource(AnyUrl("resource://test"))
+            assert isinstance(result[0], TextResourceContents)
+            assert result[0].text == "1"
+
+
 class TestResourceTemplates:
     async def test_resource_with_params_not_in_uri(self):
         """Test that a resource with function parameters raises an error if the URI
@@ -1026,144 +1181,19 @@ class TestResourceTemplates:
             assert result[0].text == "Template resource 1: a/b"
 
 
-class TestContextInjection:
-    """Test context injection in tools."""
-
-    async def test_context_detection(self):
-        """Test that context parameters are properly detected."""
+class TestResourceTemplateContext:
+    async def test_resource_template_context(self):
         mcp = FastMCP()
 
-        def tool_with_context(x: int, ctx: Context) -> str:
-            return f"Request {ctx.request_id}: {x}"
-
-        mcp.add_tool(tool_with_context)
-        async with Client(mcp) as client:
-            tools = await client.list_tools()
-            assert len(tools) == 1
-            assert tools[0].name == "tool_with_context"
-
-    async def test_context_injection(self):
-        """Test that context is properly injected into tool calls."""
-        mcp = FastMCP()
-
-        def tool_with_context(x: int, ctx: Context) -> str:
-            assert ctx.request_id is not None
-            return f"Request {ctx.request_id}: {x}"
-
-        mcp.add_tool(tool_with_context)
-        async with Client(mcp) as client:
-            result = await client.call_tool("tool_with_context", {"x": 42})
-            assert len(result) == 1
-            content = result[0]
-            assert isinstance(content, TextContent)
-            assert "Request" in content.text
-            assert "42" in content.text
-
-    async def test_async_context(self):
-        """Test that context works in async functions."""
-        mcp = FastMCP()
-
-        async def async_tool(x: int, ctx: Context) -> str:
-            assert ctx.request_id is not None
-            return f"Async request {ctx.request_id}: {x}"
-
-        mcp.add_tool(async_tool)
-        async with Client(mcp) as client:
-            result = await client.call_tool("async_tool", {"x": 42})
-            assert len(result) == 1
-            content = result[0]
-            assert isinstance(content, TextContent)
-            assert "Async request" in content.text
-            assert "42" in content.text
-
-    async def test_context_logging(self):
-        from unittest.mock import patch
-
-        import mcp.server.session
-
-        """Test that context logging methods work."""
-        mcp = FastMCP()
-
-        async def logging_tool(msg: str, ctx: Context) -> str:
-            await ctx.debug("Debug message")
-            await ctx.info("Info message")
-            await ctx.warning("Warning message")
-            await ctx.error("Error message")
-            return f"Logged messages for {msg}"
-
-        mcp.add_tool(logging_tool)
-
-        with patch("mcp.server.session.ServerSession.send_log_message") as mock_log:
-            async with Client(mcp) as client:
-                result = await client.call_tool("logging_tool", {"msg": "test"})
-                assert len(result) == 1
-                content = result[0]
-                assert isinstance(content, TextContent)
-                assert "Logged messages for test" in content.text
-
-                assert mock_log.call_count == 4
-                mock_log.assert_any_call(
-                    level="debug", data="Debug message", logger=None
-                )
-                mock_log.assert_any_call(level="info", data="Info message", logger=None)
-                mock_log.assert_any_call(
-                    level="warning", data="Warning message", logger=None
-                )
-                mock_log.assert_any_call(
-                    level="error", data="Error message", logger=None
-                )
-
-    async def test_optional_context(self):
-        """Test that context is optional."""
-        mcp = FastMCP()
-
-        def no_context(x: int) -> int:
-            return x * 2
-
-        mcp.add_tool(no_context)
-        async with Client(mcp) as client:
-            result = await client.call_tool("no_context", {"x": 21})
-            assert len(result) == 1
-            content = result[0]
-            assert isinstance(content, TextContent)
-            assert content.text == "42"
-
-    async def test_context_resource_access(self):
-        """Test that context can access resources."""
-        mcp = FastMCP()
-
-        @mcp.resource("test://data")
-        def test_resource() -> str:
-            return "resource data"
-
-        @mcp.tool()
-        async def tool_with_resource(ctx: Context) -> str:
-            r_iter = await ctx.read_resource("test://data")
-            r_list = list(r_iter)
-            assert len(r_list) == 1
-            r = r_list[0]
-            return f"Read resource: {r.content} with mime type {r.mime_type}"
-
-        async with Client(mcp) as client:
-            result = await client.call_tool("tool_with_resource", {})
-            assert len(result) == 1
-            content = result[0]
-            assert isinstance(content, TextContent)
-            assert "Read resource: resource data" in content.text
-
-    async def test_tool_decorator_with_tags(self):
-        """Test that the tool decorator properly sets tags."""
-        mcp = FastMCP()
-
-        @mcp.tool(tags={"example", "test-tag"})
-        def sample_tool(x: int) -> int:
-            return x * 2
+        @mcp.resource("resource://{param}")
+        def resource_template(param: str, ctx: Context) -> str:
+            assert isinstance(ctx, Context)
+            return f"Resource template: {param} {ctx.request_id}"
 
-        # Verify the tool exists
         async with Client(mcp) as client:
-            tools = await client.list_tools()
-            assert len(tools) == 1
-            # Note: MCPTool from the client API doesn't expose tags
+            result = await client.read_resource(AnyUrl("resource://test"))
+            assert isinstance(result[0], TextResourceContents)
+            assert result[0].text == "Resource template: test 1"
 
 
 class TestPrompts:
@@ -1350,3 +1380,19 @@ class TestPrompts:
         assert len(prompts_dict) == 1
         prompt = prompts_dict["sample_prompt"]
         assert prompt.tags == {"example", "test-tag"}
+
+
+class TestPromptContext:
+    async def test_prompt_context(self):
+        mcp = FastMCP()
+
+        @mcp.prompt()
+        def prompt_fn(name: str, ctx: Context) -> str:
+            assert isinstance(ctx, Context)
+            return f"Hello, {name}! {ctx.request_id}"
+
+        async with Client(mcp) as client:
+            result = await client.get_prompt("prompt_fn", {"name": "World"})
+            assert len(result) == 1
+            message = result[0]
+            assert message.role == "user"