Add route_map_fn for fine control

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
+
+<VersionBadge version="2.5.0" />
+
+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
+
+<Warning>
+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`.
+</Warning>
 
 ## Request Parameter Handling
 

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:

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,
         )
 

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)