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

Merge pull request #347 from jlowin/context

Browse source 
Context refactor
This commit is contained in:
Jeremiah Lowin 2025-05-06 21:40:34 -04:00 • committed by GitHub
commit 1c1ded03c7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
22 changed files with 479 additions and 598 deletions
Download patch file Download diff file Expand all files Collapse all files

133
docs/servers/context.mdx
View file

@ -23,6 +23,17 @@ The `Context` object provides a clean interface to access MCP features within yo
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.
**Key Points:**
- 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; it will not be exposed to MCP clients as a valid parameter.
- The context is optional - functions that don't need it can omit the parameter entirely.
- Context methods are async, so your function usually needs to be async as well.
- The type hint can be a union (`Context | None`) or use `Annotated[]` and it will still work properly.
- Context is only available during a request; attempting to use context methods outside a request will raise errors. If you need to debug or call your context methods outside of a request, you can type your variable as `Context | None=None` to avoid missing argument errors.
### Tools
```python
from fastmcp import FastMCP, Context
@ -31,43 +42,40 @@ mcp = FastMCP(name="ContextDemo")
@mcp.tool()
async def process_file(file_uri: str, ctx: Context) -> str:
"""Processes a file, using context for logging and resource access."""
request_id = ctx.request_id
await ctx.info(f"[{request_id}] Starting processing for {file_uri}")
try:
# Use context to read a resource
contents_list = await ctx.read_resource(file_uri)
if not contents_list:
await ctx.warning(f"Resource {file_uri} is empty.")
return "Resource empty"
data = contents_list[0].content # Assuming TextResourceContents
await ctx.debug(f"Read {len(data)} bytes from {file_uri}")
# Report progress
await ctx.report_progress(progress=50, total=100)
# Simulate work
processed_data = data.upper() # Example processing
await ctx.report_progress(progress=100, total=100)
await ctx.info(f"Processing complete for {file_uri}")
return f"Processed data length: {len(processed_data)}"
except Exception as e:
# Use context to log errors
await ctx.error(f"Error processing {file_uri}: {str(e)}")
raise # Re-raise to send error back to client
# Context is available as the ctx parameter
return "Processed file"
```
**Key Points:**
### Resources and Templates
<VersionBadge version="2.2.5" />
```python
@mcp.resource("resource://user-data")
async def get_user_data(ctx: Context) -> dict:
"""Fetch personalized user data based on the request context."""
# Context is available as the ctx parameter
return {"user_id": "example"}
@mcp.resource("resource://users/{user_id}/profile")
async def get_user_profile(user_id: str, ctx: Context) -> dict:
"""Fetch user profile with context-aware logging."""
# Context is available as the ctx parameter
return {"id": user_id}
```
### Prompts
<VersionBadge version="2.2.5" />
```python
@mcp.prompt()
async def data_analysis_request(dataset: str, ctx: Context) -> str:
"""Generate a request to analyze data with contextual information."""
# Context is available as the ctx parameter
return f"Please analyze the following dataset: {dataset}"
```
- 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 - 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
@ -305,60 +313,3 @@ async def handle_web_request(ctx: Context) -> dict:
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 Different Components
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.

3
src/fastmcp/__init__.py
View file

@ -2,9 +2,10 @@
from importlib.metadata import version
from fastmcp.server.server import FastMCP
from fastmcp.server.context import Context
import fastmcp.server
from fastmcp.client import Client
from fastmcp.utilities.types import Image
from . import client, settings

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

@ -12,6 +12,7 @@ from mcp.types import Prompt as MCPPrompt
from mcp.types import PromptArgument as MCPPromptArgument
from pydantic import BaseModel, BeforeValidator, Field, TypeAdapter, validate_call
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.json_schema import prune_params
from fastmcp.utilities.types import (
_convert_set_defaults,
@ -20,10 +21,7 @@ from fastmcp.utilities.types import (
)
if TYPE_CHECKING:
from mcp.server.session import ServerSessionT
from mcp.shared.context import LifespanContextT
from fastmcp.server import Context
pass
CONTENT_TYPES = TextContent | ImageContent | EmbeddedResource
@ -76,9 +74,6 @@ 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(
@ -87,7 +82,6 @@ class Prompt(BaseModel):
name: str | None = None,
description: str | None = None,
tags: set[str] | None = None,
context_kwarg: str | None = None,
) -> Prompt:
"""Create a Prompt from a function.
@ -97,7 +91,7 @@ class Prompt(BaseModel):
- A dict (converted to a message)
- A sequence of any of the above
"""
from fastmcp import Context
from fastmcp.server.context import Context
func_name = name or fn.__name__
@ -115,8 +109,8 @@ class Prompt(BaseModel):
parameters = type_adapter.json_schema()
# Auto-detect context parameter if not provided
if context_kwarg is None:
context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context)
context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context)
if context_kwarg:
parameters = prune_params(parameters, params=[context_kwarg])
@ -141,15 +135,15 @@ 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,
context: Context[ServerSessionT, LifespanContextT] | None = None,
) -> list[PromptMessage]:
"""Render the prompt with arguments."""
from fastmcp.server.context import Context
# Validate required arguments
if self.arguments:
required = {arg.name for arg in self.arguments if arg.required}
@ -161,8 +155,9 @@ class Prompt(BaseModel):
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
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg and context_kwarg not in kwargs:
kwargs[context_kwarg] = get_context()
# Call function and check if result is a coroutine
result = self.fn(**kwargs)

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

@ -13,10 +13,7 @@ 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
pass
logger = get_logger(__name__)
@ -82,19 +79,15 @@ class PromptManager:
self,
name: str,
arguments: dict[str, Any] | None = None,
context: Context[ServerSessionT, LifespanContextT] | None = None,
) -> GetPromptResult:
"""Render a prompt by name with arguments."""
prompt = self.get_prompt(name)
if not prompt:
raise NotFoundError(f"Unknown prompt: {name}")
messages = await prompt.render(arguments, context=context)
messages = await prompt.render(arguments)
return GetPromptResult(
description=prompt.description,
messages=messages,
)
return GetPromptResult(description=prompt.description, messages=messages)
def has_prompt(self, key: str) -> bool:
"""Check if a prompt exists."""

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

