Add openapi parsing utilities

src/fastmcp/utilities/openapi.py

@@ -0,0 +1,468 @@
+import logging
+from typing import Any, Dict, List, Literal, Optional, Tuple, Union, cast
+
+# Using the recommended library: openapi-pydantic
+from openapi_pydantic import (
+    MediaType,
+    OpenAPI,
+    Operation,
+    Parameter,
+    PathItem,
+    Reference,
+    RequestBody,
+    Schema,
+)
+from pydantic import BaseModel, Field, ValidationError
+
+logger = logging.getLogger(__name__)
+
+# --- Intermediate Representation (IR) Definition ---
+# (IR models remain the same)
+
+HttpMethod = Literal[
+    "GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD", "TRACE"
+]
+ParameterLocation = Literal["path", "query", "header", "cookie"]
+JsonSchema = Dict[str, Any]
+
+
+class ParameterInfo(BaseModel):
+    """Represents a single parameter for an HTTP operation in our IR."""
+
+    name: str
+    location: ParameterLocation  # Mapped from 'in' field of openapi-pydantic Parameter
+    required: bool = False
+    schema_: JsonSchema = Field(..., alias="schema")  # Target name in IR
+    description: Optional[str] = None
+
+    # No model_config needed here if we populate manually after accessing 'in'
+
+
+class RequestBodyInfo(BaseModel):
+    """Represents the request body for an HTTP operation in our IR."""
+
+    required: bool = False
+    content_schema: Dict[str, JsonSchema] = Field(
+        default_factory=dict
+    )  # Key: media type
+    description: Optional[str] = None
+
+
+class HTTPRoute(BaseModel):
+    """Intermediate Representation for a single OpenAPI operation."""
+
+    path: str
+    method: HttpMethod
+    operation_id: Optional[str] = None
+    summary: Optional[str] = None
+    description: Optional[str] = None
+    tags: List[str] = Field(default_factory=list)
+    parameters: List[ParameterInfo] = Field(default_factory=list)
+    request_body: Optional[RequestBodyInfo] = None
+
+
+# --- Helper Functions ---
+
+
+def _resolve_ref(
+    item: Union[Reference, Schema, Parameter, RequestBody, Any], openapi: OpenAPI
+) -> Any:
+    """Resolves a potential Reference object to its target definition (no changes needed here)."""
+    if isinstance(item, Reference):
+        ref_str = item.ref
+        try:
+            if not ref_str.startswith("#/"):
+                raise ValueError(
+                    f"External or non-local reference not supported: {ref_str}"
+                )
+            parts = ref_str.strip("#/").split("/")
+            target = openapi
+            for part in parts:
+                if part.isdigit() and isinstance(target, list):
+                    target = target[int(part)]
+                elif isinstance(target, BaseModel):
+                    # Use model_extra for fields not explicitly defined (like components types)
+                    # Check class fields first, then model_extra
+                    if part in target.model_fields:  # Access class attribute here
+                        target = getattr(target, part, None)
+                    elif target.model_extra and part in target.model_extra:
+                        target = target.model_extra[part]
+                    else:
+                        # Special handling for components sub-types common structure
+                        if part == "components" and hasattr(target, "components"):
+                            target = getattr(target, "components")
+                        elif hasattr(target, part):  # Fallback check
+                            target = getattr(target, part, None)
+                        else:
+                            target = None  # Part not found
+                elif isinstance(target, dict):
+                    target = target.get(part)
+                else:
+                    raise ValueError(
+                        f"Cannot traverse part '{part}' in reference '{ref_str}' from type {type(target)}"
+                    )
+                if target is None:
+                    raise ValueError(
+                        f"Reference part '{part}' not found in path '{ref_str}'"
+                    )
+            if isinstance(target, Reference):
+                return _resolve_ref(target, openapi)
+            return target
+        except (AttributeError, KeyError, IndexError, TypeError, ValueError) as e:
+            raise ValueError(f"Failed to resolve reference '{ref_str}': {e}") from e
+    return item
+
+
+def _extract_schema_as_dict(
+    schema_obj: Union[Schema, Reference], openapi: OpenAPI
+) -> JsonSchema:
+    """Resolves a schema/reference and returns it as a dictionary."""
+    resolved_schema = _resolve_ref(schema_obj, openapi)
+    if isinstance(resolved_schema, Schema):
+        # Using exclude_none=True might be better than exclude_unset sometimes
+        return resolved_schema.model_dump(mode="json", by_alias=True, exclude_none=True)
+    elif isinstance(resolved_schema, dict):
+        logger.warning(
+            "Resolved schema reference resulted in a dict, not a Schema model."
+        )
+        return resolved_schema
+    else:
+        ref_str = getattr(schema_obj, "ref", "unknown")
+        logger.warning(
+            f"Expected Schema after resolving ref '{ref_str}', got {type(resolved_schema)}. Returning empty dict."
+        )
+        return {}
+
+
+def _convert_to_parameter_location(param_in: str) -> ParameterLocation:
+    """Convert string parameter location to our ParameterLocation type."""
+    if param_in == "path":
+        return "path"
+    elif param_in == "query":
+        return "query"
+    elif param_in == "header":
+        return "header"
+    elif param_in == "cookie":
+        return "cookie"
+    else:
+        logger.warning(f"Unknown parameter location: {param_in}, defaulting to 'query'")
+        return "query"
+
+
+def _extract_parameters(
+    operation_params: Optional[List[Union[Parameter, Reference]]],
+    path_item_params: Optional[List[Union[Parameter, Reference]]],
+    openapi: OpenAPI,
+) -> List[ParameterInfo]:
+    """Extracts and resolves parameters using corrected attribute names."""
+    extracted_params: List[ParameterInfo] = []
+    seen_params: Dict[
+        Tuple[str, str], bool
+    ] = {}  # Use string keys to avoid type issues
+    all_params_refs = (operation_params or []) + (path_item_params or [])
+
+    for param_or_ref in all_params_refs:
+        try:
+            parameter = cast(Parameter, _resolve_ref(param_or_ref, openapi))
+            if not isinstance(parameter, Parameter):
+                # ... (error logging remains the same)
+                continue
+
+            # --- *** CORRECTED ATTRIBUTE ACCESS HERE *** ---
+            param_in = parameter.param_in  # CORRECTED: Use 'param_in'
+            param_location = _convert_to_parameter_location(param_in)
+            param_schema_obj = parameter.param_schema  # CORRECTED: Use 'param_schema'
+            # --- *** ---
+
+            param_key = (parameter.name, param_in)
+            if param_key in seen_params:
+                continue
+            seen_params[param_key] = True
+
+            param_schema_dict = {}
+            if param_schema_obj:  # Check if schema exists
+                param_schema_dict = _extract_schema_as_dict(param_schema_obj, openapi)
+            elif parameter.content:
+                # Handle complex parameters with 'content'
+                first_media_type = next(iter(parameter.content.values()), None)
+                if (
+                    first_media_type and first_media_type.media_type_schema
+                ):  # CORRECTED: Use 'media_type_schema'
+                    param_schema_dict = _extract_schema_as_dict(
+                        first_media_type.media_type_schema, openapi
+                    )
+                    logger.debug(
+                        f"Parameter '{parameter.name}' using schema from 'content' field."
+                    )
+
+            # Manually create ParameterInfo instance using correct field names
+            param_info = ParameterInfo(
+                name=parameter.name,
+                location=param_location,  # Use converted parameter location
+                required=parameter.required,
+                schema=param_schema_dict,  # Populate 'schema' field in IR
+                description=parameter.description,
+            )
+            extracted_params.append(param_info)
+
+        except (
+            ValidationError,
+            ValueError,
+            AttributeError,
+            TypeError,
+        ) as e:  # Added TypeError
+            param_name = getattr(
+                param_or_ref, "name", getattr(param_or_ref, "ref", "unknown")
+            )
+            logger.error(
+                f"Failed to extract parameter '{param_name}': {e}", exc_info=False
+            )
+
+    return extracted_params
+
+
+def _extract_request_body(
+    request_body_or_ref: Optional[Union[RequestBody, Reference]], openapi: OpenAPI
+) -> Optional[RequestBodyInfo]:
+    """Extracts and resolves the request body using corrected attribute names."""
+    if not request_body_or_ref:
+        return None
+    try:
+        request_body = cast(RequestBody, _resolve_ref(request_body_or_ref, openapi))
+        if not isinstance(request_body, RequestBody):
+            # ... (error logging remains the same)
+            return None
+
+        content_schemas: Dict[str, JsonSchema] = {}
+        if request_body.content:
+            for media_type_str, media_type_obj in request_body.content.items():
+                # --- *** CORRECTED ATTRIBUTE ACCESS HERE *** ---
+                if (
+                    isinstance(media_type_obj, MediaType)
+                    and media_type_obj.media_type_schema
+                ):  # CORRECTED: Use 'media_type_schema'
+                    # --- *** ---
+                    try:
+                        # Use the corrected attribute here as well
+                        schema_dict = _extract_schema_as_dict(
+                            media_type_obj.media_type_schema, openapi
+                        )
+                        content_schemas[media_type_str] = schema_dict
+                    except ValueError as schema_err:
+                        logger.error(
+                            f"Failed to extract schema for media type '{media_type_str}' in request body: {schema_err}"
+                        )
+                elif not isinstance(media_type_obj, MediaType):
+                    logger.warning(
+                        f"Skipping invalid media type object for '{media_type_str}' (type: {type(media_type_obj)}) in request body."
+                    )
+                elif not media_type_obj.media_type_schema:  # Corrected check
+                    logger.warning(
+                        f"Skipping media type '{media_type_str}' in request body because it lacks a schema."
+                    )
+
+        return RequestBodyInfo(
+            required=request_body.required,
+            content_schema=content_schemas,
+            description=request_body.description,
+        )
+    except (ValidationError, ValueError, AttributeError) as e:
+        ref_name = getattr(request_body_or_ref, "ref", "unknown")
+        logger.error(
+            f"Failed to extract request body '{ref_name}': {e}", exc_info=False
+        )
+        return None
+
+
+# --- Main Parsing Function ---
+# (No changes needed in the main loop logic, only in the helpers it calls)
+def parse_openapi_to_http_routes(openapi_dict: Dict[str, Any]) -> List[HTTPRoute]:
+    """
+    Parses an OpenAPI schema dictionary into a list of HTTPRoute objects
+    using the openapi-pydantic library.
+    """
+    routes: List[HTTPRoute] = []
+    try:
+        openapi: OpenAPI = OpenAPI.model_validate(openapi_dict)
+        logger.info(f"Successfully parsed OpenAPI schema version: {openapi.openapi}")
+    except ValidationError as e:
+        logger.error(f"OpenAPI schema validation failed: {e}")
+        error_details = e.errors()
+        logger.error(f"Validation errors: {error_details}")
+        raise ValueError(f"Invalid OpenAPI schema: {error_details}") from e
+
+    if not openapi.paths:
+        logger.warning("OpenAPI schema has no paths defined.")
+        return []
+
+    for path_str, path_item_obj in openapi.paths.items():
+        if not isinstance(path_item_obj, PathItem):
+            logger.warning(
+                f"Skipping invalid path item object for path '{path_str}' (type: {type(path_item_obj)})"
+            )
+            continue
+
+        path_level_params = path_item_obj.parameters
+
+        # Iterate through possible HTTP methods defined in the PathItem model fields
+        # Use model_fields from the class, not the instance
+        for method_lower in PathItem.model_fields.keys():
+            if method_lower not in [
+                "get",
+                "put",
+                "post",
+                "delete",
+                "options",
+                "head",
+                "patch",
+                "trace",
+            ]:
+                continue
+
+            operation: Optional[Operation] = getattr(path_item_obj, method_lower, None)
+
+            if operation and isinstance(operation, Operation):
+                method_upper = cast(HttpMethod, method_lower.upper())
+                logger.debug(f"Processing operation: {method_upper} {path_str}")
+                try:
+                    parameters = _extract_parameters(
+                        operation.parameters, path_level_params, openapi
+                    )
+                    request_body_info = _extract_request_body(
+                        operation.requestBody, openapi
+                    )
+
+                    route = HTTPRoute(
+                        path=path_str,
+                        method=method_upper,
+                        operation_id=operation.operationId,
+                        summary=operation.summary,
+                        description=operation.description,
+                        tags=operation.tags or [],
+                        parameters=parameters,
+                        request_body=request_body_info,
+                    )
+                    routes.append(route)
+                    logger.info(
+                        f"Successfully extracted route: {method_upper} {path_str}"
+                    )
+                except Exception as op_error:
+                    op_id = operation.operationId or "unknown"
+                    logger.error(
+                        f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}",
+                        exc_info=True,
+                    )
+
+    logger.info(f"Finished parsing. Extracted {len(routes)} HTTP routes.")
+    return routes
+
+
+# --- Example Usage (Optional) ---
+if __name__ == "__main__":
+    import json
+
+    logging.basicConfig(
+        level=logging.INFO, format="%(levelname)s:%(name)s:%(message)s"
+    )  # Set to INFO
+
+    petstore_schema = {
+        "openapi": "3.1.0",  # Keep corrected version
+        "info": {"title": "Simple Pet Store API", "version": "1.0.0"},
+        "paths": {
+            "/pets": {
+                "get": {
+                    "summary": "List all pets",
+                    "operationId": "listPets",
+                    "tags": ["pets"],
+                    "parameters": [
+                        {
+                            "name": "limit",
+                            "in": "query",
+                            "description": "How many items to return",
+                            "required": False,
+                            "schema": {"type": "integer", "format": "int32"},
+                        }
+                    ],
+                    "responses": {"200": {"description": "A paged array of pets"}},
+                },
+                "post": {
+                    "summary": "Create a pet",
+                    "operationId": "createPet",
+                    "tags": ["pets"],
+                    "requestBody": {"$ref": "#/components/requestBodies/PetBody"},
+                    "responses": {"201": {"description": "Null response"}},
+                },
+            },
+            "/pets/{petId}": {
+                "get": {
+                    "summary": "Info for a specific pet",
+                    "operationId": "showPetById",
+                    "tags": ["pets"],
+                    "parameters": [
+                        {
+                            "name": "petId",
+                            "in": "path",
+                            "required": True,
+                            "description": "The id of the pet",
+                            "schema": {"type": "string"},
+                        },
+                        {
+                            "name": "X-Request-ID",
+                            "in": "header",
+                            "required": False,
+                            "schema": {"type": "string", "format": "uuid"},
+                        },
+                    ],
+                    "responses": {"200": {"description": "Information about the pet"}},
+                },
+                "parameters": [  # Path level parameter example
+                    {
+                        "name": "traceId",
+                        "in": "header",
+                        "description": "Common trace ID",
+                        "required": False,
+                        "schema": {"type": "string"},
+                    }
+                ],
+            },
+        },
+        "components": {
+            "schemas": {
+                "Pet": {
+                    "type": "object",
+                    "required": ["id", "name"],
+                    "properties": {
+                        "id": {"type": "integer", "format": "int64"},
+                        "name": {"type": "string"},
+                        "tag": {"type": "string"},
+                    },
+                }
+            },
+            "requestBodies": {
+                "PetBody": {
+                    "description": "Pet object",
+                    "required": True,
+                    "content": {
+                        "application/json": {
+                            "schema": {"$ref": "#/components/schemas/Pet"}
+                        }
+                    },
+                }
+            },
+        },
+    }
+
+    print("--- Parsing Pet Store Schema using openapi-pydantic (Corrected) ---")
+    try:
+        http_routes = parse_openapi_to_http_routes(petstore_schema)
+        print(f"\n--- Extracted {len(http_routes)} Routes ---")
+        for i, route in enumerate(http_routes):
+            print(f"\nRoute {i + 1}:")
+            # Use model_dump for clean JSON-like output, show aliases from IR model
+            print(
+                json.dumps(route.model_dump(by_alias=True, exclude_none=True), indent=2)
+            )  # exclude_none is often cleaner
+    except ValueError as e:
+        print(f"\nError parsing schema: {e}")
+    except Exception as e:
+        print(f"\nAn unexpected error occurred: {e}")

