src/fastmcp/utilities/openapi.py

import json
from typing import Any, Generic, Literal, TypeVar, cast

from openapi_pydantic import (
    OpenAPI,
    Operation,
    Parameter,
    PathItem,
    Reference,
    RequestBody,
    Response,
    Schema,
)

# Import OpenAPI 3.0 models as well
from openapi_pydantic.v3.v3_0 import OpenAPI as OpenAPI_30
from openapi_pydantic.v3.v3_0 import Operation as Operation_30
from openapi_pydantic.v3.v3_0 import Parameter as Parameter_30
from openapi_pydantic.v3.v3_0 import PathItem as PathItem_30
from openapi_pydantic.v3.v3_0 import Reference as Reference_30
from openapi_pydantic.v3.v3_0 import RequestBody as RequestBody_30
from openapi_pydantic.v3.v3_0 import Response as Response_30
from openapi_pydantic.v3.v3_0 import Schema as Schema_30
from pydantic import BaseModel, Field, ValidationError

from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import FastMCPBaseModel

logger = get_logger(__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]


def format_array_parameter(
    values: list, parameter_name: str, is_query_parameter: bool = False
) -> str | list:
    """
    Format an array parameter according to OpenAPI specifications.

    Args:
        values: List of values to format
        parameter_name: Name of the parameter (for error messages)
        is_query_parameter: If True, can return list for explode=True behavior

    Returns:
        String (comma-separated) or list (for query params with explode=True)
    """
    # For arrays of simple types (strings, numbers, etc.), join with commas
    if all(isinstance(item, str | int | float | bool) for item in values):
        return ",".join(str(v) for v in values)

    # For complex types, try to create a simpler representation
    try:
        # Try to create a simple string representation
        formatted_parts = []
        for item in values:
            if isinstance(item, dict):
                # For objects, serialize key-value pairs
                item_parts = []
                for k, v in item.items():
                    item_parts.append(f"{k}:{v}")
                formatted_parts.append(".".join(item_parts))
            else:
                formatted_parts.append(str(item))

        return ",".join(formatted_parts)
    except Exception as e:
        param_type = "query" if is_query_parameter else "path"
        logger.warning(
            f"Failed to format complex array {param_type} parameter '{parameter_name}': {e}"
        )

        if is_query_parameter:
            # For query parameters, fallback to original list
            return values
        else:
            # For path parameters, fallback to string representation without Python syntax
            # Use str.translate() for efficient character removal
            translation_table = str.maketrans("", "", "[]'\"")
            str_value = str(values).translate(translation_table)
            return str_value


def format_deep_object_parameter(
    param_value: dict, parameter_name: str
) -> dict[str, str]:
    """
    Format a dictionary parameter for deepObject style serialization.

    According to OpenAPI 3.0 spec, deepObject style with explode=true serializes
    object properties as separate query parameters with bracket notation.

    For example: `{"id": "123", "type": "user"}` becomes `param[id]=123&param[type]=user`.

    Args:
        param_value: Dictionary value to format
        parameter_name: Name of the parameter

    Returns:
        Dictionary with bracketed parameter names as keys
    """
    if not isinstance(param_value, dict):
        logger.warning(
            f"deepObject style parameter '{parameter_name}' expected dict, got {type(param_value)}"
        )
        return {}

    result = {}
    for key, value in param_value.items():
        # Format as param[key]=value
        bracketed_key = f"{parameter_name}[{key}]"
        result[bracketed_key] = str(value)

    return result


class ParameterInfo(FastMCPBaseModel):
    """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: str | None = None
    explode: bool | None = None  # OpenAPI explode property for array parameters
    style: str | None = None  # OpenAPI style property for parameter serialization


class RequestBodyInfo(FastMCPBaseModel):
    """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: str | None = None


class ResponseInfo(FastMCPBaseModel):
    """Represents response information in our IR."""

    description: str | None = None
    # Store schema per media type, key is media type
    content_schema: dict[str, JsonSchema] = Field(default_factory=dict)


class HTTPRoute(FastMCPBaseModel):
    """Intermediate Representation for a single OpenAPI operation."""

    path: str
    method: HttpMethod
    operation_id: str | None = None
    summary: str | None = None
    description: str | None = None
    tags: list[str] = Field(default_factory=list)
    parameters: list[ParameterInfo] = Field(default_factory=list)
    request_body: RequestBodyInfo | None = None
    responses: dict[str, ResponseInfo] = Field(
        default_factory=dict
    )  # Key: status code str
    schema_definitions: dict[str, JsonSchema] = Field(
        default_factory=dict
    )  # Store component schemas
    extensions: dict[str, Any] = Field(default_factory=dict)
    openapi_version: str | None = None


# Export public symbols
__all__ = [
    "HTTPRoute",
    "ParameterInfo",
    "RequestBodyInfo",
    "ResponseInfo",
    "HttpMethod",
    "ParameterLocation",
    "JsonSchema",
    "parse_openapi_to_http_routes",
    "extract_output_schema_from_responses",
    "format_deep_object_parameter",
    "_handle_nullable_fields",
]

