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

Merge branch 'main' into sse-bugfix

Browse source
This commit is contained in:
Jeremiah Lowin 2025-05-14 17:21:15 -04:00 • committed by GitHub
commit af7b7a9314
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
29 changed files with 1226 additions and 325 deletions
Download patch file Download diff file Expand all files Collapse all files

49
docs/servers/resources.mdx
View file

@ -147,6 +147,7 @@ async def read_important_log() -> str:
return "Log file not found."
```
### Resource Classes
While `@mcp.resource` is ideal for dynamic content, you can directly register pre-defined resources (like static files or simple text) using `mcp.add_resource()` and concrete `Resource` subclasses.
@ -403,18 +404,46 @@ In this stacked decorator pattern:
- Each parameter defaults to `None` when not included in the URI
- The function logic handles whichever parameter is provided
**How Templates Work:**
1. **Definition:** When FastMCP sees `{...}` placeholders in the `@resource` URI and matching function parameters, it registers a `ResourceTemplate`.
2. **Discovery:** Clients list templates via `resources/listResourceTemplates`.
3. **Request & Matching:** A client requests a specific URI, e.g., `weather://london/current`. FastMCP matches this to the `weather://{city}/current` template.
4. **Parameter Extraction:** It extracts the parameter value: `city="london"`.
5. **Type Conversion & Function Call:** It converts extracted values to the types hinted in the function and calls `get_weather(city="london")`.
6. **Default Values:** For any function parameters with default values not included in the URI template, FastMCP uses the default values.
7. **Response:** The function's return value is formatted (e.g., dict to JSON) and sent back as the resource content.
Templates provide a powerful way to expose parameterized data access points following REST-like principles.
## Error Handling
<VersionBadge version="2.3.4" />
If your resource function encounters an error, you can raise a standard Python exception (`ValueError`, `TypeError`, `FileNotFoundError`, custom exceptions, etc.) or a FastMCP `ResourceError`.
For security reasons, most exceptions are wrapped in a generic `ResourceError` before being sent to the client, with internal error details masked. However, if you raise a `ResourceError` directly, its contents **are** included in the response. This allows you to provide informative error messages to the client on an opt-in basis.
```python
from fastmcp import FastMCP
from fastmcp.exceptions import ResourceError
mcp = FastMCP(name="DataServer")
@mcp.resource("resource://safe-error")
def fail_with_details() -> str:
"""This resource provides detailed error information."""
# ResourceError contents are sent back to clients
raise ResourceError("Unable to retrieve data: file not found")
@mcp.resource("resource://masked-error")
def fail_with_masked_details() -> str:
"""This resource masks internal error details."""
# Other exceptions are converted to ResourceError with generic message
raise ValueError("Sensitive internal file path: /etc/secrets.conf")
@mcp.resource("data://{id}")
def get_data_by_id(id: str) -> dict:
"""Template resources also support the same error handling pattern."""
if id == "secure":
raise ValueError("Cannot access secure data")
elif id == "missing":
raise ResourceError("Data ID 'missing' not found in database")
return {"id": id, "value": "data"}
```
This error handling pattern applies to both regular resources and resource templates.
## Server Behavior
### Duplicate Resources

46
docs/servers/tools.mdx
View file