tests/utilities/__init__.py

@@ -0,0 +1 @@
+"""Tests for utilities in the fastmcp package."""

tests/utilities/openapi/__init__.py

@@ -0,0 +1 @@
+"""Tests for the OpenAPI utilities."""

tests/utilities/openapi/conftest.py

@@ -0,0 +1 @@
+

tests/utilities/openapi/test_openapi.py

@@ -0,0 +1,709 @@
+"""Tests for the OpenAPI parsing utilities."""
+
+from typing import Any, Dict
+
+import pytest
+from fastapi import Body, FastAPI, Path, Query
+from pydantic import BaseModel, Field
+
+from fastmcp.utilities.openapi import parse_openapi_to_http_routes
+
+# --- Test Data: Static OpenAPI Schema Dictionaries --- #
+
+
+@pytest.fixture
+def petstore_schema() -> Dict[str, Any]:
+    """Fixture that returns a simple Pet Store API schema."""
+    return {
+        "openapi": "3.1.0",
+        "info": {"title": "Simple Pet Store API", "version": "1.0.0"},
+        "paths": {
+            "/pets": {
+                "get": {
+                    "summary": "List all pets",
+                    "operationId": "listPets",
+                    "tags": ["pets"],
+                    "parameters": [
+                        {
+                            "name": "limit",
+                            "in": "query",
+                            "description": "How many items to return",
+                            "required": False,
+                            "schema": {"type": "integer", "format": "int32"},
+                        }
+                    ],
+                    "responses": {"200": {"description": "A paged array of pets"}},
+                },
+                "post": {
+                    "summary": "Create a pet",
+                    "operationId": "createPet",
+                    "tags": ["pets"],
+                    "requestBody": {"$ref": "#/components/requestBodies/PetBody"},
+                    "responses": {"201": {"description": "Null response"}},
+                },
+            },
+            "/pets/{petId}": {
+                "get": {
+                    "summary": "Info for a specific pet",
+                    "operationId": "showPetById",
+                    "tags": ["pets"],
+                    "parameters": [
+                        {
+                            "name": "petId",
+                            "in": "path",
+                            "required": True,
+                            "description": "The id of the pet",
+                            "schema": {"type": "string"},
+                        },
+                        {
+                            "name": "X-Request-ID",
+                            "in": "header",
+                            "required": False,
+                            "schema": {"type": "string", "format": "uuid"},
+                        },
+                    ],
+                    "responses": {"200": {"description": "Information about the pet"}},
+                },
+                "parameters": [  # Path level parameter example
+                    {
+                        "name": "traceId",
+                        "in": "header",
+                        "description": "Common trace ID",
+                        "required": False,
+                        "schema": {"type": "string"},
+                    }
+                ],
+            },
+        },
+        "components": {
+            "schemas": {
+                "Pet": {
+                    "type": "object",
+                    "required": ["id", "name"],
+                    "properties": {
+                        "id": {"type": "integer", "format": "int64"},
+                        "name": {"type": "string"},
+                        "tag": {"type": "string"},
+                    },
+                }
+            },
+            "requestBodies": {
+                "PetBody": {
+                    "description": "Pet object",
+                    "required": True,
+                    "content": {
+                        "application/json": {
+                            "schema": {"$ref": "#/components/schemas/Pet"}
+                        }
+                    },
+                }
+            },
+        },
+    }
+
+
+@pytest.fixture
+def parsed_petstore_routes(petstore_schema):
+    """Return parsed routes from the PetStore schema."""
+    return parse_openapi_to_http_routes(petstore_schema)
+
+
+@pytest.fixture
+def bookstore_schema() -> Dict[str, Any]:
+    """Fixture that returns a Book Store API schema with different parameter types."""
+    return {
+        "openapi": "3.1.0",
+        "info": {"title": "Book Store API", "version": "1.0.0"},
+        "paths": {
+            "/books": {
+                "get": {
+                    "summary": "List all books",
+                    "operationId": "listBooks",
+                    "tags": ["books"],
+                    "parameters": [
+                        {
+                            "name": "genre",
+                            "in": "query",
+                            "description": "Filter by genre",
+                            "required": False,
+                            "schema": {"type": "string"},
+                        },
+                        {
+                            "name": "published_after",
+                            "in": "query",
+                            "description": "Filter by publication date",
+                            "required": False,
+                            "schema": {"type": "string", "format": "date"},
+                        },
+                        {
+                            "name": "limit",
+                            "in": "query",
+                            "description": "Maximum number of results",
+                            "required": False,
+                            "schema": {"type": "integer", "default": 10},
+                        },
+                    ],
+                    "responses": {"200": {"description": "A list of books"}},
+                },
+                "post": {
+                    "summary": "Create a new book",
+                    "operationId": "createBook",
+                    "tags": ["books"],
+                    "requestBody": {
+                        "required": True,
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "type": "object",
+                                    "required": ["title", "author"],
+                                    "properties": {
+                                        "title": {"type": "string"},
+                                        "author": {"type": "string"},
+                                        "isbn": {"type": "string"},
+                                        "published": {
+                                            "type": "string",
+                                            "format": "date",
+                                        },
+                                        "genre": {"type": "string"},
+                                    },
+                                }
+                            }
+                        },
+                    },
+                    "responses": {"201": {"description": "Book created"}},
+                },
+            },
+            "/books/{isbn}": {
+                "get": {
+                    "summary": "Get book by ISBN",
+                    "operationId": "getBook",
+                    "tags": ["books"],
+                    "parameters": [
+                        {
+                            "name": "isbn",
+                            "in": "path",
+                            "required": True,
+                            "description": "ISBN of the book",
+                            "schema": {"type": "string"},
+                        }
+                    ],
+                    "responses": {"200": {"description": "Book details"}},
+                },
+                "delete": {
+                    "summary": "Delete a book",
+                    "operationId": "deleteBook",
+                    "tags": ["books"],
+                    "parameters": [
+                        {
+                            "name": "isbn",
+                            "in": "path",
+                            "required": True,
+                            "description": "ISBN of the book to delete",
+                            "schema": {"type": "string"},
+                        }
+                    ],
+                    "responses": {"204": {"description": "Book deleted"}},
+                },
+            },
+        },
+    }
+
+
+@pytest.fixture
+def parsed_bookstore_routes(bookstore_schema):
+    """Return parsed routes from the BookStore schema."""
+    return parse_openapi_to_http_routes(bookstore_schema)
+
+
+# --- FastAPI App Fixtures --- #
+
+
+class Item(BaseModel):
+    """Example pydantic model for API testing."""
+
+    name: str
+    description: str | None = None
+    price: float
+    tax: float | None = None
+    tags: list[str] = Field(default_factory=list)
+
+
+@pytest.fixture
+def fastapi_app() -> FastAPI:
+    """Fixture that returns a FastAPI app with various types of endpoints."""
+    app = FastAPI(title="Test API", version="1.0.0")
+
+    @app.get("/items/", operation_id="list_items")
+    async def list_items(skip: int = 0, limit: int = 10):
+        """List all items with pagination."""
+        return [
+            {"name": f"Item {i}", "price": float(i)} for i in range(skip, skip + limit)
+        ]
+
+    @app.post("/items/", operation_id="create_item")
+    async def create_item(item: Item):
+        """Create a new item."""
+        return item
+
+    @app.get("/items/{item_id}", operation_id="get_item")
+    async def get_item(
+        item_id: int = Path(..., description="The ID of the item to get"),
+        q: str | None = Query(None, description="Optional query string"),
+    ):
+        """Get an item by ID."""
+        return {"item_id": item_id, "q": q}
+
+    @app.put("/items/{item_id}", operation_id="update_item")
+    async def update_item(
+        item_id: int = Path(..., description="The ID of the item to update"),
+        item: Item = Body(..., description="The updated item data"),
+    ):
+        """Update an existing item."""
+        return {"item_id": item_id, **item.model_dump()}
+
+    @app.delete("/items/{item_id}", operation_id="delete_item")
+    async def delete_item(
+        item_id: int = Path(..., description="The ID of the item to delete"),
+    ):
+        """Delete an item by ID."""
+        return {"item_id": item_id, "deleted": True}
+
+    @app.get("/items/{item_id}/tags/{tag_id}", operation_id="get_item_tag")
+    async def get_item_tag(
+        item_id: int = Path(..., description="The ID of the item"),
+        tag_id: str = Path(..., description="The ID of the tag"),
+    ):
+        """Get a specific tag for an item."""
+        return {"item_id": item_id, "tag_id": tag_id}
+
+    @app.post("/upload/", operation_id="upload_file")
+    async def upload_file(
+        file_name: str = Query(..., description="Name of the file to upload"),
+        content_type: str = Query(..., description="Content type of the file"),
+    ):
+        """Upload a file (dummy endpoint for testing query params with POST)."""
+        return {
+            "file_name": file_name,
+            "content_type": content_type,
+            "status": "uploaded",
+        }
+
+    return app
+
+
+@pytest.fixture
+def fastapi_openapi_schema(fastapi_app) -> Dict[str, Any]:
+    """Fixture that returns the OpenAPI schema of the FastAPI app."""
+    return fastapi_app.openapi()
+
+
+@pytest.fixture
+def parsed_fastapi_routes(fastapi_openapi_schema):
+    """Return parsed routes from a FastAPI OpenAPI schema."""
+    return parse_openapi_to_http_routes(fastapi_openapi_schema)
+
+
+@pytest.fixture
+def fastapi_route_map(parsed_fastapi_routes):
+    """Return a dictionary of routes by operation ID."""
+    return {
+        r.operation_id: r for r in parsed_fastapi_routes if r.operation_id is not None
+    }
+
+
+# --- Tests for PetStore schema --- #
+
+
+def test_petstore_route_count(parsed_petstore_routes):
+    """Test that parsing the PetStore schema correctly identifies the number of routes."""
+    assert len(parsed_petstore_routes) == 3
+
+
+def test_petstore_get_pets_operation_id(parsed_petstore_routes):
+    """Test that GET /pets operation_id is correctly parsed."""
+    get_pets = next(
+        (r for r in parsed_petstore_routes if r.method == "GET" and r.path == "/pets"),
+        None,
+    )
+    assert get_pets is not None
+    assert get_pets.operation_id == "listPets"
+
+
+def test_petstore_query_parameter(parsed_petstore_routes):
+    """Test that query parameter 'limit' is correctly parsed from the schema."""
+    get_pets = next(
+        (r for r in parsed_petstore_routes if r.method == "GET" and r.path == "/pets"),
+        None,
+    )
+
+    assert get_pets is not None
+    assert len(get_pets.parameters) == 1
+    param = get_pets.parameters[0]
+    assert param.name == "limit"
+    assert param.location == "query"
+    assert param.required is False
+    assert param.schema_.get("type") == "integer"
+    assert param.schema_.get("format") == "int32"
+
+
+def test_petstore_path_parameter(parsed_petstore_routes):
+    """Test that path parameter 'petId' is correctly parsed from the schema."""
+    get_pet = next(
+        (
+            r
+            for r in parsed_petstore_routes
+            if r.method == "GET" and r.path == "/pets/{petId}"
+        ),
+        None,
+    )
+
+    assert get_pet is not None
+    path_param = next((p for p in get_pet.parameters if p.name == "petId"), None)
+    assert path_param is not None
+    assert path_param.location == "path"
+    assert path_param.required is True
+    assert path_param.schema_.get("type") == "string"
+
+
+def test_petstore_header_parameters(parsed_petstore_routes):
+    """Test that header parameters are correctly parsed from the schema."""
+    get_pet = next(
+        (
+            r
+            for r in parsed_petstore_routes
+            if r.method == "GET" and r.path == "/pets/{petId}"
+        ),
+        None,
+    )
+
+    assert get_pet is not None
+    header_params = [p for p in get_pet.parameters if p.location == "header"]
+    assert len(header_params) == 2
+
+
+def test_petstore_header_parameter_names(parsed_petstore_routes):
+    """Test that header parameter names are correctly parsed."""
+    get_pet = next(
+        (
+            r
+            for r in parsed_petstore_routes
+            if r.method == "GET" and r.path == "/pets/{petId}"
+        ),
+        None,
+    )
+
+    assert get_pet is not None
+    header_params = [p for p in get_pet.parameters if p.location == "header"]
+    header_names = [p.name for p in header_params]
+    assert "X-Request-ID" in header_names
+    assert "traceId" in header_names
+
+
+def test_petstore_path_level_parameters(parsed_petstore_routes):
+    """Test that path-level parameters are correctly merged into the operation."""
+    get_pet = next(
+        (
+            r
+            for r in parsed_petstore_routes
+            if r.method == "GET" and r.path == "/pets/{petId}"
+        ),
+        None,
+    )
+
+    assert get_pet is not None
+    trace_param = next((p for p in get_pet.parameters if p.name == "traceId"), None)
+    assert trace_param is not None
+    assert trace_param.location == "header"
+    assert trace_param.required is False
+
+
+def test_petstore_request_body_reference_resolution(parsed_petstore_routes):
+    """Test that request body references are correctly resolved."""
+    create_pet = next(
+        (r for r in parsed_petstore_routes if r.method == "POST" and r.path == "/pets"),
+        None,
+    )
+
+    assert create_pet is not None
+    assert create_pet.request_body is not None
+    assert create_pet.request_body.required is True
+    assert "application/json" in create_pet.request_body.content_schema
+
+
+def test_petstore_schema_reference_resolution(parsed_petstore_routes):
+    """Test that schema references in request bodies are correctly resolved."""
+    create_pet = next(
+        (r for r in parsed_petstore_routes if r.method == "POST" and r.path == "/pets"),
+        None,
+    )
+
+    assert create_pet is not None
+    assert create_pet.request_body is not None
+    json_schema = create_pet.request_body.content_schema["application/json"]
+    properties = json_schema.get("properties", {})
+
+    assert "id" in properties
+    assert "name" in properties
+    assert "tag" in properties
+
+
+def test_petstore_required_fields_resolution(parsed_petstore_routes):
+    """Test that required fields are correctly resolved from referenced schemas."""
+    create_pet = next(
+        (r for r in parsed_petstore_routes if r.method == "POST" and r.path == "/pets"),
+        None,
+    )
+
+    assert create_pet is not None
+    assert create_pet.request_body is not None
+    json_schema = create_pet.request_body.content_schema["application/json"]
+    assert json_schema.get("required") == ["id", "name"]
+
+
+# --- Tests for BookStore schema --- #
+
+
+def test_bookstore_route_count(parsed_bookstore_routes):
+    """Test that parsing the BookStore schema correctly identifies the number of routes."""
+    assert len(parsed_bookstore_routes) == 4
+
+
+def test_bookstore_query_parameter_count(parsed_bookstore_routes):
+    """Test that the correct number of query parameters are parsed."""
+    list_books = next(
+        (r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None
+    )
+
+    assert list_books is not None
+    assert len(list_books.parameters) == 3
+
+
+def test_bookstore_query_parameter_names(parsed_bookstore_routes):
+    """Test that query parameter names are correctly parsed."""
+    list_books = next(
+        (r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None
+    )
+
+    assert list_books is not None
+    param_map = {p.name: p for p in list_books.parameters}
+    assert "genre" in param_map
+    assert "published_after" in param_map
+    assert "limit" in param_map
+
+
+def test_bookstore_query_parameter_formats(parsed_bookstore_routes):
+    """Test that query parameter formats are correctly parsed."""
+    list_books = next(
+        (r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None
+    )
+
+    assert list_books is not None
+    param_map = {p.name: p for p in list_books.parameters}
+    assert param_map["published_after"].schema_.get("format") == "date"
+
+
+def test_bookstore_query_parameter_defaults(parsed_bookstore_routes):
+    """Test that query parameter default values are correctly parsed."""
+    list_books = next(
+        (r for r in parsed_bookstore_routes if r.operation_id == "listBooks"), None
+    )
+
+    assert list_books is not None
+    param_map = {p.name: p for p in list_books.parameters}
+    assert param_map["limit"].schema_.get("default") == 10
+
+
+def test_bookstore_inline_request_body_presence(parsed_bookstore_routes):
+    """Test that request bodies with inline schemas are present."""
+    create_book = next(
+        (r for r in parsed_bookstore_routes if r.operation_id == "createBook"), None
+    )
+
+    assert create_book is not None
+    assert create_book.request_body is not None
+    assert create_book.request_body.required is True
+    assert "application/json" in create_book.request_body.content_schema
+
+
+def test_bookstore_inline_request_body_properties(parsed_bookstore_routes):
+    """Test that request body properties are correctly parsed from inline schemas."""
+    create_book = next(
+        (r for r in parsed_bookstore_routes if r.operation_id == "createBook"), None
+    )
+
+    assert create_book is not None
+    assert create_book.request_body is not None
+
+    json_schema = create_book.request_body.content_schema["application/json"]
+    properties = json_schema.get("properties", {})
+
+    assert "title" in properties
+    assert "author" in properties
+    assert "isbn" in properties
+    assert "published" in properties
+    assert "genre" in properties
+
+
+def test_bookstore_inline_request_body_required_fields(parsed_bookstore_routes):
+    """Test that required fields in inline schema are correctly parsed."""
+    create_book = next(
+        (r for r in parsed_bookstore_routes if r.operation_id == "createBook"), None
+    )
+
+    assert create_book is not None
+    assert create_book.request_body is not None
+
+    json_schema = create_book.request_body.content_schema["application/json"]
+    assert json_schema.get("required") == ["title", "author"]
+
+
+def test_bookstore_delete_method(parsed_bookstore_routes):
+    """Test that DELETE method is correctly parsed from the schema."""
+    delete_book = next(
+        (r for r in parsed_bookstore_routes if r.method == "DELETE"), None
+    )
+
+    assert delete_book is not None
+    assert delete_book.operation_id == "deleteBook"
+    assert delete_book.path == "/books/{isbn}"
+
+
+def test_bookstore_delete_method_parameters(parsed_bookstore_routes):
+    """Test that parameters for DELETE method are correctly parsed."""
+    delete_book = next(
+        (r for r in parsed_bookstore_routes if r.method == "DELETE"), None
+    )
+
+    assert delete_book is not None
+    assert len(delete_book.parameters) == 1
+    assert delete_book.parameters[0].name == "isbn"
+
+
+# --- Tests for FastAPI Generated Schema --- #
+
+
+def test_fastapi_route_count(parsed_fastapi_routes):
+    """Test that parsing a FastAPI-generated schema correctly identifies the number of routes."""
+    assert len(parsed_fastapi_routes) == 7
+
+
+def test_fastapi_parameter_default_values(fastapi_route_map):
+    """Test that default parameter values are correctly parsed from the schema."""
+    list_items = fastapi_route_map["list_items"]
+
+    param_map = {p.name: p for p in list_items.parameters}
+    assert "skip" in param_map
+    assert "limit" in param_map
+
+
+def test_fastapi_skip_parameter_default(fastapi_route_map):
+    """Test that skip parameter default value is correctly parsed."""
+    list_items = fastapi_route_map["list_items"]
+
+    param_map = {p.name: p for p in list_items.parameters}
+    assert param_map["skip"].schema_.get("default") == 0
+
+
+def test_fastapi_limit_parameter_default(fastapi_route_map):
+    """Test that limit parameter default value is correctly parsed."""
+    list_items = fastapi_route_map["list_items"]
+
+    param_map = {p.name: p for p in list_items.parameters}
+    assert param_map["limit"].schema_.get("default") == 10
+
+
+def test_fastapi_request_body_from_pydantic(fastapi_route_map):
+    """Test that request bodies from Pydantic models are present."""
+    create_item = fastapi_route_map["create_item"]
+
+    assert create_item.request_body is not None
+    assert "application/json" in create_item.request_body.content_schema
+
+
+def test_fastapi_request_body_properties(fastapi_route_map):
+    """Test that request body properties from Pydantic models are correctly parsed."""
+    create_item = fastapi_route_map["create_item"]
+
+    json_schema = create_item.request_body.content_schema["application/json"]
+    properties = json_schema.get("properties", {})
+
+    assert "name" in properties
+    assert "description" in properties
+    assert "price" in properties
+    assert "tax" in properties
+    assert "tags" in properties
+
+
+def test_fastapi_request_body_required_fields(fastapi_route_map):
+    """Test that required fields from Pydantic models are correctly parsed."""
+    create_item = fastapi_route_map["create_item"]
+
+    json_schema = create_item.request_body.content_schema["application/json"]
+    required = json_schema.get("required", [])
+
+    assert "name" in required
+    assert "price" in required
+
+
+def test_fastapi_path_parameter_presence(fastapi_route_map):
+    """Test that path parameters are present in FastAPI schema."""
+    get_item = fastapi_route_map["get_item"]
+
+    path_params = [p for p in get_item.parameters if p.location == "path"]
+    assert len(path_params) == 1
+
+
+def test_fastapi_path_parameter_properties(fastapi_route_map):
+    """Test that path parameters properties are correctly parsed."""
+    get_item = fastapi_route_map["get_item"]
+
+    path_params = [p for p in get_item.parameters if p.location == "path"]
+    assert path_params[0].name == "item_id"
+    assert path_params[0].required is True
+
+
+def test_fastapi_optional_query_parameter(fastapi_route_map):
+    """Test that optional query parameters are correctly parsed."""
+    get_item = fastapi_route_map["get_item"]
+
+    query_params = [p for p in get_item.parameters if p.location == "query"]
+    assert len(query_params) == 1
+    assert query_params[0].name == "q"
+    assert query_params[0].required is False
+
+
+def test_fastapi_multiple_path_parameter_count(fastapi_route_map):
+    """Test that multiple path parameters count is correct."""
+    get_item_tag = fastapi_route_map["get_item_tag"]
+
+    path_params = [p for p in get_item_tag.parameters if p.location == "path"]
+    assert len(path_params) == 2
+
+
+def test_fastapi_multiple_path_parameter_names(fastapi_route_map):
+    """Test that multiple path parameter names are correctly parsed."""
+    get_item_tag = fastapi_route_map["get_item_tag"]
+
+    path_params = [p for p in get_item_tag.parameters if p.location == "path"]
+    param_names = [p.name for p in path_params]
+    assert "item_id" in param_names
+    assert "tag_id" in param_names
+
+
+def test_fastapi_post_with_query_parameters(fastapi_route_map):
+    """Test that query parameters for POST methods are correctly parsed."""
+    upload_file = fastapi_route_map["upload_file"]
+
+    assert upload_file.method == "POST"
+    query_params = [p for p in upload_file.parameters if p.location == "query"]
+    assert len(query_params) == 2
+
+
+def test_fastapi_post_query_parameter_names(fastapi_route_map):
+    """Test that query parameter names for POST methods are correctly parsed."""
+    upload_file = fastapi_route_map["upload_file"]
+
+    query_params = [p for p in upload_file.parameters if p.location == "query"]
+    param_names = [p.name for p in query_params]
+    assert "file_name" in param_names
+    assert "content_type" in param_names

tests/utilities/openapi/test_openapi_advanced.py

@@ -0,0 +1,594 @@
+"""Tests for advanced features of the OpenAPI utilities."""
+
+from typing import Any, Dict
+
+import pytest
+
+from fastmcp.utilities.openapi import parse_openapi_to_http_routes
+
+
+@pytest.fixture
+def complex_schema() -> Dict[str, Any]:
+    """Fixture that returns a complex OpenAPI schema with nested references."""
+    return {
+        "openapi": "3.1.0",
+        "info": {"title": "Complex API", "version": "1.0.0"},
+        "paths": {
+            "/users": {
+                "get": {
+                    "summary": "List all users",
+                    "operationId": "listUsers",
+                    "parameters": [
+                        {"$ref": "#/components/parameters/PageLimit"},
+                        {"$ref": "#/components/parameters/PageOffset"},
+                    ],
+                    "responses": {"200": {"description": "A list of users"}},
+                }
+            },
+            "/users/{userId}": {
+                "get": {
+                    "summary": "Get user by ID",
+                    "operationId": "getUser",
+                    "parameters": [
+                        {"$ref": "#/components/parameters/UserId"},
+                        {"$ref": "#/components/parameters/IncludeInactive"},
+                    ],
+                    "responses": {"200": {"description": "User details"}},
+                }
+            },
+            "/users/{userId}/orders": {
+                "post": {
+                    "summary": "Create order for user",
+                    "operationId": "createOrder",
+                    "parameters": [{"$ref": "#/components/parameters/UserId"}],
+                    "requestBody": {"$ref": "#/components/requestBodies/OrderRequest"},
+                    "responses": {"201": {"description": "Order created"}},
+                }
+            },
+        },
+        "components": {
+            "parameters": {
+                "UserId": {
+                    "name": "userId",
+                    "in": "path",
+                    "required": True,
+                    "schema": {"type": "string", "format": "uuid"},
+                },
+                "PageLimit": {
+                    "name": "limit",
+                    "in": "query",
+                    "schema": {"type": "integer", "default": 20, "maximum": 100},
+                },
+                "PageOffset": {
+                    "name": "offset",
+                    "in": "query",
+                    "schema": {"type": "integer", "default": 0},
+                },
+                "IncludeInactive": {
+                    "name": "include_inactive",
+                    "in": "query",
+                    "schema": {"type": "boolean", "default": False},
+                },
+            },
+            "schemas": {
+                "User": {
+                    "type": "object",
+                    "properties": {
+                        "id": {"type": "string", "format": "uuid"},
+                        "name": {"type": "string"},
+                        "email": {"type": "string", "format": "email"},
+                        "role": {"$ref": "#/components/schemas/Role"},
+                        "address": {"$ref": "#/components/schemas/Address"},
+                    },
+                },
+                "Role": {
+                    "type": "string",
+                    "enum": ["admin", "user", "guest"],
+                },
+                "Address": {
+                    "type": "object",
+                    "properties": {
+                        "street": {"type": "string"},
+                        "city": {"type": "string"},
+                        "zip": {"type": "string"},
+                        "country": {"type": "string"},
+                    },
+                },
+                "Order": {
+                    "type": "object",
+                    "properties": {
+                        "id": {"type": "string", "format": "uuid"},
+                        "items": {
+                            "type": "array",
+                            "items": {"$ref": "#/components/schemas/OrderItem"},
+                        },
+                        "total": {"type": "number"},
+                        "status": {"$ref": "#/components/schemas/OrderStatus"},
+                    },
+                },
+                "OrderItem": {
+                    "type": "object",
+                    "properties": {
+                        "product_id": {"type": "string", "format": "uuid"},
+                        "quantity": {"type": "integer"},
+                        "price": {"type": "number"},
+                    },
+                },
+                "OrderStatus": {
+                    "type": "string",
+                    "enum": [
+                        "pending",
+                        "processing",
+                        "shipped",
+                        "delivered",
+                        "cancelled",
+                    ],
+                },
+            },
+            "requestBodies": {
+                "OrderRequest": {
+                    "description": "Order to create",
+                    "required": True,
+                    "content": {
+                        "application/json": {
+                            "schema": {
+                                "type": "object",
+                                "required": ["items"],
+                                "properties": {
+                                    "items": {
+                                        "type": "array",
+                                        "items": {
+                                            "$ref": "#/components/schemas/OrderItem"
+                                        },
+                                    },
+                                    "notes": {"type": "string"},
+                                },
+                            }
+                        }
+                    },
+                }
+            },
+        },
+    }
+
+
+@pytest.fixture
+def parsed_complex_routes(complex_schema):
+    """Return parsed routes from the complex schema."""
+    return parse_openapi_to_http_routes(complex_schema)
+
+
+@pytest.fixture
+def complex_route_map(parsed_complex_routes):
+    """Return a dictionary of routes by operation ID."""
+    return {
+        r.operation_id: r for r in parsed_complex_routes if r.operation_id is not None
+    }
+
+
+@pytest.fixture
+def schema_with_invalid_reference() -> Dict[str, Any]:
+    """Fixture that returns a schema with an invalid reference."""
+    return {
+        "openapi": "3.1.0",
+        "info": {"title": "Invalid Reference API", "version": "1.0.0"},
+        "paths": {
+            "/broken-ref": {
+                "get": {
+                    "summary": "Endpoint with broken reference",
+                    "operationId": "brokenRef",
+                    "parameters": [
+                        {"$ref": "#/components/parameters/NonExistentParam"}
+                    ],
+                    "responses": {"200": {"description": "Something"}},
+                }
+            }
+        },
+        "components": {
+            "parameters": {}  # Empty parameters object to ensure the reference is broken
+        },
+    }
+
+
+@pytest.fixture
+def schema_with_content_params() -> Dict[str, Any]:
+    """Fixture that returns a schema with content-based parameters (complex parameters)."""
+    return {
+        "openapi": "3.1.0",
+        "info": {"title": "Content Params API", "version": "1.0.0"},
+        "paths": {
+            "/complex-params": {
+                "post": {
+                    "summary": "Endpoint with complex parameter",
+                    "operationId": "complexParams",
+                    "parameters": [
+                        {
+                            "name": "filter",
+                            "in": "query",
+                            "content": {
+                                "application/json": {
+                                    "schema": {
+                                        "type": "object",
+                                        "properties": {
+                                            "field": {"type": "string"},
+                                            "operator": {
+                                                "type": "string",
+                                                "enum": ["eq", "gt", "lt"],
+                                            },
+                                            "value": {"type": "string"},
+                                        },
+                                    }
+                                }
+                            },
+                        }
+                    ],
+                    "responses": {"200": {"description": "Results"}},
+                }
+            },
+        },
+    }
+
+
+@pytest.fixture
+def parsed_content_param_routes(schema_with_content_params):
+    """Return parsed routes from the schema with content parameters."""
+    return parse_openapi_to_http_routes(schema_with_content_params)
+
+
+@pytest.fixture
+def schema_all_http_methods() -> Dict[str, Any]:
+    """Fixture that returns a schema with all HTTP methods."""
+    return {
+        "openapi": "3.1.0",
+        "info": {"title": "All Methods API", "version": "1.0.0"},
+        "paths": {
+            "/resource": {
+                "get": {
+                    "operationId": "getResource",
+                    "responses": {"200": {"description": "Success"}},
+                },
+                "post": {
+                    "operationId": "createResource",
+                    "responses": {"201": {"description": "Created"}},
+                },
+                "put": {
+                    "operationId": "updateResource",
+                    "responses": {"200": {"description": "Updated"}},
+                },
+                "delete": {
+                    "operationId": "deleteResource",
+                    "responses": {"204": {"description": "Deleted"}},
+                },
+                "patch": {
+                    "operationId": "patchResource",
+                    "responses": {"200": {"description": "Patched"}},
+                },
+                "head": {
+                    "operationId": "headResource",
+                    "responses": {"200": {"description": "Headers only"}},
+                },
+                "options": {
+                    "operationId": "optionsResource",
+                    "responses": {"200": {"description": "Options"}},
+                },
+                "trace": {
+                    "operationId": "traceResource",
+                    "responses": {"200": {"description": "Trace"}},
+                },
+            },
+        },
+    }
+
+
+@pytest.fixture
+def parsed_http_methods_routes(schema_all_http_methods):
+    """Return parsed routes from the schema with all HTTP methods."""
+    return parse_openapi_to_http_routes(schema_all_http_methods)
+
+
+# --- Tests for complex schemas with references --- #
+
+
+def test_complex_schema_route_count(parsed_complex_routes):
+    """Test that parsing a schema with references successfully extracts all routes."""
+    assert len(parsed_complex_routes) == 3
+
+
+def test_complex_schema_list_users_query_param_limit(complex_route_map):
+    """Test that a reference to a limit query parameter is correctly resolved."""
+    list_users = complex_route_map["listUsers"]
+
+    limit_param = next((p for p in list_users.parameters if p.name == "limit"), None)
+    assert limit_param is not None
+    assert limit_param.location == "query"
+    assert limit_param.schema_.get("default") == 20
+
+
+def test_complex_schema_list_users_query_param_limit_maximum(complex_route_map):
+    """Test that a limit parameter's maximum value is correctly resolved."""
+    list_users = complex_route_map["listUsers"]
+
+    limit_param = next((p for p in list_users.parameters if p.name == "limit"), None)
+    assert limit_param is not None
+    assert limit_param.schema_.get("maximum") == 100
+
+
+def test_complex_schema_get_user_path_param_existence(complex_route_map):
+    """Test that a reference to a path parameter exists."""
+    get_user = complex_route_map["getUser"]
+
+    user_id_param = next((p for p in get_user.parameters if p.name == "userId"), None)
+    assert user_id_param is not None
+    assert user_id_param.location == "path"
+
+
+def test_complex_schema_get_user_path_param_required(complex_route_map):
+    """Test that a path parameter is correctly marked as required."""
+    get_user = complex_route_map["getUser"]
+
+    user_id_param = next((p for p in get_user.parameters if p.name == "userId"), None)
+    assert user_id_param is not None
+    assert user_id_param.required is True
+
+
+def test_complex_schema_get_user_path_param_format(complex_route_map):
+    """Test that a path parameter format is correctly resolved."""
+    get_user = complex_route_map["getUser"]
+
+    user_id_param = next((p for p in get_user.parameters if p.name == "userId"), None)
+    assert user_id_param is not None
+    assert user_id_param.schema_.get("format") == "uuid"
+
+
+def test_complex_schema_create_order_request_body_presence(complex_route_map):
+    """Test that a reference to a request body is resolved correctly."""
+    create_order = complex_route_map["createOrder"]
+
+    assert create_order.request_body is not None
+    assert create_order.request_body.required is True
+
+
+def test_complex_schema_create_order_request_body_content_type(complex_route_map):
+    """Test that request body content type is correctly resolved."""
+    create_order = complex_route_map["createOrder"]
+
+    assert create_order.request_body is not None
+    assert "application/json" in create_order.request_body.content_schema
+
+
+def test_complex_schema_create_order_request_body_properties(complex_route_map):
+    """Test that request body properties are correctly resolved."""
+    create_order = complex_route_map["createOrder"]
+
+    assert create_order.request_body is not None
+    json_schema = create_order.request_body.content_schema["application/json"]
+    assert "items" in json_schema.get("properties", {})
+
+
+def test_complex_schema_create_order_request_body_required_fields(complex_route_map):
+    """Test that request body required fields are correctly resolved."""
+    create_order = complex_route_map["createOrder"]
+
+    assert create_order.request_body is not None
+    json_schema = create_order.request_body.content_schema["application/json"]
+    assert json_schema.get("required") == ["items"]
+
+
+# --- Tests for schema reference resolution errors --- #
+
+
+def test_parser_handles_broken_references(schema_with_invalid_reference):
+    """Test that parser handles broken references gracefully."""
+    # We're just checking that the function doesn't throw an exception
+    routes = parse_openapi_to_http_routes(schema_with_invalid_reference)
+
+    # Should still return routes list (may be empty)
+    assert isinstance(routes, list)
+
+    # Verify that the route with broken parameter reference is still included
+    # though it may not have the parameter properly
+    broken_route = next(
+        (r for r in routes if r.path == "/broken-ref" and r.method == "GET"), None
+    )
+
+    # The route should still be present
+    assert broken_route is not None
+    assert broken_route.operation_id == "brokenRef"
+
+
+# --- Tests for content-based parameters --- #
+
+
+def test_content_param_parameter_name(parsed_content_param_routes):
+    """Test that parser correctly extracts name for content-based parameters."""
+    complex_params = parsed_content_param_routes[0]
+
+    assert len(complex_params.parameters) == 1
+    param = complex_params.parameters[0]
+    assert param.name == "filter"
+
+
+def test_content_param_parameter_location(parsed_content_param_routes):
+    """Test that parser correctly extracts location for content-based parameters."""
+    complex_params = parsed_content_param_routes[0]
+
+    assert len(complex_params.parameters) == 1
+    param = complex_params.parameters[0]
+    assert param.location == "query"
+
+
+def test_content_param_schema_properties_presence(parsed_content_param_routes):
+    """Test that parser extracts schema properties from content-based parameter."""
+    complex_params = parsed_content_param_routes[0]
+
+    param = complex_params.parameters[0]
+    properties = param.schema_.get("properties", {})
+
+    assert "field" in properties
+    assert "operator" in properties
+    assert "value" in properties
+
+
+def test_content_param_schema_enum_presence(parsed_content_param_routes):
+    """Test that parser extracts enum values from content-based parameter."""
+    complex_params = parsed_content_param_routes[0]
+
+    param = complex_params.parameters[0]
+    properties = param.schema_.get("properties", {})
+
+    assert "enum" in properties.get("operator", {})
+
+
+# --- Tests for HTTP methods --- #
+
+
+def test_http_get_method_presence(parsed_http_methods_routes):
+    """Test that GET method is correctly extracted."""
+    get_route = next((r for r in parsed_http_methods_routes if r.method == "GET"), None)
+
+    assert get_route is not None
+    assert get_route.operation_id == "getResource"
+
+
+def test_http_get_method_path(parsed_http_methods_routes):
+    """Test that GET method path is correctly extracted."""
+    get_route = next((r for r in parsed_http_methods_routes if r.method == "GET"), None)
+
+    assert get_route is not None
+    assert get_route.path == "/resource"
+
+
+def test_http_post_method_presence(parsed_http_methods_routes):
+    """Test that POST method is correctly extracted."""
+    post_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "POST"), None
+    )
+
+    assert post_route is not None
+    assert post_route.operation_id == "createResource"
+
+
+def test_http_post_method_path(parsed_http_methods_routes):
+    """Test that POST method path is correctly extracted."""
+    post_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "POST"), None
+    )
+
+    assert post_route is not None
+    assert post_route.path == "/resource"
+
+
+def test_http_put_method_presence(parsed_http_methods_routes):
+    """Test that PUT method is correctly extracted."""
+    put_route = next((r for r in parsed_http_methods_routes if r.method == "PUT"), None)
+
+    assert put_route is not None
+    assert put_route.operation_id == "updateResource"
+
+
+def test_http_put_method_path(parsed_http_methods_routes):
+    """Test that PUT method path is correctly extracted."""
+    put_route = next((r for r in parsed_http_methods_routes if r.method == "PUT"), None)
+
+    assert put_route is not None
+    assert put_route.path == "/resource"
+
+
+def test_http_delete_method_presence(parsed_http_methods_routes):
+    """Test that DELETE method is correctly extracted."""
+    delete_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "DELETE"), None
+    )
+
+    assert delete_route is not None
+    assert delete_route.operation_id == "deleteResource"
+
+
+def test_http_delete_method_path(parsed_http_methods_routes):
+    """Test that DELETE method path is correctly extracted."""
+    delete_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "DELETE"), None
+    )
+
+    assert delete_route is not None
+    assert delete_route.path == "/resource"
+
+
+def test_http_patch_method_presence(parsed_http_methods_routes):
+    """Test that PATCH method is correctly extracted."""
+    patch_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "PATCH"), None
+    )
+
+    assert patch_route is not None
+    assert patch_route.operation_id == "patchResource"
+
+
+def test_http_patch_method_path(parsed_http_methods_routes):
+    """Test that PATCH method path is correctly extracted."""
+    patch_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "PATCH"), None
+    )
+
+    assert patch_route is not None
+    assert patch_route.path == "/resource"
+
+
+def test_http_head_method_presence(parsed_http_methods_routes):
+    """Test that HEAD method is correctly extracted."""
+    head_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "HEAD"), None
+    )
+
+    assert head_route is not None
+    assert head_route.operation_id == "headResource"
+
+
+def test_http_head_method_path(parsed_http_methods_routes):
+    """Test that HEAD method path is correctly extracted."""
+    head_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "HEAD"), None
+    )
+
+    assert head_route is not None
+    assert head_route.path == "/resource"
+
+
+def test_http_options_method_presence(parsed_http_methods_routes):
+    """Test that OPTIONS method is correctly extracted."""
+    options_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "OPTIONS"), None
+    )
+
+    assert options_route is not None
+    assert options_route.operation_id == "optionsResource"
+
+
+def test_http_options_method_path(parsed_http_methods_routes):
+    """Test that OPTIONS method path is correctly extracted."""
+    options_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "OPTIONS"), None
+    )
+
+    assert options_route is not None
+    assert options_route.path == "/resource"
+
+
+def test_http_trace_method_presence(parsed_http_methods_routes):
+    """Test that TRACE method is correctly extracted."""
+    trace_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "TRACE"), None
+    )
+
+    assert trace_route is not None
+    assert trace_route.operation_id == "traceResource"
+
+
+def test_http_trace_method_path(parsed_http_methods_routes):
+    """Test that TRACE method path is correctly extracted."""
+    trace_route = next(
+        (r for r in parsed_http_methods_routes if r.method == "TRACE"), None
+    )
+
+    assert trace_route is not None
+    assert trace_route.path == "/resource"