# Type variables for generic parser
TOpenAPI = TypeVar("TOpenAPI", OpenAPI, OpenAPI_30)
TSchema = TypeVar("TSchema", Schema, Schema_30)
TReference = TypeVar("TReference", Reference, Reference_30)
TParameter = TypeVar("TParameter", Parameter, Parameter_30)
TRequestBody = TypeVar("TRequestBody", RequestBody, RequestBody_30)
TResponse = TypeVar("TResponse", Response, Response_30)
TOperation = TypeVar("TOperation", Operation, Operation_30)
TPathItem = TypeVar("TPathItem", PathItem, PathItem_30)


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.

    Supports both OpenAPI 3.0.x and 3.1.x versions.
    """
    # Check OpenAPI version to use appropriate model
    openapi_version = openapi_dict.get("openapi", "")

    try:
        if openapi_version.startswith("3.0"):
            # Use OpenAPI 3.0 models
            openapi_30 = OpenAPI_30.model_validate(openapi_dict)
            logger.debug(
                f"Successfully parsed OpenAPI 3.0 schema version: {openapi_30.openapi}"
            )
            parser = OpenAPIParser(
                openapi_30,
                Reference_30,
                Schema_30,
                Parameter_30,
                RequestBody_30,
                Response_30,
                Operation_30,
                PathItem_30,
                openapi_version,
            )
            return parser.parse()
        else:
            # Default to OpenAPI 3.1 models
            openapi_31 = OpenAPI.model_validate(openapi_dict)
            logger.debug(
                f"Successfully parsed OpenAPI 3.1 schema version: {openapi_31.openapi}"
            )
            parser = OpenAPIParser(
                openapi_31,
                Reference,
                Schema,
                Parameter,
                RequestBody,
                Response,
                Operation,
                PathItem,
                openapi_version,
            )
            return parser.parse()
    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


class OpenAPIParser(
    Generic[
        TOpenAPI,
        TReference,
        TSchema,
        TParameter,
        TRequestBody,
        TResponse,
        TOperation,
        TPathItem,
    ]
):
    """Unified parser for OpenAPI schemas with generic type parameters to handle both 3.0 and 3.1."""

    def __init__(
        self,
        openapi: TOpenAPI,
        reference_cls: type[TReference],
        schema_cls: type[TSchema],
        parameter_cls: type[TParameter],
        request_body_cls: type[TRequestBody],
        response_cls: type[TResponse],
        operation_cls: type[TOperation],
        path_item_cls: type[TPathItem],
        openapi_version: str,
    ):
        """Initialize the parser with the OpenAPI schema and type classes."""
        self.openapi = openapi
        self.reference_cls = reference_cls
        self.schema_cls = schema_cls
        self.parameter_cls = parameter_cls
        self.request_body_cls = request_body_cls
        self.response_cls = response_cls
        self.operation_cls = operation_cls
        self.path_item_cls = path_item_cls
        self.openapi_version = openapi_version

    def _convert_to_parameter_location(self, param_in: str) -> ParameterLocation:
        """Convert string parameter location to our ParameterLocation type."""
        if param_in in ["path", "query", "header", "cookie"]:
            return param_in  # type: ignore[return-value]  # Safe cast since we checked values
        logger.warning(f"Unknown parameter location: {param_in}, defaulting to 'query'")
        return "query"  # type: ignore[return-value]  # Safe cast to default value

    def _resolve_ref(self, item: Any) -> Any:
        """Resolves a reference to its target definition."""
        if isinstance(item, self.reference_cls):
            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 = self.openapi

                for part in parts:
                    if part.isdigit() and isinstance(target, list):
                        target = target[int(part)]
                    elif isinstance(target, BaseModel):
                        # Check class fields first, then model_extra
                        if part in target.__class__.model_fields:
                            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
                            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}'"
                        )

                    if target is None:
                        raise ValueError(
                            f"Reference part '{part}' not found in path '{ref_str}'"
                        )

                # Handle nested references
                if isinstance(target, self.reference_cls):
                    return self._resolve_ref(target)

                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(self, schema_obj: Any) -> JsonSchema:
        """Resolves a schema and returns it as a dictionary."""
        try:
            resolved_schema = self._resolve_ref(schema_obj)

            if isinstance(resolved_schema, (self.schema_cls)):
                # Convert schema to dictionary
                result = resolved_schema.model_dump(
                    mode="json", by_alias=True, exclude_none=True
                )
            elif isinstance(resolved_schema, dict):
                result = resolved_schema
            else:
                logger.warning(
                    f"Expected Schema after resolving, got {type(resolved_schema)}. Returning empty dict."
                )
                result = {}

            return _replace_ref_with_defs(result)
        except ValueError as e:
            # Re-raise ValueError for external reference errors and other validation issues
            if "External or non-local reference not supported" in str(e):
                raise
            logger.error(f"Failed to extract schema as dict: {e}", exc_info=False)
            return {}
        except Exception as e:
            logger.error(f"Failed to extract schema as dict: {e}", exc_info=False)
            return {}

    def _extract_parameters(
        self,
        operation_params: list[Any] | None = None,
        path_item_params: list[Any] | None = None,
    ) -> list[ParameterInfo]:
        """Extract and resolve parameters from operation and path item."""
        extracted_params: list[ParameterInfo] = []
        seen_params: dict[
            tuple[str, str], bool
        ] = {}  # Use tuple of (name, location) as key
        all_params = (operation_params or []) + (path_item_params or [])

        for param_or_ref in all_params:
            try:
                parameter = self._resolve_ref(param_or_ref)

                if not isinstance(parameter, self.parameter_cls):
                    logger.warning(
                        f"Expected Parameter after resolving, got {type(parameter)}. Skipping."
                    )
                    continue

                # Extract parameter info - handle both 3.0 and 3.1 parameter models
                param_in = parameter.param_in  # Both use param_in
                # Handle enum or string parameter locations
                from enum import Enum

                param_in_str = (
                    param_in.value if isinstance(param_in, Enum) else param_in
                )
                param_location = self._convert_to_parameter_location(param_in_str)
                param_schema_obj = parameter.param_schema  # Both use param_schema

                # Skip duplicate parameters (same name and location)
                param_key = (parameter.name, param_in_str)
                if param_key in seen_params:
                    continue
                seen_params[param_key] = True

                # Extract schema
                param_schema_dict = {}
                if param_schema_obj:
                    # Process schema object
                    param_schema_dict = self._extract_schema_as_dict(param_schema_obj)

                    # Handle default value
                    resolved_schema = self._resolve_ref(param_schema_obj)
                    if (
                        not isinstance(resolved_schema, self.reference_cls)
                        and hasattr(resolved_schema, "default")
                        and resolved_schema.default is not None
                    ):
                        param_schema_dict["default"] = resolved_schema.default

                elif hasattr(parameter, "content") and parameter.content:
                    # Handle content-based parameters
                    first_media_type = next(iter(parameter.content.values()), None)
                    if (
                        first_media_type
                        and hasattr(first_media_type, "media_type_schema")
                        and first_media_type.media_type_schema
                    ):
                        media_schema = first_media_type.media_type_schema
                        param_schema_dict = self._extract_schema_as_dict(media_schema)

                        # Handle default value in content schema
                        resolved_media_schema = self._resolve_ref(media_schema)
                        if (
                            not isinstance(resolved_media_schema, self.reference_cls)
                            and hasattr(resolved_media_schema, "default")
                            and resolved_media_schema.default is not None
                        ):
                            param_schema_dict["default"] = resolved_media_schema.default

                # Extract explode and style properties if present
                explode = getattr(parameter, "explode", None)
                style = getattr(parameter, "style", None)

                # Create parameter info object
                param_info = ParameterInfo(
                    name=parameter.name,
                    location=param_location,
                    required=parameter.required,
                    schema=param_schema_dict,
                    description=parameter.description,
                    explode=explode,
                    style=style,
                )
                extracted_params.append(param_info)
            except Exception as e:
                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(self, request_body_or_ref: Any) -> RequestBodyInfo | None:
        """Extract and resolve request body information."""
        if not request_body_or_ref:
            return None

        try:
            request_body = self._resolve_ref(request_body_or_ref)

            if not isinstance(request_body, self.request_body_cls):
                logger.warning(
                    f"Expected RequestBody after resolving, got {type(request_body)}. Returning None."
                )
                return None

            # Create request body info
            request_body_info = RequestBodyInfo(
                required=request_body.required,
                description=request_body.description,
            )

            # Extract content schemas
            if hasattr(request_body, "content") and request_body.content:
                for media_type_str, media_type_obj in request_body.content.items():
                    if (
                        media_type_obj
                        and hasattr(media_type_obj, "media_type_schema")
                        and media_type_obj.media_type_schema
                    ):
                        try:
                            schema_dict = self._extract_schema_as_dict(
                                media_type_obj.media_type_schema
                            )
                            request_body_info.content_schema[media_type_str] = (
                                schema_dict
                            )
                        except ValueError as e:
                            # Re-raise ValueError for external reference errors
                            if "External or non-local reference not supported" in str(
                                e
                            ):
                                raise
                            logger.error(
                                f"Failed to extract schema for media type '{media_type_str}': {e}"
                            )
                        except Exception as e:
                            logger.error(
                                f"Failed to extract schema for media type '{media_type_str}': {e}"
                            )

            return request_body_info
        except ValueError as e:
            # Re-raise ValueError for external reference errors
            if "External or non-local reference not supported" in str(e):
                raise
            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
        except Exception 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

    def _extract_responses(
        self, operation_responses: dict[str, Any] | None
    ) -> dict[str, ResponseInfo]:
        """Extract and resolve response information."""
        extracted_responses: dict[str, ResponseInfo] = {}

        if not operation_responses:
            return extracted_responses

        for status_code, resp_or_ref in operation_responses.items():
            try:
                response = self._resolve_ref(resp_or_ref)

                if not isinstance(response, self.response_cls):
                    logger.warning(
                        f"Expected Response after resolving for status code {status_code}, "
                        f"got {type(response)}. Skipping."
                    )
                    continue

                # Create response info
                resp_info = ResponseInfo(description=response.description)

                # Extract content schemas
                if hasattr(response, "content") and response.content:
                    for media_type_str, media_type_obj in response.content.items():
                        if (
                            media_type_obj
                            and hasattr(media_type_obj, "media_type_schema")
                            and media_type_obj.media_type_schema
                        ):
                            try:
                                schema_dict = self._extract_schema_as_dict(
                                    media_type_obj.media_type_schema
                                )
                                resp_info.content_schema[media_type_str] = schema_dict
                            except ValueError as e:
                                # Re-raise ValueError for external reference errors
                                if (
                                    "External or non-local reference not supported"
                                    in str(e)
                                ):
                                    raise
                                logger.error(
                                    f"Failed to extract schema for media type '{media_type_str}' "
                                    f"in response {status_code}: {e}"
                                )
                            except Exception as e:
                                logger.error(
                                    f"Failed to extract schema for media type '{media_type_str}' "
                                    f"in response {status_code}: {e}"
                                )

                extracted_responses[str(status_code)] = resp_info
            except ValueError as e:
                # Re-raise ValueError for external reference errors
                if "External or non-local reference not supported" in str(e):
                    raise
                ref_name = getattr(resp_or_ref, "ref", "unknown")
                logger.error(
                    f"Failed to extract response for status code {status_code} "
                    f"from reference '{ref_name}': {e}",
                    exc_info=False,
                )
            except Exception as e:
                ref_name = getattr(resp_or_ref, "ref", "unknown")
                logger.error(
                    f"Failed to extract response for status code {status_code} "
                    f"from reference '{ref_name}': {e}",
                    exc_info=False,
                )

        return extracted_responses

    def parse(self) -> list[HTTPRoute]:
        """Parse the OpenAPI schema into HTTP routes."""
        routes: list[HTTPRoute] = []

        if not hasattr(self.openapi, "paths") or not self.openapi.paths:
            logger.warning("OpenAPI schema has no paths defined.")
            return []

        # Extract component schemas
        schema_definitions = {}
        if hasattr(self.openapi, "components") and self.openapi.components:
            components = self.openapi.components
            if hasattr(components, "schemas") and components.schemas:
                for name, schema in components.schemas.items():
                    try:
                        if isinstance(schema, self.reference_cls):
                            resolved_schema = self._resolve_ref(schema)
                            schema_definitions[name] = self._extract_schema_as_dict(
                                resolved_schema
                            )
                        else:
                            schema_definitions[name] = self._extract_schema_as_dict(
                                schema
                            )
                    except Exception as e:
                        logger.warning(
                            f"Failed to extract schema definition '{name}': {e}"
                        )

        # Process paths and operations
        for path_str, path_item_obj in self.openapi.paths.items():
            if not isinstance(path_item_obj, self.path_item_cls):
                logger.warning(
                    f"Skipping invalid path item for path '{path_str}' (type: {type(path_item_obj)})"
                )
                continue

            path_level_params = (
                path_item_obj.parameters
                if hasattr(path_item_obj, "parameters")
                else None
            )

            # Get HTTP methods from the path item class fields
            http_methods = [
                "get",
                "put",
                "post",
                "delete",
                "options",
                "head",
                "patch",
                "trace",
            ]
            for method_lower in http_methods:
                operation = getattr(path_item_obj, method_lower, None)

                if operation and isinstance(operation, self.operation_cls):
                    # Cast method to HttpMethod - safe since we only use valid HTTP methods
                    method_upper = method_lower.upper()

                    try:
                        parameters = self._extract_parameters(
                            getattr(operation, "parameters", None), path_level_params
                        )

                        request_body_info = self._extract_request_body(
                            getattr(operation, "requestBody", None)
                        )

                        responses = self._extract_responses(
                            getattr(operation, "responses", None)
                        )

                        extensions = {}
                        if hasattr(operation, "model_extra") and operation.model_extra:
                            extensions = {
                                k: v
                                for k, v in operation.model_extra.items()
                                if k.startswith("x-")
                            }

                        route = HTTPRoute(
                            path=path_str,
                            method=method_upper,  # type: ignore[arg-type]  # Known valid HTTP method
                            operation_id=getattr(operation, "operationId", None),
                            summary=getattr(operation, "summary", None),
                            description=getattr(operation, "description", None),
                            tags=getattr(operation, "tags", []) or [],
                            parameters=parameters,
                            request_body=request_body_info,
                            responses=responses,
                            schema_definitions=schema_definitions,
                            extensions=extensions,
                            openapi_version=self.openapi_version,
                        )
                        routes.append(route)
                        logger.debug(
                            f"Successfully extracted route: {method_upper} {path_str}"
                        )
                    except ValueError as op_error:
                        # Re-raise ValueError for external reference errors
                        if "External or non-local reference not supported" in str(
                            op_error
                        ):
                            raise
                        op_id = getattr(operation, "operationId", "unknown")
                        logger.error(
                            f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}",
                            exc_info=True,
                        )
                    except Exception as op_error:
                        op_id = getattr(operation, "operationId", "unknown")
                        logger.error(
                            f"Failed to process operation {method_upper} {path_str} (ID: {op_id}): {op_error}",
                            exc_info=True,
                        )

        logger.debug(f"Finished parsing. Extracted {len(routes)} HTTP routes.")
        return routes


def clean_schema_for_display(schema: JsonSchema | None) -> JsonSchema | None:
    """
    Clean up a schema dictionary for display by removing internal/complex fields.
    """
    if not schema or not isinstance(schema, dict):
        return schema

    # Make a copy to avoid modifying the input schema
    cleaned = schema.copy()

    # Fields commonly removed for simpler display to LLMs or users
    fields_to_remove = [
        "allOf",
        "anyOf",
        "oneOf",
        "not",  # Composition keywords
        "nullable",  # Handled by type unions usually
        "discriminator",
        "readOnly",
        "writeOnly",
        "deprecated",
        "xml",
        "externalDocs",
        # Can be verbose, maybe remove based on flag?
        # "pattern", "minLength", "maxLength",
        # "minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum",
        # "multipleOf", "minItems", "maxItems", "uniqueItems",
        # "minProperties", "maxProperties"
    ]

    for field in fields_to_remove:
        if field in cleaned:
            cleaned.pop(field)

    # Recursively clean properties and items
    if "properties" in cleaned:
        cleaned["properties"] = {
            k: clean_schema_for_display(v) for k, v in cleaned["properties"].items()
        }
        # Remove properties section if empty after cleaning
        if not cleaned["properties"]:
            cleaned.pop("properties")

    if "items" in cleaned:
        cleaned["items"] = clean_schema_for_display(cleaned["items"])
        # Remove items section if empty after cleaning
        if not cleaned["items"]:
            cleaned.pop("items")

    if "additionalProperties" in cleaned:
        # Often verbose, can be simplified
        if isinstance(cleaned["additionalProperties"], dict):
            cleaned["additionalProperties"] = clean_schema_for_display(
                cleaned["additionalProperties"]
            )
        elif cleaned["additionalProperties"] is True:
            # Maybe keep 'true' or represent as 'Allows additional properties' text?
            pass  # Keep simple boolean for now


def generate_example_from_schema(schema: JsonSchema | None) -> Any:
    """
    Generate a simple example value from a JSON schema dictionary.
    Very basic implementation focusing on types.
    """
    if not schema or not isinstance(schema, dict):
        return "unknown"  # Or None?

    # Use default value if provided
    if "default" in schema:
        return schema["default"]
    # Use first enum value if provided
    if "enum" in schema and isinstance(schema["enum"], list) and schema["enum"]:
        return schema["enum"][0]
    # Use first example if provided
    if (
        "examples" in schema
        and isinstance(schema["examples"], list)
        and schema["examples"]
    ):
        return schema["examples"][0]
    if "example" in schema:
        return schema["example"]

    schema_type = schema.get("type")

    if schema_type == "object":
        result = {}
        properties = schema.get("properties", {})
        if isinstance(properties, dict):
            # Generate example for first few properties or required ones? Limit complexity.
            required_props = set(schema.get("required", []))
            props_to_include = list(properties.keys())[
                :3
            ]  # Limit to first 3 for brevity
            for prop_name in props_to_include:
                if prop_name in properties:
                    result[prop_name] = generate_example_from_schema(
                        properties[prop_name]
                    )
            # Ensure required props are present if possible
            for req_prop in required_props:
                if req_prop not in result and req_prop in properties:
                    result[req_prop] = generate_example_from_schema(
                        properties[req_prop]
                    )
        return result if result else {"key": "value"}  # Basic object if no props

    elif schema_type == "array":
        items_schema = schema.get("items")
        if isinstance(items_schema, dict):
            # Generate one example item
            item_example = generate_example_from_schema(items_schema)
            return [item_example] if item_example is not None else []
        return ["example_item"]  # Fallback

    elif schema_type == "string":
        format_type = schema.get("format")
        if format_type == "date-time":
            return "2024-01-01T12:00:00Z"
        if format_type == "date":
            return "2024-01-01"
        if format_type == "email":
            return "user@example.com"
        if format_type == "uuid":
            return "123e4567-e89b-12d3-a456-426614174000"
        if format_type == "byte":
            return "ZXhhbXBsZQ=="  # "example" base64
        return "string"

    elif schema_type == "integer":
        return 1
    elif schema_type == "number":
        return 1.5
    elif schema_type == "boolean":
        return True
    elif schema_type == "null":
        return None

    # Fallback if type is unknown or missing
    return "unknown_type"


def format_json_for_description(data: Any, indent: int = 2) -> str:
    """Formats Python data as a JSON string block for markdown."""
    try:
        json_str = json.dumps(data, indent=indent)
        return f"```json\n{json_str}\n```"
    except TypeError:
        return f"```\nCould not serialize to JSON: {data}\n```"


def format_description_with_responses(
    base_description: str,
    responses: dict[
        str, Any
    ],  # Changed from specific ResponseInfo type to avoid circular imports
    parameters: list[ParameterInfo] | None = None,  # Add parameters parameter
    request_body: RequestBodyInfo | None = None,  # Add request_body parameter
) -> str:
    """
    Formats the base description string with response, parameter, and request body information.

    Args:
        base_description (str): The initial description to be formatted.
        responses (dict[str, Any]): A dictionary of response information, keyed by status code.
        parameters (list[ParameterInfo] | None, optional): A list of parameter information,
            including path and query parameters. Each parameter includes details such as name,
            location, whether it is required, and a description.
        request_body (RequestBodyInfo | None, optional): Information about the request body,
            including its description, whether it is required, and its content schema.

    Returns:
        str: The formatted description string with additional details about responses, parameters,
        and the request body.
    """
    desc_parts = [base_description]

    # Add parameter information
    if parameters:
        # Process path parameters
        path_params = [p for p in parameters if p.location == "path"]
        if path_params:
            param_section = "\n\n**Path Parameters:**"
            desc_parts.append(param_section)
            for param in path_params:
                required_marker = " (Required)" if param.required else ""
                param_desc = f"\n- **{param.name}**{required_marker}: {param.description or 'No description.'}"
                desc_parts.append(param_desc)

        # Process query parameters
        query_params = [p for p in parameters if p.location == "query"]
        if query_params:
            param_section = "\n\n**Query Parameters:**"
            desc_parts.append(param_section)
            for param in query_params:
                required_marker = " (Required)" if param.required else ""
                param_desc = f"\n- **{param.name}**{required_marker}: {param.description or 'No description.'}"
                desc_parts.append(param_desc)

    # Add request body information if present
    if request_body and request_body.description:
        req_body_section = "\n\n**Request Body:**"
        desc_parts.append(req_body_section)
        required_marker = " (Required)" if request_body.required else ""
        desc_parts.append(f"\n{request_body.description}{required_marker}")

        # Add request body property descriptions if available
        if request_body.content_schema:
            media_type = (
                "application/json"
                if "application/json" in request_body.content_schema
                else next(iter(request_body.content_schema), None)
            )
            if media_type:
                schema = request_body.content_schema.get(media_type, {})
                if isinstance(schema, dict) and "properties" in schema:
                    desc_parts.append("\n\n**Request Properties:**")
                    for prop_name, prop_schema in schema["properties"].items():
                        if (
                            isinstance(prop_schema, dict)
                            and "description" in prop_schema
                        ):
                            required = prop_name in schema.get("required", [])
                            req_mark = " (Required)" if required else ""
                            desc_parts.append(
                                f"\n- **{prop_name}**{req_mark}: {prop_schema['description']}"
                            )

    # Add response information
    if responses:
        response_section = "\n\n**Responses:**"
        added_response_section = False

        # Determine success codes (common ones)
        success_codes = {"200", "201", "202", "204"}  # As strings
        success_status = next((s for s in success_codes if s in responses), None)

        # Process all responses
        responses_to_process = responses.items()

        for status_code, resp_info in sorted(responses_to_process):
            if not added_response_section:
                desc_parts.append(response_section)
                added_response_section = True

            status_marker = " (Success)" if status_code == success_status else ""
            desc_parts.append(
                f"\n- **{status_code}**{status_marker}: {resp_info.description or 'No description.'}"
            )

            # Process content schemas for this response
            if resp_info.content_schema:
                # Prioritize json, then take first available
                media_type = (
                    "application/json"
                    if "application/json" in resp_info.content_schema
                    else next(iter(resp_info.content_schema), None)
                )

                if media_type:
                    schema = resp_info.content_schema.get(media_type)
                    desc_parts.append(f"  - Content-Type: `{media_type}`")

                    # Add response property descriptions
                    if isinstance(schema, dict):
                        # Handle array responses
                        if schema.get("type") == "array" and "items" in schema:
                            items_schema = schema["items"]
                            if (
                                isinstance(items_schema, dict)
                                and "properties" in items_schema
                            ):
                                desc_parts.append("\n  - **Response Item Properties:**")
                                for prop_name, prop_schema in items_schema[
                                    "properties"
                                ].items():
                                    if (
                                        isinstance(prop_schema, dict)
                                        and "description" in prop_schema
                                    ):
                                        desc_parts.append(
                                            f"\n    - **{prop_name}**: {prop_schema['description']}"
                                        )
                        # Handle object responses
                        elif "properties" in schema:
                            desc_parts.append("\n  - **Response Properties:**")
                            for prop_name, prop_schema in schema["properties"].items():
                                if (
                                    isinstance(prop_schema, dict)
                                    and "description" in prop_schema
                                ):
                                    desc_parts.append(
                                        f"\n    - **{prop_name}**: {prop_schema['description']}"
                                    )

                    # Generate Example
                    if schema:
                        example = generate_example_from_schema(schema)
                        if example != "unknown_type" and example is not None:
                            desc_parts.append("\n  - **Example:**")
                            desc_parts.append(
                                format_json_for_description(example, indent=2)
                            )

    return "\n".join(desc_parts)


def _replace_ref_with_defs(
    info: dict[str, Any], description: str | None = None
) -> dict[str, Any]:
    """
    Replace openapi $ref with jsonschema $defs

    Examples:
    - {"type": "object", "properties": {"$ref": "#/components/schemas/..."}}
    - {"$ref": "#/components/schemas/..."}
    - {"items": {"$ref": "#/components/schemas/..."}}
    - {"anyOf": [{"$ref": "#/components/schemas/..."}]}
    - {"allOf": [{"$ref": "#/components/schemas/..."}]}
    - {"oneOf": [{"$ref": "#/components/schemas/..."}]}

    Args:
        info: dict[str, Any]
        description: str | None

    Returns:
        dict[str, Any]
    """
    schema = info.copy()
    if ref_path := schema.get("$ref"):
        if isinstance(ref_path, str):
            if ref_path.startswith("#/components/schemas/"):
                schema_name = ref_path.split("/")[-1]
                schema["$ref"] = f"#/$defs/{schema_name}"
            elif not ref_path.startswith("#/"):
                raise ValueError(
                    f"External or non-local reference not supported: {ref_path}. "
                    f"FastMCP only supports local schema references starting with '#/'. "
                    f"Please include all schema definitions within the OpenAPI document."
                )
    elif properties := schema.get("properties"):
        if "$ref" in properties:
            schema["properties"] = _replace_ref_with_defs(properties)
        else:
            schema["properties"] = {
                prop_name: _replace_ref_with_defs(prop_schema)
                for prop_name, prop_schema in properties.items()
            }
    elif item_schema := schema.get("items"):
        schema["items"] = _replace_ref_with_defs(item_schema)
    for section in ["anyOf", "allOf", "oneOf"]:
        for i, item in enumerate(schema.get(section, [])):
            schema[section][i] = _replace_ref_with_defs(item)
    if info.get("description", description) and not schema.get("description"):
        schema["description"] = description
    return schema


def _make_optional_parameter_nullable(schema: dict[str, Any]) -> dict[str, Any]:
    """
    Make an optional parameter schema nullable to allow None values.

    For optional parameters, we need to allow null values in addition to the
    specified type to handle cases where None is passed for optional parameters.
    """
    # If schema already has multiple types or is already nullable, don't modify
    if "anyOf" in schema or "oneOf" in schema or "allOf" in schema:
        return schema

    # If it's already nullable (type includes null), don't modify
    if isinstance(schema.get("type"), list) and "null" in schema["type"]:
        return schema

    # Create a new schema that allows null in addition to the original type
    if "type" in schema:
        original_type = schema["type"]

        if isinstance(original_type, str):
            # Single type - make it a union with null
            # Optimize: avoid full schema copy by building directly
            nested_non_nullable_schema = {
                "type": original_type,
            }
            nullable_schema = {}

            # Define type-specific properties that should move to nested schema
            type_specific_properties = set()
            if original_type == "array":
                # https://json-schema.org/understanding-json-schema/reference/array
                type_specific_properties = {
                    "items",
                    "prefixItems",
                    "unevaluatedItems",
                    "contains",
                    "minContains",
                    "maxContains",
                    "minItems",
                    "maxItems",
                    "uniqueItems",
                }
            elif original_type == "object":
                # https://json-schema.org/understanding-json-schema/reference/object
                type_specific_properties = {
                    "properties",
                    "patternProperties",
                    "additionalProperties",
                    "unevaluatedProperties",
                    "required",
                    "propertyNames",
                    "minProperties",
                    "maxProperties",
                }

            # Efficiently distribute properties without copying the entire schema
            for key, value in schema.items():
                if key == "type":
                    continue  # Already handled
                elif key in type_specific_properties:
                    nested_non_nullable_schema[key] = value
                else:
                    nullable_schema[key] = value

            nullable_schema["anyOf"] = [nested_non_nullable_schema, {"type": "null"}]
            return nullable_schema

    return schema


def _add_null_to_type(schema: dict[str, Any]) -> None:
    """Add 'null' to the schema's type field or handle oneOf/anyOf/allOf constructs if not already present."""
    if "type" in schema:
        current_type = schema["type"]

        if isinstance(current_type, str):
            # Convert string type to array with null
            schema["type"] = [current_type, "null"]
        elif isinstance(current_type, list):
            # Add null to array if not already present
            if "null" not in current_type:
                schema["type"] = current_type + ["null"]
    elif "oneOf" in schema:
        # Convert oneOf to anyOf with null type
        schema["anyOf"] = schema.pop("oneOf") + [{"type": "null"}]
    elif "anyOf" in schema:
        # Add null type to anyOf if not already present
        if not any(item.get("type") == "null" for item in schema["anyOf"]):
            schema["anyOf"].append({"type": "null"})
    elif "allOf" in schema:
        # For allOf, wrap in anyOf with null - this means (all conditions) OR null
        schema["anyOf"] = [{"allOf": schema.pop("allOf")}, {"type": "null"}]