@ -20,10 +20,7 @@ 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
pass
class Resource(BaseModel, abc.ABC):
@ -66,9 +63,7 @@ class Resource(BaseModel, abc.ABC):
raise ValueError("Either name or uri must be provided")
@abc.abstractmethod
async def read(
self, context: Context[ServerSessionT, LifespanContextT] | None = None
) -> str | bytes:
async def read(self) -> str | bytes:
"""Read the resource content."""
pass

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

@ -109,7 +109,7 @@ class ResourceManager:
The added resource. If a resource with the same URI already exists,
returns the existing resource.
"""
resource = FunctionResource.from_function(
resource = FunctionResource(
fn=fn,
uri=AnyUrl(uri),
name=name,
@ -219,12 +219,11 @@ class ResourceManager:
return True
return False
async def get_resource(self, uri: AnyUrl | str, context=None) -> Resource:
async def get_resource(self, uri: AnyUrl | str) -> 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.
@ -244,7 +243,6 @@ class ResourceManager:
return await template.create_resource(
uri_str,
params=params,
context=context,
)
except Exception as e:
raise ValueError(f"Error creating resource from template: {e}")

35
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 TYPE_CHECKING, Annotated, Any
from typing import Annotated, Any
from urllib.parse import unquote
from mcp.types import ResourceTemplate as MCPResourceTemplate
@ -20,17 +20,12 @@ from pydantic import (
)
from fastmcp.resources.types import FunctionResource, Resource
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.types import (
_convert_set_defaults,
find_kwarg_by_type,
)
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)
@ -79,9 +74,6 @@ 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
@ -100,10 +92,9 @@ 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
from fastmcp.server.context import Context
func_name = name or fn.__name__
if func_name == "<lambda>":
@ -119,8 +110,8 @@ class ResourceTemplate(BaseModel):
)
# Auto-detect context parameter if not provided
if context_kwarg is None:
context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context)
context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context)
# Validate that URI params match function params
uri_params = set(re.findall(r"{(\w+)(?:\*)?}", uri_template))
@ -170,25 +161,22 @@ 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],
context: Context[ServerSessionT, LifespanContextT] | None = None,
) -> Resource:
async def create_resource(self, uri: str, params: dict[str, Any]) -> Resource:
"""Create a resource from the template with the given parameters."""
from fastmcp.server.context import Context
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
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg and context_kwarg not in kwargs:
kwargs[context_kwarg] = get_context()
# Call function and check if result is a coroutine
result = self.fn(**kwargs)
@ -202,7 +190,6 @@ class ResourceTemplate(BaseModel):
mime_type=self.mime_type,
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}")

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

@ -15,14 +15,12 @@ import pydantic.json
import pydantic_core
from pydantic import Field, ValidationInfo
import fastmcp
from fastmcp.resources.resource import Resource
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.types import find_kwarg_by_type
if TYPE_CHECKING:
from mcp.server.session import ServerSessionT
from mcp.shared.context import LifespanContextT
from fastmcp.server import Context
pass
class TextResource(Resource):
@ -30,9 +28,7 @@ class TextResource(Resource):
text: str = Field(description="Text content of the resource")
async def read(
self, context: Context[ServerSessionT, LifespanContextT] | None = None
) -> str:
async def read(self) -> str:
"""Read the text content."""
return self.text
@ -42,9 +38,7 @@ class BinaryResource(Resource):
data: bytes = Field(description="Binary content of the resource")
async def read(
self, context: Context[ServerSessionT, LifespanContextT] | None = None
) -> bytes:
async def read(self) -> bytes:
"""Read the binary content."""
return self.data
@ -63,40 +57,23 @@ class FunctionResource(Resource):
"""
fn: Callable[[], Any]
context_kwarg: str | None = Field(
default=None, description="Name of the kwarg that should receive context"
)
@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:
async def read(self) -> str | bytes:
"""Read the resource by calling the wrapped function."""
from fastmcp.server.context import Context
try:
kwargs = {}
if self.context_kwarg is not None:
kwargs[self.context_kwarg] = context
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg is not None:
kwargs[context_kwarg] = get_context()
result = self.fn(**kwargs)
if inspect.iscoroutinefunction(self.fn):
result = await result
if isinstance(result, Resource):
return await result.read(context=context)
return await result.read()
elif isinstance(result, bytes):
return result
elif isinstance(result, str):
@ -140,9 +117,7 @@ class FileResource(Resource):
mime_type = info.data.get("mime_type", "text/plain")
return not mime_type.startswith("text/")
async def read(
self, context: Context[ServerSessionT, LifespanContextT] | None = None
) -> str | bytes:
async def read(self) -> str | bytes:
"""Read the file content."""
try:
if self.is_binary:
@ -160,9 +135,7 @@ class HttpResource(Resource):
default="application/json", description="MIME type of the resource content"
)
async def read(
self, context: Context[ServerSessionT, LifespanContextT] | None = None
) -> str | bytes:
async def read(self) -> str | bytes:
"""Read the HTTP content."""
async with httpx.AsyncClient() as client:
response = await client.get(self.url)
@ -214,9 +187,7 @@ class DirectoryResource(Resource):
except Exception as e:
raise ValueError(f"Error listing directory {self.path}: {e}")
async def read(
self, context: Context[ServerSessionT, LifespanContextT] | None = None
) -> str: # Always returns JSON string
async def read(self) -> str: # Always returns JSON string
"""Read the directory listing."""
try:
files = await anyio.to_thread.run_sync(self.list_files)

1
src/fastmcp/server/__init__.py
View file

@ -1,5 +1,6 @@
from .server import FastMCP
from .context import Context
from . import dependencies
__all__ = ["FastMCP", "Context"]

78
src/fastmcp/server/context.py
View file

