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

Merge branch 'main' into elicitation

Browse source
This commit is contained in:
Jeremiah Lowin 2025-06-28 08:26:55 -04:00
commit 11d2bacbb4
190 changed files with 14832 additions and 1888 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/fastmcp/cli/cli.py
View file

@ -235,7 +235,7 @@ def run(
typer.Option(
"--transport",
"-t",
help="Transport protocol to use (stdio, streamable-http, or sse)",
help="Transport protocol to use (stdio, http, or sse)",
),
] = None,
host: Annotated[

4
src/fastmcp/cli/run.py
View file

@ -4,14 +4,12 @@ import importlib.util
import re
import sys
from pathlib import Path
from typing import Any, Literal
from typing import Any
from fastmcp.utilities.logging import get_logger
logger = get_logger("cli.run")
TransportType = Literal["stdio", "streamable-http", "sse"]
def is_url(path: str) -> bool:
"""Check if a string is a URL."""

2
src/fastmcp/client/auth/oauth.py
View file

@ -80,7 +80,7 @@ class OAuthClientProvider(_MCPOAuthClientProvider):
ServerOAuthMetadata instead of the restrictive MCP OAuthMetadata.
"""
# Extract base URL per MCP spec
auth_base_url = self._get_authorization_base_url(server_url)
auth_base_url = self.context.get_authorization_base_url(server_url)
url = urljoin(auth_base_url, "/.well-known/oauth-authorization-server")
from mcp.types import LATEST_PROTOCOL_VERSION

84
src/fastmcp/client/client.py
View file

@ -1,6 +1,9 @@
from __future__ import annotations
import asyncio
import datetime
from contextlib import AsyncExitStack, asynccontextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Generic, Literal, cast, overload
@ -10,17 +13,16 @@ import mcp.types
import pydantic_core
from exceptiongroup import catch
from mcp import ClientSession
from mcp.types import ContentBlock
from pydantic import AnyUrl
import fastmcp
from fastmcp.client.elicitation import ElicitationHandler, create_elicitation_callback
from fastmcp.client.logging import (
LogHandler,
MessageHandler,
create_log_callback,
default_log_handler,
)
from fastmcp.client.messages import MessageHandler, MessageHandlerT
from fastmcp.client.progress import ProgressHandler, default_progress_handler
from fastmcp.client.roots import (
RootsHandler,
@ -31,7 +33,10 @@ from fastmcp.client.sampling import SamplingHandler, create_sampling_callback
from fastmcp.exceptions import ToolError
from fastmcp.server import FastMCP
from fastmcp.utilities.exceptions import get_catch_handlers
from fastmcp.utilities.json_schema_type import json_schema_to_type
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_config import MCPConfig
from fastmcp.utilities.types import get_cached_typeadapter
from .transports import (
ClientTransportT,
@ -58,6 +63,8 @@ __all__ = [
"ProgressHandler",
]
logger = get_logger(__name__)
class Client(Generic[ClientTransportT]):
"""
@ -101,34 +108,39 @@ class Client(Generic[ClientTransportT]):
cls,
transport: ClientTransportT,
**kwargs: Any,
) -> "Client[ClientTransportT]": ...
) -> Client[ClientTransportT]: ...
@overload
def __new__(
cls, transport: AnyUrl, **kwargs
) -> "Client[SSETransport|StreamableHttpTransport]": ...
) -> Client[SSETransport | StreamableHttpTransport]: ...
@overload
def __new__(
cls, transport: FastMCP | FastMCP1Server, **kwargs
) -> "Client[FastMCPTransport]": ...
) -> Client[FastMCPTransport]: ...
@overload
def __new__(
cls, transport: Path, **kwargs
) -> "Client[PythonStdioTransport|NodeStdioTransport]": ...
) -> Client[PythonStdioTransport | NodeStdioTransport]: ...
@overload
def __new__(
cls, transport: MCPConfig | dict[str, Any], **kwargs
) -> "Client[MCPConfigTransport]": ...
) -> Client[MCPConfigTransport]: ...
@overload
def __new__(
cls, transport: str, **kwargs
) -> "Client[PythonStdioTransport|NodeStdioTransport|SSETransport|StreamableHttpTransport]": ...
) -> Client[
PythonStdioTransport
| NodeStdioTransport
| SSETransport
| StreamableHttpTransport
]: ...
def __new__(cls, transport, **kwargs) -> "Client":
def __new__(cls, transport, **kwargs) -> Client:
instance = super().__new__(cls)
return instance
@ -146,7 +158,7 @@ class Client(Generic[ClientTransportT]):
sampling_handler: SamplingHandler | None = None,
elicitation_handler: ElicitationHandler | None = None,
log_handler: LogHandler | None = None,
message_handler: MessageHandler | None = None,
message_handler: MessageHandlerT | MessageHandler | None = None,
progress_handler: ProgressHandler | None = None,
timeout: datetime.timedelta | float | int | None = None,
init_timeout: datetime.timedelta | float | int | None = None,
@ -689,7 +701,8 @@ class Client(Generic[ClientTransportT]):
arguments: dict[str, Any] | None = None,
timeout: datetime.timedelta | float | int | None = None,
progress_handler: ProgressHandler | None = None,
) -> list[ContentBlock]:
raise_on_error: bool = True,
) -> CallToolResult:
"""Call a tool on the server.
Unlike call_tool_mcp, this method raises a ToolError if the tool call results in an error.
@ -701,8 +714,13 @@ class Client(Generic[ClientTransportT]):
progress_handler (ProgressHandler | None, optional): The progress handler to use for the tool call. Defaults to None.
Returns:
list[mcp.types.TextContent | mcp.types.ImageContent | mcp.types.AudioContent | mcp.types.EmbeddedResource]:
The content returned by the tool.
CallToolResult:
The content returned by the tool. If the tool returns structured
outputs, they are returned as a dataclass (if an output schema
is available) or a dictionary; otherwise, a list of content
blocks is returned. Note: to receive both structured and
unstructured outputs, use call_tool_mcp instead and access the
raw result object.
Raises:
ToolError: If the tool call results in an error.
@ -714,7 +732,43 @@ class Client(Generic[ClientTransportT]):
timeout=timeout,
progress_handler=progress_handler,
)
if result.isError:
data = None
if result.isError and raise_on_error:
msg = cast(mcp.types.TextContent, result.content[0]).text
raise ToolError(msg)
return result.content
elif result.structuredContent:
try:
if name not in self.session._tool_output_schemas:
await self.session.list_tools()
if name in self.session._tool_output_schemas:
output_schema = self.session._tool_output_schemas.get(name)
if output_schema:
if output_schema.get("x-fastmcp-wrap-result"):
output_schema = output_schema.get("properties", {}).get(
"result"
)
structured_content = result.structuredContent.get("result")
else:
structured_content = result.structuredContent
output_type = json_schema_to_type(output_schema)
type_adapter = get_cached_typeadapter(output_type)
data = type_adapter.validate_python(structured_content)
else:
data = result.structuredContent
except Exception as e:
logger.error(f"Error parsing structured content: {e}")
return CallToolResult(
content=result.content,
structured_content=result.structuredContent,
data=data,
is_error=result.isError,
)
@dataclass
class CallToolResult:
content: list[mcp.types.ContentBlock]
structured_content: dict[str, Any] | None
data: Any = None
is_error: bool = False

3
src/fastmcp/client/logging.py
View file

@ -1,7 +1,7 @@
from collections.abc import Awaitable, Callable
from typing import TypeAlias
from mcp.client.session import LoggingFnT, MessageHandlerFnT
from mcp.client.session import LoggingFnT
from mcp.types import LoggingMessageNotificationParams
from fastmcp.utilities.logging import get_logger
@ -10,7 +10,6 @@ logger = get_logger(__name__)
LogMessage: TypeAlias = LoggingMessageNotificationParams
LogHandler: TypeAlias = Callable[[LogMessage], Awaitable[None]]
MessageHandler: TypeAlias = MessageHandlerFnT
async def default_log_handler(message: LogMessage) -> None:

126
src/fastmcp/client/messages.py Normal file
View file

@ -0,0 +1,126 @@
from typing import TypeAlias
import mcp.types
from mcp.client.session import MessageHandlerFnT
from mcp.shared.session import RequestResponder
Message: TypeAlias = (
RequestResponder[mcp.types.ServerRequest, mcp.types.ClientResult]
| mcp.types.ServerNotification
| Exception
)
MessageHandlerT: TypeAlias = MessageHandlerFnT
class MessageHandler:
"""
This class is used to handle MCP messages sent to the client. It is used to handle all messages,
requests, notifications, and exceptions. Users can override any of the hooks
"""
async def __call__(
self,
message: RequestResponder[mcp.types.ServerRequest, mcp.types.ClientResult]
| mcp.types.ServerNotification
| Exception,
) -> None:
return await self.dispatch(message)
async def dispatch(self, message: Message) -> None:
# handle all messages
await self.on_message(message)
match message:
# requests
case RequestResponder():
# handle all requests
await self.on_request(message)
# handle specific requests
match message.request.root:
case mcp.types.PingRequest():
await self.on_ping(message.request.root)
case mcp.types.ListRootsRequest():
await self.on_list_roots(message.request.root)
case mcp.types.CreateMessageRequest():
await self.on_create_message(message.request.root)
# notifications
case mcp.types.ServerNotification():
# handle all notifications
await self.on_notification(message)
# handle specific notifications
match message.root:
case mcp.types.CancelledNotification():
await self.on_cancelled(message.root)
case mcp.types.ProgressNotification():
await self.on_progress(message.root)
case mcp.types.LoggingMessageNotification():
await self.on_logging_message(message.root)
case mcp.types.ToolListChangedNotification():
await self.on_tool_list_changed(message.root)
case mcp.types.ResourceListChangedNotification():
await self.on_resource_list_changed(message.root)
case mcp.types.PromptListChangedNotification():
await self.on_prompt_list_changed(message.root)
case mcp.types.ResourceUpdatedNotification():
await self.on_resource_updated(message.root)
case Exception():
await self.on_exception(message)
async def on_message(self, message: Message) -> None:
pass
async def on_request(
self, message: RequestResponder[mcp.types.ServerRequest, mcp.types.ClientResult]
) -> None:
pass
async def on_ping(self, message: mcp.types.PingRequest) -> None:
pass
async def on_list_roots(self, message: mcp.types.ListRootsRequest) -> None:
pass
async def on_create_message(self, message: mcp.types.CreateMessageRequest) -> None:
pass
async def on_notification(self, message: mcp.types.ServerNotification) -> None:
pass
async def on_exception(self, message: Exception) -> None:
pass
async def on_progress(self, message: mcp.types.ProgressNotification) -> None:
pass
async def on_logging_message(
self, message: mcp.types.LoggingMessageNotification
) -> None:
pass
async def on_tool_list_changed(
self, message: mcp.types.ToolListChangedNotification
) -> None:
pass
async def on_resource_list_changed(
self, message: mcp.types.ResourceListChangedNotification
) -> None:
pass
async def on_prompt_list_changed(
self, message: mcp.types.PromptListChangedNotification
) -> None:
pass
async def on_resource_updated(
self, message: mcp.types.ResourceUpdatedNotification
) -> None:
pass
async def on_cancelled(self, message: mcp.types.CancelledNotification) -> None:
pass

8
src/fastmcp/client/transports.py
View file

@ -8,7 +8,7 @@ import sys
import warnings
from collections.abc import AsyncIterator, Callable
from pathlib import Path
from typing import Any, Literal, TypedDict, TypeVar, cast, overload
from typing import Any, Literal, TypeVar, cast, overload
from urllib.parse import urlparse, urlunparse
import anyio
@ -19,7 +19,7 @@ from mcp.client.session import ListRootsFnT, LoggingFnT, MessageHandlerFnT, Samp
from mcp.server.fastmcp import FastMCP as FastMCP1Server
from mcp.shared.memory import create_client_server_memory_streams
from pydantic import AnyUrl
from typing_extensions import Unpack
from typing_extensions import TypedDict, Unpack
import fastmcp
from fastmcp.client.auth.bearer import BearerAuth
@ -736,11 +736,11 @@ class MCPConfigTransport(ClientTransport):
"mcpServers": {
"weather": {
"url": "https://weather-api.example.com/mcp",
"transport": "streamable-http"
"transport": "http"
},
"calendar": {
"url": "https://calendar-api.example.com/mcp",
"transport": "streamable-http"
"transport": "http"
}
}
}

170
src/fastmcp/contrib/component_manager/README.md Normal file
View file

@ -0,0 +1,170 @@
# Component Manager – Contrib Module for FastMCP
The **Component Manager** provides a unified API for enabling and disabling tools, resources, and prompts at runtime in a FastMCP server. This module is useful for dynamic control over which components are active, enabling advanced features like feature toggling, admin interfaces, or automation workflows.
---
## 🔧 Features
- Enable/disable **tools**, **resources**, and **prompts** via HTTP endpoints.
- Supports **local** and **mounted (server)** components.
- Customizable **API root path**.
- Optional **Auth scopes** for secured access.
- Fully integrates with FastMCP with minimal configuration.
---
## 📦 Installation
This module is part of the `fastmcp.contrib` package. No separate installation is required if you're already using **FastMCP**.
---
## 🚀 Usage
### Basic Setup
```python
from fastmcp import FastMCP
from fastmcp.contrib.component_manager import set_up_component_manager
mcp = FastMCP(name="Component Manager", instructions="This is a test server with component manager.")
set_up_component_manager(server=mcp)
```
---
## 🔗 API Endpoints
All endpoints are registered at `/` by default, or under the custom path if one is provided.
### Tools
```http
POST /tools/{tool_name}/enable
POST /tools/{tool_name}/disable
```
### Resources
```http
POST /resources/{uri:path}/enable
POST /resources/{uri:path}/disable
```
* Supports template URIs as well
```http
POST /resources/example://test/{id}/enable
POST /resources/example://test/{id}/disable
```
### Prompts
```http
POST /prompts/{prompt_name}/enable
POST /prompts/{prompt_name}/disable
```
---
#### 🧪 Example Response
```http
HTTP/1.1 200 OK
Content-Type: application/json
{
"message": "Disabled tool: example_tool"
}
```
---
## ⚙️ Configuration Options
### Custom Root Path
To mount the API under a different path:
```python
set_up_component_manager(server=mcp, path="/admin")
```
### Securing Endpoints with Auth Scopes
If your server uses authentication:
```python
mcp = FastMCP(name="Component Manager", instructions="This is a test server with component manager.", auth=auth)
set_up_component_manager(server=mcp, required_scopes=["write", "read"])
```
---
## 🧪 Example: Enabling a Tool with Curl
```bash
curl -X POST \
-H "Authorization: Bearer YOUR_TOKEN_HERE" \
-H "Content-Type: application/json" \
http://localhost:8001/tools/example_tool/enable
```
---
## 🧱 Working with Mounted Servers
You can also combine different configurations when working with mounted servers — for example, using different scopes:
```python
mcp = FastMCP(name="Component Manager", instructions="This is a test server with component manager.", auth=auth)
set_up_component_manager(server=mcp, required_scopes=["mcp:write"])
mounted = FastMCP(name="Component Manager", instructions="This is a test server with component manager.", auth=auth)
set_up_component_manager(server=mounted, required_scopes=["mounted:write"])
mcp.mount(server=mounted, prefix="mo")
```
This allows you to grant different levels of access:
```bash
# Accessing the main server gives you control over both local and mounted components
curl -X POST \
-H "Authorization: Bearer YOUR_TOKEN_HERE" \
-H "Content-Type: application/json" \
http://localhost:8001/tools/mo_example_tool/enable
# Accessing the mounted server gives you control only over its own components
curl -X POST \
-H "Authorization: Bearer YOUR_TOKEN_HERE" \
-H "Content-Type: application/json" \
http://localhost:8002/tools/example_tool/enable
```
---
## ⚙️ How It Works
- `set_up_component_manager()` registers API routes for tools, resources, and prompts.
- The `ComponentService` class exposes async methods to enable/disable components.
- Each endpoint returns a success message in JSON or a 404 error if the component isn't found.
---
## 🧩 Extending
You can subclass `ComponentService` for custom behavior or mount its routes elsewhere as needed.
---
## Maintenance Notice
This module is not officially maintained by the core FastMCP team. It is an independent extension developed by [gorocode](https://github.com/gorocode).
If you encounter any issues or wish to contribute, please feel free to open an issue or submit a pull request, and kindly notify me. I'd love to stay up to date.
## 📄 License
This module follows the license of the main [FastMCP](https://github.com/jlowin/fastmcp) project.

4
src/fastmcp/contrib/component_manager/__init__.py Normal file
View file

@ -0,0 +1,4 @@
from .component_manager import set_up_component_manager
from .component_service import ComponentService
__all__ = ["set_up_component_manager", "ComponentService"]

186
src/fastmcp/contrib/component_manager/component_manager.py Normal file
View file

@ -0,0 +1,186 @@
"""
Routes and helpers for managing tools, resources, and prompts in FastMCP.
Provides endpoints for enabling/disabling components via HTTP, with optional authentication scopes.
"""
from typing import Any
from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware
from starlette.applications import Starlette
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Mount, Route
from fastmcp.contrib.component_manager.component_service import ComponentService
from fastmcp.exceptions import NotFoundError
from fastmcp.server.server import FastMCP
def set_up_component_manager(
server: FastMCP, path: str = "/", required_scopes: list[str] | None = None
):
"""Set up routes for enabling/disabling tools, resources, and prompts.
Args:
server: The FastMCP server instance
path: Path used to mount all component-related routes on the server
required_scopes: Optional list of scopes required for these routes. Applies only if authentication is enabled.
"""
service = ComponentService(server)
routes: list[Route] = []
mounts: list[Mount] = []
route_configs = {
"tool": {
"param": "tool_name",
"enable": service._enable_tool,
"disable": service._disable_tool,
},
"resource": {
"param": "uri:path",
"enable": service._enable_resource,
"disable": service._disable_resource,
},
"prompt": {
"param": "prompt_name",
"enable": service._enable_prompt,
"disable": service._disable_prompt,
},
}
if required_scopes is None:
routes.extend(build_component_manager_endpoints(route_configs, path))
else:
if path != "/":
mounts.append(
build_component_manager_mount(route_configs, path, required_scopes)
)
else:
mounts.append(
build_component_manager_mount(
{"tool": route_configs["tool"]}, "/tools", required_scopes
)
)
mounts.append(
build_component_manager_mount(
{"resource": route_configs["resource"]},
"/resources",
required_scopes,
)
)
mounts.append(
build_component_manager_mount(
{"prompt": route_configs["prompt"]}, "/prompts", required_scopes
)
)
server._additional_http_routes.extend(routes)
server._additional_http_routes.extend(mounts)
def make_endpoint(action, component, config):
"""
Factory for creating Starlette endpoint functions for enabling/disabling a component.
Args:
action: 'enable' or 'disable'
component: The component type (e.g., 'tool', 'resource', or 'prompt')
config: Dict with param and handler functions for the component
Returns:
An async endpoint function for Starlette.
"""
async def endpoint(request: Request):
name = request.path_params[config["param"].split(":")[0]]
try:
await config[action](name)
return JSONResponse(
{"message": f"{action.capitalize()}d {component}: {name}"}
)
except NotFoundError:
raise StarletteHTTPException(
status_code=404,
detail=f"Unknown {component}: {name}",
)
return endpoint
def make_route(action, component, config, required_scopes, root_path) -> Route:
"""
Creates a Starlette Route for enabling/disabling a component.
Args:
action: 'enable' or 'disable'
component: The component type
config: Dict with param and handler functions
required_scopes: Optional list of required auth scopes
root_path: The base path for the route
Returns:
A Starlette Route object.
"""
endpoint = make_endpoint(action, component, config)
if required_scopes is not None and root_path in [
"/tools",
"/resources",
"/prompts",
]:
path = f"/{{{config['param']}}}/{action}"
else:
if root_path != "/" and required_scopes is None:
path = f"{root_path}/{component}s/{{{config['param']}}}/{action}"
else:
path = f"/{component}s/{{{config['param']}}}/{action}"
return Route(path, endpoint=endpoint, methods=["POST"])
def build_component_manager_endpoints(
route_configs, root_path, required_scopes=None
) -> list[Route]:
"""
Build a list of Starlette Route objects for all components/actions.
Args:
route_configs: Dict describing component types and their handlers
root_path: The base path for the routes
required_scopes: Optional list of required auth scopes
Returns:
List of Starlette Route objects for component management.
"""
component_management_routes: list[Route] = []
for component in route_configs:
config: dict[str, Any] = route_configs[component]
for action in ["enable", "disable"]:
component_management_routes.append(
make_route(action, component, config, required_scopes, root_path)
)
return component_management_routes
def build_component_manager_mount(route_configs, root_path, required_scopes) -> Mount:
"""
Build a Starlette Mount with authentication for component management routes.
Args:
route_configs: Dict describing component types and their handlers
root_path: The base path for the mount
required_scopes: List of required auth scopes
Returns:
A Starlette Mount object with authentication middleware.
"""
component_management_routes: list[Route] = []
for component in route_configs:
config: dict[str, Any] = route_configs[component]
for action in ["enable", "disable"]:
component_management_routes.append(
make_route(action, component, config, required_scopes, root_path)
)
return Mount(
f"{root_path}",
app=RequireAuthMiddleware(
Starlette(routes=component_management_routes), required_scopes
),
)

225
src/fastmcp/contrib/component_manager/component_service.py Normal file
View file

@ -0,0 +1,225 @@
"""
ComponentService: Provides async management of tools, resources, and prompts for FastMCP servers.
Handles enabling/disabling components both locally and across mounted servers.
"""
from fastmcp.exceptions import NotFoundError
from fastmcp.prompts.prompt import Prompt
from fastmcp.resources.resource import Resource
from fastmcp.resources.template import ResourceTemplate
from fastmcp.server.server import FastMCP, has_resource_prefix, remove_resource_prefix
from fastmcp.tools.tool import Tool
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
class ComponentService:
"""Service for managing components like tools, resources, and prompts."""
def __init__(self, server: FastMCP):
self._server = server
self._tool_manager = server._tool_manager
self._resource_manager = server._resource_manager
self._prompt_manager = server._prompt_manager
async def _enable_tool(self, key: str) -> Tool:
"""Handle 'enableTool' requests.
Args:
key: The key of the tool to enable
Returns:
The tool that was enabled
"""
logger.debug("Enabling tool: %s", key)
# 1. Check local tools first. The server will have already applied its filter.
if key in self._server._tool_manager._tools:
tool: Tool = await self._server.get_tool(key)
tool.enable()
return tool
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._tool_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
tool_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
tool = await mounted_service._enable_tool(tool_key)
return tool
else:
continue
raise NotFoundError(f"Unknown tool: {key}")
async def _disable_tool(self, key: str) -> Tool:
"""Handle 'disableTool' requests.
Args:
key: The key of the tool to disable
Returns:
The tool that was disabled
"""
logger.debug("Disable tool: %s", key)
# 1. Check local tools first. The server will have already applied its filter.
if key in self._server._tool_manager._tools:
tool: Tool = await self._server.get_tool(key)
tool.disable()
return tool
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._tool_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
tool_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
tool = await mounted_service._disable_tool(tool_key)
return tool
else:
continue
raise NotFoundError(f"Unknown tool: {key}")
async def _enable_resource(self, key: str) -> Resource | ResourceTemplate:
"""Handle 'enableResource' requests.
Args:
key: The key of the resource to enable
Returns:
The resource that was enabled
"""
logger.debug("Enabling resource: %s", key)
# 1. Check local resources first. The server will have already applied its filter.
if key in self._resource_manager._resources:
resource: Resource = await self._server.get_resource(key)
resource.enable()
return resource
if key in self._resource_manager._templates:
template: ResourceTemplate = await self._server.get_resource_template(key)
template.enable()
return template
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._resource_manager._mounted_servers):
if mounted.prefix:
if has_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
):
key = remove_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
)
mounted_service = ComponentService(mounted.server)
mounted_resource: (
Resource | ResourceTemplate
) = await mounted_service._enable_resource(key)
return mounted_resource
else:
continue
raise NotFoundError(f"Unknown resource: {key}")
async def _disable_resource(self, key: str) -> Resource | ResourceTemplate:
"""Handle 'disableResource' requests.
Args:
key: The key of the resource to disable
Returns:
The resource that was disabled
"""
logger.debug("Disable resource: %s", key)
# 1. Check local resources first. The server will have already applied its filter.
if key in self._resource_manager._resources:
resource: Resource = await self._server.get_resource(key)
resource.disable()
return resource
if key in self._resource_manager._templates:
template: ResourceTemplate = await self._server.get_resource_template(key)
template.disable()
return template
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._resource_manager._mounted_servers):
if mounted.prefix:
if has_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
):
key = remove_resource_prefix(
key,
mounted.prefix,
mounted.resource_prefix_format,
)
mounted_service = ComponentService(mounted.server)
mounted_resource: (
Resource | ResourceTemplate
) = await mounted_service._disable_resource(key)
return mounted_resource
else:
continue
raise NotFoundError(f"Unknown resource: {key}")
async def _enable_prompt(self, key: str) -> Prompt:
"""Handle 'enablePrompt' requests.
Args:
key: The key of the prompt to enable
Returns:
The prompt that was enable
"""
logger.debug("Enabling prompt: %s", key)
# 1. Check local prompts first. The server will have already applied its filter.
if key in self._server._prompt_manager._prompts:
prompt: Prompt = await self._server.get_prompt(key)
prompt.enable()
return prompt
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._prompt_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
prompt_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
prompt = await mounted_service._enable_prompt(prompt_key)
return prompt
else:
continue
raise NotFoundError(f"Unknown prompt: {key}")
async def _disable_prompt(self, key: str) -> Prompt:
"""Handle 'disablePrompt' requests.
Args:
key: The key of the prompt to disable
Returns:
The prompt that was disabled
"""
# 1. Check local prompts first. The server will have already applied its filter.
if key in self._server._prompt_manager._prompts:
prompt: Prompt = await self._server.get_prompt(key)
prompt.disable()
return prompt
# 2. Check mounted servers using the filtered protocol path.
for mounted in reversed(self._prompt_manager._mounted_servers):
if mounted.prefix:
if key.startswith(f"{mounted.prefix}_"):
prompt_key = key.removeprefix(f"{mounted.prefix}_")
mounted_service = ComponentService(mounted.server)
prompt = await mounted_service._disable_prompt(prompt_key)
return prompt
else:
continue
raise NotFoundError(f"Unknown prompt: {key}")

59
src/fastmcp/contrib/component_manager/example.py Normal file
View file

@ -0,0 +1,59 @@
from fastmcp import FastMCP
from fastmcp.contrib.component_manager import set_up_component_manager
from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
key_pair = RSAKeyPair.generate()
auth = BearerAuthProvider(
public_key=key_pair.public_key,
issuer="https://dev.example.com",
audience="my-dev-server",
required_scopes=["mcp:read"],
)
# Build main server
mcp_token = key_pair.create_token(
subject="dev-user",
issuer="https://dev.example.com",
audience="my-dev-server",
scopes=["mcp:write", "mcp:read"],
)
mcp = FastMCP(
name="Component Manager",
instructions="This is a test server with component manager.",
auth=auth,
)
# Set up main server component manager
set_up_component_manager(server=mcp, required_scopes=["mcp:write"])
# Build mounted server
mounted_token = key_pair.create_token(
subject="dev-user",
issuer="https://dev.example.com",
audience="my-dev-server",
scopes=["mounted:write", "mcp:read"],
)
mounted = FastMCP(
name="Component Manager",
instructions="This is a test server with component manager.",
auth=auth,
)
# Set up mounted server component manager
set_up_component_manager(server=mounted, required_scopes=["mounted:write"])
# Mount
mcp.mount(server=mounted, prefix="mo")
@mcp.resource("resource://greeting")
def get_greeting() -> str:
"""Provides a simple greeting message."""
return "Hello from FastMCP Resources!"
@mounted.tool("greeting")
def get_info() -> str:
"""Provides a simple info."""
return "You are using component manager contrib module!"

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

@ -69,6 +69,22 @@ class Prompt(FastMCPComponent, ABC):
default=None, description="Arguments that can be passed to the prompt"
)
def enable(self) -> None:
super().enable()
try:
context = get_context()
context._queue_prompt_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
def disable(self) -> None:
super().disable()
try:
context = get_context()
context._queue_prompt_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
def to_mcp_prompt(self, **overrides: Any) -> MCPPrompt:
"""Convert the prompt to an MCP prompt."""
arguments = [
@ -338,6 +354,6 @@ class FunctionPrompt(Prompt):
raise PromptError("Could not convert prompt result to message.")
return messages
except Exception as e:
logger.exception(f"Error rendering prompt {self.name}: {e}")
except Exception:
logger.exception(f"Error rendering prompt {self.name}")
raise PromptError(f"Error rendering prompt {self.name}.")

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

@ -172,12 +172,12 @@ class PromptManager:
# Pass through PromptErrors as-is
except PromptError as e:
logger.exception(f"Error rendering prompt {name!r}: {e}")
logger.exception(f"Error rendering prompt {name!r}")
raise e
# Handle other exceptions
except Exception as e:
logger.exception(f"Error rendering prompt {name!r}: {e}")
logger.exception(f"Error rendering prompt {name!r}")
if self.mask_error_details:
# Mask internal details
raise PromptError(f"Error rendering prompt {name!r}") from e

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

@ -44,6 +44,22 @@ class Resource(FastMCPComponent, abc.ABC):
pattern=r"^[a-zA-Z0-9]+/[a-zA-Z0-9\-+.]+$",
)
def enable(self) -> None:
super().enable()
try:
context = get_context()
context._queue_resource_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
def disable(self) -> None:
super().disable()
try:
context = get_context()
context._queue_resource_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
@staticmethod
def from_function(
fn: Callable[[], Any],

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

@ -422,12 +422,12 @@ class ResourceManager:
# raise ResourceErrors as-is
except ResourceError as e:
logger.exception(f"Error reading resource {uri_str!r}: {e}")
logger.exception(f"Error reading resource {uri_str!r}")
raise e
# Handle other exceptions
except Exception as e:
logger.exception(f"Error reading resource {uri_str!r}: {e}")
logger.exception(f"Error reading resource {uri_str!r}")
if self.mask_error_details:
# Mask internal details
raise ResourceError(f"Error reading resource {uri_str!r}") from e
@ -445,12 +445,12 @@ class ResourceManager:
return await resource.read()
except ResourceError as e:
logger.exception(
f"Error reading resource from template {uri_str!r}: {e}"
f"Error reading resource from template {uri_str!r}"
)
raise e
except Exception as e:
logger.exception(
f"Error reading resource from template {uri_str!r}: {e}"
f"Error reading resource from template {uri_str!r}"
)
if self.mask_error_details:
raise ResourceError(

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

@ -15,7 +15,7 @@ from pydantic import (
validate_call,
)
from fastmcp.resources.types import Resource
from fastmcp.resources.resource import Resource
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.components import FastMCPComponent
from fastmcp.utilities.json_schema import compress_schema
@ -65,6 +65,22 @@ class ResourceTemplate(FastMCPComponent):
def __repr__(self) -> str:
return f"{self.__class__.__name__}(uri_template={self.uri_template!r}, name={self.name!r}, description={self.description!r}, tags={self.tags})"
def enable(self) -> None:
super().enable()
try:
context = get_context()
context._queue_resource_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
def disable(self) -> None:
super().disable()
try:
context = get_context()
context._queue_resource_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
@staticmethod
def from_function(
fn: Callable[..., Any],

15
src/fastmcp/server/auth/auth.py
View file

@ -43,3 +43,18 @@ class OAuthProvider(
self.client_registration_options = client_registration_options
self.revocation_options = revocation_options
self.required_scopes = required_scopes
async def verify_token(self, token: str) -> AccessToken | None:
"""
Verify a bearer token and return access info if valid.
This method implements the TokenVerifier protocol by delegating
to our existing load_access_token method.
Args:
token: The token string to validate
Returns:
AccessToken object if valid, None if invalid or expired
"""
return await self.load_access_token(token)

67
src/fastmcp/server/auth/providers/bearer.py
View file

@ -1,6 +1,6 @@
import time
from dataclasses import dataclass
from typing import Any, TypedDict
from typing import Any
import httpx
from authlib.jose import JsonWebKey, JsonWebToken
@ -18,12 +18,14 @@ from mcp.shared.auth import (
OAuthToken,
)
from pydantic import AnyHttpUrl, SecretStr, ValidationError
from typing_extensions import TypedDict
from fastmcp.server.auth.auth import (
ClientRegistrationOptions,
OAuthProvider,
RevocationOptions,
)
from fastmcp.utilities.logging import get_logger
class JWKData(TypedDict, total=False):
@ -199,6 +201,7 @@ class BearerAuthProvider(OAuthProvider):
self.public_key = public_key
self.jwks_uri = jwks_uri
self.jwt = JsonWebToken(["RS256"])
self.logger = get_logger(__name__)
# Simple JWKS cache
self._jwks_cache: dict[str, str] = {}
@ -265,6 +268,9 @@ class BearerAuthProvider(OAuthProvider):
# Select the appropriate key
if kid:
if kid not in self._jwks_cache:
self.logger.debug(
"JWKS key lookup failed: key ID '%s' not found", kid
)
raise ValueError(f"Key ID '{kid}' not found in JWKS")
return self._jwks_cache[kid]
else:
@ -279,6 +285,7 @@ class BearerAuthProvider(OAuthProvider):
raise ValueError("No keys found in JWKS")
except Exception as e:
self.logger.debug("JWKS fetch failed: %s", str(e))
raise ValueError(f"Failed to fetch JWKS: {e}")
async def load_access_token(self, token: str) -> AccessToken | None:
@ -298,15 +305,27 @@ class BearerAuthProvider(OAuthProvider):
# Decode and verify the JWT token
claims = self.jwt.decode(token, verification_key)
# Extract client ID early for logging
client_id = claims.get("client_id") or claims.get("sub") or "unknown"
# Validate expiration
exp = claims.get("exp")
if exp and exp < time.time():
self.logger.debug(
"Token validation failed: expired token for client %s", client_id
)
self.logger.info("Bearer token rejected for client %s", client_id)
return None
# Validate issuer - note we use issuer instead of issuer_url here because
# issuer is optional, allowing users to make this check optional
if self.issuer:
if claims.get("iss") != self.issuer:
self.logger.debug(
"Token validation failed: issuer mismatch for client %s",
client_id,
)
self.logger.info("Bearer token rejected for client %s", client_id)
return None
# Validate audience if configured
@ -314,26 +333,33 @@ class BearerAuthProvider(OAuthProvider):
aud = claims.get("aud")
# Handle different combinations of audience types
audience_valid = False
if isinstance(self.audience, list):
# self.audience is a list - check if any expected audience is present
if isinstance(aud, list):
# Both are lists - check for intersection
if not any(expected in aud for expected in self.audience):
return None
audience_valid = any(
expected in aud for expected in self.audience
)
else:
# aud is a string - check if it's in our expected list
if aud not in self.audience:
return None
audience_valid = aud in self.audience
else:
# self.audience is a string - use original logic
if isinstance(aud, list):
if self.audience not in aud:
return None
elif aud != self.audience:
return None
audience_valid = self.audience in aud
else:
audience_valid = aud == self.audience
# Extract claims - prefer client_id over sub for OAuth application identification
client_id = claims.get("client_id") or claims.get("sub") or "unknown"
if not audience_valid:
self.logger.debug(
"Token validation failed: audience mismatch for client %s",
client_id,
)
self.logger.info("Bearer token rejected for client %s", client_id)
return None
# Extract scopes
scopes = self._extract_scopes(claims)
return AccessToken(
@ -344,8 +370,10 @@ class BearerAuthProvider(OAuthProvider):
)
except JoseError:
self.logger.debug("Token validation failed: JWT signature/format invalid")
return None
except Exception:
except Exception as e:
self.logger.debug("Token validation failed: %s", str(e))
return None
def _extract_scopes(self, claims: dict[str, Any]) -> list[str]:
@ -357,6 +385,21 @@ class BearerAuthProvider(OAuthProvider):
return scope_claim
return []
async def verify_token(self, token: str) -> AccessToken | None:
"""
Verify a bearer token and return access info if valid.
This method implements the TokenVerifier protocol by delegating
to our existing load_access_token method.
Args:
token: The JWT token string to validate
Returns:
AccessToken object if valid, None if invalid or expired
"""
return await self.load_access_token(token)
# --- Unused OAuth server methods ---
async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
raise NotImplementedError("Client management not supported")

15
src/fastmcp/server/auth/providers/in_memory.py
View file

@ -271,6 +271,21 @@ class InMemoryOAuthProvider(OAuthProvider):
return token_obj
return None
async def verify_token(self, token: str) -> AccessToken | None:
"""
Verify a bearer token and return access info if valid.
This method implements the TokenVerifier protocol by delegating
to our existing load_access_token method.
Args:
token: The token string to validate
Returns:
AccessToken object if valid, None if invalid or expired
"""
return await self.load_access_token(token)
def _revoke_internal(
self, access_token_str: str | None = None, refresh_token_str: str | None = None
):

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

@ -1,5 +1,6 @@
from __future__ import annotations
import asyncio
import warnings
from collections.abc import Generator
from contextlib import contextmanager
@ -40,6 +41,7 @@ logger = get_logger(__name__)
T = TypeVar("T")
_current_context: ContextVar[Context | None] = ContextVar("context", default=None)
_flush_lock = asyncio.Lock()
@contextmanager
@ -90,16 +92,20 @@ class Context:
def __init__(self, fastmcp: FastMCP):
self.fastmcp = fastmcp
self._tokens: list[Token] = []
self._notification_queue: set[str] = set() # Dedupe notifications
def __enter__(self) -> Context:
async def __aenter__(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 __exit__(self, exc_type, exc_val, exc_tb) -> None:
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
"""Exit the context manager and reset the most recent token."""
# Flush any remaining notifications before exiting
await self._flush_notifications()
if self._tokens:
token = self._tokens.pop()
_current_context.reset(token)
@ -115,56 +121,6 @@ class Context:
except LookupError:
raise ValueError("Context is not available outside of a request")
@property
def session(self) -> ServerSession:
"""Access to the underlying session for advanced usage."""
return self.request_context.session
@property
def client_id(self) -> str | None:
"""Get the client ID if available."""
return (
getattr(self.request_context.meta, "client_id", None)
if self.request_context.meta
else None
)
@property
def request_id(self) -> str:
"""Get the unique ID for this request."""
return str(self.request_context.request_id)
@property
def session_id(self) -> str | None:
"""Get the MCP session ID for HTTP transports.
Returns the session ID that can be used as a key for session-based
data storage (e.g., Redis) to share data between tool calls within
the same client session.
Returns:
The session ID for HTTP transports (SSE, StreamableHTTP), or None
for stdio and in-memory transports which don't use session IDs.
Example:
```python
@server.tool
def store_data(data: dict, ctx: Context) -> str:
if session_id := ctx.session_id:
redis_client.set(f"session:{session_id}:data", json.dumps(data))
return f"Data stored for session {session_id}"
return "No session ID available (stdio/memory transport)"
```
"""
try:
from fastmcp.server.dependencies import get_http_headers
headers = get_http_headers(include_all=True)
return headers.get("mcp-session-id")
except RuntimeError:
# No HTTP context available (stdio/in-memory transport)
return None
async def report_progress(
self, progress: float, total: float | None = None, message: str | None = None
) -> None:
@ -227,6 +183,56 @@ class Context:
related_request_id=self.request_id,
)
@property
def client_id(self) -> str | None:
"""Get the client ID if available."""
return (
getattr(self.request_context.meta, "client_id", None)
if self.request_context.meta
else None
)
@property
def request_id(self) -> str:
"""Get the unique ID for this request."""
return str(self.request_context.request_id)
@property
def session_id(self) -> str | None:
"""Get the MCP session ID for HTTP transports.
Returns the session ID that can be used as a key for session-based
data storage (e.g., Redis) to share data between tool calls within
the same client session.
Returns:
The session ID for HTTP transports (SSE, StreamableHTTP), or None
for stdio and in-memory transports which don't use session IDs.
Example:
```python
@server.tool
def store_data(data: dict, ctx: Context) -> str:
if session_id := ctx.session_id:
redis_client.set(f"session:{session_id}:data", json.dumps(data))
return f"Data stored for session {session_id}"
return "No session ID available (stdio/memory transport)"
```
"""
try:
from fastmcp.server.dependencies import get_http_headers
headers = get_http_headers(include_all=True)
return headers.get("mcp-session-id")
except RuntimeError:
# No HTTP context available (stdio/in-memory transport)
return None
@property
def session(self) -> ServerSession:
"""Access to the underlying session for advanced usage."""
return self.request_context.session
# Convenience methods for common log levels
async def debug(self, message: str, logger_name: str | None = None) -> None:
"""Send a debug log message."""
@ -249,6 +255,18 @@ class Context:
result = await self.session.list_roots()
return result.roots
async def send_tool_list_changed(self) -> None:
"""Send a tool list changed notification to the client."""
await self.session.send_tool_list_changed()
async def send_resource_list_changed(self) -> None:
"""Send a resource list changed notification to the client."""
await self.session.send_resource_list_changed()
async def send_prompt_list_changed(self) -> None:
"""Send a prompt list changed notification to the client."""
await self.session.send_prompt_list_changed()
async def sample(
self,
messages: str | list[str | SamplingMessage],
@ -364,6 +382,52 @@ class Context:
return fastmcp.server.dependencies.get_http_request()
def _queue_tool_list_changed(self) -> None:
"""Queue a tool list changed notification."""
self._notification_queue.add("notifications/tools/list_changed")
self._try_flush_notifications()
def _queue_resource_list_changed(self) -> None:
"""Queue a resource list changed notification."""
self._notification_queue.add("notifications/resources/list_changed")
self._try_flush_notifications()
def _queue_prompt_list_changed(self) -> None:
"""Queue a prompt list changed notification."""
self._notification_queue.add("notifications/prompts/list_changed")
self._try_flush_notifications()
def _try_flush_notifications(self) -> None:
"""Synchronous method that attempts to flush notifications if we're in an async context."""
try:
# Check if we're in an async context
loop = asyncio.get_running_loop()
if loop and not loop.is_running():
return
# Schedule flush as a task (fire-and-forget)
asyncio.create_task(self._flush_notifications())
except RuntimeError:
# No event loop - will flush later
pass
async def _flush_notifications(self) -> None:
"""Send all queued notifications."""
async with _flush_lock:
if not self._notification_queue:
return
try:
if "notifications/tools/list_changed" in self._notification_queue:
await self.session.send_tool_list_changed()
if "notifications/resources/list_changed" in self._notification_queue:
await self.session.send_resource_list_changed()
if "notifications/prompts/list_changed" in self._notification_queue:
await self.session.send_prompt_list_changed()
self._notification_queue.clear()
except Exception:
# Don't let notification failures break the request
pass
def _parse_model_preferences(
self, model_preferences: ModelPreferences | str | list[str] | None
) -> ModelPreferences | None:

2
src/fastmcp/server/http.py
View file

@ -87,7 +87,7 @@ def setup_auth_middleware_and_routes(
middleware = [
Middleware(
AuthenticationMiddleware,
backend=BearerAuthBackend(provider=auth),
backend=BearerAuthBackend(auth),
),
Middleware(AuthContextMiddleware),
]

37
src/fastmcp/server/low_level.py Normal file
View file

@ -0,0 +1,37 @@
from typing import Any
from mcp.server.lowlevel.server import (
LifespanResultT,
NotificationOptions,
RequestT,
)
from mcp.server.lowlevel.server import (
Server as _Server,
)
from mcp.server.models import InitializationOptions
class LowLevelServer(_Server[LifespanResultT, RequestT]):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# FastMCP servers support notifications for all components
self.notification_options = NotificationOptions(
prompts_changed=True,
resources_changed=True,
tools_changed=True,
)
def create_initialization_options(
self,
notification_options: NotificationOptions | None = None,
experimental_capabilities: dict[str, dict[str, Any]] | None = None,
**kwargs: Any,
) -> InitializationOptions:
# ensure we use the FastMCP notification options
if notification_options is None:
notification_options = self.notification_options
return super().create_initialization_options(
notification_options=notification_options,
experimental_capabilities=experimental_capabilities,
**kwargs,
)

6
src/fastmcp/server/middleware/__init__.py Normal file
View file

@ -0,0 +1,6 @@
from .middleware import Middleware, MiddlewareContext
__all__ = [
"Middleware",
"MiddlewareContext",
]

206
src/fastmcp/server/middleware/error_handling.py Normal file
View file

@ -0,0 +1,206 @@
"""Error handling middleware for consistent error responses and tracking."""
import asyncio
import logging
import traceback
from collections.abc import Callable
from typing import Any
from mcp import McpError
from mcp.types import ErrorData
from .middleware import CallNext, Middleware, MiddlewareContext
class ErrorHandlingMiddleware(Middleware):
"""Middleware that provides consistent error handling and logging.
Catches exceptions, logs them appropriately, and converts them to
proper MCP error responses. Also tracks error patterns for monitoring.
Example:
```python
from fastmcp.server.middleware.error_handling import ErrorHandlingMiddleware
import logging
# Configure logging to see error details
logging.basicConfig(level=logging.ERROR)
mcp = FastMCP("MyServer")
mcp.add_middleware(ErrorHandlingMiddleware())
```
"""
def __init__(
self,
logger: logging.Logger | None = None,
include_traceback: bool = False,
error_callback: Callable[[Exception, MiddlewareContext], None] | None = None,
transform_errors: bool = True,
):
"""Initialize error handling middleware.
Args:
logger: Logger instance for error logging. If None, uses 'fastmcp.errors'
include_traceback: Whether to include full traceback in error logs
error_callback: Optional callback function called for each error
transform_errors: Whether to transform non-MCP errors to McpError
"""
self.logger = logger or logging.getLogger("fastmcp.errors")
self.include_traceback = include_traceback
self.error_callback = error_callback
self.transform_errors = transform_errors
self.error_counts = {}
def _log_error(self, error: Exception, context: MiddlewareContext) -> None:
"""Log error with appropriate detail level."""
error_type = type(error).__name__
method = context.method or "unknown"
# Track error counts
error_key = f"{error_type}:{method}"
self.error_counts[error_key] = self.error_counts.get(error_key, 0) + 1
base_message = f"Error in {method}: {error_type}: {str(error)}"
if self.include_traceback:
self.logger.error(f"{base_message}\n{traceback.format_exc()}")
else:
self.logger.error(base_message)
# Call custom error callback if provided
if self.error_callback:
try:
self.error_callback(error, context)
except Exception as callback_error:
self.logger.error(f"Error in error callback: {callback_error}")
def _transform_error(self, error: Exception) -> Exception:
"""Transform non-MCP errors to proper MCP errors."""
if isinstance(error, McpError):
return error
if not self.transform_errors:
return error
# Map common exceptions to appropriate MCP error codes
error_type = type(error)
if error_type in (ValueError, TypeError):
return McpError(
ErrorData(code=-32602, message=f"Invalid params: {str(error)}")
)
elif error_type in (FileNotFoundError, KeyError):
return McpError(
ErrorData(code=-32001, message=f"Resource not found: {str(error)}")
)
elif error_type is PermissionError:
return McpError(
ErrorData(code=-32000, message=f"Permission denied: {str(error)}")
)
elif error_type in (TimeoutError, asyncio.TimeoutError):
return McpError(
ErrorData(code=-32000, message=f"Request timeout: {str(error)}")
)
else:
return McpError(
ErrorData(code=-32603, message=f"Internal error: {str(error)}")
)
async def on_message(self, context: MiddlewareContext, call_next: CallNext) -> Any:
"""Handle errors for all messages."""
try:
return await call_next(context)
except Exception as error:
self._log_error(error, context)
# Transform and re-raise
transformed_error = self._transform_error(error)
raise transformed_error
def get_error_stats(self) -> dict[str, int]:
"""Get error statistics for monitoring."""
return self.error_counts.copy()
class RetryMiddleware(Middleware):
"""Middleware that implements automatic retry logic for failed requests.
Retries requests that fail with transient errors, using exponential
backoff to avoid overwhelming the server or external dependencies.
Example:
```python
from fastmcp.server.middleware.error_handling import RetryMiddleware
# Retry up to 3 times with exponential backoff
retry_middleware = RetryMiddleware(
max_retries=3,
retry_exceptions=(ConnectionError, TimeoutError)
)
mcp = FastMCP("MyServer")
mcp.add_middleware(retry_middleware)
```
"""
def __init__(
self,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
backoff_multiplier: float = 2.0,
retry_exceptions: tuple[type[Exception], ...] = (ConnectionError, TimeoutError),
logger: logging.Logger | None = None,
):
"""Initialize retry middleware.
Args:
max_retries: Maximum number of retry attempts
base_delay: Initial delay between retries in seconds
max_delay: Maximum delay between retries in seconds
backoff_multiplier: Multiplier for exponential backoff
retry_exceptions: Tuple of exception types that should trigger retries
logger: Logger for retry attempts
"""
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.backoff_multiplier = backoff_multiplier
self.retry_exceptions = retry_exceptions
self.logger = logger or logging.getLogger("fastmcp.retry")
def _should_retry(self, error: Exception) -> bool:
"""Determine if an error should trigger a retry."""
return isinstance(error, self.retry_exceptions)
def _calculate_delay(self, attempt: int) -> float:
"""Calculate delay for the given attempt number."""
delay = self.base_delay * (self.backoff_multiplier**attempt)
return min(delay, self.max_delay)
async def on_request(self, context: MiddlewareContext, call_next: CallNext) -> Any:
"""Implement retry logic for requests."""
last_error = None
for attempt in range(self.max_retries + 1):
try:
return await call_next(context)
except Exception as error:
last_error = error
# Don't retry on the last attempt or if it's not a retryable error
if attempt == self.max_retries or not self._should_retry(error):
break
delay = self._calculate_delay(attempt)
self.logger.warning(
f"Request {context.method} failed (attempt {attempt + 1}/{self.max_retries + 1}): "
f"{type(error).__name__}: {str(error)}. Retrying in {delay:.1f}s..."
)
await asyncio.sleep(delay)
# Re-raise the last error if all retries failed
if last_error:
raise last_error

176
src/fastmcp/server/middleware/logging.py Normal file
View file

@ -0,0 +1,176 @@
"""Comprehensive logging middleware for FastMCP servers."""
import json
import logging
from typing import Any
from .middleware import CallNext, Middleware, MiddlewareContext
class LoggingMiddleware(Middleware):
"""Middleware that provides comprehensive request and response logging.
Logs all MCP messages with configurable detail levels. Useful for debugging,
monitoring, and understanding server usage patterns.
Example:
```python
from fastmcp.server.middleware.logging import LoggingMiddleware
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
mcp = FastMCP("MyServer")
mcp.add_middleware(LoggingMiddleware())
```
"""
def __init__(
self,
logger: logging.Logger | None = None,
log_level: int = logging.INFO,
include_payloads: bool = False,
max_payload_length: int = 1000,
methods: list[str] | None = None,
):
"""Initialize logging middleware.
Args:
logger: Logger instance to use. If None, creates a logger named 'fastmcp.requests'
log_level: Log level for messages (default: INFO)
include_payloads: Whether to include message payloads in logs
max_payload_length: Maximum length of payload to log (prevents huge logs)
methods: List of methods to log. If None, logs all methods.
"""
self.logger = logger or logging.getLogger("fastmcp.requests")
self.log_level = log_level
self.include_payloads = include_payloads
self.max_payload_length = max_payload_length
self.methods = methods
def _format_message(self, context: MiddlewareContext) -> str:
"""Format a message for logging."""
parts = [
f"source={context.source}",
f"type={context.type}",
f"method={context.method or 'unknown'}",
]
if self.include_payloads and hasattr(context.message, "__dict__"):
try:
payload = json.dumps(context.message.__dict__, default=str)
if len(payload) > self.max_payload_length:
payload = payload[: self.max_payload_length] + "..."
parts.append(f"payload={payload}")
except (TypeError, ValueError):
parts.append("payload=<non-serializable>")
return " ".join(parts)
async def on_message(self, context: MiddlewareContext, call_next: CallNext) -> Any:
"""Log all messages."""
message_info = self._format_message(context)
if self.methods and context.method not in self.methods:
return await call_next(context)
self.logger.log(self.log_level, f"Processing message: {message_info}")
try:
result = await call_next(context)
self.logger.log(
self.log_level, f"Completed message: {context.method or 'unknown'}"
)
return result
except Exception as e:
self.logger.log(
logging.ERROR, f"Failed message: {context.method or 'unknown'} - {e}"
)
raise
class StructuredLoggingMiddleware(Middleware):
"""Middleware that provides structured JSON logging for better log analysis.
Outputs structured logs that are easier to parse and analyze with log
aggregation tools like ELK stack, Splunk, or cloud logging services.
Example:
```python
from fastmcp.server.middleware.logging import StructuredLoggingMiddleware
import logging
mcp = FastMCP("MyServer")
mcp.add_middleware(StructuredLoggingMiddleware())
```
"""
def __init__(
self,
logger: logging.Logger | None = None,
log_level: int = logging.INFO,
include_payloads: bool = False,
methods: list[str] | None = None,
):
"""Initialize structured logging middleware.
Args:
logger: Logger instance to use. If None, creates a logger named 'fastmcp.structured'
log_level: Log level for messages (default: INFO)
include_payloads: Whether to include message payloads in logs
methods: List of methods to log. If None, logs all methods.
"""
self.logger = logger or logging.getLogger("fastmcp.structured")
self.log_level = log_level
self.include_payloads = include_payloads
self.methods = methods
def _create_log_entry(
self, context: MiddlewareContext, event: str, **extra_fields
) -> dict:
"""Create a structured log entry."""
entry = {
"event": event,
"timestamp": context.timestamp.isoformat(),
"source": context.source,
"type": context.type,
"method": context.method,
**extra_fields,
}
if self.include_payloads and hasattr(context.message, "__dict__"):
try:
entry["payload"] = context.message.__dict__
except (TypeError, ValueError):
entry["payload"] = "<non-serializable>"
return entry
async def on_message(self, context: MiddlewareContext, call_next: CallNext) -> Any:
"""Log structured message information."""
start_entry = self._create_log_entry(context, "request_start")
if self.methods and context.method not in self.methods:
return await call_next(context)
self.logger.log(self.log_level, json.dumps(start_entry))
try:
result = await call_next(context)
success_entry = self._create_log_entry(
context,
"request_success",
result_type=type(result).__name__ if result else None,
)
self.logger.log(self.log_level, json.dumps(success_entry))
return result
except Exception as e:
error_entry = self._create_log_entry(
context,
"request_error",
error_type=type(e).__name__,
error_message=str(e),
)
self.logger.log(logging.ERROR, json.dumps(error_entry))
raise

0
src/fastmcp/server/middleware.py → src/fastmcp/server/middleware/middleware.py
View file

231
src/fastmcp/server/middleware/rate_limiting.py Normal file
View file

@ -0,0 +1,231 @@
"""Rate limiting middleware for protecting FastMCP servers from abuse."""
import asyncio
import time
from collections import defaultdict, deque
from collections.abc import Callable
from typing import Any
from mcp import McpError
from mcp.types import ErrorData
from .middleware import CallNext, Middleware, MiddlewareContext
class RateLimitError(McpError):
"""Error raised when rate limit is exceeded."""
def __init__(self, message: str = "Rate limit exceeded"):
super().__init__(ErrorData(code=-32000, message=message))
class TokenBucketRateLimiter:
"""Token bucket implementation for rate limiting."""
def __init__(self, capacity: int, refill_rate: float):
"""Initialize token bucket.
Args:
capacity: Maximum number of tokens in the bucket
refill_rate: Tokens added per second
"""
self.capacity = capacity
self.refill_rate = refill_rate
self.tokens = capacity
self.last_refill = time.time()
self._lock = asyncio.Lock()
async def consume(self, tokens: int = 1) -> bool:
"""Try to consume tokens from the bucket.
Args:
tokens: Number of tokens to consume
Returns:
True if tokens were available and consumed, False otherwise
"""
async with self._lock:
now = time.time()
elapsed = now - self.last_refill
# Add tokens based on elapsed time
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
self.last_refill = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
class SlidingWindowRateLimiter:
"""Sliding window rate limiter implementation."""
def __init__(self, max_requests: int, window_seconds: int):
"""Initialize sliding window rate limiter.
Args:
max_requests: Maximum requests allowed in the time window
window_seconds: Time window in seconds
"""
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = deque()
self._lock = asyncio.Lock()
async def is_allowed(self) -> bool:
"""Check if a request is allowed."""
async with self._lock:
now = time.time()
cutoff = now - self.window_seconds
# Remove old requests outside the window
while self.requests and self.requests[0] < cutoff:
self.requests.popleft()
if len(self.requests) < self.max_requests:
self.requests.append(now)
return True
return False
class RateLimitingMiddleware(Middleware):
"""Middleware that implements rate limiting to prevent server abuse.
Uses a token bucket algorithm by default, allowing for burst traffic
while maintaining a sustainable long-term rate.
Example:
```python
from fastmcp.server.middleware.rate_limiting import RateLimitingMiddleware
# Allow 10 requests per second with bursts up to 20
rate_limiter = RateLimitingMiddleware(
max_requests_per_second=10,
burst_capacity=20
)
mcp = FastMCP("MyServer")
mcp.add_middleware(rate_limiter)
```
"""
def __init__(
self,
max_requests_per_second: float = 10.0,
burst_capacity: int | None = None,
get_client_id: Callable[[MiddlewareContext], str] | None = None,
global_limit: bool = False,
):
"""Initialize rate limiting middleware.
Args:
max_requests_per_second: Sustained requests per second allowed
burst_capacity: Maximum burst capacity. If None, defaults to 2x max_requests_per_second
get_client_id: Function to extract client ID from context. If None, uses global limiting
global_limit: If True, apply limit globally; if False, per-client
"""
self.max_requests_per_second = max_requests_per_second
self.burst_capacity = burst_capacity or int(max_requests_per_second * 2)
self.get_client_id = get_client_id
self.global_limit = global_limit
# Storage for rate limiters per client
self.limiters: dict[str, TokenBucketRateLimiter] = defaultdict(
lambda: TokenBucketRateLimiter(
self.burst_capacity, self.max_requests_per_second
)
)
# Global rate limiter
if self.global_limit:
self.global_limiter = TokenBucketRateLimiter(
self.burst_capacity, self.max_requests_per_second
)
def _get_client_identifier(self, context: MiddlewareContext) -> str:
"""Get client identifier for rate limiting."""
if self.get_client_id:
return self.get_client_id(context)
return "global"
async def on_request(self, context: MiddlewareContext, call_next: CallNext) -> Any:
"""Apply rate limiting to requests."""
if self.global_limit:
# Global rate limiting
allowed = await self.global_limiter.consume()
if not allowed:
raise RateLimitError("Global rate limit exceeded")
else:
# Per-client rate limiting
client_id = self._get_client_identifier(context)
limiter = self.limiters[client_id]
allowed = await limiter.consume()
if not allowed:
raise RateLimitError(f"Rate limit exceeded for client: {client_id}")
return await call_next(context)
class SlidingWindowRateLimitingMiddleware(Middleware):
"""Middleware that implements sliding window rate limiting.
Uses a sliding window approach which provides more precise rate limiting
but uses more memory to track individual request timestamps.
Example:
```python
from fastmcp.server.middleware.rate_limiting import SlidingWindowRateLimitingMiddleware
# Allow 100 requests per minute
rate_limiter = SlidingWindowRateLimitingMiddleware(
max_requests=100,
window_minutes=1
)
mcp = FastMCP("MyServer")
mcp.add_middleware(rate_limiter)
```
"""
def __init__(
self,
max_requests: int,
window_minutes: int = 1,
get_client_id: Callable[[MiddlewareContext], str] | None = None,
):
"""Initialize sliding window rate limiting middleware.
Args:
max_requests: Maximum requests allowed in the time window
window_minutes: Time window in minutes
get_client_id: Function to extract client ID from context
"""
self.max_requests = max_requests
self.window_seconds = window_minutes * 60
self.get_client_id = get_client_id
# Storage for rate limiters per client
self.limiters: dict[str, SlidingWindowRateLimiter] = defaultdict(
lambda: SlidingWindowRateLimiter(self.max_requests, self.window_seconds)
)
def _get_client_identifier(self, context: MiddlewareContext) -> str:
"""Get client identifier for rate limiting."""
if self.get_client_id:
return self.get_client_id(context)
return "global"
async def on_request(self, context: MiddlewareContext, call_next: CallNext) -> Any:
"""Apply sliding window rate limiting to requests."""
client_id = self._get_client_identifier(context)
limiter = self.limiters[client_id]
allowed = await limiter.is_allowed()
if not allowed:
raise RateLimitError(
f"Rate limit exceeded: {self.max_requests} requests per "
f"{self.window_seconds // 60} minutes for client: {client_id}"
)
return await call_next(context)

156
src/fastmcp/server/middleware/timing.py Normal file
View file

@ -0,0 +1,156 @@
"""Timing middleware for measuring and logging request performance."""
import logging
import time
from typing import Any
from .middleware import CallNext, Middleware, MiddlewareContext
class TimingMiddleware(Middleware):
"""Middleware that logs the execution time of requests.
Only measures and logs timing for request messages (not notifications).
Provides insights into performance characteristics of your MCP server.
Example:
```python
from fastmcp.server.middleware.timing import TimingMiddleware
mcp = FastMCP("MyServer")
mcp.add_middleware(TimingMiddleware())
# Now all requests will be timed and logged
```
"""
def __init__(
self, logger: logging.Logger | None = None, log_level: int = logging.INFO
):
"""Initialize timing middleware.
Args:
logger: Logger instance to use. If None, creates a logger named 'fastmcp.timing'
log_level: Log level for timing messages (default: INFO)
"""
self.logger = logger or logging.getLogger("fastmcp.timing")
self.log_level = log_level
async def on_request(self, context: MiddlewareContext, call_next: CallNext) -> Any:
"""Time request execution and log the results."""
method = context.method or "unknown"
start_time = time.perf_counter()
try:
result = await call_next(context)
duration_ms = (time.perf_counter() - start_time) * 1000
self.logger.log(
self.log_level, f"Request {method} completed in {duration_ms:.2f}ms"
)
return result
except Exception as e:
duration_ms = (time.perf_counter() - start_time) * 1000
self.logger.log(
self.log_level,
f"Request {method} failed after {duration_ms:.2f}ms: {e}",
)
raise
class DetailedTimingMiddleware(Middleware):
"""Enhanced timing middleware with per-operation breakdowns.
Provides detailed timing information for different types of MCP operations,
allowing you to identify performance bottlenecks in specific operations.
Example:
```python
from fastmcp.server.middleware.timing import DetailedTimingMiddleware
import logging
# Configure logging to see the output
logging.basicConfig(level=logging.INFO)
mcp = FastMCP("MyServer")
mcp.add_middleware(DetailedTimingMiddleware())
```
"""
def __init__(
self, logger: logging.Logger | None = None, log_level: int = logging.INFO
):
"""Initialize detailed timing middleware.
Args:
logger: Logger instance to use. If None, creates a logger named 'fastmcp.timing.detailed'
log_level: Log level for timing messages (default: INFO)
"""
self.logger = logger or logging.getLogger("fastmcp.timing.detailed")
self.log_level = log_level
async def _time_operation(
self, context: MiddlewareContext, call_next: CallNext, operation_name: str
) -> Any:
"""Helper method to time any operation."""
start_time = time.perf_counter()
try:
result = await call_next(context)
duration_ms = (time.perf_counter() - start_time) * 1000
self.logger.log(
self.log_level, f"{operation_name} completed in {duration_ms:.2f}ms"
)
return result
except Exception as e:
duration_ms = (time.perf_counter() - start_time) * 1000
self.logger.log(
self.log_level,
f"{operation_name} failed after {duration_ms:.2f}ms: {e}",
)
raise
async def on_call_tool(
self, context: MiddlewareContext, call_next: CallNext
) -> Any:
"""Time tool execution."""
tool_name = getattr(context.message, "name", "unknown")
return await self._time_operation(context, call_next, f"Tool '{tool_name}'")
async def on_read_resource(
self, context: MiddlewareContext, call_next: CallNext
) -> Any:
"""Time resource reading."""
resource_uri = getattr(context.message, "uri", "unknown")
return await self._time_operation(
context, call_next, f"Resource '{resource_uri}'"
)
async def on_get_prompt(
self, context: MiddlewareContext, call_next: CallNext
) -> Any:
"""Time prompt retrieval."""
prompt_name = getattr(context.message, "name", "unknown")
return await self._time_operation(context, call_next, f"Prompt '{prompt_name}'")
async def on_list_tools(
self, context: MiddlewareContext, call_next: CallNext
) -> Any:
"""Time tool listing."""
return await self._time_operation(context, call_next, "List tools")
async def on_list_resources(
self, context: MiddlewareContext, call_next: CallNext
) -> Any:
"""Time resource listing."""
return await self._time_operation(context, call_next, "List resources")
async def on_list_resource_templates(
self, context: MiddlewareContext, call_next: CallNext
) -> Any:
"""Time resource template listing."""
return await self._time_operation(context, call_next, "List resource templates")
async def on_list_prompts(
self, context: MiddlewareContext, call_next: CallNext
) -> Any:
"""Time prompt listing."""
return await self._time_operation(context, call_next, "List prompts")

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

@ -13,7 +13,7 @@ from re import Pattern
from typing import TYPE_CHECKING, Any, Literal
import httpx
from mcp.types import ContentBlock, ToolAnnotations
from mcp.types import ToolAnnotations
from pydantic.networks import AnyUrl
import fastmcp
@ -21,7 +21,7 @@ from fastmcp.exceptions import ToolError
from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.server.dependencies import get_http_headers
from fastmcp.server.server import FastMCP
from fastmcp.tools.tool import Tool, _convert_to_content
from fastmcp.tools.tool import Tool, ToolResult
from fastmcp.utilities import openapi
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.openapi import (
@ -254,7 +254,7 @@ class OpenAPITool(Tool):
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPITool(name={self.name!r}, method={self._route.method}, path={self._route.path})"
async def run(self, arguments: dict[str, Any]) -> list[ContentBlock]:
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""Execute the HTTP request based on the route configuration."""
# Prepare URL
@ -450,10 +450,11 @@ class OpenAPITool(Tool):
# Try to parse as JSON first
try:
result = response.json()
except (json.JSONDecodeError, ValueError):
# Return text content if not JSON
result = response.text
return _convert_to_content(result)
if not isinstance(result, dict):
result = {"result": result}
return ToolResult(structured_content=result)
except json.JSONDecodeError:
return ToolResult(content=response.text)
except httpx.HTTPStatusError as e:
# Handle HTTP errors (4xx, 5xx)

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

@ -8,7 +8,6 @@ from mcp.shared.exceptions import McpError
from mcp.types import (
METHOD_NOT_FOUND,
BlobResourceContents,
ContentBlock,
GetPromptResult,
TextResourceContents,
)
@ -23,7 +22,7 @@ from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.resources.resource_manager import ResourceManager
from fastmcp.server.context import Context
from fastmcp.server.server import FastMCP
from fastmcp.tools.tool import Tool
from fastmcp.tools.tool import Tool, ToolResult
from fastmcp.tools.tool_manager import ToolManager
from fastmcp.utilities.logging import get_logger
@ -67,9 +66,7 @@ class ProxyToolManager(ToolManager):
tools_dict = await self.get_tools()
return list(tools_dict.values())
async def call_tool(
self, key: str, arguments: dict[str, Any]
) -> list[ContentBlock]:
async def call_tool(self, key: str, arguments: dict[str, Any]) -> ToolResult:
"""Calls a tool, trying local/mounted first, then proxy if not found."""
try:
# First try local and mounted tools
@ -77,7 +74,11 @@ class ProxyToolManager(ToolManager):
except NotFoundError:
# If not found locally, try proxy
async with self.client:
return await self.client.call_tool(key, arguments)
result = await self.client.call_tool(key, arguments)
return ToolResult(
content=result.content,
structured_content=result.structured_content,
)
class ProxyResourceManager(ResourceManager):
@ -226,13 +227,14 @@ class ProxyTool(Tool):
description=mcp_tool.description,
parameters=mcp_tool.inputSchema,
annotations=mcp_tool.annotations,
output_schema=mcp_tool.outputSchema,
)
async def run(
self,
arguments: dict[str, Any],
context: Context | None = None,
) -> list[ContentBlock]:
) -> ToolResult:
"""Executes the tool by making a call through the client."""
# This is where the remote execution logic lives.
async with self._client:
@ -242,7 +244,10 @@ class ProxyTool(Tool):
)
if result.isError:
raise ToolError(cast(mcp.types.TextContent, result.content[0]).text)
return result.content
return ToolResult(
content=result.content,
structured_content=result.structuredContent,
)
class ProxyResource(Resource):

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

@ -23,7 +23,6 @@ import mcp.types
import uvicorn
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.lowlevel.server import LifespanResultT, NotificationOptions
from mcp.server.lowlevel.server import Server as MCPServer
from mcp.server.stdio import stdio_server
from mcp.types import (
AnyFunction,
@ -55,14 +54,16 @@ from fastmcp.server.http import (
create_sse_app,
create_streamable_http_app,
)
from fastmcp.server.low_level import LowLevelServer
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.settings import Settings
from fastmcp.tools import ToolManager
from fastmcp.tools.tool import FunctionTool, Tool
from fastmcp.tools.tool import FunctionTool, Tool, ToolResult
from fastmcp.utilities.cache import TimedCache
from fastmcp.utilities.components import FastMCPComponent
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.mcp_config import MCPConfig
from fastmcp.utilities.types import NotSet, NotSetT
if TYPE_CHECKING:
from fastmcp.client import Client
@ -74,6 +75,7 @@ if TYPE_CHECKING:
logger = get_logger(__name__)
DuplicateBehavior = Literal["warn", "error", "replace", "ignore"]
Transport = Literal["stdio", "http", "sse", "streamable-http"]
# Compiled URI parsing regex to split a URI into protocol and path components
URI_PATTERN = re.compile(r"^([^:]+://)(.*?)$")
@ -98,10 +100,12 @@ def _lifespan_wrapper(
[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]
],
) -> Callable[
[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]
[LowLevelServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]
]:
@asynccontextmanager
async def wrap(s: MCPServer[LifespanResultT]) -> AsyncIterator[LifespanResultT]:
async def wrap(
s: LowLevelServer[LifespanResultT],
) -> AsyncIterator[LifespanResultT]:
async with AsyncExitStack() as stack:
context = await stack.enter_async_context(lifespan(app))
yield context
@ -178,7 +182,7 @@ class FastMCP(Generic[LifespanResultT]):
lifespan = default_lifespan
else:
self._has_lifespan = True
self._mcp_server = MCPServer[LifespanResultT](
self._mcp_server = LowLevelServer[LifespanResultT](
name=name or "FastMCP",
version=version,
instructions=instructions,
@ -280,7 +284,7 @@ class FastMCP(Generic[LifespanResultT]):
async def run_async(
self,
transport: Literal["stdio", "streamable-http", "sse"] | None = None,
transport: Transport | None = None,
**transport_kwargs: Any,
) -> None:
"""Run the FastMCP server asynchronously.
@ -290,19 +294,19 @@ class FastMCP(Generic[LifespanResultT]):
"""
if transport is None:
transport = "stdio"
if transport not in {"stdio", "streamable-http", "sse"}:
if transport not in {"stdio", "http", "sse", "streamable-http"}:
raise ValueError(f"Unknown transport: {transport}")
if transport == "stdio":
await self.run_stdio_async(**transport_kwargs)
elif transport in {"streamable-http", "sse"}:
elif transport in {"http", "sse", "streamable-http"}:
await self.run_http_async(transport=transport, **transport_kwargs)
else:
raise ValueError(f"Unknown transport: {transport}")
def run(
self,
transport: Literal["stdio", "streamable-http", "sse"] | None = None,
transport: Transport | None = None,
**transport_kwargs: Any,
) -> None:
"""Run the FastMCP server. Note this is a synchronous function.
@ -362,6 +366,7 @@ class FastMCP(Generic[LifespanResultT]):
return await self._resource_manager.get_resource_templates()
async def get_resource_template(self, key: str) -> ResourceTemplate:
"""Get a registered resource template by key."""
templates = await self.get_resource_templates()
if key not in templates:
raise NotFoundError(f"Unknown resource template: {key}")
@ -402,9 +407,12 @@ class FastMCP(Generic[LifespanResultT]):
include_in_schema: Whether to include in OpenAPI schema, defaults to True
Example:
Register a custom HTTP route for a health check endpoint:
```python
@server.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> Response:
return JSONResponse({"status": "ok"})
```
"""
def decorator(
@ -426,7 +434,7 @@ class FastMCP(Generic[LifespanResultT]):
async def _mcp_list_tools(self) -> list[MCPTool]:
logger.debug("Handler called: list_tools")
with fastmcp.server.context.Context(fastmcp=self):
async with fastmcp.server.context.Context(fastmcp=self):
tools = await self._list_tools()
return [tool.to_mcp_tool(name=tool.key) for tool in tools]
@ -434,7 +442,6 @@ class FastMCP(Generic[LifespanResultT]):
"""
List all available tools, in the format expected by the low-level MCP
server.
"""
async def _handler(
@ -449,7 +456,7 @@ class FastMCP(Generic[LifespanResultT]):
return mcp_tools
with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
# Create the middleware context.
mw_context = MiddlewareContext(
message=mcp.types.ListToolsRequest(method="tools/list"),
@ -465,7 +472,7 @@ class FastMCP(Generic[LifespanResultT]):
async def _mcp_list_resources(self) -> list[MCPResource]:
logger.debug("Handler called: list_resources")
with fastmcp.server.context.Context(fastmcp=self):
async with fastmcp.server.context.Context(fastmcp=self):
resources = await self._list_resources()
return [
resource.to_mcp_resource(uri=resource.key) for resource in resources
@ -490,7 +497,7 @@ class FastMCP(Generic[LifespanResultT]):
return mcp_resources
with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
# Create the middleware context.
mw_context = MiddlewareContext(
message={}, # List resources doesn't have parameters
@ -506,7 +513,7 @@ class FastMCP(Generic[LifespanResultT]):
async def _mcp_list_resource_templates(self) -> list[MCPResourceTemplate]:
logger.debug("Handler called: list_resource_templates")
with fastmcp.server.context.Context(fastmcp=self):
async with fastmcp.server.context.Context(fastmcp=self):
templates = await self._list_resource_templates()
return [
template.to_mcp_template(uriTemplate=template.key)
@ -532,7 +539,7 @@ class FastMCP(Generic[LifespanResultT]):
return mcp_templates
with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
# Create the middleware context.
mw_context = MiddlewareContext(
message={}, # List resource templates doesn't have parameters
@ -548,7 +555,7 @@ class FastMCP(Generic[LifespanResultT]):
async def _mcp_list_prompts(self) -> list[MCPPrompt]:
logger.debug("Handler called: list_prompts")
with fastmcp.server.context.Context(fastmcp=self):
async with fastmcp.server.context.Context(fastmcp=self):
prompts = await self._list_prompts()
return [prompt.to_mcp_prompt(name=prompt.key) for prompt in prompts]
@ -571,7 +578,7 @@ class FastMCP(Generic[LifespanResultT]):
return mcp_prompts
with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
async with fastmcp.server.context.Context(fastmcp=self) as fastmcp_ctx:
# Create the middleware context.
mw_context = MiddlewareContext(
message=mcp.types.ListPromptsRequest(method="prompts/list"),
@ -586,7 +593,7 @@ class FastMCP(Generic[LifespanResultT]):
async def _mcp_call_tool(
self, key: str, arguments: dict[str, Any]
) -> list[ContentBlock]:
) -> list[ContentBlock] | tuple[list[ContentBlock], dict[str, Any]]:
"""
Handle MCP 'callTool' requests.
@ -601,24 +608,23 @@ class FastMCP(Generic[LifespanResultT]):
"""
logger.debug("Handler called: call_tool %s with %s", key, arguments)
with fastmcp.server.context.Context(fastmcp=self):
async with fastmcp.server.context.Context(fastmcp=self):
try:
return await self._call_tool(key, arguments)
result = await self._call_tool(key, arguments)
return result.to_mcp_result()
except DisabledError:
raise NotFoundError(f"Unknown tool: {key}")
except NotFoundError:
raise NotFoundError(f"Unknown tool: {key}")
async def _call_tool(
self, key: str, arguments: dict[str, Any]
) -> list[ContentBlock]:
async def _call_tool(self, key: str, arguments: dict[str, Any]) -> ToolResult:
"""
Applies this server's middleware and delegates the filtered call to the manager.
"""
async def _handler(
context: MiddlewareContext[mcp.types.CallToolRequestParams],
) -> list[ContentBlock]:
) -> ToolResult:
tool = await self._tool_manager.get_tool(context.message.name)
if not self._should_enable_component(tool):
raise NotFoundError(f"Unknown tool: {context.message.name!r}")
@ -644,7 +650,7 @@ class FastMCP(Generic[LifespanResultT]):
"""
logger.debug("Handler called: read_resource %s", uri)
with fastmcp.server.context.Context(fastmcp=self):
async with fastmcp.server.context.Context(fastmcp=self):
try:
return await self._read_resource(uri)
except DisabledError:
@ -699,7 +705,7 @@ class FastMCP(Generic[LifespanResultT]):
"""
logger.debug("Handler called: get_prompt %s with %s", name, arguments)
with fastmcp.server.context.Context(fastmcp=self):
async with fastmcp.server.context.Context(fastmcp=self):
try:
return await self._get_prompt(name, arguments)
except DisabledError:
@ -748,6 +754,15 @@ class FastMCP(Generic[LifespanResultT]):
self._tool_manager.add_tool(tool)
self._cache.clear()
# Send notification if we're in a request context
try:
from fastmcp.server.dependencies import get_context
context = get_context()
context._queue_tool_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
def remove_tool(self, name: str) -> None:
"""Remove a tool from the server.
@ -760,6 +775,15 @@ class FastMCP(Generic[LifespanResultT]):
self._tool_manager.remove_tool(name)
self._cache.clear()
# Send notification if we're in a request context
try:
from fastmcp.server.dependencies import get_context
context = get_context()
context._queue_tool_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
@overload
def tool(
self,
@ -768,6 +792,7 @@ class FastMCP(Generic[LifespanResultT]):
name: str | None = None,
description: str | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | None | NotSetT = NotSet,
annotations: ToolAnnotations | dict[str, Any] | None = None,
exclude_args: list[str] | None = None,
enabled: bool | None = None,
@ -781,6 +806,7 @@ class FastMCP(Generic[LifespanResultT]):
name: str | None = None,
description: str | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | None | NotSetT = NotSet,
annotations: ToolAnnotations | dict[str, Any] | None = None,
exclude_args: list[str] | None = None,
enabled: bool | None = None,
@ -793,6 +819,7 @@ class FastMCP(Generic[LifespanResultT]):
name: str | None = None,
description: str | None = None,
tags: set[str] | None = None,
output_schema: dict[str, Any] | None | NotSetT = NotSet,
annotations: ToolAnnotations | dict[str, Any] | None = None,
exclude_args: list[str] | None = None,
enabled: bool | None = None,
@ -815,15 +842,19 @@ class FastMCP(Generic[LifespanResultT]):
name: Optional name for the tool (keyword-only, alternative to name_or_fn)
description: Optional description of what the tool does
tags: Optional set of tags for categorizing the tool
annotations: Optional annotations about the tool's behavior (e.g. {"is_async": True})
output_schema: Optional JSON schema for the tool's output
annotations: Optional annotations about the tool's behavior
exclude_args: Optional list of argument names to exclude from the tool schema
enabled: Optional boolean to enable or disable the tool
Example:
Examples:
Register a tool with a custom name:
```python
@server.tool
def my_tool(x: int) -> str:
return str(x)
# Register a tool with a custom name
@server.tool
def my_tool(x: int) -> str:
return str(x)
@ -838,6 +869,7 @@ class FastMCP(Generic[LifespanResultT]):
# Direct function call
server.tool(my_function, name="custom_name")
```
"""
if isinstance(annotations, dict):
annotations = ToolAnnotations(**annotations)
@ -867,6 +899,7 @@ class FastMCP(Generic[LifespanResultT]):
name=tool_name,
description=description,
tags=tags,
output_schema=output_schema,
annotations=annotations,
exclude_args=exclude_args,
serializer=self._tool_serializer,
@ -897,6 +930,7 @@ class FastMCP(Generic[LifespanResultT]):
name=tool_name,
description=description,
tags=tags,
output_schema=output_schema,
annotations=annotations,
exclude_args=exclude_args,
enabled=enabled,
@ -912,6 +946,15 @@ class FastMCP(Generic[LifespanResultT]):
self._resource_manager.add_resource(resource)
self._cache.clear()
# Send notification if we're in a request context
try:
from fastmcp.server.dependencies import get_context
context = get_context()
context._queue_resource_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
def add_template(self, template: ResourceTemplate) -> None:
"""Add a resource template to the server.
@ -920,6 +963,15 @@ class FastMCP(Generic[LifespanResultT]):
"""
self._resource_manager.add_template(template)
# Send notification if we're in a request context
try:
from fastmcp.server.dependencies import get_context
context = get_context()
context._queue_resource_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
def add_resource_fn(
self,
fn: AnyFunction,
@ -992,7 +1044,9 @@ class FastMCP(Generic[LifespanResultT]):
tags: Optional set of tags for categorizing the resource
enabled: Optional boolean to enable or disable the resource
Example:
Examples:
Register a resource with a custom name:
```python
@server.resource("resource://my-resource")
def get_data() -> str:
return "Hello, world!"
@ -1015,6 +1069,7 @@ class FastMCP(Generic[LifespanResultT]):
async def get_weather(city: str) -> str:
data = await fetch_weather(city)
return f"Weather for {city}: {data}"
```
"""
# Check if user passed function directly instead of calling decorator
if inspect.isroutine(uri):
@ -1088,6 +1143,15 @@ class FastMCP(Generic[LifespanResultT]):
self._prompt_manager.add_prompt(prompt)
self._cache.clear()
# Send notification if we're in a request context
try:
from fastmcp.server.dependencies import get_context
context = get_context()
context._queue_prompt_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
@overload
def prompt(
self,
@ -1139,7 +1203,9 @@ class FastMCP(Generic[LifespanResultT]):
tags: Optional set of tags for categorizing the prompt
enabled: Optional boolean to enable or disable the prompt
Example:
Examples:
```python
@server.prompt
def analyze_table(table_name: str) -> list[Message]:
schema = read_table_schema(table_name)
@ -1183,6 +1249,7 @@ class FastMCP(Generic[LifespanResultT]):
# Direct function call
server.prompt(my_function, name="custom_name")
```
"""
if isinstance(name_or_fn, classmethod):
@ -1255,7 +1322,7 @@ class FastMCP(Generic[LifespanResultT]):
async def run_http_async(
self,
transport: Literal["streamable-http", "sse"] = "streamable-http",
transport: Literal["http", "streamable-http", "sse"] = "http",
host: str | None = None,
port: int | None = None,
log_level: str | None = None,
@ -1386,7 +1453,7 @@ class FastMCP(Generic[LifespanResultT]):
middleware: list[ASGIMiddleware] | None = None,
json_response: bool | None = None,
stateless_http: bool | None = None,
transport: Literal["streamable-http", "sse"] = "streamable-http",
transport: Literal["http", "streamable-http", "sse"] = "http",
) -> StarletteWithLifespan:
"""Create a Starlette app using the specified HTTP transport.
@ -1399,7 +1466,7 @@ class FastMCP(Generic[LifespanResultT]):
A Starlette application configured with the specified transport
"""
if transport == "streamable-http":
if transport in ("streamable-http", "http"):
return create_streamable_http_app(
server=self,
streamable_http_path=path
@ -1446,7 +1513,7 @@ class FastMCP(Generic[LifespanResultT]):
stacklevel=2,
)
await self.run_http_async(
transport="streamable-http",
transport="http",
host=host,
port=port,
log_level=log_level,
@ -1788,10 +1855,10 @@ class FastMCP(Generic[LifespanResultT]):
) -> FastMCPProxy:
"""Create a FastMCP proxy server for the given backend.
The ``backend`` argument can be either an existing :class:`~fastmcp.client.Client`
instance or any value accepted as the ``transport`` argument of
:class:`~fastmcp.client.Client`. This mirrors the convenience of the
``Client`` constructor.
The `backend` argument can be either an existing `fastmcp.client.Client`
instance or any value accepted as the `transport` argument of
`fastmcp.client.Client`. This mirrors the convenience of the
`fastmcp.client.Client` constructor.
"""
from fastmcp.client.client import Client
from fastmcp.server.proxy import FastMCPProxy
@ -1828,14 +1895,14 @@ class FastMCP(Generic[LifespanResultT]):
Given a component, determine if it should be enabled. Returns True if it should be enabled; False if it should not.
Rules:
• If the component's enabled property is False, always return False.
• If both include_tags and exclude_tags are None, return True.
• If exclude_tags is provided, check each exclude tag:
- If the component's enabled property is False, always return False.
- If both include_tags and exclude_tags are None, return True.
- If exclude_tags is provided, check each exclude tag:
- If the exclude tag is a string, it must be present in the input tags to exclude.
• If include_tags is provided, check each include tag:
- If include_tags is provided, check each include tag:
- If the include tag is a string, it must be present in the input tags to include.
• If include_tags is provided and none of the include tags match, return False.
• If include_tags is not provided, return True.
- If include_tags is provided and none of the include tags match, return False.
- If include_tags is not provided, return True.
"""
if not component.enabled:
return False
@ -1876,12 +1943,21 @@ def add_resource_prefix(
The resource URI with the prefix added
Examples:
>>> add_resource_prefix("resource://path/to/resource", "prefix")
"resource://prefix/path/to/resource" # with new style
>>> add_resource_prefix("resource://path/to/resource", "prefix")
"prefix+resource://path/to/resource" # with legacy style
>>> add_resource_prefix("resource:///absolute/path", "prefix")
"resource://prefix//absolute/path" # with new style
With new style:
```python
add_resource_prefix("resource://path/to/resource", "prefix")
"resource://prefix/path/to/resource"
```
With legacy style:
```python
add_resource_prefix("resource://path/to/resource", "prefix")
"prefix+resource://path/to/resource"
```
With absolute path:
```python
add_resource_prefix("resource:///absolute/path", "prefix")
"resource://prefix//absolute/path"
```
Raises:
ValueError: If the URI doesn't match the expected protocol://path format
@ -1927,12 +2003,21 @@ def remove_resource_prefix(
The resource URI with the prefix removed
Examples:
>>> remove_resource_prefix("resource://prefix/path/to/resource", "prefix")
"resource://path/to/resource" # with new style
>>> remove_resource_prefix("prefix+resource://path/to/resource", "prefix")
"resource://path/to/resource" # with legacy style
>>> remove_resource_prefix("resource://prefix//absolute/path", "prefix")
"resource:///absolute/path" # with new style
With new style:
```python
remove_resource_prefix("resource://prefix/path/to/resource", "prefix")
"resource://path/to/resource"
```
With legacy style:
```python
remove_resource_prefix("prefix+resource://path/to/resource", "prefix")
"resource://path/to/resource"
```
With absolute path:
```python
remove_resource_prefix("resource://prefix//absolute/path", "prefix")
"resource:///absolute/path"
```
Raises:
ValueError: If the URI doesn't match the expected protocol://path format
@ -1985,12 +2070,21 @@ def has_resource_prefix(
True if the URI has the specified prefix, False otherwise
Examples:
>>> has_resource_prefix("resource://prefix/path/to/resource", "prefix")
True # with new style
>>> has_resource_prefix("prefix+resource://path/to/resource", "prefix")
True # with legacy style
>>> has_resource_prefix("resource://other/path/to/resource", "prefix")
With new style:
```python
has_resource_prefix("resource://prefix/path/to/resource", "prefix")
True
```
With legacy style:
```python
has_resource_prefix("prefix+resource://path/to/resource", "prefix")
True
```
With other path:
```python
has_resource_prefix("resource://other/path/to/resource", "prefix")
False
```
Raises:
ValueError: If the URI doesn't match the expected protocol://path format

17
src/fastmcp/settings.py
View file

@ -154,23 +154,6 @@ class Settings(BaseSettings):
),
] = "path"
tool_attempt_parse_json_args: Annotated[
bool,
Field(
default=False,
description=inspect.cleandoc(
"""
Note: this enables a legacy behavior. If True, will attempt to parse
stringified JSON lists and objects strings in tool arguments before
passing them to the tool. This is an old behavior that can create
unexpected type coercion issues, but may be helpful for less powerful
LLMs that stringify JSON instead of passing actual lists and objects.
Defaults to False.
"""
),
),
] = False
client_init_timeout: Annotated[
float | None,
Field(

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

@ -1,17 +1,16 @@
from __future__ import annotations
import inspect
import json
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Annotated, Any, Literal
import mcp.types
import pydantic_core
from mcp.types import ContentBlock, TextContent, ToolAnnotations
from mcp.types import Tool as MCPTool
from pydantic import Field
from pydantic import Field, PydanticSchemaGenerationError
import fastmcp
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.components import FastMCPComponent
from fastmcp.utilities.json_schema import compress_schema
@ -20,8 +19,11 @@ from fastmcp.utilities.types import (
Audio,
File,
Image,
NotSet,
NotSetT,
find_kwarg_by_type,
get_cached_typeadapter,
replace_type,
)
if TYPE_CHECKING:
@ -30,26 +32,114 @@ if TYPE_CHECKING:
logger = get_logger(__name__)
class _UnserializableType:
pass
def default_serializer(data: Any) -> str:
return pydantic_core.to_json(data, fallback=str, indent=2).decode()
def _wrap_schema_if_needed(schema: dict[str, Any] | None) -> dict[str, Any] | None:
"""Wrap non-object schemas with result property for structured output.
This wrapping allows primitive types (int, str, etc.) to be returned as
structured content by placing them under a "result" key.
Args:
schema: The JSON schema to potentially wrap
Returns:
Wrapped schema if needed, or original schema if already an object type
"""
if schema and schema.get("type") != "object":
return {
"type": "object",
"properties": {"result": schema},
"x-fastmcp-wrap-result": True,
}
return schema
class ToolResult:
def __init__(
self,
content: list[ContentBlock] | Any | None = None,
structured_content: dict[str, Any] | Any | None = None,
):
if content is None and structured_content is None:
raise ValueError("Either content or structured_content must be provided")
elif content is None:
content = structured_content
self.content = _convert_to_content(content)
if structured_content is not None:
try:
structured_content = pydantic_core.to_jsonable_python(
structured_content
)
except pydantic_core.PydanticSerializationError as e:
logger.error(
f"Could not serialize structured content. If this is unexpected, set your tool's output_schema to None to disable automatic serialization: {e}"
)
raise
if not isinstance(structured_content, dict):
raise ValueError(
"structured_content must be a dict or None. "
f"Got {type(structured_content).__name__}: {structured_content!r}. "
"Tools should wrap non-dict values based on their output_schema."
)
self.structured_content: dict[str, Any] | None = structured_content
def to_mcp_result(
self,
) -> list[ContentBlock] | tuple[list[ContentBlock], dict[str, Any]]:
if self.structured_content is None:
return self.content
return self.content, self.structured_content
class Tool(FastMCPComponent):
"""Internal tool registration info."""
parameters: dict[str, Any] = Field(description="JSON schema for tool parameters")
annotations: ToolAnnotations | None = Field(
default=None, description="Additional annotations about the tool"
)
serializer: Callable[[Any], str] | None = Field(
default=None, description="Optional custom serializer for tool results"
)
parameters: Annotated[
dict[str, Any], Field(description="JSON schema for tool parameters")
]
output_schema: Annotated[
dict[str, Any] | None, Field(description="JSON schema for tool output")
] = None
annotations: Annotated[
ToolAnnotations | None,
Field(description="Additional annotations about the tool"),
] = None
serializer: Annotated[
Callable[[Any], str] | None,
Field(description="Optional custom serializer for tool results"),
] = None
def enable(self) -> None:
super().enable()
try:
context = get_context()
context._queue_tool_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
def disable(self) -> None:
super().disable()
try:
context = get_context()
context._queue_tool_list_changed() # type: ignore[private-use]
except RuntimeError:
pass # No context available
def to_mcp_tool(self, **overrides: Any) -> MCPTool:
kwargs = {
"name": self.name,
"description": self.description,
"inputSchema": self.parameters,
"outputSchema": self.output_schema,
"annotations": self.annotations,
}
return MCPTool(**kwargs | overrides)
@ -62,6 +152,7 @@ class Tool(FastMCPComponent):
tags: set[str] | None = None,
annotations: ToolAnnotations | None = None,
exclude_args: list[str] | None = None,
output_schema: dict[str, Any] | None | NotSetT | Literal[False] = NotSet,
serializer: Callable[[Any], str] | None = None,
enabled: bool | None = None,
) -> FunctionTool:
@ -73,12 +164,21 @@ class Tool(FastMCPComponent):
tags=tags,
annotations=annotations,
exclude_args=exclude_args,
output_schema=output_schema,
serializer=serializer,
enabled=enabled,
)
async def run(self, arguments: dict[str, Any]) -> list[ContentBlock]:
"""Run the tool with arguments."""
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""
Run the tool with arguments.
This method is not implemented in the base Tool class and must be
implemented by subclasses.
`run()` can EITHER return a list of ContentBlocks, or a tuple of
(list of ContentBlocks, dict of structured output).
"""
raise NotImplementedError("Subclasses must implement run()")
@classmethod
@ -91,6 +191,7 @@ class Tool(FastMCPComponent):
description: str | None = None,
tags: set[str] | None = None,
annotations: ToolAnnotations | None = None,
output_schema: dict[str, Any] | None | Literal[False] = None,
serializer: Callable[[Any], str] | None = None,
enabled: bool | None = None,
) -> TransformedTool:
@ -104,6 +205,7 @@ class Tool(FastMCPComponent):
description=description,
tags=tags,
annotations=annotations,
output_schema=output_schema,
serializer=serializer,
enabled=enabled,
)
@ -121,6 +223,7 @@ class FunctionTool(Tool):
tags: set[str] | None = None,
annotations: ToolAnnotations | None = None,
exclude_args: list[str] | None = None,
output_schema: dict[str, Any] | None | NotSetT | Literal[False] = NotSet,
serializer: Callable[[Any], str] | None = None,
enabled: bool | None = None,
) -> FunctionTool:
@ -131,18 +234,32 @@ class FunctionTool(Tool):
if name is None and parsed_fn.name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
if isinstance(output_schema, NotSetT):
output_schema = _wrap_schema_if_needed(parsed_fn.output_schema)
elif output_schema is False:
output_schema = None
# Note: explicit schemas (dict) are used as-is without auto-wrapping
# Validate that explicit schemas are object type for structured content
if output_schema is not None and isinstance(output_schema, dict):
if output_schema.get("type") != "object":
raise ValueError(
f'Output schemas must have "type" set to "object" due to MCP spec limitations. Received: {output_schema!r}'
)
return cls(
fn=parsed_fn.fn,
name=name or parsed_fn.name,
description=description or parsed_fn.description,
parameters=parsed_fn.parameters,
tags=tags or set(),
parameters=parsed_fn.input_schema,
output_schema=output_schema,
annotations=annotations,
tags=tags or set(),
serializer=serializer,
enabled=enabled if enabled is not None else True,
)
async def run(self, arguments: dict[str, Any]) -> list[ContentBlock]:
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""Run the tool with arguments."""
from fastmcp.server.context import Context
@ -152,41 +269,39 @@ class FunctionTool(Tool):
if context_kwarg and context_kwarg not in arguments:
arguments[context_kwarg] = get_context()
if fastmcp.settings.tool_attempt_parse_json_args:
# Pre-parse data from JSON in order to handle cases like `["a", "b", "c"]`
# being passed in as JSON inside a string rather than an actual list.
#
# Claude desktop is prone to this - in fact it seems incapable of NOT doing
# this. For sub-models, it tends to pass dicts (JSON objects) as JSON strings,
# which can be pre-parsed here.
signature = inspect.signature(self.fn)
for param_name in self.parameters["properties"]:
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
# if not a string, we won't have a JSON to parse, so skip logic
if not isinstance(arg, str):
continue
# skip if the type is a simple type (int, float, bool)
if signature.parameters[param_name].annotation in (
int,
float,
bool,
):
continue
try:
arguments[param_name] = json.loads(arg)
except json.JSONDecodeError:
pass
type_adapter = get_cached_typeadapter(self.fn)
result = type_adapter.validate_python(arguments)
if inspect.isawaitable(result):
result = await result
return _convert_to_content(result, serializer=self.serializer)
if isinstance(result, ToolResult):
return result
unstructured_result = _convert_to_content(result, serializer=self.serializer)
structured_output = None
# First handle structured content based on output schema, if any
if self.output_schema is not None:
if self.output_schema.get("x-fastmcp-wrap-result"):
# Schema says wrap - always wrap in result key
structured_output = {"result": result}
else:
structured_output = result
# If no output schema, try to serialize the result. If it is a dict, use
# it as structured content. If it is not a dict, ignore it.
if structured_output is None:
try:
structured_output = pydantic_core.to_jsonable_python(result)
if not isinstance(structured_output, dict):
structured_output = None
except Exception:
pass
return ToolResult(
content=unstructured_result,
structured_content=structured_output,
)
@dataclass
@ -194,13 +309,15 @@ class ParsedFunction:
fn: Callable[..., Any]
name: str
description: str | None
parameters: dict[str, Any]
input_schema: dict[str, Any]
output_schema: dict[str, Any] | None
@classmethod
def from_function(
cls,
fn: Callable[..., Any],
exclude_args: list[str] | None = None,
ignore_response_types: list[type] | None = None,
validate: bool = True,
) -> ParsedFunction:
from fastmcp.server.context import Context
@ -240,9 +357,6 @@ class ParsedFunction:
if isinstance(fn, staticmethod):
fn = fn.__func__
type_adapter = get_cached_typeadapter(fn)
schema = type_adapter.json_schema()
prune_params: list[str] = []
context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context)
if context_kwarg:
@ -250,12 +364,65 @@ class ParsedFunction:
if exclude_args:
prune_params.extend(exclude_args)
schema = compress_schema(schema, prune_params=prune_params)
input_type_adapter = get_cached_typeadapter(fn)
input_schema = input_type_adapter.json_schema()
input_schema = compress_schema(input_schema, prune_params=prune_params)
output_schema = None
output_type = inspect.signature(fn).return_annotation
if output_type not in (inspect._empty, None, Any, ...):
# there are a variety of types that we don't want to attempt to
# serialize because they are either used by FastMCP internally,
# or are MCP content types that explicitly don't form structured
# content. By replacing them with an explicitly unserializable type,
# we ensure that no output schema is automatically generated.
output_type = replace_type(
output_type,
{
t: _UnserializableType
for t in (
Image,
Audio,
File,
ToolResult,
mcp.types.TextContent,
mcp.types.ImageContent,
mcp.types.AudioContent,
mcp.types.ResourceLink,
mcp.types.EmbeddedResource,
)
},
)
try:
output_type_adapter = get_cached_typeadapter(output_type)
output_schema = output_type_adapter.json_schema()
except PydanticSchemaGenerationError as e:
if "_UnserializableType" not in str(e):
logger.debug(f"Unable to generate schema for type {output_type!r}")
return cls(
fn=fn,
name=fn_name,
description=fn_doc,
parameters=schema,
input_schema=input_schema,
output_schema=output_schema or None,
)
try:
output_type_adapter = get_cached_typeadapter(output_type)
output_schema = output_type_adapter.json_schema()
except PydanticSchemaGenerationError as e:
if "_UnserializableType" not in str(e):
logger.debug(f"Unable to generate schema for type {output_type!r}")
return cls(
fn=fn,
name=fn_name,
description=fn_doc,
input_schema=input_schema,
output_schema=output_schema or None,
)

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

@ -4,12 +4,12 @@ import warnings
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
from mcp.types import ContentBlock, ToolAnnotations
from mcp.types import ToolAnnotations
from fastmcp import settings
from fastmcp.exceptions import NotFoundError, ToolError
from fastmcp.settings import DuplicateBehavior
from fastmcp.tools.tool import Tool
from fastmcp.tools.tool import Tool, ToolResult
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
@ -169,9 +169,7 @@ class ToolManager:
else:
raise NotFoundError(f"Tool {key!r} not found")
async def call_tool(
self, key: str, arguments: dict[str, Any]
) -> list[ContentBlock]:
async def call_tool(self, key: str, arguments: dict[str, Any]) -> ToolResult:
"""
Internal API for servers: Finds and calls a tool, respecting the
filtered protocol path.
@ -187,12 +185,12 @@ class ToolManager:
# raise ToolErrors as-is
except ToolError as e:
logger.exception(f"Error calling tool {key!r}: {e}")
logger.exception(f"Error calling tool {key!r}")
raise e
# Handle other exceptions
except Exception as e:
logger.exception(f"Error calling tool {key!r}: {e}")
logger.exception(f"Error calling tool {key!r}")
if self.mask_error_details:
# Mask internal details
raise ToolError(f"Error calling tool {key!r}") from e

211
src/fastmcp/tools/tool_transform.py
View file

@ -4,20 +4,17 @@ import inspect
from collections.abc import Callable
from contextvars import ContextVar
from dataclasses import dataclass
from types import EllipsisType
from typing import Any, Literal
from mcp.types import ContentBlock, ToolAnnotations
from mcp.types import ToolAnnotations
from pydantic import ConfigDict
from fastmcp.tools.tool import ParsedFunction, Tool
from fastmcp.tools.tool import ParsedFunction, Tool, ToolResult, _wrap_schema_if_needed
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import get_cached_typeadapter
from fastmcp.utilities.types import NotSet, NotSetT, get_cached_typeadapter
logger = get_logger(__name__)
NotSet = ...
# Context variable to store current transformed tool
_current_tool: ContextVar[TransformedTool | None] = ContextVar(
@ -25,7 +22,7 @@ _current_tool: ContextVar[TransformedTool | None] = ContextVar(
)
async def forward(**kwargs) -> Any:
async def forward(**kwargs) -> ToolResult:
"""Forward to parent tool with argument transformation applied.
This function can only be called from within a transformed tool's custom
@ -41,7 +38,7 @@ async def forward(**kwargs) -> Any:
**kwargs: Arguments to forward to the parent tool (using transformed names).
Returns:
The result from the parent tool execution.
The ToolResult from the parent tool execution.
Raises:
RuntimeError: If called outside a transformed tool context.
@ -55,7 +52,7 @@ async def forward(**kwargs) -> Any:
return await tool.forwarding_fn(**kwargs)
async def forward_raw(**kwargs) -> Any:
async def forward_raw(**kwargs) -> ToolResult:
"""Forward directly to parent tool without transformation.
This function bypasses all argument transformation and validation, calling the parent
@ -69,7 +66,7 @@ async def forward_raw(**kwargs) -> Any:
**kwargs: Arguments to pass directly to the parent tool (using original names).
Returns:
The result from the parent tool execution.
The ToolResult from the parent tool execution.
Raises:
RuntimeError: If called outside a transformed tool context.
@ -100,45 +97,65 @@ class ArgTransform:
examples: Examples for the argument. Use ... for no change.
Examples:
# Rename argument 'old_name' to 'new_name'
Rename argument 'old_name' to 'new_name'
```python
ArgTransform(name="new_name")
```
# Change description only
Change description only
```python
ArgTransform(description="Updated description")
```
# Add a default value (makes argument optional)
Add a default value (makes argument optional)
```python
ArgTransform(default=42)
```
# Add a default factory (makes argument optional)
Add a default factory (makes argument optional)
```python
ArgTransform(default_factory=lambda: time.time())
```
# Change the type
Change the type
```python
ArgTransform(type=str)
```
# Hide the argument entirely from clients
Hide the argument entirely from clients
```python
ArgTransform(hide=True)
```
# Hide argument but pass a constant value to parent
Hide argument but pass a constant value to parent
```python
ArgTransform(hide=True, default="constant_value")
```
# Hide argument but pass a factory-generated value to parent
Hide argument but pass a factory-generated value to parent
```python
ArgTransform(hide=True, default_factory=lambda: uuid.uuid4().hex)
```
# Make an optional parameter required (removes any default)
Make an optional parameter required (removes any default)
```python
ArgTransform(required=True)
```
# Combine multiple transformations
Combine multiple transformations
```python
ArgTransform(name="new_name", description="New desc", default=None, type=int)
```
"""
name: str | EllipsisType = NotSet
description: str | EllipsisType = NotSet
default: Any | EllipsisType = NotSet
default_factory: Callable[[], Any] | EllipsisType = NotSet
type: Any | EllipsisType = NotSet
name: str | NotSetT = NotSet
description: str | NotSetT = NotSet
default: Any | NotSetT = NotSet
default_factory: Callable[[], Any] | NotSetT = NotSet
type: Any | NotSetT = NotSet
hide: bool = False
required: Literal[True] | EllipsisType = NotSet
examples: Any | EllipsisType = NotSet
required: Literal[True] | NotSetT = NotSet
examples: Any | NotSetT = NotSet
def __post_init__(self):
"""Validate that only one of default or default_factory is provided."""
@ -181,11 +198,12 @@ class TransformedTool(Tool):
This class represents a tool that has been created by transforming another tool.
It supports argument renaming, schema modification, custom function injection,
and provides context for the forward() and forward_raw() functions.
structured output control, and provides context for the forward() and forward_raw() functions.
The transformation can be purely schema-based (argument renaming, dropping, etc.)
or can include a custom function that uses forward() to call the parent tool
with transformed arguments.
with transformed arguments. Output schemas and structured outputs are automatically
inherited from the parent tool but can be overridden or disabled.
Attributes:
parent_tool: The original tool that this tool was transformed from.
@ -202,7 +220,7 @@ class TransformedTool(Tool):
forwarding_fn: Callable[..., Any] # Always present, handles arg transformation
transform_args: dict[str, ArgTransform]
async def run(self, arguments: dict[str, Any]) -> list[ContentBlock]:
async def run(self, arguments: dict[str, Any]) -> ToolResult:
"""Run the tool with context set for forward() functions.
This method executes the tool's function while setting up the context
@ -213,8 +231,7 @@ class TransformedTool(Tool):
arguments: Dictionary of arguments to pass to the tool's function.
Returns:
List of content objects (text, image, or embedded resources) representing
the tool's output.
ToolResult object containing content and optional structured output.
"""
from fastmcp.tools.tool import _convert_to_content
@ -252,7 +269,57 @@ class TransformedTool(Tool):
token = _current_tool.set(self)
try:
result = await self.fn(**arguments)
return _convert_to_content(result, serializer=self.serializer)
# If transform function returns ToolResult, respect our output_schema setting
if isinstance(result, ToolResult):
if self.output_schema is None:
# Check if this is from a custom function that returns ToolResult
import inspect
return_annotation = inspect.signature(self.fn).return_annotation
if return_annotation is ToolResult:
# Custom function returns ToolResult - preserve its content
return result
else:
# Forwarded call with disabled schema - strip structured content
return ToolResult(
content=result.content,
structured_content=None,
)
elif self.output_schema.get(
"type"
) != "object" and not self.output_schema.get("x-fastmcp-wrap-result"):
# Non-object explicit schemas disable structured content
return ToolResult(
content=result.content,
structured_content=None,
)
else:
return result
# Otherwise convert to content and create ToolResult with proper structured content
from fastmcp.tools.tool import _convert_to_content
unstructured_result = _convert_to_content(
result, serializer=self.serializer
)
# Handle structured content based on output schema
if self.output_schema is not None:
if self.output_schema.get("x-fastmcp-wrap-result"):
# Schema says wrap - always wrap in result key
structured_output = {"result": result}
else:
# Object schemas - use result directly
# User is responsible for returning dict-compatible data
structured_output = result
else:
structured_output = None
return ToolResult(
content=unstructured_result,
structured_content=structured_output,
)
finally:
_current_tool.reset(token)
@ -266,6 +333,7 @@ class TransformedTool(Tool):
transform_fn: Callable[..., Any] | None = None,
transform_args: dict[str, ArgTransform] | None = None,
annotations: ToolAnnotations | None = None,
output_schema: dict[str, Any] | None | Literal[False] = None,
serializer: Callable[[Any], str] | None = None,
enabled: bool | None = None,
) -> TransformedTool:
@ -279,34 +347,64 @@ class TransformedTool(Tool):
name: New name for the tool. Defaults to parent tool's name.
transform_args: Optional transformations for parent tool arguments.
Only specified arguments are transformed, others pass through unchanged:
- str: Simple rename
- ArgTransform: Complex transformation (rename/description/default/drop)
- None: Drop the argument
- Simple rename (str)
- Complex transformation (rename/description/default/drop) (ArgTransform)
- Drop the argument (None)
description: New description. Defaults to parent's description.
tags: New tags. Defaults to parent's tags.
annotations: New annotations. Defaults to parent's annotations.
output_schema: Control output schema for structured outputs:
- None (default): Inherit from transform_fn if available, then parent tool
- dict: Use custom output schema
- False: Disable output schema and structured outputs
serializer: New serializer. Defaults to parent's serializer.
Returns:
TransformedTool with the specified transformations.
Examples:
Examples:
# Transform specific arguments only
```python
Tool.from_tool(parent, transform_args={"old": "new"}) # Others unchanged
```
# Custom function with partial transforms
```python
async def custom(x: int, y: int) -> str:
result = await forward(x=x, y=y)
return f"Custom: {result}"
Tool.from_tool(parent, transform_fn=custom, transform_args={"a": "x", "b": "y"})
```
# Using **kwargs (gets all args, transformed and untransformed)
```python
async def flexible(**kwargs) -> str:
result = await forward(**kwargs)
return f"Got: {kwargs}"
Tool.from_tool(parent, transform_fn=flexible, transform_args={"a": "x"})
```
# Control structured outputs and schemas
```python
# Custom output schema
Tool.from_tool(parent, output_schema={
"type": "object",
"properties": {"status": {"type": "string"}}
})
# Disable structured outputs
Tool.from_tool(parent, output_schema=False)
# Return ToolResult for full control
async def custom_output(**kwargs) -> ToolResult:
result = await forward(**kwargs)
return ToolResult(
content=[TextContent(text="Summary")],
structured_content={"processed": True}
)
```
"""
transform_args = transform_args or {}
@ -322,19 +420,45 @@ class TransformedTool(Tool):
# Always create the forwarding transform
schema, forwarding_fn = cls._create_forwarding_transform(tool, transform_args)
# Handle output schema with smart fallback
if output_schema is False:
final_output_schema = None
elif output_schema is not None:
# Explicit schema provided - use as-is
final_output_schema = output_schema
else:
# Smart fallback: try custom function, then parent, then None
if transform_fn is not None:
parsed_fn = ParsedFunction.from_function(transform_fn, validate=False)
final_output_schema = _wrap_schema_if_needed(parsed_fn.output_schema)
if final_output_schema is None:
# Check if function returns ToolResult - if so, don't fall back to parent
import inspect
return_annotation = inspect.signature(
transform_fn
).return_annotation
if return_annotation is ToolResult:
final_output_schema = None
else:
final_output_schema = tool.output_schema
else:
final_output_schema = tool.output_schema
if transform_fn is None:
# User wants pure transformation - use forwarding_fn as the main function
final_fn = forwarding_fn
final_schema = schema
else:
# User provided custom function - merge schemas
parsed_fn = ParsedFunction.from_function(transform_fn, validate=False)
if "parsed_fn" not in locals():
parsed_fn = ParsedFunction.from_function(transform_fn, validate=False)
final_fn = transform_fn
has_kwargs = cls._function_has_kwargs(transform_fn)
# Validate function parameters against transformed schema
fn_params = set(parsed_fn.parameters.get("properties", {}).keys())
fn_params = set(parsed_fn.input_schema.get("properties", {}).keys())
transformed_params = set(schema.get("properties", {}).keys())
if not has_kwargs:
@ -351,7 +475,7 @@ class TransformedTool(Tool):
# ArgTransform takes precedence over function signature
# Start with function schema as base, then override with transformed schema
final_schema = cls._merge_schema_with_precedence(
parsed_fn.parameters, schema
parsed_fn.input_schema, schema
)
else:
# With **kwargs, function can access all transformed params
@ -360,7 +484,7 @@ class TransformedTool(Tool):
# Start with function schema as base, then override with transformed schema
final_schema = cls._merge_schema_with_precedence(
parsed_fn.parameters, schema
parsed_fn.input_schema, schema
)
# Additional validation: check for naming conflicts after transformation
@ -396,6 +520,7 @@ class TransformedTool(Tool):
name=name or tool.name,
description=final_description,
parameters=final_schema,
output_schema=final_output_schema,
tags=tags or tool.tags,
annotations=annotations or tool.annotations,
serializer=serializer or tool.serializer,
@ -423,8 +548,8 @@ class TransformedTool(Tool):
Returns:
A tuple containing:
- dict: The new JSON schema for the transformed tool
- Callable: Async function that validates and forwards calls to the parent tool
- The new JSON schema for the transformed tool as a dictionary
- Async function that validates and forwards calls to the parent tool
"""
# Build transformed schema and mapping

646
src/fastmcp/utilities/json_schema_type.py Normal file
View file

@ -0,0 +1,646 @@
"""Convert JSON Schema to Python types with validation.
The json_schema_to_type function converts a JSON Schema into a Python type that can be used
for validation with Pydantic. It supports:
- Basic types (string, number, integer, boolean, null)
- Complex types (arrays, objects)
- Format constraints (date-time, email, uri)
- Numeric constraints (minimum, maximum, multipleOf)
- String constraints (minLength, maxLength, pattern)
- Array constraints (minItems, maxItems, uniqueItems)
- Object properties with defaults
- References and recursive schemas
- Enums and constants
- Union types
Example:
```python
schema = {
"type": "object",
"properties": {
"name": {"type": "string", "minLength": 1},
"age": {"type": "integer", "minimum": 0},
"email": {"type": "string", "format": "email"}
},
"required": ["name", "age"]
}
# Name is optional and will be inferred from schema's "title" property if not provided
Person = json_schema_to_type(schema)
# Creates a validated dataclass with name, age, and optional email fields
```
"""
from __future__ import annotations
import hashlib
import json
import re
from collections.abc import Callable, Mapping
from copy import deepcopy
from dataclasses import MISSING, field, make_dataclass
from datetime import datetime
from enum import Enum
from typing import (
Annotated,
Any,
ForwardRef,
Literal,
Union,
)
from pydantic import (
AnyUrl,
BaseModel,
ConfigDict,
EmailStr,
Field,
Json,
StringConstraints,
model_validator,
)
from typing_extensions import NotRequired, TypedDict
__all__ = ["json_schema_to_type", "JSONSchema"]
FORMAT_TYPES: dict[str, Any] = {
"date-time": datetime,
"email": EmailStr,
"uri": AnyUrl,
"json": Json,
}
_classes: dict[tuple[str, Any], type | None] = {}
class JSONSchema(TypedDict):
type: NotRequired[str | list[str]]
properties: NotRequired[dict[str, JSONSchema]]
required: NotRequired[list[str]]
additionalProperties: NotRequired[bool | JSONSchema]
items: NotRequired[JSONSchema | list[JSONSchema]]
enum: NotRequired[list[Any]]
const: NotRequired[Any]
default: NotRequired[Any]
description: NotRequired[str]
title: NotRequired[str]
examples: NotRequired[list[Any]]
format: NotRequired[str]
allOf: NotRequired[list[JSONSchema]]
anyOf: NotRequired[list[JSONSchema]]
oneOf: NotRequired[list[JSONSchema]]
not_: NotRequired[JSONSchema]
definitions: NotRequired[dict[str, JSONSchema]]
dependencies: NotRequired[dict[str, JSONSchema | list[str]]]
pattern: NotRequired[str]
minLength: NotRequired[int]
maxLength: NotRequired[int]
minimum: NotRequired[int | float]
maximum: NotRequired[int | float]
exclusiveMinimum: NotRequired[int | float]
exclusiveMaximum: NotRequired[int | float]
multipleOf: NotRequired[int | float]
uniqueItems: NotRequired[bool]
minItems: NotRequired[int]
maxItems: NotRequired[int]
additionalItems: NotRequired[bool | JSONSchema]
def json_schema_to_type(
schema: Mapping[str, Any],
name: str | None = None,
) -> type:
"""Convert JSON schema to appropriate Python type with validation.
Args:
schema: A JSON Schema dictionary defining the type structure and validation rules
name: Optional name for object schemas. Only allowed when schema type is "object".
If not provided for objects, name will be inferred from schema's "title"
property or default to "Root".
Returns:
A Python type (typically a dataclass for objects) with Pydantic validation
Raises:
ValueError: If a name is provided for a non-object schema
Examples:
Create a dataclass from an object schema:
```python
schema = {
"type": "object",
"title": "Person",
"properties": {
"name": {"type": "string", "minLength": 1},
"age": {"type": "integer", "minimum": 0},
"email": {"type": "string", "format": "email"}
},
"required": ["name", "age"]
}
Person = json_schema_to_type(schema)
# Creates a dataclass with name, age, and optional email fields:
# @dataclass
# class Person:
# name: str
# age: int
# email: str | None = None
```
Person(name="John", age=30)
Create a scalar type with constraints:
```python
schema = {
"type": "string",
"minLength": 3,
"pattern": "^[A-Z][a-z]+$"
}
NameType = json_schema_to_type(schema)
# Creates Annotated[str, StringConstraints(min_length=3, pattern="^[A-Z][a-z]+$")]
@dataclass
class Name:
name: NameType
```
"""
# Always use the top-level schema for references
if schema.get("type") == "object":
# If no properties defined but has additionalProperties, return typed dict
if not schema.get("properties") and schema.get("additionalProperties"):
additional_props = schema["additionalProperties"]
if additional_props is True:
return dict[str, Any] # type: ignore - additionalProperties: true means dict[str, Any]
else:
# Handle typed dictionaries like dict[str, str]
value_type = _schema_to_type(additional_props, schemas=schema)
return dict[str, value_type] # type: ignore
# If no properties and no additionalProperties, default to dict[str, Any] for safety
elif not schema.get("properties") and not schema.get("additionalProperties"):
return dict[str, Any] # type: ignore
# If has properties AND additionalProperties is True, use Pydantic BaseModel
elif schema.get("properties") and schema.get("additionalProperties") is True:
return _create_pydantic_model(schema, name, schemas=schema)
# Otherwise use fast dataclass
return _create_dataclass(schema, name, schemas=schema)
elif name:
raise ValueError(f"Can not apply name to non-object schema: {name}")
result = _schema_to_type(schema, schemas=schema)
return result # type: ignore[return-value]
def _hash_schema(schema: Mapping[str, Any]) -> str:
"""Generate a deterministic hash for schema caching."""
return hashlib.sha256(json.dumps(schema, sort_keys=True).encode()).hexdigest()
def _resolve_ref(ref: str, schemas: Mapping[str, Any]) -> Mapping[str, Any]:
"""Resolve JSON Schema reference to target schema."""
path = ref.replace("#/", "").split("/")
current = schemas
for part in path:
current = current.get(part, {})
return current
def _create_string_type(schema: Mapping[str, Any]) -> type | Annotated[Any, ...]:
"""Create string type with optional constraints."""
if "const" in schema:
return Literal[schema["const"]] # type: ignore
if fmt := schema.get("format"):
if fmt == "uri":
return AnyUrl
elif fmt == "uri-reference":
return str
return FORMAT_TYPES.get(fmt, str)
constraints = {
k: v
for k, v in {
"min_length": schema.get("minLength"),
"max_length": schema.get("maxLength"),
"pattern": schema.get("pattern"),
}.items()
if v is not None
}
return Annotated[str, StringConstraints(**constraints)] if constraints else str
def _create_numeric_type(
base: type[int | float], schema: Mapping[str, Any]
) -> type | Annotated[Any, ...]:
"""Create numeric type with optional constraints."""
if "const" in schema:
return Literal[schema["const"]] # type: ignore
constraints = {
k: v
for k, v in {
"gt": schema.get("exclusiveMinimum"),
"ge": schema.get("minimum"),
"lt": schema.get("exclusiveMaximum"),
"le": schema.get("maximum"),
"multiple_of": schema.get("multipleOf"),
}.items()
if v is not None
}
return Annotated[base, Field(**constraints)] if constraints else base
def _create_enum(name: str, values: list[Any]) -> type:
"""Create enum type from list of values."""
if all(isinstance(v, str) for v in values):
return Enum(name, {v.upper(): v for v in values}) # type: ignore[return-value]
return Literal[tuple(values)] # type: ignore[return-value]
def _create_array_type(
schema: Mapping[str, Any], schemas: Mapping[str, Any]
) -> type | Annotated[Any, ...]:
"""Create list/set type with optional constraints."""
items = schema.get("items", {})
if isinstance(items, list):
# Handle positional item schemas
item_types = [_schema_to_type(s, schemas) for s in items]
combined = Union[tuple(item_types)] # type: ignore # noqa: UP007
base = list[combined]
else:
# Handle single item schema
item_type = _schema_to_type(items, schemas)
base_class = set if schema.get("uniqueItems") else list
base = base_class[item_type] # type: ignore[misc]
constraints = {
k: v
for k, v in {
"min_length": schema.get("minItems"),
"max_length": schema.get("maxItems"),
}.items()
if v is not None
}
return Annotated[base, Field(**constraints)] if constraints else base
def _return_Any() -> Any:
return Any
def _get_from_type_handler(
schema: Mapping[str, Any], schemas: Mapping[str, Any]
) -> Callable[..., Any]:
"""Get the appropriate type handler for the schema."""
type_handlers: dict[str, Callable[..., Any]] = { # TODO
"string": lambda s: _create_string_type(s), # type: ignore
"integer": lambda s: _create_numeric_type(int, s), # type: ignore
"number": lambda s: _create_numeric_type(float, s), # type: ignore
"boolean": lambda _: bool, # type: ignore
"null": lambda _: type(None), # type: ignore
"array": lambda s: _create_array_type(s, schemas), # type: ignore
"object": lambda s: (
_create_pydantic_model(s, s.get("title"), schemas)
if s.get("properties") and s.get("additionalProperties") is True
else _create_dataclass(s, s.get("title"), schemas)
), # type: ignore
}
return type_handlers.get(schema.get("type", None), _return_Any)
def _schema_to_type(
schema: Mapping[str, Any],
schemas: Mapping[str, Any],
) -> type | ForwardRef:
"""Convert schema to appropriate Python type."""
if not schema:
return object
if "type" not in schema and "properties" in schema:
return _create_dataclass(schema, schema.get("title", "<unknown>"), schemas)
# Handle references first
if "$ref" in schema:
ref = schema["$ref"]
# Handle self-reference
if ref == "#":
return ForwardRef(schema.get("title", "Root")) # type: ignore[return-value]
return _schema_to_type(_resolve_ref(ref, schemas), schemas)
if "const" in schema:
return Literal[schema["const"]] # type: ignore
if "enum" in schema:
return _create_enum(f"Enum_{len(_classes)}", schema["enum"])
# Handle anyOf unions
if "anyOf" in schema:
types: list[type | Any] = []
for subschema in schema["anyOf"]:
# Special handling for dict-like objects in unions
if (
subschema.get("type") == "object"
and not subschema.get("properties")
and subschema.get("additionalProperties")
):
# This is a dict type, handle it directly
additional_props = subschema["additionalProperties"]
if additional_props is True:
types.append(dict[str, Any]) # type: ignore
else:
value_type = _schema_to_type(additional_props, schemas)
types.append(dict[str, value_type]) # type: ignore
else:
types.append(_schema_to_type(subschema, schemas))
# Check if one of the types is None (null)
has_null = type(None) in types
types = [t for t in types if t is not type(None)]
if len(types) == 0:
return type(None)
elif len(types) == 1:
if has_null:
return types[0] | None # type: ignore
else:
return types[0]
else:
if has_null:
return Union[tuple(types + [type(None)])] # type: ignore # noqa: UP007
else:
return Union[tuple(types)] # type: ignore # noqa: UP007
schema_type = schema.get("type")
if not schema_type:
return Any # type: ignore[return-value]
if isinstance(schema_type, list):
# Create a copy of the schema for each type, but keep all constraints
types: list[type | Any] = []
for t in schema_type:
type_schema = dict(schema)
type_schema["type"] = t
types.append(_schema_to_type(type_schema, schemas))
has_null = type(None) in types
types = [t for t in types if t is not type(None)]
if has_null:
if len(types) == 1:
return types[0] | None # type: ignore
else:
return Union[tuple(types + [type(None)])] # type: ignore # noqa: UP007
return Union[tuple(types)] # type: ignore # noqa: UP007
return _get_from_type_handler(schema, schemas)(schema)
def _sanitize_name(name: str) -> str:
"""Convert string to valid Python identifier."""
# Step 1: replace everything except [0-9a-zA-Z_] with underscores
cleaned = re.sub(r"[^0-9a-zA-Z_]", "_", name)
# Step 2: deduplicate underscores
cleaned = re.sub(r"__+", "_", cleaned)
# Step 3: if the first char of original name isn't a letter, prepend field_
if not name or not re.match(r"[a-zA-Z]", name[0]):
cleaned = f"field_{cleaned}"
# Step 4: deduplicate again and strip trailing underscores
cleaned = re.sub(r"__+", "_", cleaned).strip("_")
return cleaned
def _get_default_value(
schema: dict[str, Any],
prop_name: str,
parent_default: dict[str, Any] | None = None,
) -> Any:
"""Get default value with proper priority ordering.
1. Value from parent's default if it exists
2. Property's own default if it exists
3. None
"""
if parent_default is not None and prop_name in parent_default:
return parent_default[prop_name]
return schema.get("default")
def _create_field_with_default(
field_type: type,
default_value: Any,
schema: dict[str, Any],
) -> Any:
"""Create a field with simplified default handling."""
# Always use None as default for complex types
if isinstance(default_value, dict | list) or default_value is None:
return field(default=None)
# For simple types, use the value directly
return field(default=default_value)
def _create_pydantic_model(
schema: Mapping[str, Any],
name: str | None = None,
schemas: Mapping[str, Any] | None = None,
) -> type:
"""Create Pydantic BaseModel from object schema with additionalProperties."""
name = name or schema.get("title", "Root")
assert name is not None # Should not be None after the or operation
sanitized_name = _sanitize_name(name)
schema_hash = _hash_schema(schema)
cache_key = (schema_hash, sanitized_name)
# Return existing class if already built
if cache_key in _classes:
existing = _classes[cache_key]
if existing is None:
return ForwardRef(sanitized_name) # type: ignore[return-value]
return existing
# Place placeholder for recursive references
_classes[cache_key] = None
properties = schema.get("properties", {})
required = schema.get("required", [])
# Build field annotations and defaults
annotations = {}
defaults = {}
for prop_name, prop_schema in properties.items():
field_type = _schema_to_type(prop_schema, schemas or {})
# Handle defaults
default_value = prop_schema.get("default", MISSING)
if default_value is not MISSING:
defaults[prop_name] = default_value
annotations[prop_name] = field_type
elif prop_name in required:
annotations[prop_name] = field_type
else:
annotations[prop_name] = Union[field_type, type(None)] # type: ignore[misc] # noqa: UP007
defaults[prop_name] = None
# Create Pydantic model class
cls_dict = {
"__annotations__": annotations,
"model_config": ConfigDict(extra="allow"),
**defaults,
}
cls = type(sanitized_name, (BaseModel,), cls_dict)
# Store completed class
_classes[cache_key] = cls
return cls
def _create_dataclass(
schema: Mapping[str, Any],
name: str | None = None,
schemas: Mapping[str, Any] | None = None,
) -> type:
"""Create dataclass from object schema."""
name = name or schema.get("title", "Root")
# Sanitize name for class creation
assert name is not None # Should not be None after the or operation
sanitized_name = _sanitize_name(name)
schema_hash = _hash_schema(schema)
cache_key = (schema_hash, sanitized_name)
original_schema = dict(schema) # Store copy for validator
# Return existing class if already built
if cache_key in _classes:
existing = _classes[cache_key]
if existing is None:
return ForwardRef(sanitized_name) # type: ignore[return-value]
return existing
# Place placeholder for recursive references
_classes[cache_key] = None
if "$ref" in schema:
ref = schema["$ref"]
if ref == "#":
return ForwardRef(sanitized_name) # type: ignore[return-value]
schema = _resolve_ref(ref, schemas or {})
properties = schema.get("properties", {})
required = schema.get("required", [])
fields: list[tuple[Any, ...]] = []
for prop_name, prop_schema in properties.items():
field_name = _sanitize_name(prop_name)
# Check for self-reference in property
if prop_schema.get("$ref") == "#":
field_type = ForwardRef(sanitized_name)
else:
field_type = _schema_to_type(prop_schema, schemas or {})
default_val = prop_schema.get("default", MISSING)
is_required = prop_name in required
# Include alias in field metadata
meta = {"alias": prop_name}
if default_val is not MISSING:
if isinstance(default_val, dict | list):
field_def = field(
default_factory=lambda d=default_val: deepcopy(d), metadata=meta
)
else:
field_def = field(default=default_val, metadata=meta)
else:
if is_required:
field_def = field(metadata=meta)
else:
field_def = field(default=None, metadata=meta)
if is_required and default_val is not MISSING:
fields.append((field_name, field_type, field_def))
elif is_required:
fields.append((field_name, field_type, field_def))
else:
fields.append((field_name, Union[field_type, type(None)], field_def)) # type: ignore[misc] # noqa: UP007
cls = make_dataclass(sanitized_name, fields, kw_only=True)
# Add model validator for defaults
@model_validator(mode="before")
@classmethod
def _apply_defaults(cls, data: Mapping[str, Any]):
if isinstance(data, dict):
return _merge_defaults(data, original_schema)
return data
setattr(cls, "_apply_defaults", _apply_defaults)
# Store completed class
_classes[cache_key] = cls
return cls
def _merge_defaults(
data: Mapping[str, Any],
schema: Mapping[str, Any],
parent_default: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
"""Merge defaults with provided data at all levels."""
# If we have no data
if not data:
# Start with parent default if available
if parent_default:
result = dict(parent_default)
# Otherwise use schema default if available
elif "default" in schema:
result = dict(schema["default"])
# Otherwise start empty
else:
result = {}
# If we have data and a parent default, merge them
elif parent_default:
result = dict(parent_default)
for key, value in data.items():
if (
isinstance(value, dict)
and key in result
and isinstance(result[key], dict)
):
# recursively merge nested dicts
result[key] = _merge_defaults(value, {"properties": {}}, result[key])
else:
result[key] = value
# Otherwise just use the data
else:
result = dict(data)
# For each property in the schema
for prop_name, prop_schema in schema.get("properties", {}).items():
# If property is missing, apply defaults in priority order
if prop_name not in result:
if parent_default and prop_name in parent_default:
result[prop_name] = parent_default[prop_name]
elif "default" in prop_schema:
result[prop_name] = prop_schema["default"]
# If property exists and is an object, recursively merge
if (
prop_name in result
and isinstance(result[prop_name], dict)
and prop_schema.get("type") == "object"
):
# Get the appropriate default for this nested object
nested_default = None
if parent_default and prop_name in parent_default:
nested_default = parent_default[prop_name]
elif "default" in prop_schema:
nested_default = prop_schema["default"]
result[prop_name] = _merge_defaults(
result[prop_name], prop_schema, nested_default
)
return result

7
src/fastmcp/utilities/mcp_config.py
View file

@ -19,7 +19,7 @@ if TYPE_CHECKING:
def infer_transport_type_from_url(
url: str | AnyUrl,
) -> Literal["streamable-http", "sse"]:
) -> Literal["http", "sse"]:
"""
Infer the appropriate transport type from the given URL.
"""
@ -34,7 +34,7 @@ def infer_transport_type_from_url(
if re.search(r"/sse(/|\?|&|$)", path):
return "sse"
else:
return "streamable-http"
return "http"
class StdioMCPServer(FastMCPBaseModel):
@ -58,7 +58,7 @@ class StdioMCPServer(FastMCPBaseModel):
class RemoteMCPServer(FastMCPBaseModel):
url: str
headers: dict[str, str] = Field(default_factory=dict)
transport: Literal["streamable-http", "sse"] | None = None
transport: Literal["http", "streamable-http", "sse"] | None = None
auth: Annotated[
str | Literal["oauth"] | httpx.Auth | None,
Field(
@ -79,6 +79,7 @@ class RemoteMCPServer(FastMCPBaseModel):
if transport == "sse":
return SSETransport(self.url, headers=self.headers, auth=self.auth)
else:
# Both "http" and "streamable-http" map to StreamableHttpTransport
return StreamableHttpTransport(
self.url, headers=self.headers, auth=self.auth
)

82
src/fastmcp/utilities/openapi.py
View file

@ -84,6 +84,7 @@ class HTTPRoute(FastMCPBaseModel):
schema_definitions: dict[str, JsonSchema] = Field(
default_factory=dict
) # Store component schemas
extensions: dict[str, Any] = Field(default_factory=dict)
# Export public symbols
@ -274,6 +275,12 @@ class OpenAPIParser(
result = {}
return _replace_ref_with_defs(result)
except ValueError as e:
# Re-raise ValueError for external reference errors and other validation issues
if "External or non-local reference not supported" in str(e):
raise
logger.error(f"Failed to extract schema as dict: {e}", exc_info=False)
return {}
except Exception as e:
logger.error(f"Failed to extract schema as dict: {e}", exc_info=False)
return {}
@ -302,11 +309,17 @@ class OpenAPIParser(
# Extract parameter info - handle both 3.0 and 3.1 parameter models
param_in = parameter.param_in # Both use param_in
param_location = self._convert_to_parameter_location(param_in)
# Handle enum or string parameter locations
from enum import Enum
param_in_str = (
param_in.value if isinstance(param_in, Enum) else param_in
)
param_location = self._convert_to_parameter_location(param_in_str)
param_schema_obj = parameter.param_schema # Both use param_schema
# Skip duplicate parameters (same name and location)
param_key = (parameter.name, param_in)
param_key = (parameter.name, param_in_str)
if param_key in seen_params:
continue
seen_params[param_key] = True
@ -400,12 +413,30 @@ class OpenAPIParser(
request_body_info.content_schema[media_type_str] = (
schema_dict
)
except ValueError as e:
# Re-raise ValueError for external reference errors
if "External or non-local reference not supported" in str(
e
):
raise
logger.error(
f"Failed to extract schema for media type '{media_type_str}': {e}"
)
except Exception as e:
logger.error(
f"Failed to extract schema for media type '{media_type_str}': {e}"
)
return request_body_info
except ValueError as e:
# Re-raise ValueError for external reference errors
if "External or non-local reference not supported" in str(e):
raise
ref_name = getattr(request_body_or_ref, "ref", "unknown")
logger.error(
f"Failed to extract request body '{ref_name}': {e}", exc_info=False
)
return None
except Exception as e:
ref_name = getattr(request_body_or_ref, "ref", "unknown")
logger.error(
@ -449,6 +480,17 @@ class OpenAPIParser(
media_type_obj.media_type_schema
)
resp_info.content_schema[media_type_str] = schema_dict
except ValueError as e:
# Re-raise ValueError for external reference errors
if (
"External or non-local reference not supported"
in str(e)
):
raise
logger.error(
f"Failed to extract schema for media type '{media_type_str}' "
f"in response {status_code}: {e}"
)
except Exception as e:
logger.error(
f"Failed to extract schema for media type '{media_type_str}' "
@ -456,6 +498,16 @@ class OpenAPIParser(
)
extracted_responses[str(status_code)] = resp_info
except ValueError as e:
# Re-raise ValueError for external reference errors
if "External or non-local reference not supported" in str(e):
raise
ref_name = getattr(resp_or_ref, "ref", "unknown")
logger.error(
f"Failed to extract response for status code {status_code} "
f"from reference '{ref_name}': {e}",
exc_info=False,
)
except Exception as e:
ref_name = getattr(resp_or_ref, "ref", "unknown")
logger.error(
@ -540,6 +592,14 @@ class OpenAPIParser(
getattr(operation, "responses", None)
)
extensions = {}
if hasattr(operation, "model_extra") and operation.model_extra:
extensions = {
k: v
for k, v in operation.model_extra.items()
if k.startswith("x-")
}
route = HTTPRoute(
path=path_str,
method=method_upper, # type: ignore[arg-type] # Known valid HTTP method
@ -551,11 +611,23 @@ class OpenAPIParser(
request_body=request_body_info,
responses=responses,
schema_definitions=schema_definitions,
extensions=extensions,
)
routes.append(route)
logger.info(
f"Successfully extracted route: {method_upper} {path_str}"
)
except ValueError as op_error:
# Re-raise ValueError for external reference errors
if "External or non-local reference not supported" in str(
op_error
):
raise
op_id = getattr(operation, "operationId", "unknown")
logger.error(
f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}",
exc_info=True,
)
except Exception as op_error:
op_id = getattr(operation, "operationId", "unknown")
logger.error(
@ -901,6 +973,12 @@ def _replace_ref_with_defs(
if ref_path.startswith("#/components/schemas/"):
schema_name = ref_path.split("/")[-1]
schema["$ref"] = f"#/$defs/{schema_name}"
elif not ref_path.startswith("#/"):
raise ValueError(
f"External or non-local reference not supported: {ref_path}. "
f"FastMCP only supports local schema references starting with '#/'. "
f"Please include all schema definitions within the OpenAPI document."
)
elif properties := schema.get("properties"):
if "$ref" in properties:
schema["properties"] = _replace_ref_with_defs(properties)

2
src/fastmcp/utilities/tests.py
View file

@ -20,7 +20,7 @@ if TYPE_CHECKING:
@contextmanager
def temporary_settings(**kwargs: Any):
"""
Temporarily override ControlFlow setting values.
Temporarily override FastMCP setting values.
Args:
**kwargs: The settings to override, including nested settings.

70
src/fastmcp/utilities/types.py
View file

@ -6,21 +6,19 @@ import mimetypes
from collections.abc import Callable
from functools import lru_cache
from pathlib import Path
from types import UnionType
from typing import Annotated, TypeVar, Union, get_args, get_origin
from types import EllipsisType, UnionType
from typing import Annotated, TypeAlias, TypeVar, Union, get_args, get_origin
from mcp.types import (
Annotations,
AudioContent,
BlobResourceContents,
EmbeddedResource,
ImageContent,
TextResourceContents,
)
import mcp.types
from mcp.types import Annotations
from pydantic import AnyUrl, BaseModel, ConfigDict, TypeAdapter, UrlConstraints
T = TypeVar("T")
# sentinel values for optional arguments
NotSet = ...
NotSetT: TypeAlias = EllipsisType
class FastMCPBaseModel(BaseModel):
"""Base model for FastMCP models."""
@ -129,7 +127,7 @@ class Image:
self,
mime_type: str | None = None,
annotations: Annotations | None = None,
) -> ImageContent:
) -> mcp.types.ImageContent:
"""Convert to MCP ImageContent."""
if self.path:
with open(self.path, "rb") as f:
@ -139,7 +137,7 @@ class Image:
else:
raise ValueError("No image data available")
return ImageContent(
return mcp.types.ImageContent(
type="image",
data=data,
mimeType=mime_type or self._mime_type,
@ -188,7 +186,7 @@ class Audio:
self,
mime_type: str | None = None,
annotations: Annotations | None = None,
) -> AudioContent:
) -> mcp.types.AudioContent:
if self.path:
with open(self.path, "rb") as f:
data = base64.b64encode(f.read()).decode()
@ -197,7 +195,7 @@ class Audio:
else:
raise ValueError("No audio data available")
return AudioContent(
return mcp.types.AudioContent(
type="audio",
data=data,
mimeType=mime_type or self._mime_type,
@ -248,7 +246,7 @@ class File:
self,
mime_type: str | None = None,
annotations: Annotations | None = None,
) -> EmbeddedResource:
) -> mcp.types.EmbeddedResource:
if self.path:
with open(self.path, "rb") as f:
raw_data = f.read()
@ -271,21 +269,57 @@ class File:
text = raw_data.decode("utf-8")
except UnicodeDecodeError:
text = raw_data.decode("latin-1")
resource = TextResourceContents(
resource = mcp.types.TextResourceContents(
text=text,
mimeType=mime,
uri=uri,
)
else:
data = base64.b64encode(raw_data).decode()
resource = BlobResourceContents(
resource = mcp.types.BlobResourceContents(
blob=data,
mimeType=mime,
uri=uri,
)
return EmbeddedResource(
return mcp.types.EmbeddedResource(
type="resource",
resource=resource,
annotations=annotations or self.annotations,
)
def replace_type(type_, type_map: dict[type, type]):
"""
Given a (possibly generic, nested, or otherwise complex) type, replaces all
instances of old_type with new_type.
This is useful for transforming types when creating tools.
Args:
type_: The type to replace instances of old_type with new_type.
old_type: The type to replace.
new_type: The type to replace old_type with.
Examples:
>>> replace_type(list[int | bool], {int: str})
list[str | bool]
>>> replace_type(list[list[int]], {int: str})
list[list[str]]
"""
if type_ in type_map:
return type_map[type_]
origin = get_origin(type_)
if not origin:
return type_
args = get_args(type_)
new_args = tuple(replace_type(arg, type_map) for arg in args)
if origin is UnionType:
return Union[new_args] # type: ignore # noqa: UP007
else:
return origin[new_args]