tests/utilities/openapi/test_openapi_fastapi.py

@@ -0,0 +1,437 @@
+"""Tests for FastAPI integration with the OpenAPI utilities."""
+
+from typing import Any, Dict
+
+import pytest
+from fastapi import FastAPI
+
+from fastmcp.utilities.openapi import parse_openapi_to_http_routes
+
+
+@pytest.fixture
+def fastapi_server() -> FastAPI:
+    """Fixture that returns a FastAPI app for live OpenAPI schema testing."""
+    from enum import Enum
+    from typing import List, Optional
+
+    from fastapi import Body, Depends, Header, HTTPException, Path, Query
+    from pydantic import BaseModel, Field
+
+    class ItemStatus(str, Enum):
+        available = "available"
+        pending = "pending"
+        sold = "sold"
+
+    class Tag(BaseModel):
+        id: int
+        name: str
+
+    class Item(BaseModel):
+        """Example pydantic model for testing OpenAPI schema generation."""
+
+        name: str
+        description: Optional[str] = None
+        price: float
+        tax: Optional[float] = None
+        tags: List[str] = Field(default_factory=list)
+        status: ItemStatus = ItemStatus.available
+        dimensions: Optional[Dict[str, float]] = None
+
+    # Create a FastAPI app with comprehensive features
+    app = FastAPI(
+        title="Comprehensive Test API",
+        description="A test API with various OpenAPI features",
+        version="1.0.0",
+    )
+
+    def get_token_header(
+        x_token: str = Header(..., description="Authentication token"),
+    ):
+        """Example dependency function for header validation."""
+        if x_token != "fake-super-secret-token":
+            raise HTTPException(status_code=400, detail="X-Token header invalid")
+        return x_token
+
+    TokenDep = Depends(get_token_header)
+
+    @app.get(
+        "/items/",
+        operation_id="list_items",
+        summary="List all items",
+        description="Get a list of all items with optional filtering",
+        tags=["items"],
+    )
+    async def list_items(
+        skip: int = Query(0, description="Number of items to skip"),
+        limit: int = Query(10, description="Max number of items to return"),
+        status: Optional[ItemStatus] = Query(
+            None, description="Filter items by status"
+        ),
+    ):
+        """List all items with pagination and optional status filtering."""
+        fake_items = [
+            {"name": f"Item {i}", "price": float(i)} for i in range(skip, skip + limit)
+        ]
+        if status:
+            fake_items = [item for item in fake_items if item.get("status") == status]
+        return fake_items
+
+    @app.post(
+        "/items/",
+        operation_id="create_item",
+        summary="Create a new item",
+        tags=["items"],
+        status_code=201,
+    )
+    async def create_item(
+        item: Item = Body(..., description="Item to create"),
+        x_token: str = TokenDep,
+    ):
+        """Create a new item (requires authentication)."""
+        return item
+
+    @app.get(
+        "/items/{item_id}",
+        operation_id="get_item",
+        summary="Get a specific item by ID",
+        tags=["items"],
+    )
+    async def get_item(
+        item_id: int = Path(..., description="The ID of the item to retrieve"),
+        include_tax: bool = Query(
+            False, description="Whether to include tax information"
+        ),
+    ):
+        """Get details about a specific item."""
+        item = {
+            "id": item_id,
+            "name": f"Item {item_id}",
+            "price": float(item_id) * 10.0,
+        }
+        if include_tax:
+            item["tax"] = item["price"] * 0.2
+        return item
+
+    @app.put(
+        "/items/{item_id}",
+        operation_id="update_item",
+        summary="Update an existing item",
+        tags=["items"],
+    )
+    async def update_item(
+        item_id: int = Path(..., description="The ID of the item to update"),
+        item: Item = Body(..., description="Updated item data"),
+        x_token: str = TokenDep,
+    ):
+        """Update an existing item (requires authentication)."""
+        return {"item_id": item_id, **item.model_dump()}
+
+    @app.delete(
+        "/items/{item_id}",
+        operation_id="delete_item",
+        summary="Delete an item",
+        tags=["items"],
+    )
+    async def delete_item(
+        item_id: int = Path(..., description="The ID of the item to delete"),
+        x_token: str = TokenDep,
+    ):
+        """Delete an item (requires authentication)."""
+        return {"item_id": item_id, "deleted": True}
+
+    @app.patch(
+        "/items/{item_id}/tags",
+        operation_id="update_item_tags",
+        summary="Update item tags",
+        tags=["items", "tags"],
+    )
+    async def update_item_tags(
+        item_id: int = Path(..., description="The ID of the item"),
+        tags: List[str] = Body(..., description="Updated tags"),
+    ):
+        """Update just the tags of an item."""
+        return {"item_id": item_id, "tags": tags}
+
+    @app.get(
+        "/items/{item_id}/tags/{tag_id}",
+        operation_id="get_item_tag",
+        summary="Get a specific tag for an item",
+        tags=["items", "tags"],
+    )
+    async def get_item_tag(
+        item_id: int = Path(..., description="The ID of the item"),
+        tag_id: str = Path(..., description="The ID of the tag"),
+    ):
+        """Get a specific tag for an item."""
+        return {"item_id": item_id, "tag_id": tag_id}
+
+    @app.post(
+        "/upload/",
+        operation_id="upload_file",
+        summary="Upload a file",
+        tags=["files"],
+    )
+    async def upload_file(
+        file_name: str = Query(..., description="Name of the file"),
+        content_type: str = Query(..., description="Content type of the file"),
+    ):
+        """Upload a file (dummy endpoint for testing query params)."""
+        return {
+            "file_name": file_name,
+            "content_type": content_type,
+            "status": "uploaded",
+        }
+
+    # Add a callback route for testing complex documentation
+    @app.post(
+        "/webhook",
+        operation_id="register_webhook",
+        summary="Register a webhook",
+        tags=["webhooks"],
+        callbacks={  # type: ignore
+            "itemProcessed": {
+                "{$request.body.callbackUrl}": {
+                    "post": {
+                        "summary": "Callback for when an item is processed",
+                        "requestBody": {
+                            "required": True,
+                            "content": {
+                                "application/json": {
+                                    "schema": {
+                                        "type": "object",
+                                        "properties": {
+                                            "item_id": {"type": "integer"},
+                                            "status": {"type": "string"},
+                                            "timestamp": {
+                                                "type": "string",
+                                                "format": "date-time",
+                                            },
+                                        },
+                                    }
+                                }
+                            },
+                        },
+                        "responses": {
+                            "200": {"description": "Webhook processed successfully"}
+                        },
+                    }
+                }
+            }
+        },
+    )
+    async def register_webhook(
+        callback_url: str = Body(
+            ..., embed=True, description="URL to call when processing completes"
+        ),
+    ):
+        """Register a webhook for processing notifications."""
+        return {"registered": True, "callback_url": callback_url}
+
+    return app
+
+
+@pytest.fixture
+def fastapi_openapi_schema(fastapi_server) -> Dict[str, Any]:
+    """Fixture that returns the OpenAPI schema from a live FastAPI server."""
+    return fastapi_server.openapi()
+
+
+@pytest.fixture
+def parsed_routes(fastapi_openapi_schema):
+    """Return parsed routes from a FastAPI OpenAPI schema."""
+    return parse_openapi_to_http_routes(fastapi_openapi_schema)
+
+
+@pytest.fixture
+def route_map(parsed_routes):
+    """Return a dictionary of routes by operation ID."""
+    return {r.operation_id: r for r in parsed_routes if r.operation_id is not None}
+
+
+def test_parse_fastapi_schema_route_count(parsed_routes):
+    """Test that all routes are parsed from the FastAPI schema."""
+    assert len(parsed_routes) == 9  # 8 endpoints + 1 callback
+
+
+def test_parse_fastapi_schema_operation_ids(route_map):
+    """Test that all expected operation IDs are present in the parsed schema."""
+    expected_operations = [
+        "list_items",
+        "create_item",
+        "get_item",
+        "update_item",
+        "delete_item",
+        "update_item_tags",
+        "get_item_tag",
+        "upload_file",
+        "register_webhook",
+    ]
+
+    for op_id in expected_operations:
+        assert op_id in route_map, f"Operation ID '{op_id}' not found in parsed routes"
+
+
+def test_path_parameter_parsing(route_map):
+    """Test that path parameters are correctly parsed."""
+    get_item = route_map["get_item"]
+    path_params = [p for p in get_item.parameters if p.location == "path"]
+
+    assert len(path_params) == 1
+    assert path_params[0].name == "item_id"
+    assert path_params[0].required is True
+
+
+def test_query_parameter_parsing(route_map):
+    """Test that query parameters are correctly parsed."""
+    list_items = route_map["list_items"]
+    query_params = [p for p in list_items.parameters if p.location == "query"]
+
+    assert len(query_params) == 3  # skip, limit, status
+    param_names = [p.name for p in query_params]
+    assert "skip" in param_names
+    assert "limit" in param_names
+    assert "status" in param_names
+
+
+def test_header_parameter_parsing(route_map):
+    """Test that header parameters from dependencies are correctly parsed."""
+    create_item = route_map["create_item"]
+    header_params = [p for p in create_item.parameters if p.location == "header"]
+
+    assert len(header_params) == 1
+    assert header_params[0].name == "x-token"
+    assert header_params[0].required is True
+
+
+def test_request_body_content_type(route_map):
+    """Test that request body content types are correctly parsed."""
+    create_item = route_map["create_item"]
+
+    assert create_item.request_body is not None
+    assert "application/json" in create_item.request_body.content_schema
+
+
+def test_request_body_properties(route_map):
+    """Test that request body properties are correctly parsed."""
+    create_item = route_map["create_item"]
+    json_schema = create_item.request_body.content_schema["application/json"]
+    properties = json_schema.get("properties", {})
+
+    assert "name" in properties
+    assert "price" in properties
+    assert "description" in properties
+    assert "tags" in properties
+    assert "status" in properties
+
+
+def test_request_body_status_schema(route_map):
+    """Test that the status schema in request body is correctly handled."""
+    create_item = route_map["create_item"]
+    json_schema = create_item.request_body.content_schema["application/json"]
+    properties = json_schema.get("properties", {})
+    status_schema = properties.get("status", {})
+
+    # FastAPI may represent enums as references or directly include enum values
+    assert "$ref" in status_schema or "enum" in status_schema
+
+
+def test_route_with_items_tag(parsed_routes):
+    """Test that routes with 'items' tag are correctly parsed."""
+    item_routes = [r for r in parsed_routes if "items" in r.tags]
+
+    assert len(item_routes) >= 6  # At least 6 endpoints with "items" tag
+
+
+def test_routes_with_multiple_tags(parsed_routes):
+    """Test that routes with multiple tags are correctly parsed."""
+    multi_tag_routes = [r for r in parsed_routes if len(r.tags) > 1]
+
+    assert len(multi_tag_routes) >= 2  # At least 2 endpoints with multiple tags
+
+
+def test_specific_route_tags(route_map):
+    """Test that specific routes have the expected tags."""
+    assert "items" in route_map["list_items"].tags
+    assert "items" in route_map["update_item_tags"].tags
+    assert "tags" in route_map["update_item_tags"].tags
+    assert "webhooks" in route_map["register_webhook"].tags
+
+
+def test_operation_summary(route_map):
+    """Test that operation summary is correctly parsed."""
+    list_items = route_map["list_items"]
+
+    assert list_items.summary == "List all items"
+
+
+def test_operation_description(route_map):
+    """Test that operation description is correctly parsed."""
+    list_items = route_map["list_items"]
+
+    assert list_items.description is not None
+    assert "optional filtering" in list_items.description
+
+
+def test_path_with_route_parameters(route_map):
+    """Test that paths with route parameters are correctly parsed."""
+    get_item = route_map["get_item"]
+
+    assert get_item.path == "/items/{item_id}"
+
+
+def test_complex_nested_paths(route_map):
+    """Test that complex nested paths are correctly parsed."""
+    get_item_tag = route_map["get_item_tag"]
+
+    assert get_item_tag.path == "/items/{item_id}/tags/{tag_id}"
+
+
+def test_http_methods(route_map):
+    """Test that HTTP methods are correctly parsed."""
+    assert route_map["list_items"].method == "GET"
+    assert route_map["create_item"].method == "POST"
+    assert route_map["update_item"].method == "PUT"
+    assert route_map["delete_item"].method == "DELETE"
+    assert route_map["update_item_tags"].method == "PATCH"
+
+
+def test_item_schema_properties(route_map):
+    """Test that Item schema properties are correctly resolved."""
+    create_item = route_map["create_item"]
+    json_schema = create_item.request_body.content_schema["application/json"]
+    properties = json_schema.get("properties", {})
+
+    assert "name" in properties
+    assert properties["name"]["type"] == "string"
+    assert "price" in properties
+    assert properties["price"]["type"] == "number"
+
+
+def test_webhook_endpoint(route_map):
+    """Test parsing of webhook registration endpoint."""
+    webhook = route_map["register_webhook"]
+
+    assert webhook.method == "POST"
+    assert webhook.path == "/webhook"
+
+
+def test_webhook_request_body(route_map):
+    """Test that webhook request body is correctly parsed."""
+    webhook = route_map["register_webhook"]
+
+    assert webhook.request_body is not None
+    assert "application/json" in webhook.request_body.content_schema
+    json_schema = webhook.request_body.content_schema["application/json"]
+    assert "callback_url" in json_schema.get("properties", {})
+
+
+def test_token_dependency_handling(route_map):
+    """Test that token dependencies are correctly handled in parsed endpoints."""
+    token_endpoints = ["create_item", "update_item", "delete_item"]
+
+    for op_id in token_endpoints:
+        route = route_map[op_id]
+        header_params = [p for p in route.parameters if p.location == "header"]
+        token_headers = [p for p in header_params if p.name == "x-token"]
+        assert len(token_headers) == 1, f"Expected x-token header in {op_id}"
+        assert token_headers[0].required is True