diff --git a/docs/servers/context.mdx b/docs/servers/context.mdx
index 685b833b0..203045184 100644
--- a/docs/servers/context.mdx
+++ b/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
-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.
-## 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.
\ No newline at end of file
+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}
+"""
+```
+
+
+
+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.
\ No newline at end of file
diff --git a/docs/servers/prompts.mdx b/docs/servers/prompts.mdx
index 5668846f4..9f276209a 100644
--- a/docs/servers/prompts.mdx
+++ b/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.
+
-```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
diff --git a/docs/servers/resources.mdx b/docs/servers/resources.mdx
index 6b1fe6bd2..b3e1c1dee 100644
--- a/docs/servers/resources.mdx
+++ b/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
+
+
+
+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
-
-
-
-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
diff --git a/docs/servers/tools.mdx b/docs/servers/tools.mdx
index 2c4fb7370..34cb00601 100644
--- a/docs/servers/tools.mdx
+++ b/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
-
-
-
-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
+
+
+
+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.
diff --git a/src/fastmcp/prompts/prompt.py b/src/fastmcp/prompts/prompt.py
index c6bcefc62..0cf37403d 100644
--- a/src/fastmcp/prompts/prompt.py
+++ b/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 == "":
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
diff --git a/src/fastmcp/prompts/prompt_manager.py b/src/fastmcp/prompts/prompt_manager.py
index a48ec91b9..db16d4398 100644
--- a/src/fastmcp/prompts/prompt_manager.py
+++ b/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."""
diff --git a/src/fastmcp/resources/resource.py b/src/fastmcp/resources/resource.py
index b79381bbd..e4d69638d 100644
--- a/src/fastmcp/resources/resource.py
+++ b/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
diff --git a/src/fastmcp/resources/resource_manager.py b/src/fastmcp/resources/resource_manager.py
index 613cd728b..d27f6247d 100644
--- a/src/fastmcp/resources/resource_manager.py
+++ b/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}")
diff --git a/src/fastmcp/resources/template.py b/src/fastmcp/resources/template.py
index 0d804f3da..d0ec24fbd 100644
--- a/src/fastmcp/resources/template.py
+++ b/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 == "":
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}")
diff --git a/src/fastmcp/resources/types.py b/src/fastmcp/resources/types.py
index 30d168bee..f7384a85a 100644
--- a/src/fastmcp/resources/types.py
+++ b/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)
diff --git a/src/fastmcp/server/openapi.py b/src/fastmcp/server/openapi.py
index 0911cae8a..b3396b368 100644
--- a/src/fastmcp/server/openapi.py
+++ b/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 = []
diff --git a/src/fastmcp/server/proxy.py b/src/fastmcp/server/proxy.py
index b3d9d8dce..e7cd35d68 100644
--- a/src/fastmcp/server/proxy.py
+++ b/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
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index 6f2167410..2983932a1 100644
--- a/src/fastmcp/server/server.py
+++ b/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)
diff --git a/src/fastmcp/tools/tool.py b/src/fastmcp/tools/tool.py
index 4b8ccee84..6f67e20a6 100644
--- a/src/fastmcp/tools/tool.py
+++ b/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:
diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py
index 506f1a65a..2653ea1a9 100644
--- a/tests/server/test_server_interactions.py
+++ b/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}"
+ @mcp.resource("resource://{param}")
+ def resource_template(param: str, ctx: Context) -> str:
+ assert isinstance(ctx, Context)
+ return f"Resource template: {param} {ctx.request_id}"
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
+ 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"