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

Clean up route maps

Browse source
This commit is contained in:
Jeremiah Lowin 2025-06-10 20:19:41 -04:00
commit b1714bcf7b
2 changed files with 76 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -103,15 +103,27 @@ class RouteType(enum.Enum):
IGNORE = "IGNORE"
@dataclass
@dataclass(kw_only=True)
class RouteMap:
"""Mapping configuration for HTTP routes to FastMCP component types."""
methods: list[HttpMethod] | Literal["*"] = field(default="*")
pattern: Pattern[str] | str = field(default=r".*")
mcp_type: MCPType | None = field(default=None)
route_type: RouteType | MCPType | None = field(default=None)
tags: set[str] = field(default_factory=set)
tags: set[str] = field(
default_factory=set,
metadata={"description": "A set of tags to match. All tags must match."},
)
mcp_type: MCPType | None = field(
default=None,
metadata={"description": "The type of FastMCP component to create."},
)
mcp_tags: set[str] = field(
default_factory=set,
metadata={
"description": "A set of tags to apply to the generated FastMCP component."
},
)
def __post_init__(self):
"""Validate and process the route map after initialization."""
@ -165,7 +177,7 @@ DEFAULT_ROUTE_MAPPINGS = [
def _determine_route_type(
route: openapi.HTTPRoute,
mappings: list[RouteMap],
) -> MCPType:
) -> RouteMap:
"""
Determines the FastMCP component type based on the route and mappings.
@ -174,7 +186,7 @@ def _determine_route_type(
mappings: List of RouteMap objects in priority order
Returns:
MCPType for this route
The RouteMap that matches the route, or a catchall "Tool" RouteMap if no match is found.
"""
# Check mappings in priority order (first match wins)
for route_map in mappings:
@ -201,10 +213,10 @@ def _determine_route_type(
logger.debug(
f"Route {route.method} {route.path} matched mapping to {route_map.mcp_type.name}"
)
return route_map.mcp_type
return route_map
# Default fallback
return MCPType.TOOL
return RouteMap(mcp_type=MCPType.TOOL)
class OpenAPITool(Tool):
@ -217,7 +229,7 @@ class OpenAPITool(Tool):
name: str,
description: str,
parameters: dict[str, Any],
tags: set[str] = set(),
tags: set[str] | None = None,
timeout: float | None = None,
annotations: ToolAnnotations | None = None,
serializer: Callable[[Any], str] | None = None,
@ -226,7 +238,7 @@ class OpenAPITool(Tool):
name=name,
description=description,
parameters=parameters,
tags=tags,
tags=tags or set(),
annotations=annotations,
serializer=serializer,
)
@ -680,6 +692,8 @@ class FastMCPOpenAPI(FastMCP):
route_map_fn: RouteMapFn | None = None,
mcp_component_fn: ComponentFn | None = None,
mcp_names: dict[str, str] | None = None,
mcp_tags: dict[str, set[str]] | None = None,
tags: set[str] | None = None,
timeout: float | None = None,
**settings: Any,
):
@ -702,6 +716,11 @@ class FastMCPOpenAPI(FastMCP):
operationId up to the first double underscore. If no operationId exists,
falls back to slugified summary or path-based naming.
All names are truncated to 56 characters maximum.
mcp_tags: Optional dictionary mapping operationId to set of tags.
If an operationId is not in the dictionary, falls back to using the
tags from the route.
tags: Optional set of tags to add to all components. Components always receive any tags
from the route.
timeout: Optional timeout (in seconds) for all requests
**settings: Additional settings for FastMCP
"""
@ -709,9 +728,7 @@ class FastMCPOpenAPI(FastMCP):
self._client = client
self._timeout = timeout
self._route_map_fn = route_map_fn
self._mcp_component_fn = mcp_component_fn
self._mcp_names = mcp_names or {}
# Keep track of names to detect collisions
self._used_names = {
@ -727,12 +744,16 @@ class FastMCPOpenAPI(FastMCP):
route_maps = (route_maps or []) + DEFAULT_ROUTE_MAPPINGS
for route in http_routes:
# Determine route type based on mappings or default rules
route_type = _determine_route_type(route, route_maps)
route_map = _determine_route_type(route, route_maps)
# TODO: remove this once RouteType is removed and mcp_type is typed as MCPType without | None
assert route_map.mcp_type is not None
route_type = route_map.mcp_type
# Call route_map_fn if provided
if self._route_map_fn is not None:
if route_map_fn is not None:
try:
result = self._route_map_fn(route, route_type)
result = route_map_fn(route, route_type)
if result is not None:
route_type = result
logger.debug(
@ -746,29 +767,34 @@ class FastMCPOpenAPI(FastMCP):
)
# Generate a default name from the route
component_name = self._generate_default_name(route, route_type)
component_name = self._generate_default_name(route, mcp_names)
route_tags = set(route.tags) | route_map.mcp_tags | (tags or set())
if route.operation_id:
route_tags |= (mcp_tags or {}).get(route.operation_id, set())
if route_type == MCPType.TOOL:
self._create_openapi_tool(route, component_name)
self._create_openapi_tool(route, component_name, tags=route_tags)
elif route_type == MCPType.RESOURCE:
self._create_openapi_resource(route, component_name)
self._create_openapi_resource(route, component_name, tags=route_tags)
elif route_type == MCPType.RESOURCE_TEMPLATE:
self._create_openapi_template(route, component_name)
self._create_openapi_template(route, component_name, tags=route_tags)
elif route_type == MCPType.EXCLUDE:
logger.info(f"Excluding route: {route.method} {route.path}")
logger.info(f"Created FastMCP OpenAPI server with {len(http_routes)} routes")
def _generate_default_name(
self, route: openapi.HTTPRoute, mcp_type: MCPType
self, route: openapi.HTTPRoute, mcp_names_map: dict[str, str] | None = None
) -> str:
"""Generate a default name from the route using the configured strategy."""
name = ""
mcp_names_map = mcp_names_map or {}
# First check if there's a custom mapping for this operationId
if route.operation_id:
if route.operation_id in self._mcp_names:
name = self._mcp_names[route.operation_id]
if route.operation_id in mcp_names_map:
name = mcp_names_map[route.operation_id]
else:
# If there's a double underscore in the operationId, use the first part
name = route.operation_id.split("__")[0]
@ -813,7 +839,12 @@ class FastMCPOpenAPI(FastMCP):
return new_name
def _create_openapi_tool(self, route: openapi.HTTPRoute, name: str):
def _create_openapi_tool(
self,
route: openapi.HTTPRoute,
name: str,
tags: set[str],
):
"""Creates and registers an OpenAPITool with enhanced description."""
combined_schema = _combine_schemas(route)
@ -840,7 +871,7 @@ class FastMCPOpenAPI(FastMCP):
name=tool_name,
description=enhanced_description,
parameters=combined_schema,
tags=set(route.tags or []),
tags=set(route.tags or []) | tags,
timeout=self._timeout,
)
@ -861,7 +892,12 @@ class FastMCPOpenAPI(FastMCP):
f"Registered TOOL: {tool_name} ({route.method} {route.path}) with tags: {route.tags}"
)
def _create_openapi_resource(self, route: openapi.HTTPRoute, name: str):
def _create_openapi_resource(
self,
route: openapi.HTTPRoute,
name: str,
tags: set[str],
):
"""Creates and registers an OpenAPIResource with enhanced description."""
# Get a unique resource name
resource_name = self._get_unique_name(name, "resource")
@ -885,7 +921,7 @@ class FastMCPOpenAPI(FastMCP):
uri=resource_uri,
name=resource_name,
description=enhanced_description,
tags=set(route.tags or []),
tags=set(route.tags or []) | tags,
timeout=self._timeout,
)
@ -906,7 +942,12 @@ class FastMCPOpenAPI(FastMCP):
f"Registered RESOURCE: {resource_uri} ({route.method} {route.path}) with tags: {route.tags}"
)
def _create_openapi_template(self, route: openapi.HTTPRoute, name: str):
def _create_openapi_template(
self,
route: openapi.HTTPRoute,
name: str,
tags: set[str],
):
"""Creates and registers an OpenAPIResourceTemplate with enhanced description."""
# Get a unique template name
template_name = self._get_unique_name(name, "resource_template")
@ -959,7 +1000,7 @@ class FastMCPOpenAPI(FastMCP):
name=template_name,
description=enhanced_description,
parameters=template_params_schema,
tags=set(route.tags or []),
tags=set(route.tags or []) | tags,
timeout=self._timeout,
)

19
tests/server/openapi/test_route_map_fn.py
View file

@ -3,7 +3,7 @@
import httpx
import pytest
from fastmcp.server.openapi import FastMCPOpenAPI, MCPType
from fastmcp.server.openapi import FastMCPOpenAPI, MCPType, RouteMap, RouteMapFn
@pytest.fixture
@ -175,8 +175,6 @@ def test_route_map_fn_returns_none(sample_openapi_spec, http_client):
def test_route_map_fn_called_for_excluded_routes(sample_openapi_spec, http_client):
"""Test that route_map_fn is called for excluded routes and can rescue them."""
from fastmcp.server.openapi import RouteMap
# Exclude all admin routes
route_maps = [
RouteMap(
@ -188,7 +186,7 @@ def test_route_map_fn_called_for_excluded_routes(sample_openapi_spec, http_clien
def track_calls_and_rescue(route, mcp_type):
"""Track which routes the function is called for and rescue some excluded routes."""
called_routes.append(route.path)
called_routes.append((route.method, route.path))
# Rescue the admin GET route by converting it to a tool
if route.path == "/admin/settings" and route.method == "GET":
@ -205,10 +203,11 @@ def test_route_map_fn_called_for_excluded_routes(sample_openapi_spec, http_clien
)
# route_map_fn should now be called for all routes, including excluded admin routes
assert "/admin/settings" in called_routes
assert "/users" in called_routes
assert "/users/{id}" in called_routes
assert "/api/data" in called_routes
assert ("GET", "/admin/settings") in called_routes
assert ("GET", "/users") in called_routes
assert ("GET", "/users/{id}") in called_routes
assert ("GET", "/api/data") in called_routes
assert ("POST", "/admin/settings") in called_routes
# The rescued admin GET route should now be a tool
tools = server._tool_manager._tools
@ -296,7 +295,7 @@ def test_combined_route_map_fn_and_component_fn(sample_openapi_spec, http_client
def test_route_map_fn_signature_validation():
"""Test that route_map_fn has the correct signature."""
from fastmcp.server.openapi import RouteMapFn
from fastmcp.utilities import openapi
# This is more of a type checking test
@ -335,8 +334,6 @@ def test_component_fn_signature_validation():
def test_route_map_fn_can_rescue_excluded_routes(sample_openapi_spec, http_client):
"""Test that route_map_fn can rescue routes that were excluded by RouteMap."""
from fastmcp.server.openapi import RouteMap
# Exclude ALL routes by default
route_maps = [
RouteMap(mcp_type=MCPType.EXCLUDE) # Catch-all exclusion