def _handle_nullable_fields(schema: dict[str, Any] | Any) -> dict[str, Any] | Any:
    """Convert OpenAPI nullable fields to JSON Schema format: {"type": "string",
    "nullable": true} -> {"type": ["string", "null"]}"""

    if not isinstance(schema, dict):
        return schema

    # Check if we need to modify anything first to avoid unnecessary copying
    has_root_nullable_field = "nullable" in schema
    has_root_nullable_true = (
        has_root_nullable_field
        and schema["nullable"]
        and (
            "type" in schema
            or "oneOf" in schema
            or "anyOf" in schema
            or "allOf" in schema
        )
    )

    has_property_nullable_field = False
    if "properties" in schema:
        for prop_schema in schema["properties"].values():
            if isinstance(prop_schema, dict) and "nullable" in prop_schema:
                has_property_nullable_field = True
                break

    # If no nullable fields at all, return original schema unchanged
    if not has_root_nullable_field and not has_property_nullable_field:
        return schema

    # Only copy if we need to modify
    result = schema.copy()

    # Handle root level nullable - always remove the field, convert type if true
    if has_root_nullable_field:
        result.pop("nullable")
        if has_root_nullable_true:
            _add_null_to_type(result)

    # Handle properties nullable fields
    if has_property_nullable_field and "properties" in result:
        for prop_name, prop_schema in result["properties"].items():
            if isinstance(prop_schema, dict) and "nullable" in prop_schema:
                nullable_value = prop_schema.pop("nullable")
                if nullable_value and (
                    "type" in prop_schema
                    or "oneOf" in prop_schema
                    or "anyOf" in prop_schema
                    or "allOf" in prop_schema
                ):
                    _add_null_to_type(prop_schema)

    return result


