diff --git a/src/fastmcp/server/openapi.py b/src/fastmcp/server/openapi.py index 8f6682799..74143cbb5 100644 --- a/src/fastmcp/server/openapi.py +++ b/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, ) diff --git a/tests/server/openapi/test_route_map_fn.py b/tests/server/openapi/test_route_map_fn.py index 2cac7e36b..f0cc8eef3 100644 --- a/tests/server/openapi/test_route_map_fn.py +++ b/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