feature: add automatic operation ID shortening for LLM compatibility

examples/01_basic_usage_example.py

@@ -1,4 +1,4 @@
-from examples.shared.apps.items import app # The FastAPI app
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi_mcp import FastApiMCP
@@ -15,4 +15,4 @@
 if __name__ == "__main__":
     import uvicorn
 
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    uvicorn.run(app, host="0.0.0.0", port=8000)

examples/02_full_schema_description_example.py

@@ -1,8 +1,8 @@
-
 """
 This example shows how to describe the full response schema instead of just a response example.
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi_mcp import FastApiMCP
@@ -22,5 +22,5 @@
 
 if __name__ == "__main__":
     import uvicorn
-    
+
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/03_custom_exposed_endpoints_example.py

@@ -6,7 +6,8 @@
 - You can combine operation filtering with tag filtering (e.g., use `include_operations` with `include_tags`)
 - When combining filters, a greedy approach will be taken. Endpoints matching either criteria will be included
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi_mcp import FastApiMCP
@@ -24,7 +25,7 @@
 
 # Filter by excluding specific operation IDs
 exclude_operations_mcp = FastApiMCP(
-    app,    
+    app,
     name="Item API MCP - Excluded Operations",
     exclude_operations=["create_item", "update_item", "delete_item"],
 )

examples/04_separate_server_example.py

@@ -2,6 +2,7 @@
 This example shows how to run the MCP server and the FastAPI app separately.
 You can create an MCP server from one FastAPI app, and mount it to a different app.
 """
+
 from fastapi import FastAPI
 
 from examples.shared.apps.items import app
@@ -30,4 +31,4 @@
 if __name__ == "__main__":
     import uvicorn
 
-    uvicorn.run(mcp_app, host="0.0.0.0", port=8000)
+    uvicorn.run(mcp_app, host="0.0.0.0", port=8000)

examples/05_reregister_tools_example.py

@@ -1,15 +1,16 @@
 """
 This example shows how to re-register tools if you add endpoints after the MCP server was created.
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi_mcp import FastApiMCP
 
 setup_logging()
 
-mcp = FastApiMCP(app) # Add MCP server to the FastAPI app
-mcp.mount() # MCP server
+mcp = FastApiMCP(app)  # Add MCP server to the FastAPI app
+mcp.mount()  # MCP server
 
 
 # This endpoint will not be registered as a tool, since it was added after the MCP instance was created
@@ -24,5 +25,5 @@ async def new_endpoint():
 
 if __name__ == "__main__":
     import uvicorn
-    
+
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/06_custom_mcp_router_example.py

@@ -1,15 +1,16 @@
 """
 This example shows how to mount the MCP server to a specific APIRouter, giving a custom mount path.
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi import APIRouter
 from fastapi_mcp import FastApiMCP
 
 setup_logging()
 
-other_router = APIRouter(prefix="/other/route")    
+other_router = APIRouter(prefix="/other/route")
 app.include_router(other_router)
 
 mcp = FastApiMCP(app)
@@ -21,5 +22,5 @@
 
 if __name__ == "__main__":
     import uvicorn
-    
+
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/07_configure_http_timeout_example.py

@@ -2,7 +2,8 @@
 This example shows how to configure the HTTP client timeout for the MCP server.
 In case you have API endpoints that take longer than 5 seconds to respond, you can increase the timeout.
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 import httpx
@@ -12,14 +13,11 @@
 setup_logging()
 
 
-mcp = FastApiMCP(
-    app,
-    http_client=httpx.AsyncClient(timeout=20)
-)
+mcp = FastApiMCP(app, http_client=httpx.AsyncClient(timeout=20))
 mcp.mount()
 
 
 if __name__ == "__main__":
     import uvicorn
-    
+
     uvicorn.run(app, host="0.0.0.0", port=8000)

examples/08_auth_example_token_passthrough.py