def _combine_schemas(route: HTTPRoute) -> dict[str, Any]:
    """
    Combines parameter and request body schemas into a single schema.
    Handles parameter name collisions by adding location suffixes.

    Args:
        route: HTTPRoute object

    Returns:
        Combined schema dictionary
    """
    properties = {}
    required = []

    # First pass: collect parameter names by location and body properties
    param_names_by_location = {
        "path": set(),
        "query": set(),
        "header": set(),
        "cookie": set(),
    }
    body_props = {}

    for param in route.parameters:
        param_names_by_location[param.location].add(param.name)

    if route.request_body and route.request_body.content_schema:
        content_type = next(iter(route.request_body.content_schema))
        body_schema = _replace_ref_with_defs(
            route.request_body.content_schema[content_type].copy(),
            route.request_body.description,
        )
        body_props = body_schema.get("properties", {})

    # Detect collisions: parameters that exist in both body and path/query/header
    all_non_body_params = set()
    for location_params in param_names_by_location.values():
        all_non_body_params.update(location_params)

    body_param_names = set(body_props.keys())
    colliding_params = all_non_body_params & body_param_names

    # Add parameters with suffixes for collisions
    for param in route.parameters:
        if param.name in colliding_params:
            # Add suffix for non-body parameters when collision detected
            suffixed_name = f"{param.name}__{param.location}"
            if param.required:
                required.append(suffixed_name)

            # Add location info to description
            param_schema = _replace_ref_with_defs(
                param.schema_.copy(), param.description
            )
            original_desc = param_schema.get("description", "")
            location_desc = f"({param.location.capitalize()} parameter)"
            if original_desc:
                param_schema["description"] = f"{original_desc} {location_desc}"
            else:
                param_schema["description"] = location_desc

            # Don't make optional parameters nullable - they can simply be omitted
            # The OpenAPI specification doesn't require optional parameters to accept null values

            properties[suffixed_name] = param_schema
        else:
            # No collision, use original name
            if param.required:
                required.append(param.name)
            param_schema = _replace_ref_with_defs(
                param.schema_.copy(), param.description
            )

            # Don't make optional parameters nullable - they can simply be omitted
            # The OpenAPI specification doesn't require optional parameters to accept null values

            properties[param.name] = param_schema

    # Add request body properties (no suffixes for body parameters)
    if route.request_body and route.request_body.content_schema:
        for prop_name, prop_schema in body_props.items():
            properties[prop_name] = prop_schema

        if route.request_body.required:
            required.extend(body_schema.get("required", []))

    result = {
        "type": "object",
        "properties": properties,
        "required": required,
    }
    # Add schema definitions if available
    if route.schema_definitions:
        result["$defs"] = route.schema_definitions.copy()

    # Use lightweight compression - prune additionalProperties and unused definitions
    if result.get("additionalProperties") is False:
        result.pop("additionalProperties")

    # Remove unused definitions (lightweight approach - just check direct $ref usage)
    if "$defs" in result:
        used_refs = set()

        def find_refs_in_value(value):
            if isinstance(value, dict):
                if "$ref" in value and isinstance(value["$ref"], str):
                    ref = value["$ref"]
                    if ref.startswith("#/$defs/"):
                        used_refs.add(ref.split("/")[-1])
                for v in value.values():
                    find_refs_in_value(v)
            elif isinstance(value, list):
                for item in value:
                    find_refs_in_value(item)

        # Find refs in the main schema (excluding $defs section)
        for key, value in result.items():
            if key != "$defs":
                find_refs_in_value(value)

        # Remove unused definitions
        if used_refs:
            result["$defs"] = {
                name: def_schema
                for name, def_schema in result["$defs"].items()
                if name in used_refs
            }
        else:
            result.pop("$defs")

    return result


