ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-22 05:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #263 from jlowin/context

Browse source 
Support context injection in resources, templates, and prompts (like tools)
This commit is contained in:
Jeremiah Lowin 2025-04-25 21:15:53 -04:00 • committed by GitHub
commit 42872a74a8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 609 additions and 285 deletions
Download patch file Download diff file Expand all files Collapse all files

82
docs/servers/context.mdx
View file

@ -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.

31
docs/servers/prompts.mdx
View file

@ -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

54
docs/servers/resources.mdx
View file

@ -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

69
docs/servers/tools.mdx
View file

@ -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.

47
src/fastmcp/prompts/prompt.py
View file

@ -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

17
src/fastmcp/prompts/prompt_manager.py
View file

@ -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."""

14
src/fastmcp/resources/resource.py
View file

@ -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

25
src/fastmcp/resources/resource_manager.py
View file

@ -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}")

47
src/fastmcp/resources/template.py
View file

@ -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}")

66
src/fastmcp/resources/types.py
View file

@ -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)

22
src/fastmcp/server/openapi.py
View file

@ -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 = []

57
src/fastmcp/server/proxy.py
View file

@ -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

34
src/fastmcp/server/server.py
View file

@ -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)

13
src/fastmcp/tools/tool.py
View file

@ -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:

316
tests/server/test_server_interactions.py
View file

@ -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"