From aa2747497c36080e1ad374c50022ad728b710d34 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Thu, 22 May 2025 22:10:45 -0400
Subject: [PATCH] Add route_map_fn for fine control
---
docs/servers/openapi.mdx | 41 ++++
src/fastmcp/server/openapi.py | 42 +++-
src/fastmcp/server/server.py | 6 +-
tests/server/openapi/test_route_map_fn.py | 284 ++++++++++++++++++++++
4 files changed, 370 insertions(+), 3 deletions(-)
create mode 100644 tests/server/openapi/test_route_map_fn.py
diff --git a/docs/servers/openapi.mdx b/docs/servers/openapi.mdx
index 4ea549f91..6c8349708 100644
--- a/docs/servers/openapi.mdx
+++ b/docs/servers/openapi.mdx
@@ -118,6 +118,47 @@ To prevent the default mappings from being applied, add a catch-all exclusion ro
To filter routes by OpenAPI tags, use `RouteMap(tags={...})`. The route must have ALL of the specified tags to be matched. If no tags are specified, all routes will be matched.
+### Advanced Route Mapping
+
+
+
+For advanced users who need fine-grained control over route mapping, you can provide a `route_map_fn` callable. This function receives each route that was matched by a route map (and wasn't excluded) along with the assigned MCP type and name, and can return either `None` to accept the defaults or a `(mcp_type, name)` tuple to override the type and/or object name.
+
+```python
+from fastmcp.server.openapi import MCPType
+
+def custom_route_mapper(route, mcp_type, name):
+ """Custom route mapping function for advanced control."""
+ # Convert all admin routes to tools regardless of HTTP method
+ if "/admin/" in route.path:
+ return MCPType.TOOL, f"admin_{name}"
+
+ # Rename all user-specific routes to have the prefix "user_"
+ if "/users/{id}" in route.path:
+ return mcp_type, f"user_{name}"
+
+ # Accept defaults for all other routes
+ return None
+
+mcp = FastMCP.from_openapi(
+ openapi_spec=spec,
+ client=api_client,
+ route_map_fn=custom_route_mapper,
+)
+```
+
+The `route_map_fn` receives:
+- `route`: The OpenAPI route object with properties like `.method`, `.path`, `.operation_id`, etc.
+- `mcp_type`: The assigned `MCPType` (based on route maps)
+- `name`: The assigned component name (derived from operation ID or path)
+
+It should return either:
+- `None` to accept the defaults
+- `(mcp_type, name)` tuple to override the type and/or name
+
+
+The `route_map_fn` is only called for routes that matched a route map and were **not** excluded. It will not be called for routes with `MCPType.EXCLUDE`.
+
## Request Parameter Handling
diff --git a/src/fastmcp/server/openapi.py b/src/fastmcp/server/openapi.py
index 11ea93878..9ac9b8d90 100644
--- a/src/fastmcp/server/openapi.py
+++ b/src/fastmcp/server/openapi.py
@@ -33,6 +33,9 @@ logger = get_logger(__name__)
HttpMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"]
+# Type definition for the route mapping function
+RouteMapFn = Callable[[openapi.HTTPRoute, "MCPType", str], tuple["MCPType", str] | None]
+
class MCPType(enum.Enum):
"""Type of FastMCP component to create from a route.
@@ -614,7 +617,7 @@ class FastMCPOpenAPI(FastMCP):
Example:
```python
- from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap, RouteType
+ from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap, MCPType
import httpx
# Define custom route mappings
@@ -633,12 +636,26 @@ class FastMCPOpenAPI(FastMCP):
),
]
- # Create server with custom mappings
+ # Advanced: Custom route mapping function for fine-grained control
+ def custom_route_mapper(route, mcp_type, name):
+ # Convert all admin routes to tools regardless of HTTP method
+ if "/admin/" in route.path:
+ return MCPType.TOOL, f"admin_{name}"
+
+ # Rename all user-specific routes to include "user_"
+ if "/users/{id}" in route.path:
+ return mcp_type, f"user_{name}"
+
+ # Accept defaults for all other routes
+ return None
+
+ # Create server with custom mappings and route mapper
server = FastMCPOpenAPI(
openapi_spec=spec,
client=httpx.AsyncClient(),
name="API Server",
route_maps=custom_mappings,
+ route_map_fn=custom_route_mapper,
)
```
"""
@@ -649,6 +666,7 @@ class FastMCPOpenAPI(FastMCP):
client: httpx.AsyncClient,
name: str | None = None,
route_maps: list[RouteMap] | None = None,
+ route_map_fn: RouteMapFn | None = None,
timeout: float | None = None,
**settings: Any,
):
@@ -660,6 +678,9 @@ class FastMCPOpenAPI(FastMCP):
client: httpx AsyncClient for making HTTP requests
name: Optional name for the server
route_maps: Optional list of RouteMap objects defining route mappings
+ route_map_fn: Optional callable for advanced users to customize route mapping.
+ Receives (route, mcp_type, name) and returns (mcp_type, name) tuple or None.
+ Only called on routes that matched a route_map and were not excluded.
timeout: Optional timeout (in seconds) for all requests
**settings: Additional settings for FastMCP
"""
@@ -667,6 +688,7 @@ class FastMCPOpenAPI(FastMCP):
self._client = client
self._timeout = timeout
+ self._route_map_fn = route_map_fn
# Keep track of names to detect collisions
self._used_names = {"tools": set(), "resources": set(), "templates": set()}
@@ -682,6 +704,22 @@ class FastMCPOpenAPI(FastMCP):
# Generate a default name from the route
component_name = self._generate_default_name(route, route_type)
+ # Call route_map_fn if provided and route is not excluded
+ if self._route_map_fn is not None and route_type != MCPType.EXCLUDE:
+ try:
+ result = self._route_map_fn(route, route_type, component_name)
+ if result is not None:
+ route_type, component_name = result
+ logger.debug(
+ f"Route {route.method} {route.path} mapping customized by route_map_fn: "
+ f"type={route_type.name}, name={component_name}"
+ )
+ except Exception as e:
+ logger.warning(
+ f"Error in route_map_fn for {route.method} {route.path}: {e}. "
+ f"Using default values."
+ )
+
if route_type == MCPType.TOOL:
self._create_openapi_tool(route, component_name)
elif route_type == MCPType.RESOURCE:
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index e9ab55b5a..3bb1e46ee 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -63,7 +63,7 @@ from fastmcp.utilities.mcp_config import MCPConfig
if TYPE_CHECKING:
from fastmcp.client import Client
from fastmcp.client.transports import ClientTransport
- from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap
+ from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap, RouteMapFn
from fastmcp.server.proxy import FastMCPProxy
logger = get_logger(__name__)
@@ -1141,6 +1141,7 @@ class FastMCP(Generic[LifespanResultT]):
openapi_spec: dict[str, Any],
client: httpx.AsyncClient,
route_maps: list[RouteMap] | None = None,
+ route_map_fn: RouteMapFn | None = None,
all_routes_as_tools: bool = False,
**settings: Any,
) -> FastMCPOpenAPI:
@@ -1168,6 +1169,7 @@ class FastMCP(Generic[LifespanResultT]):
openapi_spec=openapi_spec,
client=client,
route_maps=route_maps,
+ route_map_fn=route_map_fn,
**settings,
)
@@ -1177,6 +1179,7 @@ class FastMCP(Generic[LifespanResultT]):
app: Any,
name: str | None = None,
route_maps: list[RouteMap] | None = None,
+ route_map_fn: RouteMapFn | None = None,
all_routes_as_tools: bool = False,
**settings: Any,
) -> FastMCPOpenAPI:
@@ -1212,6 +1215,7 @@ class FastMCP(Generic[LifespanResultT]):
client=client,
name=name,
route_maps=route_maps,
+ route_map_fn=route_map_fn,
**settings,
)
diff --git a/tests/server/openapi/test_route_map_fn.py b/tests/server/openapi/test_route_map_fn.py
new file mode 100644
index 000000000..41b68fc9a
--- /dev/null
+++ b/tests/server/openapi/test_route_map_fn.py
@@ -0,0 +1,284 @@
+"""Tests for the route_map_fn functionality in FastMCPOpenAPI."""
+
+import httpx
+import pytest
+
+from fastmcp.server.openapi import FastMCPOpenAPI, MCPType
+
+
+@pytest.fixture
+def sample_openapi_spec():
+ """Sample OpenAPI spec for testing."""
+ return {
+ "openapi": "3.0.0",
+ "info": {"title": "Test API", "version": "1.0.0"},
+ "paths": {
+ "/users": {
+ "get": {
+ "summary": "List users",
+ "operationId": "listUsers",
+ "responses": {"200": {"description": "Success"}},
+ }
+ },
+ "/users/{id}": {
+ "get": {
+ "summary": "Get user by ID",
+ "operationId": "getUserById",
+ "parameters": [
+ {
+ "name": "id",
+ "in": "path",
+ "required": True,
+ "schema": {"type": "string"},
+ }
+ ],
+ "responses": {"200": {"description": "Success"}},
+ }
+ },
+ "/admin/settings": {
+ "get": {
+ "summary": "Get admin settings",
+ "operationId": "getAdminSettings",
+ "responses": {"200": {"description": "Success"}},
+ },
+ "post": {
+ "summary": "Update admin settings",
+ "operationId": "updateAdminSettings",
+ "requestBody": {
+ "content": {"application/json": {"schema": {"type": "object"}}}
+ },
+ "responses": {"200": {"description": "Success"}},
+ },
+ },
+ "/api/data": {
+ "get": {
+ "summary": "Get data",
+ "operationId": "getData",
+ "responses": {"200": {"description": "Success"}},
+ }
+ },
+ },
+ }
+
+
+@pytest.fixture
+def http_client():
+ """HTTP client for testing."""
+ return httpx.AsyncClient()
+
+
+def test_route_map_fn_none(sample_openapi_spec, http_client):
+ """Test that server works correctly when route_map_fn is None."""
+ server = FastMCPOpenAPI(
+ openapi_spec=sample_openapi_spec,
+ client=http_client,
+ name="Test Server",
+ route_map_fn=None, # Explicitly set to None
+ )
+
+ assert server.name == "Test Server"
+
+
+def test_route_map_fn_custom_type_conversion(sample_openapi_spec, http_client):
+ """Test that route_map_fn can convert route types."""
+
+ def admin_routes_to_tools(route, mcp_type, name):
+ """Convert all admin routes to tools."""
+ if "/admin/" in route.path:
+ return MCPType.TOOL, f"admin_{name}"
+ return None
+
+ server = FastMCPOpenAPI(
+ openapi_spec=sample_openapi_spec,
+ client=http_client,
+ name="Test Server",
+ route_map_fn=admin_routes_to_tools,
+ )
+
+ # Admin GET route should be converted to tool instead of resource
+ tools = server._tool_manager._tools
+ assert "admin_getAdminSettings" in tools
+
+ # Admin POST route should be renamed
+ assert "admin_updateAdminSettings" in tools
+
+
+def test_route_map_fn_custom_naming(sample_openapi_spec, http_client):
+ """Test that route_map_fn can customize naming."""
+
+ def prefix_user_routes(route, mcp_type, name):
+ """Add user_ prefix to user-related routes."""
+ if "/users/" in route.path:
+ return mcp_type, f"user_{name}"
+ return None
+
+ server = FastMCPOpenAPI(
+ openapi_spec=sample_openapi_spec,
+ client=http_client,
+ name="Test Server",
+ route_map_fn=prefix_user_routes,
+ )
+
+ # Check that user routes got renamed
+ templates = server._resource_manager._templates
+ template_names = list(templates.keys())
+
+ # The getUserById template should be renamed to user_getUserById
+ found_user_template = False
+ for uri in template_names:
+ if "user_getUserById" in uri:
+ found_user_template = True
+ break
+ assert found_user_template
+
+
+def test_route_map_fn_returns_none(sample_openapi_spec, http_client):
+ """Test that route_map_fn returning None uses defaults."""
+
+ def always_return_none(route, mcp_type, name):
+ """Always return None to use defaults."""
+ return None
+
+ server = FastMCPOpenAPI(
+ openapi_spec=sample_openapi_spec,
+ client=http_client,
+ name="Test Server",
+ route_map_fn=always_return_none,
+ )
+
+ # Should have default behavior
+ assert server.name == "Test Server"
+ # Check that components were created with default names
+ tools = server._tool_manager._tools
+ resources = server._resource_manager._resources
+ templates = server._resource_manager._templates
+
+ # Should have tools, resources, and templates based on default mapping
+ assert len(tools) > 0
+ assert len(resources) > 0
+ assert len(templates) > 0
+
+
+def test_route_map_fn_not_called_for_excluded_routes(sample_openapi_spec, http_client):
+ """Test that route_map_fn is not called for excluded routes."""
+
+ from fastmcp.server.openapi import RouteMap
+
+ # Exclude all admin routes
+ route_maps = [
+ RouteMap(
+ methods=["GET", "POST"], pattern=r".*/admin/.*", mcp_type=MCPType.EXCLUDE
+ )
+ ]
+
+ called_routes = []
+
+ def track_calls(route, mcp_type, name):
+ """Track which routes the function is called for."""
+ called_routes.append(route.path)
+ return None
+
+ FastMCPOpenAPI(
+ openapi_spec=sample_openapi_spec,
+ client=http_client,
+ name="Test Server",
+ route_maps=route_maps,
+ route_map_fn=track_calls,
+ )
+
+ # route_map_fn should not be called for excluded admin routes
+ assert "/admin/settings" not in called_routes
+ # But should be called for other routes
+ assert "/users" in called_routes
+ assert "/users/{id}" in called_routes
+ assert "/api/data" in called_routes
+
+
+def test_route_map_fn_error_handling(sample_openapi_spec, http_client):
+ """Test that errors in route_map_fn are handled gracefully."""
+
+ def error_function(route, mcp_type, name):
+ """Function that raises an error."""
+ if route.path == "/users":
+ raise ValueError("Test error")
+ return None
+
+ # Should not raise an error, but log a warning
+ server = FastMCPOpenAPI(
+ openapi_spec=sample_openapi_spec,
+ client=http_client,
+ name="Test Server",
+ route_map_fn=error_function,
+ )
+
+ # Server should still be created successfully
+ assert server.name == "Test Server"
+
+
+def test_route_map_fn_with_complex_logic(sample_openapi_spec, http_client):
+ """Test route_map_fn with complex conditional logic."""
+
+ def complex_mapper(route, mcp_type, name):
+ """Complex mapping logic."""
+ # Convert admin routes to tools
+ if "/admin/" in route.path:
+ return MCPType.TOOL, f"admin_{name}"
+
+ # Convert user parameter routes to templates with custom naming
+ if "/users/{" in route.path:
+ return MCPType.RESOURCE_TEMPLATE, f"user_template_{name}"
+
+ # Convert list routes to resources with custom naming
+ if route.path.endswith("/users") or route.path.endswith("/data"):
+ return MCPType.RESOURCE, f"list_{name}"
+
+ # Use defaults for everything else
+ return None
+
+ server = FastMCPOpenAPI(
+ openapi_spec=sample_openapi_spec,
+ client=http_client,
+ name="Test Server",
+ route_map_fn=complex_mapper,
+ )
+
+ # Check that the complex logic was applied correctly
+ tools = server._tool_manager._tools
+ resources = server._resource_manager._resources
+ templates = server._resource_manager._templates
+
+ # Admin routes should be tools
+ assert "admin_getAdminSettings" in tools
+ assert "admin_updateAdminSettings" in tools
+
+ # List routes should be resources with custom names
+ found_list_resource = False
+ for uri in resources.keys():
+ if "list_listUsers" in uri or "list_getData" in uri:
+ found_list_resource = True
+ break
+ assert found_list_resource
+
+ # User parameter route should be template with custom name
+ found_user_template = False
+ for uri in templates.keys():
+ if "user_template_getUserById" in uri:
+ found_user_template = True
+ break
+ assert found_user_template
+
+
+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
+ def valid_route_map_fn(
+ route: openapi.HTTPRoute, mcp_type: MCPType, name: str
+ ) -> tuple[MCPType, str] | None:
+ return None
+
+ # Should be assignable to RouteMapFn type
+ fn: RouteMapFn = valid_route_map_fn
+ assert callable(fn)