Merge pull request #791 from jlowin/tags

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,7 @@ class FastMCPOpenAPI(FastMCP):
         route_map_fn: RouteMapFn | None = None,
         mcp_component_fn: ComponentFn | None = None,
         mcp_names: dict[str, str] | None = None,
+        tags: set[str] | None = None,
         timeout: float | None = None,
         **settings: Any,
     ):
@@ -702,6 +715,8 @@ 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.
+            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 +724,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 +740,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 +763,32 @@ 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_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 +833,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 +865,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 +886,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 +915,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 +936,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 +994,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_openapi.py

@@ -2460,3 +2460,357 @@ class TestMCPNames:
         assert (
             len(truncated_name) == 56
         )  # Should be exactly 56 since original was longer
+
+
+class TestRouteMapMCPTags:
+    """Tests for RouteMap mcp_tags functionality."""
+
+    @pytest.fixture
+    def simple_fastapi_app(self) -> FastAPI:
+        """Create a simple FastAPI app for testing mcp_tags."""
+        app = FastAPI(title="MCP Tags Test API")
+
+        @app.get("/users", tags=["users"])
+        async def get_users():
+            """Get all users."""
+            return [{"id": 1, "name": "Alice"}]
+
+        @app.get("/users/{user_id}", tags=["users"])
+        async def get_user(user_id: int):
+            """Get user by ID."""
+            return {"id": user_id, "name": f"User {user_id}"}
+
+        @app.post("/users", tags=["users"])
+        async def create_user(name: str):
+            """Create a new user."""
+            return {"id": 99, "name": name}
+
+        return app
+
+    @pytest.fixture
+    async def mock_client(self) -> httpx.AsyncClient:
+        """Mock client for testing."""
+
+        async def _responder(request):
+            return httpx.Response(200, json={"status": "ok"})
+
+        transport = httpx.MockTransport(_responder)
+        return httpx.AsyncClient(transport=transport, base_url="http://test")
+
+    async def test_mcp_tags_added_to_tools(self, simple_fastapi_app, mock_client):
+        """Test that mcp_tags are added to Tools created from routes."""
+        # Create route map that adds custom tags to POST endpoints
+        route_maps = [
+            RouteMap(
+                methods=["POST"],
+                pattern=r".*",
+                mcp_type=MCPType.TOOL,
+                mcp_tags={"custom", "api-write"},
+            ),
+            # Default mapping for other routes
+            RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
+        ]
+
+        server = FastMCPOpenAPI(
+            openapi_spec=simple_fastapi_app.openapi(),
+            client=mock_client,
+            route_maps=route_maps,
+        )
+
+        # Get the POST tool
+        tools = server._tool_manager.list_tools()
+        create_user_tool = next((t for t in tools if "create_user" in t.name), None)
+
+        assert create_user_tool is not None, "create_user tool not found"
+
+        # Check that both original tags and mcp_tags are present
+        assert "users" in create_user_tool.tags  # Original OpenAPI tag
+        assert "custom" in create_user_tool.tags  # Added via mcp_tags
+        assert "api-write" in create_user_tool.tags  # Added via mcp_tags
+
+    async def test_mcp_tags_added_to_resources(self, simple_fastapi_app, mock_client):
+        """Test that mcp_tags are added to Resources created from routes."""
+        # Create route map that adds custom tags to GET endpoints without path params
+        route_maps = [
+            RouteMap(
+                methods=["GET"],
+                pattern=r"^/users$",  # Only match /users, not /users/{id}
+                mcp_type=MCPType.RESOURCE,
+                mcp_tags={"list-data", "public-api"},
+            ),
+            # Default mapping for other routes
+            RouteMap(
+                methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE_TEMPLATE
+            ),
+            RouteMap(methods=["POST"], pattern=r".*", mcp_type=MCPType.TOOL),
+        ]
+
+        server = FastMCPOpenAPI(
+            openapi_spec=simple_fastapi_app.openapi(),
+            client=mock_client,
+            route_maps=route_maps,
+        )
+
+        # Get the resource
+        resources = list(server._resource_manager.get_resources().values())
+        get_users_resource = next((r for r in resources if "get_users" in r.name), None)
+
+        assert get_users_resource is not None, "get_users resource not found"
+
+        # Check that both original tags and mcp_tags are present
+        assert "users" in get_users_resource.tags  # Original OpenAPI tag
+        assert "list-data" in get_users_resource.tags  # Added via mcp_tags
+        assert "public-api" in get_users_resource.tags  # Added via mcp_tags
+
+    async def test_mcp_tags_added_to_resource_templates(
+        self, simple_fastapi_app, mock_client
+    ):
+        """Test that mcp_tags are added to ResourceTemplates created from routes."""
+        # Create route map that adds custom tags to GET endpoints with path params
+        route_maps = [
+            RouteMap(
+                methods=["GET"],
+                pattern=r".*\{.*\}.*",  # Match routes with path parameters
+                mcp_type=MCPType.RESOURCE_TEMPLATE,
+                mcp_tags={"detail-view", "parameterized"},
+            ),
+            # Default mapping for other routes
+            RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
+            RouteMap(methods=["POST"], pattern=r".*", mcp_type=MCPType.TOOL),
+        ]
+
+        server = FastMCPOpenAPI(
+            openapi_spec=simple_fastapi_app.openapi(),
+            client=mock_client,
+            route_maps=route_maps,
+        )
+
+        # Get the resource template
+        templates = list(server._resource_manager.get_templates().values())
+        get_user_template = next((t for t in templates if "get_user" in t.name), None)
+
+        assert get_user_template is not None, "get_user template not found"
+
+        # Check that both original tags and mcp_tags are present
+        assert "users" in get_user_template.tags  # Original OpenAPI tag
+        assert "detail-view" in get_user_template.tags  # Added via mcp_tags
+        assert "parameterized" in get_user_template.tags  # Added via mcp_tags
+
+    async def test_multiple_route_maps_with_different_mcp_tags(
+        self, simple_fastapi_app, mock_client
+    ):
+        """Test that different route maps can add different mcp_tags."""
+        # Multiple route maps with different mcp_tags
+        route_maps = [
+            # First priority: POST requests get write-related tags
+            RouteMap(
+                methods=["POST"],
+                pattern=r".*",
+                mcp_type=MCPType.TOOL,
+                mcp_tags={"write-operation", "mutation"},
+            ),
+            # Second priority: GET with path params get detail tags
+            RouteMap(
+                methods=["GET"],
+                pattern=r".*\{.*\}.*",
+                mcp_type=MCPType.RESOURCE_TEMPLATE,
+                mcp_tags={"detail", "single-item"},
+            ),
+            # Third priority: Other GET requests get list tags
+            RouteMap(
+                methods=["GET"],
+                pattern=r".*",
+                mcp_type=MCPType.RESOURCE,
+                mcp_tags={"list", "collection"},
+            ),
+        ]
+
+        server = FastMCPOpenAPI(
+            openapi_spec=simple_fastapi_app.openapi(),
+            client=mock_client,
+            route_maps=route_maps,
+        )
+
+        # Check tool tags
+        tools = server._tool_manager.list_tools()
+        create_tool = next((t for t in tools if "create_user" in t.name), None)
+        assert create_tool is not None
+        assert "write-operation" in create_tool.tags
+        assert "mutation" in create_tool.tags
+
+        # Check resource template tags
+        templates = list(server._resource_manager.get_templates().values())
+        detail_template = next((t for t in templates if "get_user" in t.name), None)
+        assert detail_template is not None
+        assert "detail" in detail_template.tags
+        assert "single-item" in detail_template.tags
+
+        # Check resource tags
+        resources = list(server._resource_manager.get_resources().values())
+        list_resource = next((r for r in resources if "get_users" in r.name), None)
+        assert list_resource is not None
+        assert "list" in list_resource.tags
+        assert "collection" in list_resource.tags
+
+
+class TestGlobalTagsParameter:
+    """Tests for the global tags parameter on from_openapi and from_fastapi class methods."""
+
+    @pytest.fixture
+    def simple_fastapi_app(self) -> FastAPI:
+        """Create a simple FastAPI app for testing global tags."""
+        app = FastAPI(title="Global Tags Test API")
+
+        @app.get("/items", tags=["items"])
+        async def get_items():
+            """Get all items."""
+            return [{"id": 1, "name": "Item 1"}]
+
+        @app.get("/items/{item_id}", tags=["items"])
+        async def get_item(item_id: int):
+            """Get item by ID."""
+            return {"id": item_id, "name": f"Item {item_id}"}
+
+        @app.post("/items", tags=["items"])
+        async def create_item(name: str):
+            """Create a new item."""
+            return {"id": 99, "name": name}
+
+        return app
+
+    @pytest.fixture
+    async def mock_client(self) -> httpx.AsyncClient:
+        """Mock client for testing."""
+
+        async def _responder(request):
+            return httpx.Response(200, json={"status": "ok"})
+
+        transport = httpx.MockTransport(_responder)
+        return httpx.AsyncClient(transport=transport, base_url="http://test")
+
+    async def test_from_fastapi_adds_global_tags(self, simple_fastapi_app):
+        """Test that from_fastapi adds global tags to all components."""
+        global_tags = {"global", "api-v1"}
+
+        server = FastMCP.from_fastapi(
+            simple_fastapi_app,
+            tags=global_tags,
+            route_maps=[
+                RouteMap(
+                    methods=["GET"], pattern=r"^/items$", mcp_type=MCPType.RESOURCE
+                ),
+                RouteMap(
+                    methods=["GET"],
+                    pattern=r".*\{.*\}.*",
+                    mcp_type=MCPType.RESOURCE_TEMPLATE,
+                ),
+                RouteMap(methods=["POST"], pattern=r".*", mcp_type=MCPType.TOOL),
+            ],
+        )
+
+        # Check tool has both original and global tags
+        tools = server._tool_manager.list_tools()
+        create_item_tool = next((t for t in tools if "create_item" in t.name), None)
+        assert create_item_tool is not None
+        assert "items" in create_item_tool.tags  # Original OpenAPI tag
+        assert "global" in create_item_tool.tags  # Global tag
+        assert "api-v1" in create_item_tool.tags  # Global tag
+
+        # Check resource has both original and global tags
+        resources = list(server._resource_manager.get_resources().values())
+        get_items_resource = next((r for r in resources if "get_items" in r.name), None)
+        assert get_items_resource is not None
+        assert "items" in get_items_resource.tags  # Original OpenAPI tag
+        assert "global" in get_items_resource.tags  # Global tag
+        assert "api-v1" in get_items_resource.tags  # Global tag
+
+        # Check resource template has both original and global tags
+        templates = list(server._resource_manager.get_templates().values())
+        get_item_template = next((t for t in templates if "get_item" in t.name), None)
+        assert get_item_template is not None
+        assert "items" in get_item_template.tags  # Original OpenAPI tag
+        assert "global" in get_item_template.tags  # Global tag
+        assert "api-v1" in get_item_template.tags  # Global tag
+
+    async def test_from_openapi_adds_global_tags(self, simple_fastapi_app, mock_client):
+        """Test that from_openapi adds global tags to all components."""
+        global_tags = {"openapi-global", "service"}
+
+        server = FastMCP.from_openapi(
+            openapi_spec=simple_fastapi_app.openapi(),
+            client=mock_client,
+            tags=global_tags,
+            route_maps=[
+                RouteMap(
+                    methods=["GET"], pattern=r"^/items$", mcp_type=MCPType.RESOURCE
+                ),
+                RouteMap(
+                    methods=["GET"],
+                    pattern=r".*\{.*\}.*",
+                    mcp_type=MCPType.RESOURCE_TEMPLATE,
+                ),
+                RouteMap(methods=["POST"], pattern=r".*", mcp_type=MCPType.TOOL),
+            ],
+        )
+
+        # Check tool has both original and global tags
+        tools = server._tool_manager.list_tools()
+        create_item_tool = next((t for t in tools if "create_item" in t.name), None)
+        assert create_item_tool is not None
+        assert "items" in create_item_tool.tags  # Original OpenAPI tag
+        assert "openapi-global" in create_item_tool.tags  # Global tag
+        assert "service" in create_item_tool.tags  # Global tag
+
+        # Check resource has both original and global tags
+        resources = list(server._resource_manager.get_resources().values())
+        get_items_resource = next((r for r in resources if "get_items" in r.name), None)
+        assert get_items_resource is not None
+        assert "items" in get_items_resource.tags  # Original OpenAPI tag
+        assert "openapi-global" in get_items_resource.tags  # Global tag
+        assert "service" in get_items_resource.tags  # Global tag
+
+        # Check resource template has both original and global tags
+        templates = list(server._resource_manager.get_templates().values())
+        get_item_template = next((t for t in templates if "get_item" in t.name), None)
+        assert get_item_template is not None
+        assert "items" in get_item_template.tags  # Original OpenAPI tag
+        assert "openapi-global" in get_item_template.tags  # Global tag
+        assert "service" in get_item_template.tags  # Global tag
+
+    async def test_global_tags_combine_with_route_map_tags(
+        self, simple_fastapi_app, mock_client
+    ):
+        """Test that global tags combine with both OpenAPI tags and RouteMap mcp_tags."""
+        global_tags = {"global"}
+        route_map_tags = {"route-specific"}
+
+        server = FastMCP.from_openapi(
+            openapi_spec=simple_fastapi_app.openapi(),
+            client=mock_client,
+            tags=global_tags,
+            route_maps=[
+                RouteMap(
+                    methods=["POST"],
+                    pattern=r".*",
+                    mcp_type=MCPType.TOOL,
+                    mcp_tags=route_map_tags,
+                ),
+                RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
+            ],
+        )
+
+        # Check that all three types of tags are present on the tool
+        tools = server._tool_manager.list_tools()
+        create_item_tool = next((t for t in tools if "create_item" in t.name), None)
+        assert create_item_tool is not None
+        assert "items" in create_item_tool.tags  # Original OpenAPI tag
+        assert "global" in create_item_tool.tags  # Global tag
+        assert "route-specific" in create_item_tool.tags  # RouteMap mcp_tag
+
+        # Check that resource only has OpenAPI and global tags (no route-specific since different RouteMap)
+        resources = list(server._resource_manager.get_resources().values())
+        get_items_resource = next((r for r in resources if "get_items" in r.name), None)
+        assert get_items_resource is not None
+        assert "items" in get_items_resource.tags  # Original OpenAPI tag
+        assert "global" in get_items_resource.tags  # Global tag
+        assert "route-specific" not in get_items_resource.tags  # Not from this RouteMap

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