Clean up route maps

src/fastmcp/server/openapi.py

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

tests/server/openapi/test_route_map_fn.py

@@ -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