diff --git a/docs/patterns/openapi.mdx b/docs/patterns/openapi.mdx index 0934cec8e..c2a586ccd 100644 --- a/docs/patterns/openapi.mdx +++ b/docs/patterns/openapi.mdx @@ -31,21 +31,20 @@ if __name__ == "__main__": ### Timeout -You can set a timeout for all API requests: +You can set a timeout for all requests by providing a `timeout` parameter (in seconds): ```python -# Set a 5 second timeout for all requests mcp = FastMCP.from_openapi( openapi_spec=spec, - client=api_client, - timeout=5.0 + client=api_client, + timeout=30.0 # 30 second timeout ) ``` -This timeout is applied to all requests made by tools, resources, and resource templates. - ## Route Mapping + + By default, OpenAPI routes are mapped to MCP components based on these rules: | OpenAPI Route | Example |MCP Component | Notes | @@ -54,7 +53,6 @@ By default, OpenAPI routes are mapped to MCP components based on these rules: | `GET` with path params | `GET /users/{id}` | Resource Template | Path parameters become template parameters | | `POST`, `PUT`, `PATCH`, `DELETE`, etc. | `POST /users` | Tool | Operations that modify data | - Internally, FastMCP uses a priority-ordered set of `RouteMap` objects to determine the component type. Route maps indicate that a specific HTTP method (or methods) and path pattern should be treated as a specific component type. This is the default set of route maps: ```python @@ -64,75 +62,195 @@ DEFAULT_ROUTE_MAPPINGS = [ RouteMap( methods=["GET"], pattern=r".*\{.*\}.*", - route_type=RouteType.RESOURCE_TEMPLATE, + mcp_type=MCPType.RESOURCE_TEMPLATE, ), # GET without path parameters -> Resource RouteMap( methods=["GET"], pattern=r".*", - route_type=RouteType.RESOURCE, + mcp_type=MCPType.RESOURCE, ), # All other methods -> Tool - RouteMap( - methods="*", - pattern=r".*", - route_type=RouteType.TOOL, - ), + ALL_TOOLS(), ] ``` -### Custom Route Maps + +#### Custom Route Maps Users can add custom route maps to override the default mapping behavior. User-supplied route maps are always applied first, before the default route maps. ```python -from fastmcp.server.openapi import RouteMap, RouteType +from fastmcp.server.openapi import RouteMap, MCPType # Custom mapping rules custom_maps = [ # Force all analytics endpoints to be Tools RouteMap(methods=["GET"], pattern=r"^/analytics/.*", - route_type=RouteType.TOOL) + mcp_type=MCPType.TOOL) ] # Apply custom mappings -mcp = await FastMCP.from_openapi( +mcp = FastMCP.from_openapi( openapi_spec=spec, client=api_client, route_maps=custom_maps ) ``` + +For backward compatibility, FastMCP still supports the `route_type` parameter and `RouteType` enum, but they are deprecated and will be removed in a future version. You will see deprecation warnings if you use them. + -### All Routes as Tools +#### All Routes as Tools -When building AI agent backends, it's often useful to treat all routes as callable tools regardless of their HTTP method. You can use the `all_routes_as_tools` parameter to automatically map every route to a Tool: +When building AI agent backends, it's often useful to treat all routes as callable tools regardless of their HTTP method. You can use the `ALL_TOOLS()` shortcut or create a custom route map: ```python -# Make all endpoints tools, regardless of HTTP method +# Make all endpoints tools using the shortcut mcp = FastMCP.from_openapi( openapi_spec=spec, client=api_client, - all_routes_as_tools=True + route_maps=[ALL_TOOLS()] ) -``` -This is equivalent to defining a single route map that matches all routes: - -```python -# Same effect as all_routes_as_tools=True +# Same effect using a custom route map mcp = FastMCP.from_openapi( openapi_spec=spec, client=api_client, route_maps=[ - RouteMap(methods="*", pattern=r".*", route_type=RouteType.TOOL) + RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL) ] ) ``` -Note that `all_routes_as_tools` and `route_maps` cannot be used together - if you need more complex mapping rules, use `route_maps` instead. +#### Excluding Routes + +If you want to exclude certain routes from being converted to MCP components, you can map them to `MCPType.EXCLUDE`. This is useful for endpoints that should not be accessible to the agent. + +```python +from fastmcp.server.openapi import RouteMap, MCPType + +# Custom mapping rules to exclude specific routes +custom_maps = [ + # Exclude all admin endpoints + RouteMap( + methods="*", + pattern=r"^/admin/.*", + mcp_type=MCPType.EXCLUDE + ), + # Exclude analytics GET endpoints + RouteMap( + methods=["GET"], + pattern=r"^/analytics/.*", + mcp_type=MCPType.EXCLUDE + ) +] + +# Apply custom mappings +mcp = FastMCP.from_openapi( + openapi_spec=spec, + client=api_client, + route_maps=custom_maps +) +``` + +When a route is mapped to `MCPType.EXCLUDE`, FastMCP will log its presence but won't create any MCP component for it, effectively making it invisible to clients and agents using the MCP server. + +You can customize this behavior by providing a list of `RouteMap` objects: + +```python +from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap, MCPType + +# Custom route mappings +custom_mappings = [ + # Convert all user-related routes to tools + RouteMap( + methods=["GET", "POST", "PUT", "DELETE"], + pattern=r"^/users.*", + mcp_type=MCPType.TOOL + ), + # Exclude analytics routes + RouteMap( + methods=["*"], # All methods + pattern=r"^/analytics.*", + mcp_type=MCPType.EXCLUDE + ), +] + +# Create server with custom mappings +mcp = FastMCPOpenAPI( + openapi_spec=spec, + client=httpx.AsyncClient(), + route_maps=custom_mappings, +) +``` + +#### Route Map Shortcuts + + +FastMCP provides several shortcut functions to create common route maps more easily: + +```python +from fastmcp.server.openapi import ( + ALL_TOOLS, + EXCLUDE_ALL, + EXCLUDE_PATTERN, + PATTERN_AS_TOOLS, +) + +# Create an MCP server with custom route maps using shortcuts +mcp = FastMCP.from_openapi( + openapi_spec=spec, + client=api_client, + route_maps=[ + # First exclude all admin endpoints + EXCLUDE_PATTERN(r"^/admin/.*"), + + # Make all /api/v1 endpoints tools + PATTERN_AS_TOOLS(r"^/api/v1/.*"), + + # Make all remaining routes tools + ALL_TOOLS(), + ] +) +``` + +Available shortcuts: + +| Shortcut Function | Description | +|------------------|-------------| +| `ALL_TOOLS()` | Converts all matching routes to tools | +| `EXCLUDE_ALL()` | Excludes all matching routes from being converted to any component | +| `PATTERN_AS_TOOLS(pattern)` | Converts routes matching a specific pattern to tools | +| `EXCLUDE_PATTERN(pattern)` | Excludes routes matching a specific pattern | + +These shortcuts are particularly useful for: + +1. Converting all remaining unmatched routes to tools (use `ALL_TOOLS()`) +2. Excluding whole sections of your API (use `EXCLUDE_PATTERN("/path/.*")`) +3. Converting routes matching specific patterns to tools (use `PATTERN_AS_TOOLS("/path/.*")`) + + +You can use `EXCLUDE_ALL()` as the last entry in your custom route maps to completely ignore the default route maps. Since custom route maps are applied first and default maps are appended afterward, having `EXCLUDE_ALL()` at the end of your custom maps will match any routes that your earlier custom rules didn't match, preventing the default maps from having any effect. + +```python +# Create server that only uses custom route maps, ignoring defaults +mcp = FastMCP.from_openapi( + openapi_spec=spec, + client=api_client, + route_maps=[ + # Routes to keep as tools + PATTERN_AS_TOOLS(r"^/api/v1/.*"), + + # Exclude everything else (ignores default route maps) + EXCLUDE_ALL(), + ] +) +``` + ## How It Works @@ -175,89 +293,20 @@ await client.call_tool("get_product", {"product_id": 123}) await client.call_tool("get_product", {"product_id": None}) ``` -## Complete Example +## Example: Custom Authentication -```python [expandable] -import asyncio +If your API requires authentication, you can set headers on the client: +```python import httpx - from fastmcp import FastMCP -# Sample OpenAPI spec for a Pet Store API -petstore_spec = { - "openapi": "3.0.0", - "info": { - "title": "Pet Store API", - "version": "1.0.0", - "description": "A sample API for managing pets", - }, - "paths": { - "/pets": { - "get": { - "operationId": "listPets", - "summary": "List all pets", - "responses": {"200": {"description": "A list of pets"}}, - }, - "post": { - "operationId": "createPet", - "summary": "Create a new pet", - "responses": {"201": {"description": "Pet created successfully"}}, - }, - }, - "/pets/{petId}": { - "get": { - "operationId": "getPet", - "summary": "Get a pet by ID", - "parameters": [ - { - "name": "petId", - "in": "path", - "required": True, - "schema": {"type": "string"}, - } - ], - "responses": { - "200": {"description": "Pet details"}, - "404": {"description": "Pet not found"}, - }, - } - }, - }, -} +# Create a client with authentication +api_client = httpx.AsyncClient( + base_url="https://api.example.com", + headers={"Authorization": "Bearer YOUR_TOKEN"} +) - -async def check_mcp(mcp: FastMCP): - # List what components were created - tools = await mcp.get_tools() - resources = await mcp.get_resources() - templates = await mcp.get_resource_templates() - - print( - f"{len(tools)} Tool(s): {', '.join([t.name for t in tools.values()])}" - ) # Should include createPet - print( - f"{len(resources)} Resource(s): {', '.join([r.name for r in resources.values()])}" - ) # Should include listPets - print( - f"{len(templates)} Resource Template(s): {', '.join([t.name for t in templates.values()])}" - ) # Should include getPet - - return mcp - - -if __name__ == "__main__": - # Client for the Pet Store API - client = httpx.AsyncClient(base_url="https://petstore.example.com/api") - - # Create the MCP server - mcp = FastMCP.from_openapi( - openapi_spec=petstore_spec, client=client, name="PetStore" - ) - - asyncio.run(check_mcp(mcp)) - - # Start the MCP server - mcp.run() +# Create an MCP server from your OpenAPI spec +mcp = FastMCP.from_openapi(openapi_spec=spec, client=api_client) ``` - diff --git a/src/fastmcp/server/openapi.py b/src/fastmcp/server/openapi.py index 687790653..0bed5c537 100644 --- a/src/fastmcp/server/openapi.py +++ b/src/fastmcp/server/openapi.py @@ -5,8 +5,9 @@ from __future__ import annotations import enum import json import re +import warnings from collections.abc import Callable -from dataclasses import dataclass +from dataclasses import dataclass, field from re import Pattern from typing import TYPE_CHECKING, Any, Literal @@ -33,8 +34,32 @@ logger = get_logger(__name__) HttpMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"] +class MCPType(enum.Enum): + """Type of FastMCP component to create from a route. + + Enum values: + TOOL: Convert the route to a callable Tool + RESOURCE: Convert the route to a Resource (typically GET endpoints) + RESOURCE_TEMPLATE: Convert the route to a ResourceTemplate (typically GET with path params) + PROMPT: Convert the route to a Prompt (not yet implemented) + EXCLUDE: Exclude the route from being converted to any MCP component + IGNORE: Deprecated, use EXCLUDE instead + """ + + TOOL = "TOOL" + RESOURCE = "RESOURCE" + RESOURCE_TEMPLATE = "RESOURCE_TEMPLATE" + PROMPT = "PROMPT" + EXCLUDE = "EXCLUDE" + + +# Keep RouteType as an alias to MCPType for backward compatibility class RouteType(enum.Enum): - """Type of FastMCP component to create from a route.""" + """ + Deprecated: Use MCPType instead. + + This enum is kept for backward compatibility and will be removed in a future version. + """ TOOL = "TOOL" RESOURCE = "RESOURCE" @@ -47,32 +72,121 @@ class RouteType(enum.Enum): class RouteMap: """Mapping configuration for HTTP routes to FastMCP component types.""" - methods: list[HttpMethod] | Literal["*"] - pattern: Pattern[str] | str - route_type: RouteType + 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) + + def __post_init__(self): + """Validate and process the route map after initialization.""" + # Handle backward compatibility for route_type, deprecated in 2.5.0 + if self.mcp_type is None and self.route_type is not None: + warnings.warn( + "The 'route_type' parameter is deprecated and will be removed in a future version. " + "Use 'mcp_type' instead with the appropriate MCPType value.", + DeprecationWarning, + stacklevel=2, + ) + if isinstance(self.route_type, RouteType): + warnings.warn( + "The RouteType class is deprecated and will be removed in a future version. " + "Use MCPType instead.", + DeprecationWarning, + stacklevel=2, + ) + # Check for the deprecated IGNORE value + if self.route_type == RouteType.IGNORE: + warnings.warn( + "RouteType.IGNORE is deprecated and will be removed in a future version. " + "Use MCPType.EXCLUDE instead.", + DeprecationWarning, + stacklevel=2, + ) + + # Convert from RouteType to MCPType if needed + if isinstance(self.route_type, RouteType): + route_type_name = self.route_type.name + if route_type_name == "IGNORE": + route_type_name = "EXCLUDE" + self.mcp_type = getattr(MCPType, route_type_name) + else: + self.mcp_type = self.route_type + elif self.mcp_type is None: + raise ValueError("`mcp_type` must be provided") + + # Set route_type to match mcp_type for backward compatibility + if self.route_type is None: + self.route_type = self.mcp_type + + +# Common route map pattern functions +def EXCLUDE_ALL() -> RouteMap: + """ + Create a RouteMap that excludes all routes that haven't been matched by earlier rules. + + This is useful as the last route map to exclude any routes that don't match specific patterns. + + Returns: + RouteMap: A route map that excludes all routes + """ + return RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.EXCLUDE) + + +def ALL_TOOLS() -> RouteMap: + """ + Create a RouteMap that converts all routes to tools that haven't been matched by earlier rules. + + This is useful to replace the last item in the default route mappings to make all unmatched routes tools. + + Returns: + RouteMap: A route map that converts all routes to tools + """ + return RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL) + + +def PATTERN_AS_TOOLS(pattern: str) -> RouteMap: + """ + Create a RouteMap that converts routes matching a specific pattern to tools. + + Args: + pattern: Regex pattern to match routes + + Returns: + RouteMap: A route map that converts routes matching the pattern to tools + """ + return RouteMap(methods="*", pattern=pattern, mcp_type=MCPType.TOOL) + + +def EXCLUDE_PATTERN(pattern: str) -> RouteMap: + """ + Create a RouteMap that excludes routes matching a specific pattern. + + Args: + pattern: Regex pattern to match routes to exclude + + Returns: + RouteMap: A route map that excludes routes matching the pattern + """ + return RouteMap(methods="*", pattern=pattern, mcp_type=MCPType.EXCLUDE) # Default route mappings as a list, where order determines priority DEFAULT_ROUTE_MAPPINGS = [ # GET requests with path parameters go to ResourceTemplate RouteMap( - methods=["GET"], pattern=r".*\{.*\}.*", route_type=RouteType.RESOURCE_TEMPLATE + methods=["GET"], pattern=r".*\{.*\}.*", mcp_type=MCPType.RESOURCE_TEMPLATE ), # GET requests without path parameters go to Resource - RouteMap(methods=["GET"], pattern=r".*", route_type=RouteType.RESOURCE), + RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE), # All other HTTP methods go to Tool - RouteMap( - methods=["POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"], - pattern=r".*", - route_type=RouteType.TOOL, - ), + ALL_TOOLS(), ] def _determine_route_type( route: openapi.HTTPRoute, mappings: list[RouteMap], -) -> RouteType: +) -> MCPType: """ Determines the FastMCP component type based on the route and mappings. @@ -81,7 +195,7 @@ def _determine_route_type( mappings: List of RouteMap objects in priority order Returns: - RouteType for this route + MCPType for this route """ # Check mappings in priority order (first match wins) for route_map in mappings: @@ -94,20 +208,15 @@ def _determine_route_type( pattern_matches = re.search(route_map.pattern, route.path) if pattern_matches: + # We know mcp_type is not None here due to post_init validation + assert route_map.mcp_type is not None logger.debug( - f"Route {route.method} {route.path} matched mapping to {route_map.route_type.name}" + f"Route {route.method} {route.path} matched mapping to {route_map.mcp_type.name}" ) - return route_map.route_type + return route_map.mcp_type # Default fallback - return RouteType.TOOL - - -# Placeholder function to provide function metadata -async def _openapi_passthrough(*args, **kwargs): - """Placeholder function for OpenAPI endpoints.""" - # This is kept for metadata generation purposes - pass + return MCPType.TOOL class OpenAPITool(Tool): @@ -555,13 +664,13 @@ class FastMCPOpenAPI(FastMCP): RouteMap( methods=["GET", "POST", "PATCH"], pattern=r".*/users/.*", - route_type=RouteType.RESOURCE_TEMPLATE + mcp_type=MCPType.RESOURCE_TEMPLATE ), # Map all analytics endpoints to Tool RouteMap( methods=["GET"], pattern=r".*/analytics/.*", - route_type=RouteType.TOOL + mcp_type=MCPType.TOOL ), ] @@ -599,6 +708,10 @@ class FastMCPOpenAPI(FastMCP): self._client = client self._timeout = timeout + + # Keep track of names to detect collisions + self._used_names = {"tools": set(), "resources": set(), "templates": set()} + http_routes = openapi.parse_openapi_to_http_routes(openapi_spec) # Process routes @@ -607,34 +720,99 @@ class FastMCPOpenAPI(FastMCP): # Determine route type based on mappings or default rules route_type = _determine_route_type(route, route_maps) - # Use operation_id if available, otherwise generate a name - operation_id = route.operation_id - if not operation_id: - # Generate operation ID from method and path - path_parts = route.path.strip("/").split("/") - path_name = "_".join(p for p in path_parts if not p.startswith("{")) - operation_id = f"{route.method.lower()}_{path_name}" + # Generate a default name from the route + component_name = self._generate_default_name(route, route_type) - if route_type == RouteType.TOOL: - self._create_openapi_tool(route, operation_id) - elif route_type == RouteType.RESOURCE: - self._create_openapi_resource(route, operation_id) - elif route_type == RouteType.RESOURCE_TEMPLATE: - self._create_openapi_template(route, operation_id) - elif route_type == RouteType.PROMPT: + if route_type == MCPType.TOOL: + self._create_openapi_tool(route, component_name) + elif route_type == MCPType.RESOURCE: + self._create_openapi_resource(route, component_name) + elif route_type == MCPType.RESOURCE_TEMPLATE: + self._create_openapi_template(route, component_name) + elif route_type == MCPType.PROMPT: # Not implemented yet logger.warning( f"PROMPT route type not implemented: {route.method} {route.path}" ) - elif route_type == RouteType.IGNORE: - logger.info(f"Ignoring route: {route.method} {route.path}") + 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 _create_openapi_tool(self, route: openapi.HTTPRoute, operation_id: str): + def _generate_default_name( + self, route: openapi.HTTPRoute, mcp_type: MCPType + ) -> str: + """Generate a default name from the route path.""" + # First check for OpenAPI operationId which takes precedence + if route.operation_id: + return route.operation_id + + # For path-based naming, clean up the path + path_parts = route.path.strip("/").split("/") + + # Remove path parameters (parts with {}) + clean_parts = [] + for part in path_parts: + if part.startswith("{") and part.endswith("}"): + # For templates, include parameter name without braces + if mcp_type == MCPType.RESOURCE_TEMPLATE: + param_name = part[1:-1] # Remove braces + clean_parts.append(param_name) + else: + clean_parts.append(part) + + # Join the parts + resource_name = "_".join(clean_parts) + + # For tools, might be useful to keep the method for clarity on what it does + if mcp_type == MCPType.TOOL: + # Only include method if it helps distinguish (POST, PUT, PATCH, DELETE) + # For GET we don't need the method as it's implied for resources + if route.method != "GET": + resource_name = f"{route.method.lower()}_{resource_name}" + + return resource_name + + def _get_unique_name( + self, name: str, component_type: Literal["tools", "resources", "templates"] + ) -> str: + """ + Ensure the name is unique within its component type by appending numbers if needed. + + Args: + name: The proposed name + component_type: The type of component ("tools", "resources", or "templates") + + Returns: + str: A unique name for the component + """ + # Check if the name is already used + if name not in self._used_names[component_type]: + self._used_names[component_type].add(name) + return name + + # Find the next available number suffix + counter = 2 + while f"{name}_{counter}" in self._used_names[component_type]: + counter += 1 + + # Create the new name + new_name = f"{name}_{counter}" + logger.debug( + f"Name collision detected: '{name}' already exists as a {component_type[:-1]}. " + f"Using '{new_name}' instead." + ) + + self._used_names[component_type].add(new_name) + return new_name + + def _create_openapi_tool(self, route: openapi.HTTPRoute, name: str): """Creates and registers an OpenAPITool with enhanced description.""" combined_schema = _combine_schemas(route) - tool_name = operation_id + + # Get a unique tool name + tool_name = self._get_unique_name(name, "tools") + base_description = ( route.description or route.summary @@ -664,9 +842,11 @@ class FastMCPOpenAPI(FastMCP): f"Registered TOOL: {tool_name} ({route.method} {route.path}) with tags: {route.tags}" ) - def _create_openapi_resource(self, route: openapi.HTTPRoute, operation_id: str): + def _create_openapi_resource(self, route: openapi.HTTPRoute, name: str): """Creates and registers an OpenAPIResource with enhanced description.""" - resource_name = operation_id + # Get a unique resource name + resource_name = self._get_unique_name(name, "resources") + resource_uri = f"resource://openapi/{resource_name}" base_description = ( route.description or route.summary or f"Represents {route.path}" @@ -695,9 +875,11 @@ class FastMCPOpenAPI(FastMCP): f"Registered RESOURCE: {resource_uri} ({route.method} {route.path}) with tags: {route.tags}" ) - def _create_openapi_template(self, route: openapi.HTTPRoute, operation_id: str): + def _create_openapi_template(self, route: openapi.HTTPRoute, name: str): """Creates and registers an OpenAPIResourceTemplate with enhanced description.""" - template_name = operation_id + # Get a unique template name + template_name = self._get_unique_name(name, "templates") + path_params = [p.name for p in route.parameters if p.location == "path"] path_params.sort() # Sort for consistent URIs diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index 6ce1dff96..dfd1b67e1 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -1147,19 +1147,22 @@ class FastMCP(Generic[LifespanResultT]): """ Create a FastMCP server from an OpenAPI specification. """ - from .openapi import FastMCPOpenAPI, RouteMap, RouteType + from .openapi import ALL_TOOLS, FastMCPOpenAPI + + # Deprecated since 2.5.0 + if all_routes_as_tools: + warnings.warn( + "The 'all_routes_as_tools' parameter is deprecated and will be removed in a future version. " + "Use 'route_maps=[ALL_TOOLS()]' instead.", + DeprecationWarning, + stacklevel=2, + ) if all_routes_as_tools and route_maps: raise ValueError("Cannot specify both all_routes_as_tools and route_maps") elif all_routes_as_tools: - route_maps = [ - RouteMap( - methods="*", - pattern=r".*", - route_type=RouteType.TOOL, - ) - ] + route_maps = [ALL_TOOLS()] return FastMCPOpenAPI( openapi_spec=openapi_spec, @@ -1181,15 +1184,21 @@ class FastMCP(Generic[LifespanResultT]): Create a FastMCP server from a FastAPI application. """ - from .openapi import FastMCPOpenAPI, RouteMap, RouteType + from .openapi import ALL_TOOLS, FastMCPOpenAPI + + if all_routes_as_tools: + warnings.warn( + "The 'all_routes_as_tools' parameter is deprecated and will be removed in a future version. " + "Use 'route_maps=[ALL_TOOLS()]' instead.", + DeprecationWarning, + stacklevel=2, + ) if all_routes_as_tools and route_maps: raise ValueError("Cannot specify both all_routes_as_tools and route_maps") elif all_routes_as_tools: - route_maps = [ - RouteMap(methods="*", pattern=r".*", route_type=RouteType.TOOL) - ] + route_maps = [ALL_TOOLS()] client = httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://fastapi" diff --git a/tests/deprecated/test_route_type_ignore.py b/tests/deprecated/test_route_type_ignore.py new file mode 100644 index 000000000..1382d7137 --- /dev/null +++ b/tests/deprecated/test_route_type_ignore.py @@ -0,0 +1,113 @@ +"""Tests for the deprecated RouteType.IGNORE.""" + +import warnings + +import httpx +import pytest + +from fastmcp.server.openapi import ( + FastMCPOpenAPI, + MCPType, + RouteMap, + RouteType, +) + + +def test_route_type_ignore_deprecation_warning(): + """Test that using RouteType.IGNORE emits a deprecation warning.""" + # Let's manually capture the warnings + + # Record all warnings + with warnings.catch_warnings(record=True) as recorded: + # Make sure warnings are always triggered + warnings.simplefilter("always") + + # Create a RouteMap with RouteType.IGNORE + route_map = RouteMap( + methods=["GET"], pattern=r"^/analytics$", route_type=RouteType.IGNORE + ) + + # Check for the expected warnings in the recorded warnings + route_type_warning = False + ignore_warning = False + + for w in recorded: + if issubclass(w.category, DeprecationWarning): + message = str(w.message) + if "route_type' parameter is deprecated" in message: + route_type_warning = True + if "RouteType.IGNORE is deprecated" in message: + ignore_warning = True + + # Make sure both warnings were triggered + assert route_type_warning, "Missing 'route_type' deprecation warning" + assert ignore_warning, "Missing 'RouteType.IGNORE' deprecation warning" + + # Verify that RouteType.IGNORE was converted to MCPType.EXCLUDE + assert route_map.mcp_type == MCPType.EXCLUDE + + +class TestRouteTypeIgnoreDeprecation: + """Test class for the deprecated RouteType.IGNORE.""" + + @pytest.fixture + def basic_openapi_spec(self) -> dict: + """Create a simple OpenAPI spec for testing.""" + return { + "openapi": "3.0.0", + "info": {"title": "Test API", "version": "1.0.0"}, + "paths": { + "/items": { + "get": { + "operationId": "get_items", + "summary": "Get all items", + "responses": {"200": {"description": "Success"}}, + } + }, + "/analytics": { + "get": { + "operationId": "get_analytics", + "summary": "Get analytics data", + "responses": {"200": {"description": "Success"}}, + } + }, + }, + } + + @pytest.fixture + async def mock_client(self) -> httpx.AsyncClient: + """Create a mock client for testing.""" + + async def _responder(request): + return httpx.Response(200, json={"success": True}) + + return httpx.AsyncClient(transport=httpx.MockTransport(_responder)) + + async def test_route_type_ignore_conversion(self, basic_openapi_spec, mock_client): + """Test that routes with RouteType.IGNORE are properly excluded.""" + # Capture the deprecation warning without checking the exact message + with pytest.warns(DeprecationWarning): + server = FastMCPOpenAPI( + openapi_spec=basic_openapi_spec, + client=mock_client, + route_maps=[ + # Use the deprecated RouteType.IGNORE + RouteMap( + methods=["GET"], + pattern=r"^/analytics$", + route_type=RouteType.IGNORE, + ), + # Make everything else a resource + RouteMap( + methods=["GET"], pattern=r".*", route_type=RouteType.RESOURCE + ), + ], + ) + + # Check that the analytics route was excluded (converted from IGNORE to EXCLUDE) + resources = await server.get_resources() + resource_uris = [str(r.uri) for r in resources.values()] + + # Analytics should be excluded + assert "resource://openapi/get_items" in resource_uris + assert "resource://openapi/get_analytics" not in resource_uris diff --git a/tests/server/test_openapi.py b/tests/server/openapi/test_openapi.py similarity index 93% rename from tests/server/test_openapi.py rename to tests/server/openapi/test_openapi.py index 3149ac19c..4dfb721e0 100644 --- a/tests/server/test_openapi.py +++ b/tests/server/openapi/test_openapi.py @@ -18,11 +18,11 @@ from fastmcp.client import Client from fastmcp.exceptions import ToolError from fastmcp.server.openapi import ( FastMCPOpenAPI, + MCPType, OpenAPIResource, OpenAPIResourceTemplate, OpenAPITool, RouteMap, - RouteType, ) @@ -304,7 +304,7 @@ class TestTools: openapi_spec=openapi_spec, client=api_client, route_maps=[ - RouteMap(methods=["GET"], pattern=r".*", route_type=RouteType.TOOL) + RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.TOOL) ], ) async with Client(mcp_server) as client: @@ -956,9 +956,7 @@ async def test_empty_query_parameters_not_sent( mcp_server = FastMCPOpenAPI( openapi_spec=openapi_spec, client=api_client, - route_maps=[ - RouteMap(methods=["GET"], pattern=r".*", route_type=RouteType.TOOL) - ], + route_maps=[RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.TOOL)], ) # Call the search tool with mixed parameter values @@ -1499,17 +1497,15 @@ class TestFastAPIDescriptionPropagation: # Create custom route mappings route_maps = [ # Map GET /items to Resource - RouteMap( - methods=["GET"], pattern=r"^/items$", route_type=RouteType.RESOURCE - ), + RouteMap(methods=["GET"], pattern=r"^/items$", mcp_type=MCPType.RESOURCE), # Map GET /items/{item_id} to ResourceTemplate RouteMap( methods=["GET"], pattern=r"^/items/\{.*\}$", - route_type=RouteType.RESOURCE_TEMPLATE, + mcp_type=MCPType.RESOURCE_TEMPLATE, ), # Map POST /items to Tool - RouteMap(methods=["POST"], pattern=r"^/items$", route_type=RouteType.TOOL), + RouteMap(methods=["POST"], pattern=r"^/items$", mcp_type=MCPType.TOOL), ] # Create FastMCP server with the OpenAPI spec and custom route mappings @@ -1918,7 +1914,7 @@ class TestRouteMapWildcard: ): """Test that a RouteMap with methods='*' matches all HTTP methods.""" # Create a single route map with wildcard method - route_maps = [RouteMap(methods="*", pattern=r".*", route_type=RouteType.TOOL)] + route_maps = [RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL)] mcp = FastMCPOpenAPI( openapi_spec=basic_openapi_spec, @@ -1947,9 +1943,9 @@ class TestRouteMapWildcard: # Create route maps with specific method first, then wildcard route_maps = [ # GET operations should be mapped to resources - RouteMap(methods=["GET"], pattern=r".*", route_type=RouteType.RESOURCE), + RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE), # All other operations should be mapped to tools - RouteMap(methods="*", pattern=r".*", route_type=RouteType.TOOL), + RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL), ] mcp = FastMCPOpenAPI( @@ -1977,9 +1973,9 @@ class TestRouteMapWildcard: # Create route maps with wildcard first, then specific methods route_maps = [ # Wildcard first matches everything - RouteMap(methods="*", pattern=r".*", route_type=RouteType.TOOL), + RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL), # This should never be reached - RouteMap(methods=["GET"], pattern=r".*", route_type=RouteType.RESOURCE), + RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE), ] mcp = FastMCPOpenAPI( @@ -2002,9 +1998,9 @@ class TestRouteMapWildcard: """Test wildcard methods combined with specific path patterns.""" route_maps = [ # All methods on /users path -> Resources - RouteMap(methods="*", pattern=r".*/users$", route_type=RouteType.RESOURCE), + RouteMap(methods="*", pattern=r".*/users$", mcp_type=MCPType.RESOURCE), # All methods on /posts path -> Tools - RouteMap(methods="*", pattern=r".*/posts$", route_type=RouteType.TOOL), + RouteMap(methods="*", pattern=r".*/posts$", mcp_type=MCPType.TOOL), ] mcp = FastMCPOpenAPI( @@ -2063,92 +2059,169 @@ class TestAllRoutesAsTools: async def test_from_openapi_all_routes_as_tools(self, simple_api_spec, mock_client): """Test FastMCP.from_openapi with all_routes_as_tools=True.""" - # Create server with all routes as tools - server = FastMCP.from_openapi( - openapi_spec=simple_api_spec, client=mock_client, all_routes_as_tools=True - ) - # All operations (GET and POST) should be mapped to tools - tools = server._tool_manager.list_tools() - tool_names = {t.name for t in tools} + with pytest.warns(DeprecationWarning, match="all_routes_as_tools.*deprecated"): + server = FastMCP.from_openapi( + openapi_spec=simple_api_spec, + client=mock_client, + all_routes_as_tools=True, + ) - assert "getItems" in tool_names - assert "createItem" in tool_names - assert len(tools) == 2 + # Check that all routes are tools + tools = await server.get_tools() + assert len(tools) >= 2 # Should have at least the two endpoints as tools - # No resources or templates should be created - resources = server._resource_manager.get_resources() - templates = server._resource_manager.get_templates() + # Should have no resources since all routes are tools + resources = await server.get_resources() assert len(resources) == 0 + + # Should have no resource templates since all routes are tools + templates = await server.get_resource_templates() assert len(templates) == 0 async def test_from_openapi_all_routes_as_tools_conflicting_args( self, simple_api_spec, mock_client ): """Test FastMCP.from_openapi raises error when both route_maps and all_routes_as_tools are provided.""" - # Try to create server with conflicting args with pytest.raises( ValueError, match="Cannot specify both all_routes_as_tools and route_maps" ): - FastMCP.from_openapi( - openapi_spec=simple_api_spec, - client=mock_client, - all_routes_as_tools=True, - route_maps=[ - RouteMap( - methods=["GET"], pattern=r".*", route_type=RouteType.RESOURCE - ) - ], - ) + with pytest.warns( + DeprecationWarning, match="all_routes_as_tools.*deprecated" + ): + FastMCP.from_openapi( + openapi_spec=simple_api_spec, + client=mock_client, + route_maps=[ + RouteMap( + methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE + ) + ], + all_routes_as_tools=True, + ) async def test_from_fastapi_all_routes_as_tools(self): """Test FastMCP.from_fastapi with all_routes_as_tools=True.""" - # Create a simple FastAPI app - app = FastAPI(title="Test FastAPI") + + try: + import fastapi + except ImportError: + pytest.skip("FastAPI not available") + + app = fastapi.FastAPI() @app.get("/items") - async def get_items(): - return [{"id": 1, "name": "Item 1"}] + def get_items(): + return {"items": []} @app.post("/items") - async def create_item(item: dict): - return {"id": 2, **item} + def create_item(): + return {"item": "created"} - # Create server with all routes as tools - server = FastMCP.from_fastapi(app=app, all_routes_as_tools=True) + with pytest.warns(DeprecationWarning, match="all_routes_as_tools.*deprecated"): + server = FastMCP.from_fastapi(app=app, all_routes_as_tools=True) - # Both GET and POST operations should be mapped to tools - tools = server._tool_manager.list_tools() + # Check that all routes are tools + tools = await server.get_tools() + assert len(tools) >= 2 # Should have at least the two endpoints as tools - # Get tool names from the generated operation IDs - tool_names = {t.name for t in tools} - - # Check that both routes were mapped to tools - # The exact names depend on FastAPI's operation ID generation - assert len(tools) == 2 - assert any("get" in name.lower() for name in tool_names) - assert any("post" in name.lower() for name in tool_names) - - # No resources or templates should be created - resources = server._resource_manager.get_resources() - templates = server._resource_manager.get_templates() + # Should have no resources since all routes are tools + resources = await server.get_resources() assert len(resources) == 0 + + # Should have no resource templates since all routes are tools + templates = await server.get_resource_templates() assert len(templates) == 0 async def test_from_fastapi_all_routes_as_tools_conflicting_args(self): """Test FastMCP.from_fastapi raises error when both route_maps and all_routes_as_tools are provided.""" - app = FastAPI(title="Test FastAPI") + try: + import fastapi + except ImportError: + pytest.skip("FastAPI not available") + + app = fastapi.FastAPI() - # Try to create server with conflicting args with pytest.raises( ValueError, match="Cannot specify both all_routes_as_tools and route_maps" ): - FastMCP.from_fastapi( - app=app, - all_routes_as_tools=True, - route_maps=[ - RouteMap( - methods=["GET"], pattern=r".*", route_type=RouteType.RESOURCE - ) - ], - ) + with pytest.warns( + DeprecationWarning, match="all_routes_as_tools.*deprecated" + ): + FastMCP.from_fastapi( + app=app, + route_maps=[ + RouteMap( + methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE + ) + ], + all_routes_as_tools=True, + ) + + +class TestRouteTypeExclude: + @pytest.fixture + def basic_openapi_spec(self) -> dict: + return { + "openapi": "3.0.0", + "info": {"title": "Test API", "version": "1.0.0"}, + "paths": { + "/items": { + "get": { + "operationId": "get_items", + "summary": "Get all items", + "responses": {"200": {"description": "Success"}}, + } + }, + "/users": { + "get": { + "operationId": "get_users", + "summary": "Get all users", + "responses": {"200": {"description": "Success"}}, + } + }, + "/analytics": { + "get": { + "operationId": "get_analytics", + "summary": "Get analytics data", + "responses": {"200": {"description": "Success"}}, + } + }, + }, + } + + @pytest.fixture + async def mock_client(self) -> httpx.AsyncClient: + async def _responder(request): + return httpx.Response(200, json={"success": True}) + + return httpx.AsyncClient(transport=httpx.MockTransport(_responder)) + + async def test_exclude_routes(self, basic_openapi_spec, mock_client): + # Create a server with custom mappings that exclude specific routes + server = FastMCPOpenAPI( + openapi_spec=basic_openapi_spec, + client=mock_client, + route_maps=[ + # Exclude analytics endpoints + RouteMap( + methods=["GET"], + pattern=r"^/analytics$", + mcp_type=MCPType.EXCLUDE, + ), + # Make everything else a resource + RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE), + ], + ) + + # Check that resources were created for non-excluded routes + resources = await server.get_resources() + resource_uris = [str(r.uri) for r in resources.values()] + + # The /analytics endpoint should be excluded + assert "resource://openapi/get_items" in resource_uris + assert "resource://openapi/get_users" in resource_uris + assert "resource://openapi/get_analytics" not in resource_uris + + # Should only have 2 resources (analytics is excluded) + assert len(resources) == 2 diff --git a/tests/server/test_openapi_path_parameters.py b/tests/server/openapi/test_openapi_path_parameters.py similarity index 98% rename from tests/server/test_openapi_path_parameters.py rename to tests/server/openapi/test_openapi_path_parameters.py index 9940594a8..517382a2c 100644 --- a/tests/server/test_openapi_path_parameters.py +++ b/tests/server/openapi/test_openapi_path_parameters.py @@ -6,7 +6,7 @@ import pytest from fastapi import FastAPI, Query from fastmcp import Client, FastMCP -from fastmcp.server.openapi import OpenAPITool, RouteMap, RouteType +from fastmcp.server.openapi import MCPType, OpenAPITool, RouteMap from fastmcp.utilities.openapi import HTTPRoute, ParameterInfo @@ -286,9 +286,7 @@ async def test_array_query_param_with_fastapi(): # Create a FastMCP server from the FastAPI app mcp = FastMCP.from_fastapi( app, - route_maps=[ - RouteMap(methods=["GET"], pattern=r".*", route_type=RouteType.TOOL) - ], + route_maps=[RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.TOOL)], ) # Test with the client diff --git a/tests/server/test_route_map_shortcuts.py b/tests/server/test_route_map_shortcuts.py new file mode 100644 index 000000000..92cc5465f --- /dev/null +++ b/tests/server/test_route_map_shortcuts.py @@ -0,0 +1,208 @@ +"""Tests for the route map shortcut functions.""" + +import httpx +import pytest + +from fastmcp.server.openapi import ( + ALL_TOOLS, + EXCLUDE_ALL, + EXCLUDE_PATTERN, + PATTERN_AS_TOOLS, + FastMCPOpenAPI, + MCPType, + RouteMap, +) + + +class TestRouteMapShortcuts: + """Tests for the route map shortcut functions.""" + + def test_functions_return_correct_route_maps(self): + """Test that each shortcut function returns a RouteMap with the expected properties.""" + # Test EXCLUDE_ALL + exclude_all = EXCLUDE_ALL() + assert isinstance(exclude_all, RouteMap) + assert exclude_all.methods == "*" + assert exclude_all.pattern == ".*" + assert exclude_all.mcp_type == MCPType.EXCLUDE + + # Test ALL_TOOLS + all_tools = ALL_TOOLS() + assert isinstance(all_tools, RouteMap) + assert all_tools.methods == "*" + assert all_tools.pattern == ".*" + assert all_tools.mcp_type == MCPType.TOOL + + # Test PATTERN_AS_TOOLS + pattern = r"^/api/.*" + pattern_as_tools = PATTERN_AS_TOOLS(pattern) + assert isinstance(pattern_as_tools, RouteMap) + assert pattern_as_tools.methods == "*" + assert pattern_as_tools.pattern == pattern + assert pattern_as_tools.mcp_type == MCPType.TOOL + + # Test EXCLUDE_PATTERN + pattern = r"^/admin/.*" + exclude_pattern = EXCLUDE_PATTERN(pattern) + assert isinstance(exclude_pattern, RouteMap) + assert exclude_pattern.methods == "*" + assert exclude_pattern.pattern == pattern + assert exclude_pattern.mcp_type == MCPType.EXCLUDE + + def test_backward_compatibility(self): + """Test that backward compatibility with RouteType and route_type works.""" + from fastmcp.server.openapi import RouteType + + # Test creating a RouteMap with route_type + with pytest.warns(DeprecationWarning): + route_map = RouteMap( + methods=["GET"], pattern=r".*", route_type=RouteType.TOOL + ) + assert route_map.mcp_type == MCPType.TOOL + + # Test accessing fields on RouteType directly + # Note: importing RouteType already causes the deprecation warning, + # so we don't need to check for it again here + rt = RouteType.RESOURCE + assert rt.value == "RESOURCE" + assert rt.name == "RESOURCE" + + +class TestRouteMapShortcutsIntegration: + """Integration tests for the route map shortcut functions with FastMCPOpenAPI.""" + + @pytest.fixture + def basic_openapi_spec(self) -> dict: + """Create a simple OpenAPI spec for testing.""" + return { + "openapi": "3.0.0", + "info": {"title": "Test API", "version": "1.0.0"}, + "paths": { + "/items": { + "get": { + "operationId": "get_items", + "summary": "Get all items", + "responses": {"200": {"description": "Success"}}, + }, + "post": { + "operationId": "create_item", + "summary": "Create an item", + "responses": {"201": {"description": "Created"}}, + }, + }, + "/users": { + "get": { + "operationId": "get_users", + "summary": "Get all users", + "responses": {"200": {"description": "Success"}}, + }, + }, + "/admin": { + "get": { + "operationId": "get_admin", + "summary": "Admin endpoint", + "responses": {"200": {"description": "Success"}}, + }, + }, + "/items/{item_id}": { + "get": { + "operationId": "get_item", + "summary": "Get an item by ID", + "parameters": [ + { + "name": "item_id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ], + "responses": {"200": {"description": "Success"}}, + }, + }, + }, + } + + @pytest.fixture + async def mock_client(self) -> httpx.AsyncClient: + """Create a mock client for testing.""" + + async def _responder(request): + return httpx.Response(200, json={"success": True}) + + return httpx.AsyncClient(transport=httpx.MockTransport(_responder)) + + async def test_all_tools(self, basic_openapi_spec, mock_client): + """Test using ALL_TOOLS() to convert all routes to tools.""" + server = FastMCPOpenAPI( + openapi_spec=basic_openapi_spec, + client=mock_client, + route_maps=[ALL_TOOLS()], + ) + + # Check that all routes are tools + tools = await server.get_tools() + resources = await server.get_resources() + templates = await server.get_resource_templates() + + # All 5 routes should be tools + assert len(tools) == 5 + assert len(resources) == 0 + assert len(templates) == 0 + + # Check that all expected tools exist + tool_names = [t.name for t in tools.values()] + assert "get_items" in tool_names + assert "create_item" in tool_names + assert "get_users" in tool_names + assert "get_admin" in tool_names + assert "get_item" in tool_names + + async def test_exclude_pattern(self, basic_openapi_spec, mock_client): + """Test using EXCLUDE_PATTERN() to exclude specific routes.""" + server = FastMCPOpenAPI( + openapi_spec=basic_openapi_spec, + client=mock_client, + route_maps=[ + # Exclude admin endpoints + EXCLUDE_PATTERN(r"^/admin"), + # Make everything else a tool + ALL_TOOLS(), + ], + ) + + # Check that admin route is excluded + tools = await server.get_tools() + tool_names = [t.name for t in tools.values()] + + # All routes except admin should be tools + assert "get_items" in tool_names + assert "create_item" in tool_names + assert "get_users" in tool_names + assert "get_item" in tool_names + assert "get_admin" not in tool_names # This should be excluded + + async def test_pattern_as_tools(self, basic_openapi_spec, mock_client): + """Test using PATTERN_AS_TOOLS() to convert routes matching a pattern to tools.""" + server = FastMCPOpenAPI( + openapi_spec=basic_openapi_spec, + client=mock_client, + route_maps=[ + # Make /items routes tools regardless of method + PATTERN_AS_TOOLS(r"^/items"), + # Make everything else a resource + RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.RESOURCE), + ], + ) + + # Check that /items routes are tools + tools = await server.get_tools() + tool_names = [t.name for t in tools.values()] + assert "get_items" in tool_names + assert "create_item" in tool_names + assert "get_item" in tool_names + + # Check that other routes are resources + resources = await server.get_resources() + resource_names = [r.name for r in resources.values()] + assert "get_users" in resource_names + assert "get_admin" in resource_names