@ -1,11 +1,13 @@
from __future__ import annotations as _annotations
from typing import Any, Generic
from collections.abc import Generator
from contextlib import contextmanager
from contextvars import ContextVar, Token
from dataclasses import dataclass
from mcp import LoggingLevel
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.session import ServerSessionT
from mcp.shared.context import LifespanContextT, RequestContext
from mcp.shared.context import RequestContext
from mcp.types import (
CreateMessageResult,
ImageContent,
@ -13,18 +15,29 @@ from mcp.types import (
SamplingMessage,
TextContent,
)
from pydantic import BaseModel, ConfigDict
from pydantic.networks import AnyUrl
from starlette.requests import Request
import fastmcp.server.dependencies
from fastmcp.server.server import FastMCP
from fastmcp.utilities.http import get_current_starlette_request
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
_current_context: ContextVar[Context | None] = ContextVar("context", default=None)
class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
@contextmanager
def set_context(context: Context) -> Generator[Context, None, None]:
token = _current_context.set(context)
try:
yield context
finally:
_current_context.reset(token)
@dataclass
class Context:
"""Context object providing access to MCP capabilities.
This provides a cleaner interface to MCP's RequestContext functionality.
@ -56,37 +69,30 @@ class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
The context parameter name can be anything as long as it's annotated with Context.
The context is optional - tools that don't need it can omit the parameter.
"""
_request_context: RequestContext[ServerSessionT, LifespanContextT] | None
_fastmcp: FastMCP | None
def __init__(self, fastmcp: FastMCP):
self.fastmcp = fastmcp
self._tokens: list[Token] = []
model_config = ConfigDict(arbitrary_types_allowed=True)
def __enter__(self) -> Context:
"""Enter the context manager and set this context as the current context."""
# Always set this context and save the token
token = _current_context.set(self)
self._tokens.append(token)
return self
def __init__(
self,
*,
request_context: RequestContext[ServerSessionT, LifespanContextT] | None = None,
fastmcp: FastMCP | None = None,
**kwargs: Any,
):
super().__init__(**kwargs)
self._request_context = request_context
self._fastmcp = fastmcp
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""Exit the context manager and reset the most recent token."""
if self._tokens:
token = self._tokens.pop()
_current_context.reset(token)
@property
def fastmcp(self) -> FastMCP:
"""Access to the FastMCP server."""
if self._fastmcp is None:
raise ValueError("Context is not available outside of a request")
return self._fastmcp
@property
def request_context(self) -> RequestContext[ServerSessionT, LifespanContextT]:
def request_context(self) -> RequestContext:
"""Access to the underlying request context."""
if self._request_context is None:
raise ValueError("Context is not available outside of a request")
return self._request_context
return self.fastmcp._mcp_server.request_context
async def report_progress(
self, progress: float, total: float | None = None
@ -120,10 +126,8 @@ class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
Returns:
The resource content as either text or bytes
"""
assert self._fastmcp is not None, (
"Context is not available outside of a request"
)
return await self._fastmcp._mcp_read_resource(uri)
assert self.fastmcp is not None, "Context is not available outside of a request"
return await self.fastmcp._mcp_read_resource(uri)
async def log(
self,
@ -229,7 +233,5 @@ class Context(BaseModel, Generic[ServerSessionT, LifespanContextT]):
def get_http_request(self) -> Request:
"""Get the active starlette request."""
request = get_current_starlette_request()
if request is None:
raise ValueError("Request is not available outside a Starlette request")
return request
return fastmcp.server.dependencies.get_http_request()

35
src/fastmcp/server/dependencies.py Normal file
View file

@ -0,0 +1,35 @@
from __future__ import annotations
from typing import TYPE_CHECKING, ParamSpec, TypeVar
from starlette.requests import Request
if TYPE_CHECKING:
from fastmcp.server.context import Context
P = ParamSpec("P")
R = TypeVar("R")
# --- Context ---
def get_context() -> Context:
from fastmcp.server.context import _current_context
context = _current_context.get()
if context is None:
raise RuntimeError("No active context found.")
return context
# --- HTTP Request ---
def get_http_request() -> Request:
from fastmcp.server.http import _current_http_request
request = _current_http_request.get()
if request is None:
raise RuntimeError("No active HTTP request found.")
return request

38
src/fastmcp/server/http.py Normal file
View file

@ -0,0 +1,38 @@
from __future__ import annotations
from collections.abc import Generator
from contextlib import contextmanager
from contextvars import ContextVar
from starlette.requests import Request
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
_current_http_request: ContextVar[Request | None] = ContextVar(
"http_request",
default=None,
)
@contextmanager
def set_http_request(request: Request) -> Generator[Request, None, None]:
token = _current_http_request.set(request)
try:
yield request
finally:
_current_http_request.reset(token)
class RequestContextMiddleware:
"""
Middleware that stores each request in a ContextVar
"""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
with set_http_request(Request(scope)):
await self.app(scope, receive, send)

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

@ -25,9 +25,6 @@ from fastmcp.utilities.openapi import (
)
if TYPE_CHECKING:
from mcp.server.session import ServerSessionT
from mcp.shared.context import LifespanContextT
from fastmcp.server import Context
logger = get_logger(__name__)
@ -132,7 +129,6 @@ class OpenAPITool(Tool):
description=description,
parameters=parameters,
fn=self._execute_request, # We'll use an instance method instead of a global function
context_kwarg="context", # Default context keyword argument
tags=tags,
annotations=annotations,
serializer=serializer,
@ -258,12 +254,10 @@ class OpenAPITool(Tool):
raise ValueError(f"Request error: {str(e)}")
async def run(
self,
arguments: dict[str, Any],
context: Context[ServerSessionT, LifespanContextT] | None = None,
self, arguments: dict[str, Any]
) -> list[TextContent | ImageContent | EmbeddedResource]:
"""Run the tool with arguments and optional context."""
response = await self._execute_request(**arguments, context=context)
response = await self._execute_request(**arguments)
return _convert_to_content(response)
@ -292,9 +286,7 @@ class OpenAPIResource(Resource):
self._route = route
self._timeout = timeout
async def read(
self, context: Context[ServerSessionT, LifespanContextT] | None = None
) -> str | bytes:
async def read(self) -> str | bytes:
"""Fetch the resource data by making an HTTP request."""
try:
# Extract path parameters from the URI if present
@ -399,7 +391,6 @@ class OpenAPIResourceTemplate(ResourceTemplate):
fn=lambda **kwargs: None,
parameters=parameters,
tags=tags,
context_kwarg=None,
)
self._client = client
self._route = route
@ -409,7 +400,7 @@ class OpenAPIResourceTemplate(ResourceTemplate):
self,
uri: str,
params: dict[str, Any],
context: Context[ServerSessionT, LifespanContextT] | None = None,
context: Context | None = None,
) -> Resource:
"""Create a resource with the given parameters."""
# Generate a URI for this resource instance
@ -650,7 +641,5 @@ class FastMCPOpenAPI(FastMCP):
async def _mcp_call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
"""Override the call_tool method to return the raw result without converting to content."""
context = self.get_context()
result = await self._tool_manager.call_tool(name, arguments, context=context)
result = await self._tool_manager.call_tool(name, arguments)
return result

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