@ -248,27 +248,30 @@ def do_nothing() -> None:
### Error Handling
If your tool encounters an error, simply raise a standard Python exception (`ValueError`, `TypeError`, `FileNotFoundError`, custom exceptions, etc.).
<VersionBadge version="2.3.4" />
If your tool encounters an error, you can raise a standard Python exception (`ValueError`, `TypeError`, `FileNotFoundError`, custom exceptions, etc.) or a FastMCP `ToolError`.
In all cases, the exception is logged and converted into an MCP error response to be sent back to the client LLM. For security reasons, the error message is **not** included in the response by default. However, if you raise a `ToolError`, the contents of the exception **are** included in the response. This allows you to provide informative error messages to the client LLM on an opt-in basis, which can help the LLM understand failures and react appropriately.
```python {2, 10, 14}
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
```python
@mcp.tool()
def divide(a: float, b: float) -> float:
"""Divide a by b."""
if b == 0:
# Raise a standard exception
raise ValueError("Division by zero is not allowed.")
# Python exceptions raise errors but the contents are not sent to clients
if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
raise TypeError("Both arguments must be numbers.")
if b == 0:
# ToolError contents are sent back to clients
raise ToolError("Division by zero is not allowed.")
return a / b
```
FastMCP automatically catches exceptions raised within your tool function:
1. It converts the exception into an MCP error response, typically including the exception type and message.
2. This error response is sent back to the client/LLM.
3. The LLM can then inform the user or potentially try the tool again with different arguments.
Using informative exceptions helps the LLM understand failures and react appropriately.
### Annotations
<VersionBadge version="2.2.7" />
@ -709,6 +712,25 @@ The duplicate behavior options are:
- `"replace"`: Silently replaces the existing tool with the new one.
- `"ignore"`: Keeps the original tool and ignores the new registration attempt.
### Removing Tools
<VersionBadge version="2.3.4" />
You can dynamically remove tools from a server using the `remove_tool` method:
```python
from fastmcp import FastMCP
mcp = FastMCP(name="DynamicToolServer")
@mcp.tool()
def calculate_sum(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
mcp.remove_tool("calculate_sum")
```
### Legacy JSON Parsing
<VersionBadge version="2.2.10" />

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

@ -13,7 +13,8 @@ from mcp.types import PromptArgument as MCPPromptArgument
from pydantic import BaseModel, BeforeValidator, Field, TypeAdapter, validate_call
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.json_schema import prune_params
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
_convert_set_defaults,
find_kwarg_by_type,
@ -25,6 +26,8 @@ if TYPE_CHECKING:
CONTENT_TYPES = TextContent | ImageContent | EmbeddedResource
logger = get_logger(__name__)
def Message(
content: str | CONTENT_TYPES, role: Role | None = None, **kwargs: Any
@ -112,7 +115,11 @@ class Prompt(BaseModel):
context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context)
if context_kwarg:
parameters = prune_params(parameters, params=[context_kwarg])
prune_params = [context_kwarg]
else:
prune_params = None
parameters = compress_schema(parameters, prune_params=prune_params)
# Convert parameters to PromptArguments
arguments: list[PromptArgument] = []
@ -192,13 +199,12 @@ class Prompt(BaseModel):
)
)
except Exception:
raise ValueError(
f"Could not convert prompt result to message: {msg}"
)
raise ValueError("Could not convert prompt result to message.")
return messages
except Exception as e:
raise ValueError(f"Error rendering prompt {self.name}: {e}")
logger.exception(f"Error rendering prompt {self.name}: {e}")
raise ValueError(f"Error rendering prompt {self.name}.")
def __eq__(self, other: object) -> bool:
if not isinstance(other, Prompt):

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

@ -6,7 +6,7 @@ from typing import Any
from pydantic import AnyUrl
from fastmcp.exceptions import NotFoundError
from fastmcp.exceptions import NotFoundError, ResourceError
from fastmcp.resources import FunctionResource
from fastmcp.resources.resource import Resource
from fastmcp.resources.template import (
@ -244,11 +244,32 @@ class ResourceManager:
uri_str,
params=params,
)
except ResourceError as e:
logger.error(f"Error creating resource from template: {e}")
raise e
except Exception as e:
logger.error(f"Error creating resource from template: {e}")
raise ValueError(f"Error creating resource from template: {e}")
raise NotFoundError(f"Unknown resource: {uri_str}")
async def read_resource(self, uri: AnyUrl | str) -> str | bytes:
"""Read a resource contents."""
resource = await self.get_resource(uri)
try:
return await resource.read()
# raise ResourceErrors as-is
except ResourceError as e:
logger.error(f"Error reading resource {uri!r}: {e}")
raise e
# raise other exceptions as ResourceErrors without revealing internal details
except Exception as e:
logger.error(f"Error reading resource {uri!r}: {e}")
raise ResourceError(f"Error reading resource {uri!r}") from e
def get_resources(self) -> dict[str, Resource]:
"""Get all registered resources, keyed by URI."""
return self._resources

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

@ -21,6 +21,7 @@ from pydantic import (
from fastmcp.resources.types import FunctionResource, Resource
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.types import (
_convert_set_defaults,
find_kwarg_by_type,
@ -150,6 +151,10 @@ class ResourceTemplate(BaseModel):
# Get schema from TypeAdapter - will fail if function isn't properly typed
parameters = TypeAdapter(fn).json_schema()
# compress the schema
prune_params = [context_kwarg] if context_kwarg else None
parameters = compress_schema(parameters, prune_params=prune_params)
# ensure the arguments are properly cast
fn = validate_call(fn)
@ -171,28 +176,27 @@ class ResourceTemplate(BaseModel):
"""Create a resource from the template with the given parameters."""
from fastmcp.server.context import Context
try:
# Add context to parameters if needed
kwargs = params.copy()
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg and context_kwarg not in kwargs:
kwargs[context_kwarg] = get_context()
# Add context to parameters if needed
kwargs = params.copy()
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg and context_kwarg not in kwargs:
kwargs[context_kwarg] = get_context()
async def resource_read_fn() -> str | bytes:
# Call function and check if result is a coroutine
result = self.fn(**kwargs)
if inspect.iscoroutine(result):
result = await result
return result
return FunctionResource(
uri=AnyUrl(uri), # Explicitly convert to AnyUrl
name=self.name,
description=self.description,
mime_type=self.mime_type,
fn=lambda **kwargs: result, # Capture result in closure
tags=self.tags,
)
except Exception as e:
raise ValueError(f"Error creating resource from template: {e}")
return FunctionResource(
uri=AnyUrl(uri), # Explicitly convert to AnyUrl
name=self.name,
description=self.description,
mime_type=self.mime_type,
fn=resource_read_fn,
tags=self.tags,
)
def __eq__(self, other: object) -> bool:
if not isinstance(other, ResourceTemplate):

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

@ -6,7 +6,7 @@ import inspect
import json
from collections.abc import Callable
from pathlib import Path
from typing import TYPE_CHECKING, Any
from typing import Any
import anyio
import anyio.to_thread
@ -15,12 +15,13 @@ import pydantic.json
import pydantic_core
from pydantic import Field, ValidationInfo
from fastmcp.exceptions import ResourceError
from fastmcp.resources.resource import Resource
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import find_kwarg_by_type
if TYPE_CHECKING:
pass
logger = get_logger(__name__)
class TextResource(Resource):
@ -62,26 +63,23 @@ class FunctionResource(Resource):
"""Read the resource by calling the wrapped function."""
from fastmcp.server.context import Context
try:
kwargs = {}
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg is not None:
kwargs[context_kwarg] = get_context()
kwargs = {}
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg is not None:
kwargs[context_kwarg] = get_context()
result = self.fn(**kwargs)
if inspect.iscoroutinefunction(self.fn):
result = await result
result = self.fn(**kwargs)
if inspect.iscoroutinefunction(self.fn):
result = await result
if isinstance(result, Resource):
return await result.read()
elif isinstance(result, bytes):
return result
elif isinstance(result, str):
return result
else:
return pydantic_core.to_json(result, fallback=str, indent=2).decode()
except Exception as e:
raise ValueError(f"Error reading resource {self.uri}: {e}")
if isinstance(result, Resource):
return await result.read()
elif isinstance(result, bytes):
return result
elif isinstance(result, str):
return result
else:
return pydantic_core.to_json(result, fallback=str, indent=2).decode()
class FileResource(Resource):
@ -124,7 +122,7 @@ class FileResource(Resource):
return await anyio.to_thread.run_sync(self.path.read_bytes)
return await anyio.to_thread.run_sync(self.path.read_text)
except Exception as e:
raise ValueError(f"Error reading file {self.path}: {e}")
raise ResourceError(f"Error reading file {self.path}") from e
class HttpResource(Resource):
@ -185,7 +183,7 @@ class DirectoryResource(Resource):
else list(self.path.rglob("*"))
)
except Exception as e:
raise ValueError(f"Error listing directory {self.path}: {e}")
raise ResourceError(f"Error listing directory {self.path}: {e}")
async def read(self) -> str: # Always returns JSON string
"""Read the directory listing."""
@ -193,5 +191,5 @@ class DirectoryResource(Resource):
files = await anyio.to_thread.run_sync(self.list_files)
file_list = [str(f.relative_to(self.path)) for f in files if f.is_file()]
return json.dumps({"files": file_list}, indent=2)
except Exception as e:
raise ValueError(f"Error reading directory {self.path}: {e}")
except Exception:
raise ResourceError(f"Error reading directory {self.path}")

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

@ -14,6 +14,7 @@ import httpx
from mcp.types import EmbeddedResource, ImageContent, TextContent, ToolAnnotations
from pydantic.networks import AnyUrl
from fastmcp.exceptions import ToolError
from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.server.server import FastMCP
from fastmcp.tools.tool import Tool, _convert_to_content
@ -137,6 +138,10 @@ class OpenAPITool(Tool):
self._route = route
self._timeout = timeout
def __repr__(self) -> str:
"""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 _execute_request(self, *args, **kwargs):
"""Execute the HTTP request based on the route configuration."""
context = kwargs.get("context")
@ -163,7 +168,7 @@ class OpenAPITool(Tool):
}
missing_params = required_path_params - path_params.keys()
if missing_params:
raise ValueError(f"Missing required path parameters: {missing_params}")
raise ToolError(f"Missing required path parameters: {missing_params}")
for param_name, param_value in path_params.items():
path = path.replace(f"{{{param_name}}}", str(param_value))
@ -286,6 +291,10 @@ class OpenAPIResource(Resource):
self._route = route
self._timeout = timeout
def __repr__(self) -> str:
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPIResource(name={self.name!r}, uri={self.uri!r}, path={self._route.path})"
async def read(self) -> str | bytes:
"""Fetch the resource data by making an HTTP request."""
try:
@ -396,6 +405,10 @@ class OpenAPIResourceTemplate(ResourceTemplate):
self._route = route
self._timeout = timeout
def __repr__(self) -> str:
"""Custom representation to prevent recursion errors when printing."""
return f"OpenAPIResourceTemplate(name={self.name!r}, uri_template={self.uri_template!r}, path={self._route.path})"
async def create_resource(
self,
uri: str,

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

@ -18,7 +18,7 @@ from mcp.types import (
from pydantic.networks import AnyUrl
from fastmcp.client import Client
from fastmcp.exceptions import NotFoundError
from fastmcp.exceptions import NotFoundError, ResourceError, ToolError
from fastmcp.prompts import Prompt, PromptMessage
from fastmcp.resources import Resource, ResourceTemplate
from fastmcp.server.context import Context
@ -64,7 +64,7 @@ class ProxyTool(Tool):
arguments=arguments,
)
if result.isError:
raise ValueError(cast(mcp.types.TextContent, result.content[0]).text)
raise ToolError(cast(mcp.types.TextContent, result.content[0]).text)
return result.content
@ -97,7 +97,7 @@ class ProxyResource(Resource):
elif isinstance(result[0], BlobResourceContents):
return result[0].blob
else:
raise ValueError(f"Unsupported content type: {type(result[0])}")
raise ResourceError(f"Unsupported content type: {type(result[0])}")
class ProxyTemplate(ResourceTemplate):
@ -138,7 +138,7 @@ class ProxyTemplate(ResourceTemplate):
elif isinstance(result[0], BlobResourceContents):
value = result[0].blob
else:
raise ValueError(f"Unsupported content type: {type(result[0])}")
raise ResourceError(f"Unsupported content type: {type(result[0])}")
return ProxyResource(
client=self._client,

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

@ -19,7 +19,7 @@ import pydantic
import uvicorn
from mcp.server.auth.provider import OAuthAuthorizationServerProvider
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.lowlevel.server import LifespanResultT
from mcp.server.lowlevel.server import LifespanResultT, NotificationOptions
from mcp.server.lowlevel.server import Server as MCPServer
from mcp.server.stdio import stdio_server
from mcp.types import (
@ -43,7 +43,7 @@ from starlette.routing import BaseRoute, Route
import fastmcp.server
import fastmcp.settings
from fastmcp.exceptions import NotFoundError, ResourceError
from fastmcp.exceptions import NotFoundError
from fastmcp.prompts import Prompt, PromptManager
from fastmcp.prompts.prompt import PromptResult
from fastmcp.resources import Resource, ResourceManager
@ -385,16 +385,13 @@ class FastMCP(Generic[LifespanResultT]):
with fastmcp.server.context.Context(fastmcp=self):
if self._resource_manager.has_resource(uri):
resource = await self._resource_manager.get_resource(uri)
try:
content = await resource.read()
return [
ReadResourceContents(
content=content, mime_type=resource.mime_type
)
]
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
raise ResourceError(str(e))
content = await self._resource_manager.read_resource(uri)
return [
ReadResourceContents(
content=content,
mime_type=resource.mime_type,
)
]
else:
for server in self._mounted_servers.values():
if server.match_resource(str(uri)):
@ -457,6 +454,18 @@ class FastMCP(Generic[LifespanResultT]):
)
self._cache.clear()
def remove_tool(self, name: str) -> None:
"""Remove a tool from the server.
Args:
name: The name of the tool to remove
Raises:
NotFoundError: If the tool is not found
"""
self._tool_manager.remove_tool(name)
self._cache.clear()
def tool(
self,
name: str | None = None,
@ -722,7 +731,9 @@ class FastMCP(Generic[LifespanResultT]):
await self._mcp_server.run(
read_stream,
write_stream,
self._mcp_server.create_initialization_options(),
self._mcp_server.create_initialization_options(
NotificationOptions(tools_changed=True)
),
)
async def run_http_async(

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

@ -11,9 +11,8 @@ from mcp.types import Tool as MCPTool
from pydantic import BaseModel, BeforeValidator, Field
import fastmcp
from fastmcp.exceptions import ToolError
from fastmcp.server.dependencies import get_context
from fastmcp.utilities.json_schema import prune_params
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
Image,
@ -82,7 +81,11 @@ class Tool(BaseModel):
context_kwarg = find_kwarg_by_type(fn, kwarg_type=Context)
if context_kwarg:
schema = prune_params(schema, params=[context_kwarg])
prune_params = [context_kwarg]
else:
prune_params = None
schema = compress_schema(schema, prune_params=prune_params)
return cls(
fn=fn,
@ -102,49 +105,45 @@ class Tool(BaseModel):
arguments = arguments.copy()
try:
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg and context_kwarg not in arguments:
arguments[context_kwarg] = get_context()
context_kwarg = find_kwarg_by_type(self.fn, kwarg_type=Context)
if context_kwarg and context_kwarg not in arguments:
arguments[context_kwarg] = get_context()
if fastmcp.settings.settings.tool_attempt_parse_json_args:
# Pre-parse data from JSON in order to handle cases like `["a", "b", "c"]`
# 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)
if fastmcp.settings.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
except json.JSONDecodeError:
pass
type_adapter = get_cached_typeadapter(self.fn)
result = type_adapter.validate_python(arguments)
if inspect.isawaitable(result):
result = await result
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)
except Exception as e:
logger.exception(f"Tool {self.name} failed")
raise ToolError(f"Error executing tool {self.name}: {e}") from e
return _convert_to_content(result, serializer=self.serializer)
def to_mcp_tool(self, **overrides: Any) -> MCPTool:
kwargs = {

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

@ -5,7 +5,7 @@ from typing import TYPE_CHECKING, Any
from mcp.types import EmbeddedResource, ImageContent, TextContent, ToolAnnotations
from fastmcp.exceptions import NotFoundError
from fastmcp.exceptions import NotFoundError, ToolError
from fastmcp.settings import DuplicateBehavior
from fastmcp.tools.tool import Tool
from fastmcp.utilities.logging import get_logger
@ -94,6 +94,20 @@ class ToolManager:
self._tools[key] = tool
return tool
def remove_tool(self, key: str) -> None:
"""Remove a tool from the server.
Args:
key: The key of the tool to remove
Raises:
NotFoundError: If the tool is not found
"""
if key in self._tools:
del self._tools[key]
else:
raise NotFoundError(f"Unknown tool: {key}")
async def call_tool(
self, key: str, arguments: dict[str, Any]
) -> list[TextContent | ImageContent | EmbeddedResource]:
@ -102,4 +116,15 @@ class ToolManager:
if not tool:
raise NotFoundError(f"Unknown tool: {key}")
return await tool.run(arguments)
try:
return await tool.run(arguments)
# raise ToolErrors as-is
except ToolError as e:
logger.exception(f"Error calling tool {key!r}: {e}")
raise e
# raise other exceptions as ToolErrors without revealing internal details
except Exception as e:
logger.exception(f"Error calling tool {key!r}: {e}")
raise ToolError(f"Error calling tool {key!r}") from e

95
src/fastmcp/utilities/json_schema.py
View file

@ -1,7 +1,6 @@
from __future__ import annotations
import copy
from collections.abc import Mapping, Sequence
def _prune_param(schema: dict, param: str) -> dict:
@ -14,6 +13,7 @@ def _prune_param(schema: dict, param: str) -> dict:
removed = props.pop(param, None)
if removed is None: # nothing to do
return schema
# Keep empty properties object rather than removing it entirely
schema["properties"] = props
if param in schema.get("required", []):
@ -21,39 +21,100 @@ def _prune_param(schema: dict, param: str) -> dict:
if not schema["required"]:
schema.pop("required")
# ── 2. collect all remaining local $ref targets ───────────────────
return schema
def _walk_and_prune(
schema: dict,
prune_defs: bool = False,
prune_titles: bool = False,
prune_additional_properties: bool = False,
) -> dict:
"""Walk the schema and optionally prune titles, unused definitions, and additionalProperties: false."""
# Will only be used if prune_defs is True
used_defs: set[str] = set()
def walk(node: object) -> None: # depth-first traversal
if isinstance(node, Mapping):
ref = node.get("$ref")
if isinstance(ref, str) and ref.startswith("#/$defs/"):
used_defs.add(ref.split("/")[-1])
def walk(node: object) -> None:
if isinstance(node, dict):
# Process $ref for definition tracking
if prune_defs:
ref = node.get("$ref")
if isinstance(ref, str) and ref.startswith("#/$defs/"):
used_defs.add(ref.split("/")[-1])
# Remove title if requested
if prune_titles and "title" in node:
node.pop("title")
# Remove additionalProperties: false at any level if requested
if (
prune_additional_properties
and node.get("additionalProperties", None) is False
):
node.pop("additionalProperties")
# Walk children
for v in node.values():
walk(v)
elif isinstance(node, Sequence) and not isinstance(node, str | bytes):
elif isinstance(node, list):
for v in node:
walk(v)
# Traverse the schema once
walk(schema)
# ── 3. remove orphaned definitions ────────────────────────────────
defs = schema.get("$defs", {})
for def_name in list(defs):
if def_name not in used_defs:
defs.pop(def_name)
if not defs:
schema.pop("$defs", None)
# Remove orphaned definitions if requested
if prune_defs:
defs = schema.get("$defs", {})
for def_name in list(defs):
if def_name not in used_defs:
defs.pop(def_name)
if not defs:
schema.pop("$defs", None)
return schema
def prune_params(schema: dict, params: list[str]) -> dict:
def _prune_additional_properties(schema: dict) -> dict:
"""Remove additionalProperties from the schema if it is False."""
if schema.get("additionalProperties", None) is False:
schema.pop("additionalProperties")
return schema
def compress_schema(
schema: dict,
prune_params: list[str] | None = None,
prune_defs: bool = True,
prune_additional_properties: bool = True,
prune_titles: bool = False,
) -> dict:
"""
Remove the given parameters from the schema.
Args:
schema: The schema to compress
prune_params: List of parameter names to remove from properties
prune_defs: Whether to remove unused definitions
prune_additional_properties: Whether to remove additionalProperties: false
prune_titles: Whether to remove title fields from the schema
"""
# Make a copy so we don't modify the original
schema = copy.deepcopy(schema)
for param in params:
# Remove specific parameters if requested
for param in prune_params or []:
schema = _prune_param(schema, param=param)
# Do a single walk to handle pruning operations
if prune_defs or prune_titles or prune_additional_properties:
schema = _walk_and_prune(
schema,
prune_defs=prune_defs,
prune_titles=prune_titles,
prune_additional_properties=prune_additional_properties,
)
return schema

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

@ -84,6 +84,9 @@ class HTTPRoute(BaseModel):
responses: dict[str, ResponseInfo] = Field(
default_factory=dict
) # Key: status code str
schema_definitions: dict[str, JsonSchema] = Field(
default_factory=dict
) # Store component schemas
# Export public symbols
@ -221,6 +224,27 @@ class OpenAPI31Parser(BaseOpenAPIParser):
logger.warning("OpenAPI schema has no paths defined.")
return []
# Extract component schemas to add to each route
schema_definitions = {}
if hasattr(self.openapi, "components") and self.openapi.components:
components = self.openapi.components
if hasattr(components, "schemas") and components.schemas:
for name, schema in components.schemas.items():
try:
if isinstance(schema, Reference):
resolved_schema = self._resolve_ref(schema)
schema_definitions[name] = self._extract_schema_as_dict(
resolved_schema
)
else:
schema_definitions[name] = self._extract_schema_as_dict(
schema
)
except Exception as e:
logger.warning(
f"Failed to extract schema definition '{name}': {e}"
)
for path_str, path_item_obj in self.openapi.paths.items():
if not isinstance(path_item_obj, PathItem):
logger.warning(
@ -269,6 +293,7 @@ class OpenAPI31Parser(BaseOpenAPIParser):
parameters=parameters,
request_body=request_body_info,
responses=responses,
schema_definitions=schema_definitions,
)
routes.append(route)
logger.info(
@ -386,16 +411,36 @@ class OpenAPI31Parser(BaseOpenAPIParser):
param_schema_dict = {}
if param_schema_obj: # Check if schema exists
# Resolve the schema if it's a reference
resolved_schema = self._resolve_ref(param_schema_obj)
param_schema_dict = self._extract_schema_as_dict(param_schema_obj)
# Ensure default value is preserved from resolved schema
if (
not isinstance(resolved_schema, Reference)
and hasattr(resolved_schema, "default")
and resolved_schema.default is not None
):
param_schema_dict["default"] = resolved_schema.default
elif parameter.content:
# Handle complex parameters with 'content'
first_media_type = next(iter(parameter.content.values()), None)
if (
first_media_type and first_media_type.media_type_schema
): # CORRECTED: Use 'media_type_schema'
param_schema_dict = self._extract_schema_as_dict(
first_media_type.media_type_schema
)
# Resolve the schema if it's a reference
media_schema = first_media_type.media_type_schema
resolved_media_schema = self._resolve_ref(media_schema)
param_schema_dict = self._extract_schema_as_dict(media_schema)
# Ensure default value is preserved from resolved schema
if (
not isinstance(resolved_media_schema, Reference)
and hasattr(resolved_media_schema, "default")
and resolved_media_schema.default is not None
):
param_schema_dict["default"] = resolved_media_schema.default
logger.debug(
f"Parameter '{parameter.name}' using schema from 'content' field."
)
@ -543,6 +588,27 @@ class OpenAPI30Parser(BaseOpenAPIParser):
logger.warning("OpenAPI schema has no paths defined.")
return []
# Extract component schemas to add to each route
schema_definitions = {}
if hasattr(self.openapi, "components") and self.openapi.components:
components = self.openapi.components
if hasattr(components, "schemas") and components.schemas:
for name, schema in components.schemas.items():
try:
if isinstance(schema, Reference_30):
resolved_schema = self._resolve_ref(schema)
schema_definitions[name] = self._extract_schema_as_dict(
resolved_schema
)
else:
schema_definitions[name] = self._extract_schema_as_dict(
schema
)
except Exception as e:
logger.warning(
f"Failed to extract schema definition '{name}': {e}"
)
for path_str, path_item_obj in self.openapi.paths.items():
if not isinstance(path_item_obj, PathItem_30):
logger.warning(
@ -593,6 +659,7 @@ class OpenAPI30Parser(BaseOpenAPIParser):
parameters=parameters,
request_body=request_body_info,
responses=responses,
schema_definitions=schema_definitions,
)
routes.append(route)
logger.info(
@ -711,14 +778,34 @@ class OpenAPI30Parser(BaseOpenAPIParser):
param_schema_dict = {}
if param_schema_obj: # Check if schema exists
# Resolve the schema if it's a reference
resolved_schema = self._resolve_ref(param_schema_obj)
param_schema_dict = self._extract_schema_as_dict(param_schema_obj)
# Ensure default value is preserved from resolved schema
if (
not isinstance(resolved_schema, Reference_30)
and hasattr(resolved_schema, "default")
and resolved_schema.default is not None
):
param_schema_dict["default"] = resolved_schema.default
elif parameter.content:
# Handle complex parameters with 'content'
first_media_type = next(iter(parameter.content.values()), None)
if first_media_type and first_media_type.media_type_schema:
param_schema_dict = self._extract_schema_as_dict(
first_media_type.media_type_schema
)
# Resolve the schema if it's a reference
media_schema = first_media_type.media_type_schema
resolved_media_schema = self._resolve_ref(media_schema)
param_schema_dict = self._extract_schema_as_dict(media_schema)
# Ensure default value is preserved from resolved schema
if (
not isinstance(resolved_media_schema, Reference_30)
and hasattr(resolved_media_schema, "default")
and resolved_media_schema.default is not None
):
param_schema_dict["default"] = resolved_media_schema.default
logger.debug(
f"Parameter '{parameter.name}' using schema from 'content' field."
)
@ -1173,6 +1260,23 @@ def _combine_schemas(route: openapi.HTTPRoute) -> dict[str, Any]:
# Copy the schema and add description if available
param_schema = param.schema_.copy() if isinstance(param.schema_, dict) else {}
# Convert #/components/schemas references to #/$defs references
if isinstance(param_schema, dict) and "$ref" in param_schema:
ref_path = param_schema["$ref"]
if ref_path.startswith("#/components/schemas/"):
schema_name = ref_path.split("/")[-1]
param_schema["$ref"] = f"#/$defs/{schema_name}"
# Also handle anyOf, allOf, oneOf references
for section in ["anyOf", "allOf", "oneOf"]:
if section in param_schema and isinstance(param_schema[section], list):
for i, item in enumerate(param_schema[section]):
if isinstance(item, dict) and "$ref" in item:
ref_path = item["$ref"]
if ref_path.startswith("#/components/schemas/"):
schema_name = ref_path.split("/")[-1]
param_schema[section][i]["$ref"] = f"#/$defs/{schema_name}"
# Add parameter description to schema if available and not already present
if param.description and not param_schema.get("description"):
param_schema["description"] = param.description
@ -1193,8 +1297,19 @@ def _combine_schemas(route: openapi.HTTPRoute) -> dict[str, Any]:
if route.request_body.required:
required.extend(body_schema.get("required", []))
return {
result = {
"type": "object",
"properties": properties,
"required": required,
}
# Add schema definitions if available
if route.schema_definitions:
result["$defs"] = route.schema_definitions
# Use compress_schema to remove unused definitions
from fastmcp.utilities.json_schema import compress_schema
result = compress_schema(result)
return result

95
tests/client/test_client.py
View file

@ -5,6 +5,7 @@ from pydantic import AnyUrl
from fastmcp.client import Client
from fastmcp.client.transports import FastMCPTransport
from fastmcp.exceptions import ResourceError, ToolError
from fastmcp.prompts.prompt import TextContent
from fastmcp.server.server import FastMCP
@ -404,3 +405,97 @@ async def test_tagged_template_functionality(tagged_resources_server):
content_str = str(result[0])
assert '"id": "123"' in content_str
assert '"type": "template_data"' in content_str
class TestErrorHandling:
async def test_general_tool_exceptions_are_masked(self):
mcp = FastMCP("TestServer")
@mcp.tool()
def error_tool():
raise ValueError("This is a test error (abc)")
client = Client(transport=FastMCPTransport(mcp))
async with client:
result = await client.call_tool_mcp("error_tool", {})
assert result.isError
assert isinstance(result.content[0], TextContent)
assert "test error" not in result.content[0].text
assert "abc" not in result.content[0].text
async def test_specific_tool_errors_are_sent_to_client(self):
mcp = FastMCP("TestServer")
@mcp.tool()
def custom_error_tool():
raise ToolError("This is a test error (abc)")
client = Client(transport=FastMCPTransport(mcp))
async with client:
result = await client.call_tool_mcp("custom_error_tool", {})
assert result.isError
assert isinstance(result.content[0], TextContent)
assert "test error" in result.content[0].text
assert "abc" in result.content[0].text
async def test_general_resource_exceptions_are_masked(self):
mcp = FastMCP("TestServer")
@mcp.resource(uri="exception://resource")
async def exception_resource():
raise ValueError("This is an internal error (sensitive)")
client = Client(transport=FastMCPTransport(mcp))
async with client:
with pytest.raises(Exception) as excinfo:
await client.read_resource(AnyUrl("exception://resource"))
assert "Error reading resource" in str(excinfo.value)
assert "sensitive" not in str(excinfo.value)
assert "internal error" not in str(excinfo.value)
async def test_resource_errors_are_sent_to_client(self):
mcp = FastMCP("TestServer")
@mcp.resource(uri="error://resource")
async def error_resource():
raise ResourceError("This is a resource error (xyz)")
client = Client(transport=FastMCPTransport(mcp))
async with client:
with pytest.raises(Exception) as excinfo:
await client.read_resource(AnyUrl("error://resource"))
assert "This is a resource error (xyz)" in str(excinfo.value)
async def test_general_template_exceptions_are_masked(self):
mcp = FastMCP("TestServer")
@mcp.resource(uri="exception://resource/{id}")
async def exception_resource(id: str):
raise ValueError("This is an internal error (sensitive)")
client = Client(transport=FastMCPTransport(mcp))
async with client:
with pytest.raises(Exception) as excinfo:
await client.read_resource(AnyUrl("exception://resource/123"))
assert "Error reading resource" in str(excinfo.value)
assert "sensitive" not in str(excinfo.value)
assert "internal error" not in str(excinfo.value)
async def test_template_errors_are_sent_to_client(self):
mcp = FastMCP("TestServer")
@mcp.resource(uri="error://resource/{id}")
async def error_resource(id: str):
raise ResourceError("This is a resource error (xyz)")
client = Client(transport=FastMCPTransport(mcp))
async with client:
with pytest.raises(Exception) as excinfo:
await client.read_resource(AnyUrl("error://resource/123"))
assert "This is a resource error (xyz)" in str(excinfo.value)

4
tests/client/test_sse.py
View file

@ -96,7 +96,7 @@ async def test_http_headers(sse_server: str):
def run_nested_server(host: str, port: int) -> None:
try:
app = fastmcp_server().http_app(transport="sse")
app = fastmcp_server().sse_app(path="/mcp/sse", message_path="/mcp/messages")
mount = Starlette(routes=[Mount("/nest-inner", app=app)])
mount2 = Starlette(routes=[Mount("/nest-outer", app=mount)])
server = uvicorn.Server(
@ -122,7 +122,7 @@ async def test_nested_sse_server_resolves_correctly():
with run_server_in_process(run_nested_server) as url:
async with Client(
transport=SSETransport(f"{url}/nest-outer/nest-inner/sse")
transport=SSETransport(f"{url}/nest-outer/nest-inner/mcp/sse")
) as client:
result = await client.ping()
assert result is True

4
tests/client/test_streamable_http.py
View file

@ -106,7 +106,7 @@ async def test_http_headers(streamable_http_server: str):
def run_nested_server(host: str, port: int) -> None:
try:
mcp_app = fastmcp_server().http_app()
mcp_app = fastmcp_server().http_app(path="/final/mcp")
mount = Starlette(routes=[Mount("/nest-inner", app=mcp_app)])
mount2 = Starlette(
@ -135,7 +135,7 @@ async def test_nested_streamable_http_server_resolves_correctly():
with run_server_in_process(run_nested_server) as url:
async with Client(
transport=StreamableHttpTransport(f"{url}/nest-outer/nest-inner/mcp")
transport=StreamableHttpTransport(f"{url}/nest-outer/nest-inner/final/mcp")
) as client:
result = await client.ping()
assert result is True

3
tests/contrib/test_bulk_tool_caller.py
View file

@ -27,8 +27,7 @@ async def error_tool(arg1: str) -> dict[str, Any]:
def error_tool_result_factory(arg1: str) -> CallToolRequestResult:
"""Generates the expected error result for error_tool."""
# Mimic the error message format generated by BulkToolCaller when catching ToolException
exception_message = f"Error in tool with arg1: {arg1}"
formatted_error_text = f"Error executing tool error_tool: {exception_message}"
formatted_error_text = "Error calling tool 'error_tool'"
return CallToolRequestResult(
isError=True,
content=[TextContent(text=formatted_error_text, type="text")],

5
tests/resources/test_file_resources.py
View file

@ -5,6 +5,7 @@ from tempfile import NamedTemporaryFile
import pytest
from pydantic import FileUrl
from fastmcp.exceptions import ResourceError
from fastmcp.resources import FileResource
@ -94,7 +95,7 @@ class TestFileResource:
name="test",
path=missing,
)
with pytest.raises(ValueError, match="Error reading file"):
with pytest.raises(ResourceError, match="Error reading file"):
await resource.read()
@pytest.mark.skipif(
@ -109,7 +110,7 @@ class TestFileResource:
name="test",
path=temp_file,
)
with pytest.raises(ValueError, match="Error reading file"):
with pytest.raises(ResourceError, match="Error reading file"):
await resource.read()
finally:
temp_file.chmod(0o644) # Restore permissions

2
tests/resources/test_function_resources.py
View file

@ -80,7 +80,7 @@ class TestFunctionResource:
name="test",
fn=failing_func,
)
with pytest.raises(ValueError, match="Error reading resource function://test"):
with pytest.raises(ValueError, match="Test error"):
await resource.read()
async def test_basemodel_conversion(self):

86
tests/resources/test_resource_manager.py
View file

@ -4,7 +4,7 @@ from tempfile import NamedTemporaryFile
import pytest
from pydantic import AnyUrl, FileUrl
from fastmcp.exceptions import NotFoundError
from fastmcp.exceptions import NotFoundError, ResourceError
from fastmcp.resources import (
FileResource,
FunctionResource,
@ -540,3 +540,87 @@ class TestCustomResourceKeys:
# Shouldn't work with the original template pattern
with pytest.raises(NotFoundError, match="Unknown resource"):
await manager.get_resource("greet://world")
class TestResourceErrorHandling:
"""Test error handling in the ResourceManager."""
async def test_resource_error_passthrough(self):
"""Test that ResourceErrors are passed through directly."""
manager = ResourceManager()
async def error_resource():
"""Resource that raises a ResourceError."""
raise ResourceError("Specific resource error")
resource = FunctionResource(
uri=AnyUrl("error://resource"),
name="error_resource",
fn=error_resource,
)
manager.add_resource(resource)
with pytest.raises(ResourceError, match="Specific resource error"):
await manager.read_resource("error://resource")
async def test_exception_converted_to_resource_error(self):
"""Test that other exceptions are converted to ResourceError."""
manager = ResourceManager()
async def buggy_resource():
"""Resource that raises a ValueError."""
raise ValueError("Internal error details")
resource = FunctionResource(
uri=AnyUrl("buggy://resource"),
name="buggy_resource",
fn=buggy_resource,
)
manager.add_resource(resource)
with pytest.raises(ResourceError) as excinfo:
await manager.read_resource("buggy://resource")
# Exception message should contain the resource URI but not the internal details
assert "Error reading resource 'buggy://resource'" in str(excinfo.value)
assert "Internal error details" not in str(excinfo.value)
async def test_template_resource_error_passthrough(self):
"""Test that ResourceErrors from template-generated resources are passed through."""
manager = ResourceManager()
def error_template(param: str):
"""Template that raises a ResourceError."""
raise ResourceError(f"Template error with param {param}")
template = ResourceTemplate.from_function(
fn=error_template,
uri_template="error://{param}",
name="error_template",
)
manager.add_template(template)
with pytest.raises(ResourceError) as excinfo:
await manager.read_resource("error://test")
# The original error message should be included in the ValueError
assert "Template error with param test" in str(excinfo.value)
async def test_template_exception_converted_to_resource_error(self):
"""Test that other exceptions from template-generated resources are converted."""
manager = ResourceManager()
def buggy_template(param: str):
"""Template that raises a ValueError."""
raise ValueError(f"Internal template error with {param}")
template = ResourceTemplate.from_function(
fn=buggy_template,
uri_template="buggy://{param}",
name="buggy_template",
)
manager.add_template(template)
# First, the template creation will fail with ValueError
with pytest.raises(ResourceError, match="Error reading resource"):
await manager.read_resource("buggy://test")

15
tests/resources/test_resource_template.py
View file

@ -186,21 +186,6 @@ class TestResourceTemplate:
data = json.loads(content)
assert data == {"key": "foo", "value": 123}
async def test_template_error(self):
"""Test error handling in template resource creation."""
def failing_func(x: str) -> str:
raise ValueError("Test error")
template = ResourceTemplate.from_function(
fn=failing_func,
uri_template="fail://{x}",
name="fail",
)
with pytest.raises(ValueError, match="Error creating resource from template"):
await template.create_resource("fail://test", {"x": "test"})
async def test_async_text_resource(self):
"""Test creating a text resource from async function."""

119
tests/server/test_openapi.py
View file

@ -1,6 +1,7 @@
import base64
import json
import re
from enum import Enum
import httpx
import pytest
@ -1771,3 +1772,121 @@ class TestFastAPIDescriptionPropagation:
"name parameter missing from Tool schema in client API"
)
# We don't test for the description field content as it may not be consistently propagated
class TestReprMethods:
"""Tests for the custom __repr__ methods of OpenAPI objects."""
async def test_openapi_tool_repr(self, fastmcp_openapi_server: FastMCPOpenAPI):
"""Test that OpenAPITool's __repr__ method works without recursion errors."""
tools = fastmcp_openapi_server._tool_manager.list_tools()
tool = next(iter(tools))
# Verify repr doesn't cause recursion and contains expected elements
tool_repr = repr(tool)
assert "OpenAPITool" in tool_repr
assert f"name={tool.name!r}" in tool_repr
assert "method=" in tool_repr
assert "path=" in tool_repr
async def test_openapi_resource_repr(self, fastmcp_openapi_server: FastMCPOpenAPI):
"""Test that OpenAPIResource's __repr__ method works without recursion errors."""
resources = list(
fastmcp_openapi_server._resource_manager.get_resources().values()
)
resource = next(iter(resources))
# Verify repr doesn't cause recursion and contains expected elements
resource_repr = repr(resource)
assert "OpenAPIResource" in resource_repr
assert f"name={resource.name!r}" in resource_repr
assert "uri=" in resource_repr
assert "path=" in resource_repr
async def test_openapi_resource_template_repr(
self, fastmcp_openapi_server: FastMCPOpenAPI
):
"""Test that OpenAPIResourceTemplate's __repr__ method works without recursion errors."""
templates = list(
fastmcp_openapi_server._resource_manager.get_templates().values()
)
template = next(iter(templates))
# Verify repr doesn't cause recursion and contains expected elements
template_repr = repr(template)
assert "OpenAPIResourceTemplate" in template_repr
assert f"name={template.name!r}" in template_repr
assert "uri_template=" in template_repr
assert "path=" in template_repr
class TestEnumHandling:
"""Tests for handling enum parameters in OpenAPI schemas."""
async def test_enum_parameter_schema(self):
"""Test that enum parameters are properly handled in tool parameter schemas."""
# Define an enum just like in example.py
class QueryEnum(str, Enum):
foo = "foo"
bar = "bar"
baz = "baz"
# Create a minimal FastAPI app with an endpoint using the enum
app = FastAPI()
@app.post("/items/{item_id}")
def read_item(
item_id: int,
query: QueryEnum | None = None,
):
return {"item_id": item_id, "query": query}
# Create a client for the app
client = AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
# Create the FastMCPOpenAPI server from the app
openapi_spec = app.openapi()
server = FastMCPOpenAPI(
openapi_spec=openapi_spec,
client=client,
name="Enum Test",
)
# Get the tools from the server
tools = server._tool_manager.list_tools()
# Find the read_item tool
read_item_tool = next(
(t for t in tools if t.name == "read_item_items__item_id__post"), None
)
# Verify the tool exists
assert read_item_tool is not None, "read_item tool wasn't created"
# Check that the parameters include the enum reference
assert "properties" in read_item_tool.parameters
assert "query" in read_item_tool.parameters["properties"]
# Check for the anyOf with $ref to the enum definition
query_param = read_item_tool.parameters["properties"]["query"]
assert "anyOf" in query_param
# Find the ref in the anyOf list
ref_found = False
for option in query_param["anyOf"]:
if "$ref" in option and option["$ref"].startswith("#/$defs/QueryEnum"):
ref_found = True
break
assert ref_found, "Reference to enum definition not found in query parameter"
# Check that the $defs section exists and contains the enum definition
assert "$defs" in read_item_tool.parameters
assert "QueryEnum" in read_item_tool.parameters["$defs"]
# Verify the enum definition
enum_def = read_item_tool.parameters["$defs"]["QueryEnum"]
assert "enum" in enum_def
assert enum_def["enum"] == ["foo", "bar", "baz"]
assert enum_def["type"] == "string"

19
tests/server/test_server.py
View file

@ -72,6 +72,25 @@ class TestTools:
assert len(mcp_tools) == 1
assert mcp_tools[0].name == "custom_name"
async def test_remove_tool_successfully(self):
"""Test that FastMCP.remove_tool removes the tool from the registry."""
mcp = FastMCP()
@mcp.tool(name="adder")
def add(a: int, b: int) -> int:
return a + b
mcp_tools = await mcp.get_tools()
assert "adder" in mcp_tools
mcp.remove_tool("adder")
mcp_tools = await mcp.get_tools()
assert "adder" not in mcp_tools
with pytest.raises(NotFoundError, match="Unknown tool: adder"):
await mcp._mcp_call_tool("adder", {"a": 1, "b": 2})
class TestToolDecorator:
async def test_no_tools_before_decorator(self):

48
tests/server/test_server_interactions.py
View file

@ -17,6 +17,7 @@ from mcp.types import (
from pydantic import AnyUrl, Field
from fastmcp import Client, Context, FastMCP
from fastmcp.client.transports import FastMCPTransport
from fastmcp.exceptions import ClientError
from fastmcp.prompts.prompt import EmbeddedResource, PromptMessage
from fastmcp.resources import FileResource, FunctionResource
@ -94,12 +95,19 @@ class TestTools:
with pytest.raises(Exception):
await client.call_tool("error_tool", {})
async def test_call_tool_error_as_client_raw(self, tool_server: FastMCP):
async with Client(tool_server) as client:
result = await client.call_tool_mcp("error_tool", {})
assert result.isError
assert isinstance(result.content[0], TextContent)
assert "Test error" in result.content[0].text
async def test_call_tool_error_as_client_raw(self):
"""Test raising and catching errors from a tool."""
mcp = FastMCP()
client = Client(transport=FastMCPTransport(mcp))
@mcp.tool()
def error_tool():
raise ValueError("Test error")
async with client:
with pytest.raises(Exception) as excinfo:
await client.call_tool("error_tool", {})
assert "Error calling tool 'error_tool'" in str(excinfo.value)
async def test_tool_returns_list(self, tool_server: FastMCP):
async with Client(tool_server) as client:
@ -313,7 +321,7 @@ class TestToolParameters:
async with Client(mcp) as client:
with pytest.raises(
ClientError,
match="Input should be a valid integer, unable to parse string as an integer",
match="Error calling tool 'my_tool'",
):
await client.call_tool("my_tool", {"x": "not an int"})
@ -357,10 +365,7 @@ class TestToolParameters:
pass
async with Client(mcp) as client:
with pytest.raises(
ClientError,
match="Input should be greater than or equal to 1",
):
with pytest.raises(ClientError, match="Error calling tool 'analyze'"):
await client.call_tool("analyze", {"x": 0})
async def test_default_field_validation(self):
@ -371,10 +376,7 @@ class TestToolParameters:
pass
async with Client(mcp) as client:
with pytest.raises(
ClientError,
match="Input should be greater than or equal to 1",
):
with pytest.raises(ClientError, match="Error calling tool 'analyze'"):
await client.call_tool("analyze", {"x": 0})
async def test_default_field_is_still_required_if_no_default_specified(self):
@ -385,7 +387,7 @@ class TestToolParameters:
pass
async with Client(mcp) as client:
with pytest.raises(ClientError, match="Missing required argument"):
with pytest.raises(ClientError, match="Error calling tool 'analyze'"):
await client.call_tool("analyze", {})
async def test_literal_type_validation_error(self):
@ -396,7 +398,7 @@ class TestToolParameters:
pass
async with Client(mcp) as client:
with pytest.raises(ClientError, match="Input should be 'a' or 'b'"):
with pytest.raises(ClientError, match="Error calling tool 'analyze'"):
await client.call_tool("analyze", {"x": "c"})
async def test_literal_type_validation_success(self):
@ -424,9 +426,7 @@ class TestToolParameters:
return x.value
async with Client(mcp) as client:
with pytest.raises(
ClientError, match="Input should be 'red', 'green' or 'blue'"
):
with pytest.raises(ClientError, match="Error calling tool 'analyze'"):
await client.call_tool("analyze", {"x": "some-color"})
async def test_enum_type_validation_success(self):
@ -462,7 +462,7 @@ class TestToolParameters:
assert isinstance(result[0], TextContent)
assert result[0].text == "1.0"
with pytest.raises(ClientError, match="2 validation errors"):
with pytest.raises(ClientError, match="Error calling tool 'analyze'"):
await client.call_tool("analyze", {"x": "not a number"})
async def test_path_type(self):
@ -489,7 +489,7 @@ class TestToolParameters:
return str(path)
async with Client(mcp) as client:
with pytest.raises(ClientError, match="Input is not a valid path"):
with pytest.raises(ClientError, match="Error calling tool 'send_path'"):
await client.call_tool("send_path", {"path": 1})
async def test_uuid_type(self):
@ -515,7 +515,7 @@ class TestToolParameters:
return str(x)
async with Client(mcp) as client:
with pytest.raises(ClientError, match="Input should be a valid UUID"):
with pytest.raises(ClientError, match="Error calling tool 'send_uuid'"):
await client.call_tool("send_uuid", {"x": "not a uuid"})
async def test_datetime_type(self):
@ -554,7 +554,7 @@ class TestToolParameters:
return x.isoformat()
async with Client(mcp) as client:
with pytest.raises(ClientError, match="Input should be a valid datetime"):
with pytest.raises(ClientError, match="Error calling tool 'send_datetime'"):
await client.call_tool("send_datetime", {"x": "not a datetime"})
async def test_date_type(self):

2
tests/tools/test_tool.py
View file

@ -300,7 +300,7 @@ class TestLegacyToolJsonParsing:
async with Client(mcp) as client:
with pytest.raises(
ClientError,
match="Input should be a valid list",
match="Error calling tool 'process_list'",
):
await client.call_tool("process_list", {"items": "['a', 'b', 3]"})

86
tests/tools/test_tool_manager.py
View file

@ -100,6 +100,26 @@ class TestAddTools:
):
manager.add_tool_from_fn(lambda x: x)
def test_remove_tool_successfully(self):
"""Test removing an added tool by key."""
manager = ToolManager()
def add(a: int, b: int) -> int:
return a + b
manager.add_tool_from_fn(add)
assert manager.get_tool("add") is not None
manager.remove_tool("add")
with pytest.raises(NotFoundError):
manager.get_tool("add")
def test_remove_tool_missing_key(self):
"""Test removing a tool that does not exist raises NotFoundError."""
manager = ToolManager()
with pytest.raises(NotFoundError, match=f"Unknown tool: {'missing'}"):
manager.remove_tool("missing")
def test_warn_on_duplicate_tools(self, caplog):
"""Test warning on duplicate tools."""
manager = ToolManager(duplicate_behavior="warn")
@ -643,7 +663,7 @@ class TestContextHandling:
with context:
with pytest.raises(
ToolError, match="Error executing tool tool_with_context"
ToolError, match="Error calling tool 'tool_with_context'"
):
await manager.call_tool("tool_with_context", {"x": 42})
@ -740,3 +760,67 @@ class TestCustomToolNames:
# But the function is different
assert stored_tool.fn.__name__ == "replacement_fn"
class TestToolErrorHandling:
"""Test error handling in the ToolManager."""
async def test_tool_error_passthrough(self):
"""Test that ToolErrors are passed through directly."""
manager = ToolManager()
def error_tool(x: int) -> int:
"""Tool that raises a ToolError."""
raise ToolError("Specific tool error")
manager.add_tool_from_fn(error_tool)
with pytest.raises(ToolError, match="Specific tool error"):
await manager.call_tool("error_tool", {"x": 42})
async def test_exception_converted_to_tool_error(self):
"""Test that other exceptions are converted to ToolError."""
manager = ToolManager()
def buggy_tool(x: int) -> int:
"""Tool that raises a ValueError."""
raise ValueError("Internal error details")
manager.add_tool_from_fn(buggy_tool)
with pytest.raises(ToolError) as excinfo:
await manager.call_tool("buggy_tool", {"x": 42})
# Exception message should contain the tool name but not the internal details
assert "Error calling tool 'buggy_tool'" in str(excinfo.value)
assert "Internal error details" not in str(excinfo.value)
async def test_async_tool_error_passthrough(self):
"""Test that ToolErrors from async tools are passed through directly."""
manager = ToolManager()
async def async_error_tool(x: int) -> int:
"""Async tool that raises a ToolError."""
raise ToolError("Async tool error")
manager.add_tool_from_fn(async_error_tool)
with pytest.raises(ToolError, match="Async tool error"):
await manager.call_tool("async_error_tool", {"x": 42})
async def test_async_exception_converted_to_tool_error(self):
"""Test that other exceptions from async tools are converted to ToolError."""
manager = ToolManager()
async def async_buggy_tool(x: int) -> int:
"""Async tool that raises a ValueError."""
raise ValueError("Internal async error details")
manager.add_tool_from_fn(async_buggy_tool)
with pytest.raises(ToolError) as excinfo:
await manager.call_tool("async_buggy_tool", {"x": 42})
# Exception message should contain the tool name but not the internal details
assert "Error calling tool 'async_buggy_tool'" in str(excinfo.value)
assert "Internal async error details" not in str(excinfo.value)

17
tests/utilities/openapi/test_openapi_fastapi.py
View file

@ -520,3 +520,20 @@ def test_duplicate_tags_handling(fastapi_server):
# We'll test both possibilities to be safe
assert "duplicate" in test_route.tags, "Tag 'duplicate' should be present"
assert "items" in test_route.tags, "Tag 'items' should be present"
def test_repr_http_routes(parsed_routes):
"""Test that HTTPRoute objects can be represented without recursion errors."""
# Test repr on all parsed routes
for route in parsed_routes:
route_repr = repr(route)
# Verify repr contains essential information
assert route.method in route_repr, f"Method {route.method} missing from repr"
assert route.path in route_repr, f"Path {route.path} missing from repr"
# If operation_id exists, it should be in the repr
if route.operation_id:
assert route.operation_id in route_repr, (
f"Operation ID {route.operation_id} missing from repr"
)

394
tests/utilities/test_json_schema.py
View file

@ -1,110 +1,304 @@
from fastmcp.utilities.json_schema import _prune_param, prune_params
from fastmcp.utilities.json_schema import (
_prune_param,
_walk_and_prune,
compress_schema,
)
def test_prune_param_nonexistent():
"""Test pruning a parameter that doesn't exist."""
schema = {"properties": {"foo": {"type": "string"}}}
result = _prune_param(schema, "bar")
assert result == schema # Schema should be unchanged
# Create wrappers for backward compatibility with tests
def _prune_unused_defs(schema):
"""Wrapper for _walk_and_prune that only prunes definitions."""
return _walk_and_prune(schema, prune_defs=True)
def test_prune_param_exists():
"""Test pruning a parameter that exists."""
schema = {"properties": {"foo": {"type": "string"}, "bar": {"type": "integer"}}}
result = _prune_param(schema, "bar")
assert result["properties"] == {"foo": {"type": "string"}}
def _prune_additional_properties(schema):
"""Wrapper for _walk_and_prune that only prunes additionalProperties: false."""
return _walk_and_prune(schema, prune_additional_properties=True)
def test_prune_param_last_property():
"""Test pruning the only/last parameter, should leave empty properties object."""
schema = {"properties": {"foo": {"type": "string"}}}
result = _prune_param(schema, "foo")
assert "properties" in result
assert result["properties"] == {}
class TestPruneParam:
"""Tests for the _prune_param function."""
def test_nonexistent(self):
"""Test pruning a parameter that doesn't exist."""
schema = {"properties": {"foo": {"type": "string"}}}
result = _prune_param(schema, "bar")
assert result == schema # Schema should be unchanged
def test_exists(self):
"""Test pruning a parameter that exists."""
schema = {"properties": {"foo": {"type": "string"}, "bar": {"type": "integer"}}}
result = _prune_param(schema, "bar")
assert result["properties"] == {"foo": {"type": "string"}}
def test_last_property(self):
"""Test pruning the only/last parameter, should leave empty properties object."""
schema = {"properties": {"foo": {"type": "string"}}}
result = _prune_param(schema, "foo")
assert "properties" in result
assert result["properties"] == {}
def test_from_required(self):
"""Test pruning a parameter that's in the required list."""
schema = {
"properties": {"foo": {"type": "string"}, "bar": {"type": "integer"}},
"required": ["foo", "bar"],
}
result = _prune_param(schema, "bar")
assert result["required"] == ["foo"]
def test_last_required(self):
"""Test pruning the last required parameter, should remove required field."""
schema = {
"properties": {"foo": {"type": "string"}, "bar": {"type": "integer"}},
"required": ["foo"],
}
result = _prune_param(schema, "foo")
assert "required" not in result
def test_prune_param_from_required():
"""Test pruning a parameter that's in the required list."""
schema = {
"properties": {"foo": {"type": "string"}, "bar": {"type": "integer"}},
"required": ["foo", "bar"],
}
result = _prune_param(schema, "bar")
assert result["required"] == ["foo"]
class TestPruneUnusedDefs:
"""Tests for the _prune_unused_defs function."""
def test_prune_param_last_required():
"""Test pruning the last required parameter, should remove required field."""
schema = {
"properties": {"foo": {"type": "string"}, "bar": {"type": "integer"}},
"required": ["foo"],
}
result = _prune_param(schema, "foo")
assert "required" not in result
def test_prune_param_with_refs():
"""Test pruning a parameter that has references in $defs."""
schema = {
"properties": {
"foo": {"$ref": "#/$defs/foo_def"},
"bar": {"$ref": "#/$defs/bar_def"},
},
"$defs": {
"foo_def": {"type": "string"},
"bar_def": {"type": "integer"},
},
}
result = _prune_param(schema, "bar")
assert "bar_def" not in result["$defs"]
assert "foo_def" in result["$defs"]
def test_prune_param_all_refs():
"""Test pruning all parameters with refs, should remove $defs."""
schema = {
"properties": {
"foo": {"$ref": "#/$defs/foo_def"},
},
"$defs": {
"foo_def": {"type": "string"},
},
}
result = _prune_param(schema, "foo")
assert "$defs" not in result
def test_prune_params_multiple():
"""Test pruning multiple parameters at once."""
schema = {
"properties": {
"foo": {"type": "string"},
"bar": {"type": "integer"},
"baz": {"type": "boolean"},
},
"required": ["foo", "bar"],
}
result = prune_params(schema, ["foo", "baz"])
assert result["properties"] == {"bar": {"type": "integer"}}
assert result["required"] == ["bar"]
def test_prune_params_nested_refs():
"""Test pruning with nested references."""
schema = {
"properties": {
"foo": {
"type": "object",
"properties": {"nested": {"$ref": "#/$defs/nested_def"}},
def test_removes_unreferenced_defs(self):
"""Test that unreferenced definitions are removed."""
schema = {
"properties": {
"foo": {"$ref": "#/$defs/foo_def"},
},
"bar": {"$ref": "#/$defs/bar_def"},
},
"$defs": {
"nested_def": {"type": "string"},
"bar_def": {"type": "integer"},
},
}
# Removing foo should keep nested_def as it's not referenced anymore
result = _prune_param(schema, "foo")
assert "nested_def" not in result["$defs"]
assert "bar_def" in result["$defs"]
"$defs": {
"foo_def": {"type": "string"},
"unused_def": {"type": "integer"},
},
}
result = _prune_unused_defs(schema)
assert "foo_def" in result["$defs"]
assert "unused_def" not in result["$defs"]
def test_nested_references_kept(self):
"""Test that definitions referenced via nesting are kept."""
schema = {
"properties": {
"foo": {"$ref": "#/$defs/foo_def"},
},
"$defs": {
"foo_def": {
"type": "object",
"properties": {"nested": {"$ref": "#/$defs/nested_def"}},
},
"nested_def": {"type": "string"},
"unused_def": {"type": "integer"},
},
}
result = _prune_unused_defs(schema)
assert "foo_def" in result["$defs"]
assert "nested_def" in result["$defs"]
assert "unused_def" not in result["$defs"]
def test_array_references_kept(self):
"""Test that definitions referenced in array items are kept."""
schema = {
"properties": {
"items": {"type": "array", "items": {"$ref": "#/$defs/item_def"}},
},
"$defs": {
"item_def": {"type": "string"},
"unused_def": {"type": "integer"},
},
}
result = _prune_unused_defs(schema)
assert "item_def" in result["$defs"]
assert "unused_def" not in result["$defs"]
def test_removes_defs_field_when_empty(self):
"""Test that $defs field is removed when all definitions are unused."""
schema = {
"properties": {
"foo": {"type": "string"},
},
"$defs": {
"unused_def": {"type": "integer"},
},
}
result = _prune_unused_defs(schema)
assert "$defs" not in result
class TestPruneAdditionalProperties:
"""Tests for the _prune_additional_properties function."""
def test_removes_when_false(self):
"""Test that additionalProperties is removed when it's false."""
schema = {
"type": "object",
"properties": {"foo": {"type": "string"}},
"additionalProperties": False,
}
result = _prune_additional_properties(schema)
assert "additionalProperties" not in result
def test_keeps_when_true(self):
"""Test that additionalProperties is kept when it's true."""
schema = {
"type": "object",
"properties": {"foo": {"type": "string"}},
"additionalProperties": True,
}
result = _prune_additional_properties(schema)
assert "additionalProperties" in result
assert result["additionalProperties"] is True
def test_keeps_when_object(self):
"""Test that additionalProperties is kept when it's an object schema."""
schema = {
"type": "object",
"properties": {"foo": {"type": "string"}},
"additionalProperties": {"type": "string"},
}
result = _prune_additional_properties(schema)
assert "additionalProperties" in result
assert result["additionalProperties"] == {"type": "string"}
class TestCompressSchema:
"""Tests for the compress_schema function."""
def test_prune_params(self):
"""Test pruning parameters with compress_schema."""
schema = {
"properties": {
"foo": {"type": "string"},
"bar": {"type": "integer"},
"baz": {"type": "boolean"},
},
"required": ["foo", "bar"],
}
result = compress_schema(schema, prune_params=["foo", "baz"])
assert result["properties"] == {"bar": {"type": "integer"}}
assert result["required"] == ["bar"]
def test_prune_defs(self):
"""Test pruning unused definitions with compress_schema."""
schema = {
"properties": {
"foo": {"$ref": "#/$defs/foo_def"},
"bar": {"type": "integer"},
},
"$defs": {
"foo_def": {"type": "string"},
"unused_def": {"type": "number"},
},
}
result = compress_schema(schema)
assert "foo_def" in result["$defs"]
assert "unused_def" not in result["$defs"]
def test_disable_prune_defs(self):
"""Test disabling pruning of unused definitions."""
schema = {
"properties": {
"foo": {"$ref": "#/$defs/foo_def"},
"bar": {"type": "integer"},
},
"$defs": {
"foo_def": {"type": "string"},
"unused_def": {"type": "number"},
},
}
result = compress_schema(schema, prune_defs=False)
assert "foo_def" in result["$defs"]
assert "unused_def" in result["$defs"]
def test_pruning_additional_properties(self):
"""Test pruning additionalProperties when False."""
schema = {
"type": "object",
"properties": {"foo": {"type": "string"}},
"additionalProperties": False,
}
result = compress_schema(schema)
assert "additionalProperties" not in result
def test_disable_pruning_additional_properties(self):
"""Test disabling pruning of additionalProperties."""
schema = {
"type": "object",
"properties": {"foo": {"type": "string"}},
"additionalProperties": False,
}
result = compress_schema(schema, prune_additional_properties=False)
assert "additionalProperties" in result
assert result["additionalProperties"] is False
def test_combined_operations(self):
"""Test all pruning operations together."""
schema = {
"type": "object",
"properties": {
"keep": {"type": "string"},
"remove": {"$ref": "#/$defs/remove_def"},
},
"required": ["keep", "remove"],
"additionalProperties": False,
"$defs": {
"remove_def": {"type": "string"},
"unused_def": {"type": "number"},
},
}
result = compress_schema(schema, prune_params=["remove"])
# Check that parameter was removed
assert "remove" not in result["properties"]
# Check that required list was updated
assert result["required"] == ["keep"]
# Check that unused definitions were removed
assert "$defs" not in result # Both defs should be gone
# Check that additionalProperties was removed
assert "additionalProperties" not in result
def test_prune_titles(self):
"""Test pruning title fields."""
schema = {
"title": "Root Schema",
"type": "object",
"properties": {
"foo": {"title": "Foo Property", "type": "string"},
"bar": {
"title": "Bar Property",
"type": "object",
"properties": {
"nested": {"title": "Nested Property", "type": "string"}
},
},
},
}
result = compress_schema(schema, prune_titles=True)
assert "title" not in result
assert "title" not in result["properties"]["foo"]
assert "title" not in result["properties"]["bar"]
assert "title" not in result["properties"]["bar"]["properties"]["nested"]
def test_prune_nested_additional_properties(self):
"""Test pruning additionalProperties: false at all levels."""
schema = {
"type": "object",
"additionalProperties": False,
"properties": {
"foo": {
"type": "object",
"additionalProperties": False,
"properties": {
"nested": {
"type": "object",
"additionalProperties": False,
}
},
},
},
}
result = compress_schema(schema)
assert "additionalProperties" not in result
assert "additionalProperties" not in result["properties"]["foo"]
assert (
"additionalProperties"
not in result["properties"]["foo"]["properties"]["nested"]
)

4
tests/utilities/test_typeadapter.py
View file

@ -13,7 +13,7 @@ import annotated_types
import pytest
from pydantic import BaseModel, Field
from fastmcp.utilities.json_schema import prune_params
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.types import get_cached_typeadapter
@ -175,7 +175,7 @@ def test_skip_names():
# Get schema and prune parameters
type_adapter = get_cached_typeadapter(func_with_many_params)
schema = type_adapter.json_schema()
pruned_schema = prune_params(schema, params=["skip_this", "also_skip"])
pruned_schema = compress_schema(schema, prune_params=["skip_this", "also_skip"])
# Check that only the desired parameters remain
assert "keep_this" in pruned_schema["properties"]