def _adjust_union_types(
    schema: dict[str, Any] | list[Any],
) -> dict[str, Any] | list[Any]:
    """Recursively replace 'oneOf' with 'anyOf' in schema to handle overlapping unions."""
    if isinstance(schema, dict):
        # Optimize: only copy if we need to modify something
        has_one_of = "oneOf" in schema
        needs_recursive_processing = False

        # Check if we need recursive processing
        for v in schema.values():
            if isinstance(v, dict | list):
                needs_recursive_processing = True
                break

        # If nothing to change, return original
        if not has_one_of and not needs_recursive_processing:
            return schema

        # Work on a copy only when modification is needed
        result = schema.copy()
        if has_one_of:
            result["anyOf"] = result.pop("oneOf")

        # Only recurse where needed
        if needs_recursive_processing:
            for k, v in result.items():
                if isinstance(v, dict | list):
                    result[k] = _adjust_union_types(v)

        return result
    elif isinstance(schema, list):
        return [_adjust_union_types(item) for item in schema]
    return schema


def extract_output_schema_from_responses(
    responses: dict[str, ResponseInfo],
    schema_definitions: dict[str, Any] | None = None,
    openapi_version: str | None = None,
) -> dict[str, Any] | None:
    """
    Extract output schema from OpenAPI responses for use as MCP tool output schema.

    This function finds the first successful response (200, 201, 202, 204) with a
    JSON-compatible content type and extracts its schema. If the schema is not an
    object type, it wraps it to comply with MCP requirements.

    Args:
        responses: Dictionary of ResponseInfo objects keyed by status code
        schema_definitions: Optional schema definitions to include in the output schema
        openapi_version: OpenAPI version string, used to optimize nullable field handling

    Returns:
        dict: MCP-compliant output schema with potential wrapping, or None if no suitable schema found
    """
    if not responses:
        return None

    # Priority order for success status codes
    success_codes = ["200", "201", "202", "204"]

    # Find the first successful response
    response_info = None
    for status_code in success_codes:
        if status_code in responses:
            response_info = responses[status_code]
            break

    # If no explicit success codes, try any 2xx response
    if response_info is None:
        for status_code, resp_info in responses.items():
            if status_code.startswith("2"):
                response_info = resp_info
                break

    if response_info is None or not response_info.content_schema:
        return None

    # Prefer application/json, then fall back to other JSON-compatible types
    json_compatible_types = [
        "application/json",
        "application/vnd.api+json",
        "application/hal+json",
        "application/ld+json",
        "text/json",
    ]

    schema = None
    for content_type in json_compatible_types:
        if content_type in response_info.content_schema:
            schema = response_info.content_schema[content_type]
            break

    # If no JSON-compatible type found, try the first available content type
    if schema is None and response_info.content_schema:
        first_content_type = next(iter(response_info.content_schema))
        schema = response_info.content_schema[first_content_type]
        logger.debug(
            f"Using non-JSON content type for output schema: {first_content_type}"
        )

    if not schema or not isinstance(schema, dict):
        return None

    # Clean and copy the schema
    output_schema = schema.copy()

    # If schema has a $ref, resolve it first before processing nullable fields
    if "$ref" in output_schema and schema_definitions:
        ref_path = output_schema["$ref"]
        if ref_path.startswith("#/components/schemas/"):
            schema_name = ref_path.split("/")[-1]
            if schema_name in schema_definitions:
                # Replace $ref with the actual schema definition
                output_schema = schema_definitions[schema_name].copy()

    # Handle OpenAPI nullable fields by converting them to JSON Schema format
    # This prevents "None is not of type 'string'" validation errors
    # Only needed for OpenAPI 3.0 - 3.1 uses standard JSON Schema null types
    if openapi_version and openapi_version.startswith("3.0"):
        output_schema = _handle_nullable_fields(output_schema)

    # MCP requires output schemas to be objects. If this schema is not an object,
    # we need to wrap it similar to how ParsedFunction.from_function() does it
    if output_schema.get("type") != "object":
        # Create a wrapped schema that contains the original schema under a "result" key
        wrapped_schema = {
            "type": "object",
            "properties": {"result": output_schema},
            "required": ["result"],
            "x-fastmcp-wrap-result": True,
        }
        output_schema = wrapped_schema

    # Add schema definitions if available and handle nullable fields in them
    # Only add $defs if we didn't resolve the $ref inline above
    if schema_definitions and "$ref" not in schema.copy():
        processed_defs = {}
        for def_name, def_schema in schema_definitions.items():
            # Only handle nullable fields for OpenAPI 3.0 - 3.1 uses standard JSON Schema null types
            if openapi_version and openapi_version.startswith("3.0"):
                processed_defs[def_name] = _handle_nullable_fields(def_schema)
            else:
                processed_defs[def_name] = def_schema
        output_schema["$defs"] = processed_defs

    # Use lightweight compression - prune additionalProperties and unused definitions
    if output_schema.get("additionalProperties") is False:
        output_schema.pop("additionalProperties")

    # Remove unused definitions (lightweight approach - just check direct $ref usage)
    if "$defs" in output_schema:
        used_refs = set()

        def find_refs_in_value(value):
            if isinstance(value, dict):
                if "$ref" in value and isinstance(value["$ref"], str):
                    ref = value["$ref"]
                    if ref.startswith("#/$defs/"):
                        used_refs.add(ref.split("/")[-1])
                for v in value.values():
                    find_refs_in_value(v)
            elif isinstance(value, list):
                for item in value:
                    find_refs_in_value(item)

        # Find refs in the main schema (excluding $defs section)
        for key, value in output_schema.items():
            if key != "$defs":
                find_refs_in_value(value)

        # Remove unused definitions
        if used_refs:
            output_schema["$defs"] = {
                name: def_schema
                for name, def_schema in output_schema["$defs"].items()
                if name in used_refs
            }
        else:
            output_schema.pop("$defs")

    # Adjust union types to handle overlapping unions
    output_schema = cast(dict[str, Any], _adjust_union_types(output_schema))

    return output_schema