@ -27,9 +27,6 @@ from fastmcp.tools.tool import Tool
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__)
@ -57,7 +54,7 @@ class ProxyTool(Tool):
async def run(
self,
arguments: dict[str, Any],
context: Context[ServerSessionT, LifespanContextT] | None = None,
context: Context | None = None,
) -> list[TextContent | ImageContent | EmbeddedResource]:
# the client context manager will swallow any exceptions inside a TaskGroup
# so we return the raw result and raise an exception ourselves
@ -89,9 +86,7 @@ class ProxyResource(Resource):
mime_type=resource.mimeType,
)
async def read(
self, context: Context[ServerSessionT, LifespanContextT] | None = None
) -> str | bytes:
async def read(self) -> str | bytes:
if self._value is not None:
return self._value
@ -127,7 +122,7 @@ class ProxyTemplate(ResourceTemplate):
self,
uri: str,
params: dict[str, Any],
context: Context[ServerSessionT, LifespanContextT] | None = None,
context: Context | None = None,
) -> ProxyResource:
# dont use the provided uri, because it may not be the same as the
# uri_template on the remote server.
@ -171,11 +166,7 @@ class ProxyPrompt(Prompt):
fn=_proxy_passthrough,
)
async def render(
self,
arguments: dict[str, Any],
context: Context[ServerSessionT, LifespanContextT] | None = None,
) -> list[PromptMessage]:
async def render(self, arguments: dict[str, Any]) -> list[PromptMessage]:
async with self._client:
result = await self._client.get_prompt(self.name, arguments)
return result.messages

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