@@ -21,7 +21,8 @@
 }
 ```
 """
-from examples.shared.apps.items import app # The FastAPI app
+
+from examples.shared.apps.items import app  # The FastAPI app
 from examples.shared.setup import setup_logging
 
 from fastapi import Depends
@@ -34,11 +35,13 @@
 # Scheme for the Authorization header
 token_auth_scheme = HTTPBearer()
 
+
 # Create a private endpoint
 @app.get("/private")
-async def private(token = Depends(token_auth_scheme)):
+async def private(token=Depends(token_auth_scheme)):
     return token.credentials
 
+
 # Create the MCP server with the token auth scheme
 mcp = FastApiMCP(
     app,
@@ -54,5 +57,5 @@ async def private(token = Depends(token_auth_scheme)):
 
 if __name__ == "__main__":
     import uvicorn
-    
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+
+    uvicorn.run(app, host="0.0.0.0", port=8000)

fastapi_mcp/openapi/convert.py

@@ -1,6 +1,6 @@
 import json
 import logging
-from typing import Any, Dict, List, Tuple
+from typing import Any, Dict, List, Tuple, Optional
 
 import mcp.types as types
 
@@ -9,6 +9,7 @@
     generate_example_from_schema,
     resolve_schema_references,
     get_single_param_type_from_schema,
+    shorten_operation_id,
 )
 
 logger = logging.getLogger(__name__)
@@ -18,25 +19,32 @@ def convert_openapi_to_mcp_tools(
     openapi_schema: Dict[str, Any],
     describe_all_responses: bool = False,
     describe_full_response_schema: bool = False,
-) -> Tuple[List[types.Tool], Dict[str, Dict[str, Any]]]:
+    max_operation_id_length: Optional[int] = None,
+) -> Tuple[List[types.Tool], Dict[str, Dict[str, Any]], Dict[str, str]]:
     """
     Convert OpenAPI operations to MCP tools.
 
     Args:
         openapi_schema: The OpenAPI schema
         describe_all_responses: Whether to include all possible response schemas in tool descriptions
         describe_full_response_schema: Whether to include full response schema in tool descriptions
+        max_operation_id_length: Maximum length for operation IDs. IDs longer than this will be shortened.
 
     Returns:
         A tuple containing:
         - A list of MCP tools
         - A mapping of operation IDs to operation details for HTTP execution
+        - A mapping of shortened operation IDs to original operation IDs (for debugging)
     """
     # Resolve all references in the schema at once
     resolved_openapi_schema = resolve_schema_references(openapi_schema, openapi_schema)
 
     tools = []
     operation_map = {}
+    operation_id_mappings = {}  # Maps shortened IDs to original IDs
+
+    # Track seen operation IDs for collision detection
+    seen_operation_ids = set()
 
     # Process each path in the OpenAPI schema
     for path, path_item in resolved_openapi_schema.get("paths", {}).items():
@@ -52,12 +60,36 @@ def convert_openapi_to_mcp_tools(
                 logger.warning(f"Skipping operation with no operationId: {operation}")
                 continue
 
+            # Store the original operation ID for mapping
+            original_operation_id = operation_id
+
+            # Apply shortening if max_operation_id_length is set
+            if max_operation_id_length is not None:
+                shortened_id = shorten_operation_id(operation_id, max_operation_id_length)
+                if shortened_id != operation_id:
+                    logger.debug(f"Shortened operation ID: {operation_id} -> {shortened_id}")
+                    # Store the mapping for debugging
+                    operation_id_mappings[shortened_id] = original_operation_id
+                operation_id = shortened_id
+
+            # Check for collisions
+            if operation_id in seen_operation_ids:
+                logger.warning(
+                    f"Collision detected! Operation ID '{operation_id}' already exists. "
+                    f"Original operation ID was '{original_operation_id}'. "
+                    f"Consider using unique operation IDs in your FastAPI app or adjusting max_operation_id_length."
+                )
+            else:
+                seen_operation_ids.add(operation_id)
+
             # Save operation details for later HTTP calls
+            # Use the shortened operation_id as the key, but store the original for reference
             operation_map[operation_id] = {
                 "path": path,
                 "method": method,
                 "parameters": operation.get("parameters", []),
                 "request_body": operation.get("requestBody", {}),
+                "original_operation_id": original_operation_id,
             }
 
             summary = operation.get("summary", "")
@@ -264,4 +296,11 @@ def convert_openapi_to_mcp_tools(
 
             tools.append(tool)
 
-    return tools, operation_map
+    # Log summary of operation ID shortening if it occurred
+    if operation_id_mappings:
+        logger.debug(
+            f"Operation ID shortening summary: {len(operation_id_mappings)} IDs shortened "
+            f"to fit within {max_operation_id_length} character limit"
+        )
+
+    return tools, operation_map, operation_id_mappings

fastapi_mcp/openapi/utils.py

@@ -1,6 +1,117 @@
+import hashlib
 from typing import Any, Dict
 
 
+def shorten_operation_id(operation_id: str, max_length: int) -> str:
+    """
+    Shorten an operation ID to fit within a maximum length while preserving semantic meaning.
+
+    This function implements a collision-resistant shortening algorithm that:
+    1. Preserves operation IDs that are already within the limit
+    2. For longer IDs, extracts components (function name, path segments, HTTP method)
+    3. Generates a 6-character MD5 hash for uniqueness
+    4. Prioritizes the most specific path segments (rightmost)
+    5. Intelligently truncates function names if needed
+
+    Args:
+        operation_id: The original operation ID to potentially shorten
+        max_length: The maximum allowed length for the operation ID
+
+    Returns:
+        The original operation ID if within limit, or a shortened version
+
+    Example:
+        >>> shorten_operation_id("get_user_profile_api_v1_users_profiles_user_id_get", 50)
+        'get_user_profile_users_profiles_user_id_get_e1e2e3'
+    """
+    # Return as-is if already within limit
+    if len(operation_id) <= max_length:
+        return operation_id
+
+    # Generate 6-character hash from the full original operation ID
+    hash_value = hashlib.md5(operation_id.encode()).hexdigest()[:6]
+
+    # Extract components based on FastAPI's default pattern
+    # Pattern: {function_name}_{path_segments}_{http_method}
+    parts = operation_id.split("_")
+
+    # Extract HTTP method (last part)
+    http_method = ""
+    if parts and parts[-1] in ["get", "post", "put", "delete", "patch", "head", "options"]:
+        http_method = parts[-1]
+        parts = parts[:-1]
+
+    # Extract function name (first part, may be multiple words)
+    # Function names often contain underscores, so we need to identify where the path starts
+    # Heuristic: path segments are usually shorter and may include version indicators like 'v1'
+    function_name_parts = []
+    path_segments = []
+
+    # Common path indicators
+    path_indicators = {"api", "v1", "v2", "v3", "admin", "auth", "public", "private"}
+
+    found_path = False
+    for i, part in enumerate(parts):
+        if not found_path and part.lower() in path_indicators:
+            found_path = True
+            path_segments = parts[i:]
+            break
+        elif not found_path:
+            function_name_parts.append(part)
+
+    # If no path indicators found, assume first part is function name, rest is path
+    if not found_path and parts:
+        function_name_parts = [parts[0]] if parts else []
+        path_segments = parts[1:] if len(parts) > 1 else []
+
+    function_name = "_".join(function_name_parts) if function_name_parts else ""
+
+    # Calculate available space for path segments
+    # Format: {function_name}_{truncated_path}_{http_method}_{hash}
+    fixed_parts_length = len(function_name) + len(http_method) + len(hash_value) + 3  # 3 underscores
+
+    # Handle case where fixed parts alone exceed max_length
+    if fixed_parts_length >= max_length:
+        # Truncate function name intelligently
+        available_for_function = max_length - len(http_method) - len(hash_value) - 3
+        if available_for_function > 10:  # Ensure we have reasonable space
+            # Preserve start and end of function name
+            half = (available_for_function - 3) // 2  # -3 for "..."
+            function_name = function_name[:half] + "..." + function_name[-(available_for_function - half - 3) :]
+        else:
+            # Very extreme case, just truncate
+            function_name = function_name[:available_for_function]
+
+        # No room for path segments
+        return f"{function_name}_{http_method}_{hash_value}"
+
+    # Calculate space available for path segments
+    available_for_path = max_length - fixed_parts_length
+
+    # Build path from right to left (most specific segments first)
+    selected_segments: list[str] = []
+    current_length = 0
+
+    for segment in reversed(path_segments):
+        segment_length = len(segment) + (1 if selected_segments else 0)  # +1 for underscore if not first
+        if current_length + segment_length <= available_for_path:
+            selected_segments.append(segment)
+            current_length += segment_length
+        else:
+            break
+
+    # Reverse to get correct order
+    selected_segments.reverse()
+
+    # Build final shortened operation ID
+    path_part = "_".join(selected_segments) if selected_segments else ""
+
+    if path_part:
+        return f"{function_name}_{path_part}_{http_method}_{hash_value}"
+    else:
+        return f"{function_name}_{http_method}_{hash_value}"
+
+
 def get_single_param_type_from_schema(param_schema: Dict[str, Any]) -> str:
     """
     Get the type of a parameter from the schema.