@ -25,7 +25,6 @@ from mcp.server.auth.provider import OAuthAuthorizationServerProvider
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.lowlevel.server import LifespanResultT
from mcp.server.lowlevel.server import Server as MCPServer
from mcp.server.session import ServerSession
from mcp.server.sse import SseServerTransport
from mcp.server.stdio import stdio_server
from mcp.types import (
@ -49,120 +48,27 @@ from starlette.responses import Response
from starlette.routing import Mount, Route
from starlette.types import Receive, Scope, Send
import fastmcp
import fastmcp.server
import fastmcp.settings
from fastmcp.exceptions import NotFoundError, ResourceError
from fastmcp.prompts import Prompt, PromptManager
from fastmcp.prompts.prompt import PromptResult
from fastmcp.resources import Resource, ResourceManager
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.http import RequestContextMiddleware
from fastmcp.tools import ToolManager
from fastmcp.tools.tool import Tool
from fastmcp.utilities.cache import TimedCache
from fastmcp.utilities.decorators import DecoratedFunction
from fastmcp.utilities.http import RequestMiddleware
from fastmcp.utilities.logging import configure_logging, get_logger
if TYPE_CHECKING:
from fastmcp.client import Client
from fastmcp.server.context import Context
from fastmcp.server.openapi import FastMCPOpenAPI
from fastmcp.server.proxy import FastMCPProxy
logger = get_logger(__name__)
NOT_FOUND = object()
class MountedServer:
def __init__(
self,
prefix: str,
server: FastMCP,
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
):
if tool_separator is None:
tool_separator = "_"
if resource_separator is None:
resource_separator = "+"
if prompt_separator is None:
prompt_separator = "_"
_validate_resource_prefix(f"{prefix}{resource_separator}")
self.server = server
self.prefix = prefix
self.tool_separator = tool_separator
self.resource_separator = resource_separator
self.prompt_separator = prompt_separator
async def get_tools(self) -> dict[str, Tool]:
tools = await self.server.get_tools()
return {
f"{self.prefix}{self.tool_separator}{key}": tool
for key, tool in tools.items()
}
async def get_resources(self) -> dict[str, Resource]:
resources = await self.server.get_resources()
return {
f"{self.prefix}{self.resource_separator}{key}": resource
for key, resource in resources.items()
}
async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
templates = await self.server.get_resource_templates()
return {
f"{self.prefix}{self.resource_separator}{key}": template
for key, template in templates.items()
}
async def get_prompts(self) -> dict[str, Prompt]:
prompts = await self.server.get_prompts()
return {
f"{self.prefix}{self.prompt_separator}{key}": prompt
for key, prompt in prompts.items()
}
def match_tool(self, key: str) -> bool:
return key.startswith(f"{self.prefix}{self.tool_separator}")
def strip_tool_prefix(self, key: str) -> str:
return key.removeprefix(f"{self.prefix}{self.tool_separator}")
def match_resource(self, key: str) -> bool:
return key.startswith(f"{self.prefix}{self.resource_separator}")
def strip_resource_prefix(self, key: str) -> str:
return key.removeprefix(f"{self.prefix}{self.resource_separator}")
def match_prompt(self, key: str) -> bool:
return key.startswith(f"{self.prefix}{self.prompt_separator}")
def strip_prompt_prefix(self, key: str) -> str:
return key.removeprefix(f"{self.prefix}{self.prompt_separator}")
class TimedCache:
def __init__(self, expiration: datetime.timedelta):
self.expiration = expiration
self.cache: dict[Any, tuple[Any, datetime.datetime]] = {}
def set(self, key: Any, value: Any) -> None:
expires = datetime.datetime.now() + self.expiration
self.cache[key] = (value, expires)
def get(self, key: Any) -> Any:
value = self.cache.get(key)
if value is not None and value[1] > datetime.datetime.now():
return value[0]
else:
return NOT_FOUND
def clear(self) -> None:
self.cache.clear()
@asynccontextmanager
async def default_lifespan(server: FastMCP) -> AsyncIterator[Any]:
@ -309,23 +215,9 @@ class FastMCP(Generic[LifespanResultT]):
self._mcp_server.get_prompt()(self._mcp_get_prompt)
self._mcp_server.list_resource_templates()(self._mcp_list_resource_templates)
def get_context(self) -> Context[ServerSession, LifespanResultT]:
"""
Returns a Context object. Note that the context will only be valid
during a request; outside a request, most methods will error.
"""
try:
request_context = self._mcp_server.request_context
except LookupError:
request_context = None
from fastmcp.server.context import Context
return Context(request_context=request_context, fastmcp=self)
async def get_tools(self) -> dict[str, Tool]:
"""Get all registered tools, indexed by registered key."""
if (tools := self._cache.get("tools")) is NOT_FOUND:
if (tools := self._cache.get("tools")) is self._cache.NOT_FOUND:
tools = {}
for server in self._mounted_servers.values():
server_tools = await server.get_tools()
@ -336,7 +228,7 @@ class FastMCP(Generic[LifespanResultT]):
async def get_resources(self) -> dict[str, Resource]:
"""Get all registered resources, indexed by registered key."""
if (resources := self._cache.get("resources")) is NOT_FOUND:
if (resources := self._cache.get("resources")) is self._cache.NOT_FOUND:
resources = {}
for server in self._mounted_servers.values():
server_resources = await server.get_resources()
@ -347,7 +239,9 @@ class FastMCP(Generic[LifespanResultT]):
async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
"""Get all registered resource templates, indexed by registered key."""
if (templates := self._cache.get("resource_templates")) is NOT_FOUND:
if (
templates := self._cache.get("resource_templates")
) is self._cache.NOT_FOUND:
templates = {}
for server in self._mounted_servers.values():
server_templates = await server.get_resource_templates()
@ -360,7 +254,7 @@ class FastMCP(Generic[LifespanResultT]):
"""
List all available prompts.
"""
if (prompts := self._cache.get("prompts")) is NOT_FOUND:
if (prompts := self._cache.get("prompts")) is self._cache.NOT_FOUND:
prompts = {}
for server in self._mounted_servers.values():
server_prompts = await server.get_prompts()
@ -458,43 +352,46 @@ class FastMCP(Generic[LifespanResultT]):
self, key: str, arguments: dict[str, Any]
) -> list[TextContent | ImageContent | EmbeddedResource]:
"""Call a tool by name with arguments."""
if self._tool_manager.has_tool(key):
context = self.get_context()
result = await self._tool_manager.call_tool(key, arguments, context=context)
else:
for server in self._mounted_servers.values():
if server.match_tool(key):
new_key = server.strip_tool_prefix(key)
result = await server.server._mcp_call_tool(new_key, arguments)
break
with fastmcp.server.context.Context(fastmcp=self):
if self._tool_manager.has_tool(key):
result = await self._tool_manager.call_tool(key, arguments)
else:
raise NotFoundError(f"Unknown tool: {key}")
return result
for server in self._mounted_servers.values():
if server.match_tool(key):
new_key = server.strip_tool_prefix(key)
result = await server.server._mcp_call_tool(new_key, arguments)
break
else:
raise NotFoundError(f"Unknown tool: {key}")
return result
async def _mcp_read_resource(self, uri: AnyUrl | str) -> list[ReadResourceContents]:
"""
Read a resource by URI, in the format expected by the low-level MCP
server.
"""
if self._resource_manager.has_resource(uri):
context = self.get_context()
resource = await self._resource_manager.get_resource(uri, context=context)
try:
content = await resource.read(context=context)
return [
ReadResourceContents(content=content, mime_type=resource.mime_type)
]
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
raise ResourceError(str(e))
else:
for server in self._mounted_servers.values():
if server.match_resource(str(uri)):
new_uri = server.strip_resource_prefix(str(uri))
return await server.server._mcp_read_resource(new_uri)
with fastmcp.server.context.Context(fastmcp=self):
if self._resource_manager.has_resource(uri):
resource = await self._resource_manager.get_resource(uri)
try:
content = await resource.read()
return [
ReadResourceContents(
content=content, mime_type=resource.mime_type
)
]
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
raise ResourceError(str(e))
else:
raise NotFoundError(f"Unknown resource: {uri}")
for server in self._mounted_servers.values():
if server.match_resource(str(uri)):
new_uri = server.strip_resource_prefix(str(uri))
return await server.server._mcp_read_resource(new_uri)
else:
raise NotFoundError(f"Unknown resource: {uri}")
async def _mcp_get_prompt(
self, name: str, arguments: dict[str, Any] | None = None
@ -504,19 +401,19 @@ class FastMCP(Generic[LifespanResultT]):
MCP server.
"""
if self._prompt_manager.has_prompt(name):
context = self.get_context()
prompt_result = await self._prompt_manager.render_prompt(
name, arguments=arguments or {}, context=context
)
return prompt_result
else:
for server in self._mounted_servers.values():
if server.match_prompt(name):
new_key = server.strip_prompt_prefix(name)
return await server.server._mcp_get_prompt(new_key, arguments)
with fastmcp.server.context.Context(fastmcp=self):
if self._prompt_manager.has_prompt(name):
prompt_result = await self._prompt_manager.render_prompt(
name, arguments=arguments or {}
)
return prompt_result
else:
raise NotFoundError(f"Unknown prompt: {name}")
for server in self._mounted_servers.values():
if server.match_prompt(name):
new_key = server.strip_prompt_prefix(name)
return await server.server._mcp_get_prompt(new_key, arguments)
else:
raise NotFoundError(f"Unknown prompt: {name}")
def add_tool(
self,
@ -827,10 +724,11 @@ class FastMCP(Generic[LifespanResultT]):
) -> None:
"""Run the server using SSE transport."""
uvicorn_config = uvicorn_config or {}
# the SSE app hangs even when a signal is sent, so we disable the timeout to make it possible to close immediately.
# see https://github.com/jlowin/fastmcp/issues/296
# the SSE app hangs even when a signal is sent, so we disable the
# timeout to make it possible to close immediately. see
# https://github.com/jlowin/fastmcp/issues/296
uvicorn_config.setdefault("timeout_graceful_shutdown", 0)
app = RequestMiddleware(self.sse_app())
app = RequestContextMiddleware(self.sse_app())
config = uvicorn.Config(
app,
@ -1145,3 +1043,74 @@ def _validate_resource_prefix(prefix: str) -> None:
raise ValueError(
f"Resource prefix or separator would result in an invalid resource URI: {e}"
)
class MountedServer:
def __init__(
self,
prefix: str,
server: FastMCP,
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
):
if tool_separator is None:
tool_separator = "_"
if resource_separator is None:
resource_separator = "+"
if prompt_separator is None:
prompt_separator = "_"
_validate_resource_prefix(f"{prefix}{resource_separator}")
self.server = server
self.prefix = prefix
self.tool_separator = tool_separator
self.resource_separator = resource_separator
self.prompt_separator = prompt_separator
async def get_tools(self) -> dict[str, Tool]:
tools = await self.server.get_tools()
return {
f"{self.prefix}{self.tool_separator}{key}": tool
for key, tool in tools.items()
}
async def get_resources(self) -> dict[str, Resource]:
resources = await self.server.get_resources()
return {
f"{self.prefix}{self.resource_separator}{key}": resource
for key, resource in resources.items()
}
async def get_resource_templates(self) -> dict[str, ResourceTemplate]:
templates = await self.server.get_resource_templates()
return {
f"{self.prefix}{self.resource_separator}{key}": template
for key, template in templates.items()
}
async def get_prompts(self) -> dict[str, Prompt]:
prompts = await self.server.get_prompts()
return {
f"{self.prefix}{self.prompt_separator}{key}": prompt
for key, prompt in prompts.items()
}
def match_tool(self, key: str) -> bool:
return key.startswith(f"{self.prefix}{self.tool_separator}")
def strip_tool_prefix(self, key: str) -> str:
return key.removeprefix(f"{self.prefix}{self.tool_separator}")
def match_resource(self, key: str) -> bool:
return key.startswith(f"{self.prefix}{self.resource_separator}")
def strip_resource_prefix(self, key: str) -> str:
return key.removeprefix(f"{self.prefix}{self.resource_separator}")
def match_prompt(self, key: str) -> bool:
return key.startswith(f"{self.prefix}{self.prompt_separator}")
def strip_prompt_prefix(self, key: str) -> str:
return key.removeprefix(f"{self.prefix}{self.prompt_separator}")

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

@ -12,6 +12,7 @@ from pydantic import BaseModel, BeforeValidator, Field
import fastmcp
from fastmcp.exceptions import ToolError
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.json_schema import prune_params
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
@ -22,10 +23,7 @@ from fastmcp.utilities.types import (
)
if TYPE_CHECKING:
from mcp.server.session import ServerSessionT
from mcp.shared.context import LifespanContextT
from fastmcp.server import Context
pass
logger = get_logger(__name__)
@ -41,9 +39,6 @@ class Tool(BaseModel):
name: str = Field(description="Name of the tool")
description: str = Field(description="Description of what the tool does")
parameters: dict[str, Any] = Field(description="JSON schema for tool parameters")
context_kwarg: str | None = Field(
None, description="Name of the kwarg that should receive context"
)
tags: Annotated[set[str], BeforeValidator(_convert_set_defaults)] = Field(
default_factory=set, description="Tags for the tool"
)
@ -60,13 +55,12 @@ class Tool(BaseModel):
fn: Callable[..., Any],
name: str | None = None,
description: str | None = None,
context_kwarg: str | None = None,
tags: set[str] | None = None,
annotations: ToolAnnotations | None = None,
serializer: Callable[[Any], str] | None = None,
) -> Tool:
"""Create a Tool from a function."""
from fastmcp import Context
from fastmcp.server.context import Context
# Reject functions with *args or **kwargs
sig = inspect.signature(fn)
@ -86,8 +80,7 @@ class Tool(BaseModel):
type_adapter = get_cached_typeadapter(fn)
schema = type_adapter.json_schema()
if context_kwarg is None:
context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context)
context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context)
if context_kwarg:
schema = prune_params(schema, params=[context_kwarg])
@ -96,25 +89,23 @@ class Tool(BaseModel):
name=func_name,
description=func_doc,
parameters=schema,
context_kwarg=context_kwarg,
tags=tags or set(),
annotations=annotations,
serializer=serializer,
)
async def run(
self,
arguments: dict[str, Any],
context: Context[ServerSessionT, LifespanContextT] | None = None,
self, arguments: dict[str, Any]
) -> list[TextContent | ImageContent | EmbeddedResource]:
"""Run the tool with arguments."""
from fastmcp.server.context import Context
arguments = arguments.copy()
try:
injected_args = (
{self.context_kwarg: context} if self.context_kwarg is not None else {}
)
parsed_args = arguments.copy()
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg and context_kwarg not in arguments:
arguments[context_kwarg] = get_context()
if fastmcp.settings.settings.tool_attempt_parse_json_args:
# Pre-parse data from JSON in order to handle cases like `["a", "b", "c"]`
@ -125,7 +116,7 @@ class Tool(BaseModel):
# which can be pre-parsed here.
signature = inspect.signature(self.fn)
for param_name in self.parameters["properties"]:
arg = parsed_args.get(param_name, None)
arg = arguments.get(param_name, None)
# if not in signature, we won't have annotations, so skip logic
if param_name not in signature.parameters:
continue
@ -140,13 +131,13 @@ class Tool(BaseModel):
):
continue
try:
parsed_args[param_name] = json.loads(arg)
arguments[param_name] = json.loads(arg)
except json.JSONDecodeError:
pass
type_adapter = get_cached_typeadapter(self.fn)
result = type_adapter.validate_python(parsed_args | injected_args)
result = type_adapter.validate_python(arguments)
if inspect.isawaitable(result):
result = await result

12
src/fastmcp/tools/tool_manager.py
View file

@ -3,7 +3,6 @@ from __future__ import annotations as _annotations
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from mcp.shared.context import LifespanContextT
from mcp.types import EmbeddedResource, ImageContent, TextContent, ToolAnnotations
from fastmcp.exceptions import NotFoundError
@ -12,9 +11,7 @@ from fastmcp.tools.tool import Tool
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
from mcp.server.session import ServerSessionT
from fastmcp.server import Context
pass
logger = get_logger(__name__)
@ -98,14 +95,11 @@ class ToolManager:
return tool
async def call_tool(
self,
key: str,
arguments: dict[str, Any],
context: Context[ServerSessionT, LifespanContextT] | None = None,
self, key: str, arguments: dict[str, Any]
) -> list[TextContent | ImageContent | EmbeddedResource]:
"""Call a tool by name with arguments."""
tool = self.get_tool(key)
if not tool:
raise NotFoundError(f"Unknown tool: {key}")
return await tool.run(arguments, context=context)
return await tool.run(arguments)

26
src/fastmcp/utilities/cache.py Normal file
View file

@ -0,0 +1,26 @@
import datetime
from typing import Any
UTC = datetime.timezone.utc
class TimedCache:
NOT_FOUND = object()
def __init__(self, expiration: datetime.timedelta):
self.expiration = expiration
self.cache: dict[Any, tuple[Any, datetime.datetime]] = {}
def set(self, key: Any, value: Any) -> None:
expires = datetime.datetime.now(UTC) + self.expiration
self.cache[key] = (value, expires)
def get(self, key: Any) -> Any:
value = self.cache.get(key)
if value is not None and value[1] > datetime.datetime.now(UTC):
return value[0]
else:
return self.NOT_FOUND
def clear(self) -> None:
self.cache.clear()

44
src/fastmcp/utilities/http.py
View file

@ -1,44 +0,0 @@
from __future__ import annotations
from contextlib import (
asynccontextmanager,
)
from contextvars import ContextVar
from starlette.requests import Request
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
_current_starlette_request: ContextVar[Request | None] = ContextVar(
"starlette_request",
default=None,
)
@asynccontextmanager
async def starlette_request_context(request: Request):
token = _current_starlette_request.set(request)
try:
yield
finally:
_current_starlette_request.reset(token)
def get_current_starlette_request() -> Request | None:
return _current_starlette_request.get()
class RequestMiddleware:
"""
Middleware that stores each request in a ContextVar
"""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
async with starlette_request_context(Request(scope)):
await self.app(scope, receive, send)

51
tests/prompts/test_prompt_manager.py
View file

@ -1,8 +1,6 @@
from typing import Annotated
import pytest
from mcp.server.session import ServerSessionT
from mcp.shared.context import LifespanContextT
from fastmcp import Context
from fastmcp.exceptions import NotFoundError
@ -308,38 +306,30 @@ class TestContextHandling:
def prompt_with_context(x: int, ctx: Context) -> str:
return str(x)
prompt = Prompt.from_function(prompt_with_context)
assert prompt.context_kwarg == "ctx"
Prompt.from_function(prompt_with_context)
def prompt_without_context(x: int) -> str:
return str(x)
prompt = Prompt.from_function(prompt_without_context)
assert prompt.context_kwarg is None
Prompt.from_function(prompt_without_context)
def test_parameterized_context_parameter_detection(self):
"""Test that parameterized context parameters are properly detected in
Prompt.from_function()."""
def prompt_with_context(
x: int, ctx: Context[ServerSessionT, LifespanContextT]
) -> str:
def prompt_with_context(x: int, ctx: Context) -> str:
return str(x)
prompt = Prompt.from_function(prompt_with_context)
assert prompt.context_kwarg == "ctx"
Prompt.from_function(prompt_with_context)
def test_parameterized_union_context_parameter_detection(self):
"""Test that context parameters in a union are properly detected in
Prompt.from_function()."""
def prompt_with_context(
x: int, ctx: Context[ServerSessionT, LifespanContextT] | None
) -> str:
def prompt_with_context(x: int, ctx: Context | None) -> str:
return str(x)
prompt = Prompt.from_function(prompt_with_context)
assert prompt.context_kwarg == "ctx"
Prompt.from_function(prompt_with_context)
async def test_context_injection(self):
"""Test that context is properly injected during prompt rendering."""
@ -349,17 +339,15 @@ class TestContextHandling:
return str(x)
prompt = Prompt.from_function(prompt_with_context)
assert prompt.context_kwarg == "ctx"
from fastmcp import FastMCP
mcp = FastMCP()
ctx = mcp.get_context()
context = Context(fastmcp=mcp)
with context:
messages = await prompt.render(arguments={"x": 42})
messages = await prompt.render(
arguments={"x": 42},
context=ctx,
)
assert len(messages) == 1
assert isinstance(messages[0].content, TextContent)
assert messages[0].content.text == "42"
@ -371,12 +359,18 @@ class TestContextHandling:
return str(x)
prompt = Prompt.from_function(prompt_with_context)
assert prompt.context_kwarg == "ctx"
# Should not raise an error when context is not provided
messages = await prompt.render(
arguments={"x": 42},
)
# Even for optional context, we need to provide a context
from fastmcp import FastMCP
mcp = FastMCP()
context = Context(fastmcp=mcp)
with context:
messages = await prompt.render(
arguments={"x": 42},
)
assert len(messages) == 1
assert isinstance(messages[0].content, TextContent)
assert messages[0].content.text == "42"
@ -388,5 +382,4 @@ class TestContextHandling:
def prompt_with_context(x: int, ctx: Annotated[Context, "ctx"]) -> str:
return str(x)
prompt = Prompt.from_function(prompt_with_context)
assert prompt.context_kwarg == "ctx"
Prompt.from_function(prompt_with_context)

54
tests/resources/test_resource_template.py
View file

@ -2,8 +2,6 @@ import json
from urllib.parse import quote
import pytest
from mcp.server.session import ServerSessionT
from mcp.shared.context import LifespanContextT
from pydantic import BaseModel
from fastmcp import Context
@ -560,54 +558,46 @@ class TestContextHandling:
def template_with_context(x: int, ctx: Context) -> str:
return str(x)
template = ResourceTemplate.from_function(
ResourceTemplate.from_function(
fn=template_with_context,
uri_template="test://{x}",
name="test",
)
assert template.context_kwarg == "ctx"
def template_without_context(x: int) -> str:
return str(x)
template = ResourceTemplate.from_function(
ResourceTemplate.from_function(
fn=template_without_context,
uri_template="test://{x}",
name="test",
)
assert template.context_kwarg is None
def test_parameterized_context_parameter_detection(self):
"""Test that parameterized context parameters are properly detected in
ResourceTemplate.from_function()."""
def template_with_context(
x: int, ctx: Context[ServerSessionT, LifespanContextT]
) -> str:
def template_with_context(x: int, ctx: Context) -> str:
return str(x)
template = ResourceTemplate.from_function(
ResourceTemplate.from_function(
fn=template_with_context,
uri_template="test://{x}",
name="test",
)
assert template.context_kwarg == "ctx"
def test_parameterized_union_context_parameter_detection(self):
"""Test that context parameters in a union are properly detected in
ResourceTemplate.from_function()."""
def template_with_context(
x: int, ctx: Context[ServerSessionT, LifespanContextT] | None
) -> str:
def template_with_context(x: int, ctx: Context | None) -> str:
return str(x)
template = ResourceTemplate.from_function(
ResourceTemplate.from_function(
fn=template_with_context,
uri_template="test://{x}",
name="test",
)
assert template.context_kwarg == "ctx"
async def test_context_injection(self):
"""Test that context is properly injected during resource creation."""
@ -621,18 +611,18 @@ class TestContextHandling:
uri_template="test://{x}",
name="test",
)
assert template.context_kwarg == "ctx"
from fastmcp import FastMCP
mcp = FastMCP()
ctx = mcp.get_context()
context = Context(fastmcp=mcp)
with context:
resource = await template.create_resource(
"test://42",
{"x": 42},
)
resource = await template.create_resource(
"test://42",
{"x": 42},
context=ctx,
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "42"
@ -648,13 +638,19 @@ class TestContextHandling:
uri_template="test://{x}",
name="test",
)
assert template.context_kwarg == "ctx"
# Should not raise an error when context is not provided
resource = await template.create_resource(
"test://42",
{"x": 42},
)
# Even for optional context, we need to provide a context
from fastmcp import FastMCP
mcp = FastMCP()
context = Context(fastmcp=mcp)
with context:
resource = await template.create_resource(
"test://42",
{"x": 42},
)
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "42"

105
tests/tools/test_tool_manager.py
View file

@ -5,8 +5,6 @@ from typing import Annotated, Any
import pydantic_core
import pytest
from mcp.server.session import ServerSessionT
from mcp.shared.context import LifespanContextT
from mcp.types import ImageContent, TextContent
from pydantic import BaseModel
@ -403,10 +401,20 @@ class TestCallTools:
manager = ToolManager()
manager.add_tool_from_fn(name_shrimp)
result = await manager.call_tool(
"name_shrimp",
{"tank": {"x": None, "shrimp": [{"name": "rex"}, {"name": "gertrude"}]}},
)
mcp = FastMCP()
context = Context(fastmcp=mcp)
with context:
result = await manager.call_tool(
"name_shrimp",
{
"tank": {
"x": None,
"shrimp": [{"name": "rex"}, {"name": "gertrude"}],
}
},
)
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
@ -498,14 +506,12 @@ class TestContextHandling:
return str(x)
manager = ToolManager()
tool = manager.add_tool_from_fn(tool_with_context)
assert tool.context_kwarg == "ctx"
manager.add_tool_from_fn(tool_with_context)
def tool_without_context(x: int) -> str:
return str(x)
tool = manager.add_tool_from_fn(tool_without_context)
assert tool.context_kwarg is None
manager.add_tool_from_fn(tool_without_context)
async def test_context_injection(self):
"""Test that context is properly injected during tool execution."""
@ -515,16 +521,17 @@ class TestContextHandling:
return str(x)
manager = ToolManager()
tool = manager.add_tool_from_fn(tool_with_context)
assert tool.context_kwarg == "ctx"
manager.add_tool_from_fn(tool_with_context)
mcp = FastMCP()
ctx = mcp.get_context()
result = await manager.call_tool("tool_with_context", {"x": 42}, context=ctx)
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
context = Context(fastmcp=mcp)
with context:
result = await manager.call_tool("tool_with_context", {"x": 42})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
async def test_context_injection_async(self):
"""Test that context is properly injected in async tools."""
@ -534,16 +541,17 @@ class TestContextHandling:
return str(x)
manager = ToolManager()
tool = manager.add_tool_from_fn(async_tool)
assert tool.context_kwarg == "ctx"
manager.add_tool_from_fn(async_tool)
mcp = FastMCP()
ctx = mcp.get_context()
result = await manager.call_tool("async_tool", {"x": 42}, context=ctx)
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
context = Context(fastmcp=mcp)
with context:
result = await manager.call_tool("async_tool", {"x": 42})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
async def test_context_optional(self):
"""Test that context is optional when calling tools."""
@ -553,48 +561,45 @@ class TestContextHandling:
return x
manager = ToolManager()
tool = manager.add_tool_from_fn(tool_with_context)
assert tool.context_kwarg == "ctx"
manager.add_tool_from_fn(tool_with_context)
# Should not raise an error when context is not provided
result = await manager.call_tool("tool_with_context", {"x": 42})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
mcp = FastMCP()
context = Context(fastmcp=mcp)
with context:
result = await manager.call_tool("tool_with_context", {"x": 42})
assert isinstance(result, list)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert result[0].text == "42"
def test_parameterized_context_parameter_detection(self):
"""Test that context parameters are properly detected in
Tool.from_function()."""
def tool_with_context(
x: int, ctx: Context[ServerSessionT, LifespanContextT]
) -> str:
def tool_with_context(x: int, ctx: Context) -> str:
return str(x)
manager = ToolManager()
tool = manager.add_tool_from_fn(tool_with_context)
assert tool.context_kwarg == "ctx"
manager.add_tool_from_fn(tool_with_context)
def test_annotated_context_parameter_detection(self):
def tool_with_context(x: int, ctx: Annotated[Context, "ctx"]) -> str:
return str(x)
manager = ToolManager()
tool = manager.add_tool_from_fn(tool_with_context)
assert tool.context_kwarg == "ctx"
manager.add_tool_from_fn(tool_with_context)
def test_parameterized_union_context_parameter_detection(self):
"""Test that context parameters are properly detected in
Tool.from_function()."""
def tool_with_context(
x: int, ctx: Context[ServerSessionT, LifespanContextT] | None
) -> str:
def tool_with_context(x: int, ctx: Context | None) -> str:
return str(x)
manager = ToolManager()
tool = manager.add_tool_from_fn(tool_with_context)
assert tool.context_kwarg == "ctx"
manager.add_tool_from_fn(tool_with_context)
async def test_context_error_handling(self):
"""Test error handling when context injection fails."""
@ -606,9 +611,13 @@ class TestContextHandling:
manager.add_tool_from_fn(tool_with_context)
mcp = FastMCP()
ctx = mcp.get_context()
with pytest.raises(ToolError, match="Error executing tool tool_with_context"):
await manager.call_tool("tool_with_context", {"x": 42}, context=ctx)
context = Context(fastmcp=mcp)
with context:
with pytest.raises(
ToolError, match="Error executing tool tool_with_context"
):
await manager.call_tool("tool_with_context", {"x": 42})
class TestCustomToolNames: