Improve OpenAPI-to-JSONSchema conversion utilities (#1283)

src/fastmcp/experimental/server/openapi/server.py

@@ -5,7 +5,7 @@ from collections import Counter
 from typing import Any, Literal
 
 import httpx
-from openapi_core import Spec
+from jsonschema_path import SchemaPath
 
 # Import from our new utilities and components
 from fastmcp.experimental.utilities.openapi import (
@@ -149,7 +149,7 @@ class FastMCPOpenAPI(FastMCP):
 
         # Create openapi-core Spec and RequestDirector for stateless request building
         try:
-            self._spec = Spec.from_dict(openapi_spec)  # type: ignore[arg-type]
+            self._spec = SchemaPath.from_dict(openapi_spec)  # type: ignore[arg-type]
             self._director = RequestDirector(self._spec)
             logger.debug(
                 "Initialized OpenAPI RequestDirector for stateless request building"
@@ -270,7 +270,7 @@ class FastMCPOpenAPI(FastMCP):
 
         # Extract output schema from OpenAPI responses
         output_schema = extract_output_schema_from_responses(
-            route.responses, route.schema_definitions
+            route.responses, route.schema_definitions, route.openapi_version
         )
 
         # Get a unique tool name

src/fastmcp/experimental/utilities/openapi/README.md

@@ -36,13 +36,13 @@ The new implementation follows a **stateless request building strategy** using `
 ### Initialization Process
 
 ```
-OpenAPI Spec → Parser → HTTPRoute with Pre-calculated Fields → RequestDirector + openapi-core Spec
+OpenAPI Spec → Parser → HTTPRoute with Pre-calculated Fields → RequestDirector + SchemaPath
 ```
 
 1. **Input**: Raw OpenAPI specification (dict)
 2. **Parsing**: Extract operations to `HTTPRoute` models
 3. **Pre-calculation**: Generate combined schemas and parameter maps during parsing
-4. **Director Setup**: Create `RequestDirector` with `openapi-core` Spec for request building
+4. **Director Setup**: Create `RequestDirector` with `SchemaPath` for request building
 
 ### Request Processing
 
@@ -125,10 +125,10 @@ async with httpx.AsyncClient() as client:
 
 ```python
 from fastmcp.experimental.utilities.openapi.director import RequestDirector
-from openapi_core import Spec
+from jsonschema_path import SchemaPath
 
 # Create RequestDirector manually
-spec = Spec.from_dict(openapi_spec)
+spec = SchemaPath.from_dict(openapi_spec)
 director = RequestDirector(spec)
 
 # Build HTTP request
@@ -210,7 +210,7 @@ Tests are located in `/tests/server/openapi_new/`:
 ### Common Issues
 
 1. **RequestDirector Initialization Fails**
-   - Check OpenAPI spec validity with `openapi-core`
+   - Check OpenAPI spec validity with `jsonschema-path`
    - Verify spec format is correct JSON/YAML
    - Ensure all required OpenAPI fields are present
 

src/fastmcp/experimental/utilities/openapi/__init__.py

@@ -30,7 +30,12 @@ from .schemas import (
     clean_schema_for_display,
     _replace_ref_with_defs,
     _make_optional_parameter_nullable,
-    _adjust_union_types,
+)
+
+# Import from json_schema_converter
+from .json_schema_converter import (
+    convert_openapi_schema_to_json_schema,
+    convert_schema_definitions,
 )
 
 # Export public symbols - maintaining backward compatibility
@@ -57,5 +62,7 @@ __all__ = [
     "clean_schema_for_display",
     "_replace_ref_with_defs",
     "_make_optional_parameter_nullable",
-    "_adjust_union_types",
+    # JSON Schema Converter
+    "convert_openapi_schema_to_json_schema",
+    "convert_schema_definitions",
 ]

src/fastmcp/experimental/utilities/openapi/director.py

@@ -4,7 +4,7 @@ from typing import Any
 from urllib.parse import urljoin
 
 import httpx
-from openapi_core import Spec
+from jsonschema_path import SchemaPath
 
 from fastmcp.utilities.logging import get_logger
 
@@ -16,8 +16,8 @@ logger = get_logger(__name__)
 class RequestDirector:
     """Builds httpx.Request objects from HTTPRoute and arguments using openapi-core."""
 
-    def __init__(self, spec: Spec):
-        """Initialize with a parsed openapi-core Spec object."""
+    def __init__(self, spec: SchemaPath):
+        """Initialize with a parsed SchemaPath object."""
         self._spec = spec
 
     def build(
@@ -66,7 +66,7 @@ class RequestDirector:
             if isinstance(body, dict) or isinstance(body, list):
                 request_data["json"] = body
             else:
-                request_data["data"] = body
+                request_data["content"] = body
 
         # Step 5: Create httpx.Request
         return httpx.Request(**{k: v for k, v in request_data.items() if v is not None})

src/fastmcp/experimental/utilities/openapi/json_schema_converter.py

@@ -0,0 +1,339 @@
+"""
+Clean OpenAPI 3.0 to JSON Schema converter for the experimental parser.
+
+This module provides a systematic approach to converting OpenAPI 3.0 schemas
+to JSON Schema, inspired by py-openapi-schema-to-json-schema but optimized
+for our specific use case.
+"""
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# OpenAPI-specific fields that should be removed from JSON Schema
+OPENAPI_SPECIFIC_FIELDS = {
+    "nullable",  # Handled by converting to type arrays
+    "discriminator",  # OpenAPI-specific
+    "readOnly",  # OpenAPI-specific metadata
+    "writeOnly",  # OpenAPI-specific metadata
+    "xml",  # OpenAPI-specific metadata
+    "externalDocs",  # OpenAPI-specific metadata
+    "deprecated",  # Can be kept but not part of JSON Schema core
+}
+
+# Fields that should be recursively processed
+RECURSIVE_FIELDS = {
+    "properties": dict,
+    "items": dict,
+    "additionalProperties": dict,
+    "allOf": list,
+    "anyOf": list,
+    "oneOf": list,
+    "not": dict,
+}
+
+
+def convert_openapi_schema_to_json_schema(
+    schema: dict[str, Any],
+    openapi_version: str | None = None,
+    remove_read_only: bool = False,
+    remove_write_only: bool = False,
+    convert_one_of_to_any_of: bool = True,
+) -> dict[str, Any]:
+    """
+    Convert an OpenAPI schema to JSON Schema format.
+
+    This is a clean, systematic approach that:
+    1. Removes OpenAPI-specific fields
+    2. Converts nullable fields to type arrays (for OpenAPI 3.0 only)
+    3. Converts oneOf to anyOf for overlapping union handling
+    4. Recursively processes nested schemas
+    5. Optionally removes readOnly/writeOnly properties
+
+    Args:
+        schema: OpenAPI schema dictionary
+        openapi_version: OpenAPI version for optimization
+        remove_read_only: Whether to remove readOnly properties
+        remove_write_only: Whether to remove writeOnly properties
+        convert_one_of_to_any_of: Whether to convert oneOf to anyOf
+
+    Returns:
+        JSON Schema compatible dictionary
+    """
+    if not isinstance(schema, dict):
+        return schema
+
+    # Early exit optimization - check if conversion is needed
+    needs_conversion = (
+        any(field in schema for field in OPENAPI_SPECIFIC_FIELDS)
+        or (remove_read_only and _has_read_only_properties(schema))
+        or (remove_write_only and _has_write_only_properties(schema))
+        or (convert_one_of_to_any_of and "oneOf" in schema)
+        or _needs_recursive_processing(
+            schema,
+            openapi_version,
+            remove_read_only,
+            remove_write_only,
+            convert_one_of_to_any_of,
+        )
+    )
+
+    if not needs_conversion:
+        return schema
+
+    # Work on a copy to avoid mutation
+    result = schema.copy()
+
+    # Step 1: Handle nullable field conversion (OpenAPI 3.0 only)
+    if openapi_version and openapi_version.startswith("3.0"):
+        result = _convert_nullable_field(result)
+
+    # Step 2: Convert oneOf to anyOf if requested
+    if convert_one_of_to_any_of and "oneOf" in result:
+        result["anyOf"] = result.pop("oneOf")
+
+    # Step 3: Remove OpenAPI-specific fields
+    for field in OPENAPI_SPECIFIC_FIELDS:
+        result.pop(field, None)
+
+    # Step 4: Handle readOnly/writeOnly property removal
+    if remove_read_only or remove_write_only:
+        result = _filter_properties_by_access(
+            result, remove_read_only, remove_write_only
+        )
+
+    # Step 5: Recursively process nested schemas
+    for field_name, field_type in RECURSIVE_FIELDS.items():
+        if field_name in result:
+            if field_type is dict and isinstance(result[field_name], dict):
+                if field_name == "properties":
+                    # Handle properties specially - each property is a schema
+                    result[field_name] = {
+                        prop_name: convert_openapi_schema_to_json_schema(
+                            prop_schema,
+                            openapi_version,
+                            remove_read_only,
+                            remove_write_only,
+                            convert_one_of_to_any_of,
+                        )
+                        if isinstance(prop_schema, dict)
+                        else prop_schema
+                        for prop_name, prop_schema in result[field_name].items()
+                    }
+                else:
+                    result[field_name] = convert_openapi_schema_to_json_schema(
+                        result[field_name],
+                        openapi_version,
+                        remove_read_only,
+                        remove_write_only,
+                        convert_one_of_to_any_of,
+                    )
+            elif field_type is list and isinstance(result[field_name], list):
+                result[field_name] = [
+                    convert_openapi_schema_to_json_schema(
+                        item,
+                        openapi_version,
+                        remove_read_only,
+                        remove_write_only,
+                        convert_one_of_to_any_of,
+                    )
+                    if isinstance(item, dict)
+                    else item
+                    for item in result[field_name]
+                ]
+
+    return result
+
+
+def _convert_nullable_field(schema: dict[str, Any]) -> dict[str, Any]:
+    """Convert OpenAPI nullable field to JSON Schema type array."""
+    if "nullable" not in schema:
+        return schema
+
+    result = schema.copy()
+    nullable_value = result.pop("nullable")
+
+    # Only convert if nullable is True and we have a type structure
+    if not nullable_value:
+        return result
+
+    if "type" in result:
+        current_type = result["type"]
+        if isinstance(current_type, str):
+            result["type"] = [current_type, "null"]
+        elif isinstance(current_type, list) and "null" not in current_type:
+            result["type"] = current_type + ["null"]
+    elif "oneOf" in result:
+        # Convert oneOf to anyOf with null
+        result["anyOf"] = result.pop("oneOf") + [{"type": "null"}]
+    elif "anyOf" in result:
+        # Add null to anyOf if not present
+        if not any(item.get("type") == "null" for item in result["anyOf"]):
+            result["anyOf"].append({"type": "null"})
+    elif "allOf" in result:
+        # Wrap allOf in anyOf with null option
+        result["anyOf"] = [{"allOf": result.pop("allOf")}, {"type": "null"}]
+
+    return result
+
+
+def _has_read_only_properties(schema: dict[str, Any]) -> bool:
+    """Quick check if schema has any readOnly properties."""
+    if "properties" not in schema:
+        return False
+    return any(
+        isinstance(prop, dict) and prop.get("readOnly")
+        for prop in schema["properties"].values()
+    )
+
+
+def _has_write_only_properties(schema: dict[str, Any]) -> bool:
+    """Quick check if schema has any writeOnly properties."""
+    if "properties" not in schema:
+        return False
+    return any(
+        isinstance(prop, dict) and prop.get("writeOnly")
+        for prop in schema["properties"].values()
+    )
+
+
+def _needs_recursive_processing(
+    schema: dict[str, Any],
+    openapi_version: str | None,
+    remove_read_only: bool,
+    remove_write_only: bool,
+    convert_one_of_to_any_of: bool,
+) -> bool:
+    """Check if the schema needs recursive processing (smarter than just checking for recursive fields)."""
+    for field_name, field_type in RECURSIVE_FIELDS.items():
+        if field_name in schema:
+            if field_type is dict and isinstance(schema[field_name], dict):
+                if field_name == "properties":
+                    # Check if any property needs conversion
+                    for prop_schema in schema[field_name].values():
+                        if isinstance(prop_schema, dict):
+                            nested_needs_conversion = (
+                                any(
+                                    field in prop_schema
+                                    for field in OPENAPI_SPECIFIC_FIELDS
+                                )
+                                or (remove_read_only and prop_schema.get("readOnly"))
+                                or (remove_write_only and prop_schema.get("writeOnly"))
+                                or (convert_one_of_to_any_of and "oneOf" in prop_schema)
+                                or _needs_recursive_processing(
+                                    prop_schema,
+                                    openapi_version,
+                                    remove_read_only,
+                                    remove_write_only,
+                                    convert_one_of_to_any_of,
+                                )
+                            )
+                            if nested_needs_conversion:
+                                return True
+                else:
+                    # Check if nested schema needs conversion
+                    nested_needs_conversion = (
+                        any(
+                            field in schema[field_name]
+                            for field in OPENAPI_SPECIFIC_FIELDS
+                        )
+                        or (
+                            remove_read_only
+                            and _has_read_only_properties(schema[field_name])
+                        )
+                        or (
+                            remove_write_only
+                            and _has_write_only_properties(schema[field_name])
+                        )
+                        or (convert_one_of_to_any_of and "oneOf" in schema[field_name])
+                        or _needs_recursive_processing(
+                            schema[field_name],
+                            openapi_version,
+                            remove_read_only,
+                            remove_write_only,
+                            convert_one_of_to_any_of,
+                        )
+                    )
+                    if nested_needs_conversion:
+                        return True
+            elif field_type is list and isinstance(schema[field_name], list):
+                # Check if any list item needs conversion
+                for item in schema[field_name]:
+                    if isinstance(item, dict):
+                        nested_needs_conversion = (
+                            any(field in item for field in OPENAPI_SPECIFIC_FIELDS)
+                            or (remove_read_only and _has_read_only_properties(item))
+                            or (remove_write_only and _has_write_only_properties(item))
+                            or (convert_one_of_to_any_of and "oneOf" in item)
+                            or _needs_recursive_processing(
+                                item,
+                                openapi_version,
+                                remove_read_only,
+                                remove_write_only,
+                                convert_one_of_to_any_of,
+                            )
+                        )
+                        if nested_needs_conversion:
+                            return True
+    return False
+
+
+def _filter_properties_by_access(
+    schema: dict[str, Any], remove_read_only: bool, remove_write_only: bool
+) -> dict[str, Any]:
+    """Remove readOnly and/or writeOnly properties from schema."""
+    if "properties" not in schema:
+        return schema
+
+    result = schema.copy()
+    filtered_properties = {}
+
+    for prop_name, prop_schema in result["properties"].items():
+        if not isinstance(prop_schema, dict):
+            filtered_properties[prop_name] = prop_schema
+            continue
+
+        should_remove = (remove_read_only and prop_schema.get("readOnly")) or (
+            remove_write_only and prop_schema.get("writeOnly")
+        )
+
+        if not should_remove:
+            filtered_properties[prop_name] = prop_schema
+
+    result["properties"] = filtered_properties
+
+    # Clean up required array if properties were removed
+    if "required" in result and filtered_properties:
+        result["required"] = [
+            prop for prop in result["required"] if prop in filtered_properties
+        ]
+        if not result["required"]:
+            result.pop("required")
+
+    return result
+
+
+def convert_schema_definitions(
+    schema_definitions: dict[str, Any] | None,
+    openapi_version: str | None = None,
+    **kwargs,
+) -> dict[str, Any]:
+    """
+    Convert a dictionary of OpenAPI schema definitions to JSON Schema.
+
+    Args:
+        schema_definitions: Dictionary of schema definitions
+        openapi_version: OpenAPI version for optimization
+        **kwargs: Additional arguments passed to convert_openapi_schema_to_json_schema
+
+    Returns:
+        Dictionary of converted schema definitions
+    """
+    if not schema_definitions:
+        return {}
+
+    return {
+        name: convert_openapi_schema_to_json_schema(schema, openapi_version, **kwargs)
+        for name, schema in schema_definitions.items()
+    }

src/fastmcp/experimental/utilities/openapi/models.py

@@ -62,6 +62,7 @@ class HTTPRoute(FastMCPBaseModel):
         default_factory=dict
     )  # Store component schemas
     extensions: dict[str, Any] = Field(default_factory=dict)
+    openapi_version: str | None = None
 
     # Pre-calculated fields for performance
     flat_param_schema: JsonSchema = Field(

src/fastmcp/experimental/utilities/openapi/parser.py

@@ -74,6 +74,7 @@ def parse_openapi_to_http_routes(openapi_dict: dict[str, Any]) -> list[HTTPRoute
                 Response_30,
                 Operation_30,
                 PathItem_30,
+                openapi_version,
             )
             return parser.parse()
         else:
@@ -91,6 +92,7 @@ def parse_openapi_to_http_routes(openapi_dict: dict[str, Any]) -> list[HTTPRoute
                 Response,
                 Operation,
                 PathItem,
+                openapi_version,
             )
             return parser.parse()
     except ValidationError as e:
@@ -124,6 +126,7 @@ class OpenAPIParser(
         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
@@ -134,6 +137,7 @@ class OpenAPIParser(
         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."""
@@ -560,6 +564,7 @@ class OpenAPIParser(
                             responses=responses,
                             schema_definitions=schema_definitions,
                             extensions=extensions,
+                            openapi_version=self.openapi_version,
                         )
 
                         # Pre-calculate schema and parameter mapping for performance

src/fastmcp/experimental/utilities/openapi/schemas.py

@@ -1,7 +1,7 @@
 """Schema manipulation utilities for OpenAPI operations."""
 
 import logging
-from typing import Any, cast
+from typing import Any
 
 from .models import HTTPRoute, JsonSchema, ResponseInfo
 
@@ -199,86 +199,6 @@ def _make_optional_parameter_nullable(schema: dict[str, Any]) -> dict[str, Any]:
     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_and_map_params(
     route: HTTPRoute,
 ) -> tuple[dict[str, Any], dict[str, dict[str, str]]]:
@@ -450,68 +370,10 @@ def _combine_schemas(route: HTTPRoute) -> dict[str, Any]:
     return schema
 
 
-def _has_one_of(obj: dict[str, Any] | list[Any]) -> bool:
-    """Quickly check if schema contains any 'oneOf' keys without deep traversal."""
-    if isinstance(obj, dict):
-        if "oneOf" in obj:
-            return True
-        # Only check likely schema containers, skip examples/defaults
-        for k, v in obj.items():
-            if k in [
-                "properties",
-                "items",
-                "allOf",
-                "anyOf",
-                "additionalProperties",
-            ] and isinstance(v, dict | list):
-                if _has_one_of(v):
-                    return True
-    elif isinstance(obj, list):
-        for item in obj:
-            if isinstance(item, dict | list) and _has_one_of(item):
-                return True
-    return False
-
-
-def _adjust_union_types(
-    schema: dict[str, Any] | list[Any], _depth: int = 0
-) -> dict[str, Any] | list[Any]:
-    """Recursively replace 'oneOf' with 'anyOf' in schema to handle overlapping unions."""
-    # MAJOR OPTIMIZATION: Skip entirely if schema has no oneOf keys
-    if _depth == 0 and not _has_one_of(schema):
-        return schema
-
-    # OPTIMIZATION: Early termination for very deep structures to prevent exponential slowdown
-    if _depth > 30:  # Reduced from 50 for better performance
-        return schema
-
-    if isinstance(schema, dict):
-        # Work on a copy to avoid mutating the input
-        result = schema.copy()
-        if "oneOf" in result:
-            result["anyOf"] = result.pop("oneOf")
-        # OPTIMIZATION: Only recurse into values that could contain more schemas
-        for k, v in result.items():
-            if isinstance(v, dict | list) and k not in [
-                "examples",
-                "example",
-                "default",
-            ]:
-                result[k] = _adjust_union_types(v, _depth + 1)
-        return result
-    elif isinstance(schema, list):
-        # Process list items without mutating the input list
-        return [
-            _adjust_union_types(item, _depth + 1)
-            if isinstance(item, dict | list)
-            else item
-            for item in schema
-        ]
-    return schema
-
-
 def extract_output_schema_from_responses(
-    responses: dict[str, ResponseInfo], schema_definitions: dict[str, Any] | None = None
+    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.
@@ -523,6 +385,7 @@ def extract_output_schema_from_responses(
     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
@@ -588,9 +451,14 @@ def extract_output_schema_from_responses(
                 # 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
-    output_schema = _handle_nullable_fields(output_schema)
+    # Convert OpenAPI schema to JSON Schema format
+    # Only needed for OpenAPI 3.0 - 3.1 uses standard JSON Schema null types
+    if openapi_version and openapi_version.startswith("3.0"):
+        from .json_schema_converter import convert_openapi_schema_to_json_schema
+
+        output_schema = convert_openapi_schema_to_json_schema(
+            output_schema, openapi_version
+        )
 
     # 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
@@ -609,7 +477,15 @@ def extract_output_schema_from_responses(
     if schema_definitions and "$ref" not in schema.copy():
         processed_defs = {}
         for def_name, def_schema in schema_definitions.items():
-            processed_defs[def_name] = _handle_nullable_fields(def_schema)
+            # Convert OpenAPI schema definitions to JSON Schema format
+            if openapi_version and openapi_version.startswith("3.0"):
+                from .json_schema_converter import convert_openapi_schema_to_json_schema
+
+                processed_defs[def_name] = convert_openapi_schema_to_json_schema(
+                    def_schema, openapi_version
+                )
+            else:
+                processed_defs[def_name] = def_schema
         output_schema["$defs"] = processed_defs
 
     # Use lightweight compression - prune additionalProperties and unused definitions
@@ -647,9 +523,6 @@ def extract_output_schema_from_responses(
         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
 
 
@@ -661,6 +534,4 @@ __all__ = [
     "extract_output_schema_from_responses",
     "_replace_ref_with_defs",
     "_make_optional_parameter_nullable",
-    "_handle_nullable_fields",
-    "_adjust_union_types",
 ]

src/fastmcp/server/openapi.py

@@ -892,7 +892,7 @@ class FastMCPOpenAPI(FastMCP):
 
         # Extract output schema from OpenAPI responses
         output_schema = extract_output_schema_from_responses(
-            route.responses, route.schema_definitions
+            route.responses, route.schema_definitions, route.openapi_version
         )
 
         # Get a unique tool name

src/fastmcp/utilities/openapi.py

@@ -173,6 +173,7 @@ class HTTPRoute(FastMCPBaseModel):
         default_factory=dict
     )  # Store component schemas
     extensions: dict[str, Any] = Field(default_factory=dict)
+    openapi_version: str | None = None
 
 
 # Export public symbols
@@ -227,6 +228,7 @@ def parse_openapi_to_http_routes(openapi_dict: dict[str, Any]) -> list[HTTPRoute
                 Response_30,
                 Operation_30,
                 PathItem_30,
+                openapi_version,
             )
             return parser.parse()
         else:
@@ -244,6 +246,7 @@ def parse_openapi_to_http_routes(openapi_dict: dict[str, Any]) -> list[HTTPRoute
                 Response,
                 Operation,
                 PathItem,
+                openapi_version,
             )
             return parser.parse()
     except ValidationError as e:
@@ -277,6 +280,7 @@ class OpenAPIParser(
         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
@@ -287,6 +291,7 @@ class OpenAPIParser(
         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."""
@@ -709,6 +714,7 @@ class OpenAPIParser(
                             responses=responses,
                             schema_definitions=schema_definitions,
                             extensions=extensions,
+                            openapi_version=self.openapi_version,
                         )
                         routes.append(route)
                         logger.info(
@@ -1401,7 +1407,9 @@ def _adjust_union_types(
 
 
 def extract_output_schema_from_responses(
-    responses: dict[str, ResponseInfo], schema_definitions: dict[str, Any] | None = None
+    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.
@@ -1413,6 +1421,7 @@ def extract_output_schema_from_responses(
     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
@@ -1480,7 +1489,9 @@ def extract_output_schema_from_responses(
 
     # Handle OpenAPI nullable fields by converting them to JSON Schema format
     # This prevents "None is not of type 'string'" validation errors
-    output_schema = _handle_nullable_fields(output_schema)
+    # 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
@@ -1499,7 +1510,11 @@ def extract_output_schema_from_responses(
     if schema_definitions and "$ref" not in schema.copy():
         processed_defs = {}
         for def_name, def_schema in schema_definitions.items():
-            processed_defs[def_name] = _handle_nullable_fields(def_schema)
+            # 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

tests/experimental/utilities/openapi/test_director.py

@@ -1,7 +1,7 @@
 """Unit tests for RequestDirector."""
 
 import pytest
-from openapi_core import Spec
+from jsonschema_path import SchemaPath
 
 from fastmcp.experimental.utilities.openapi.director import RequestDirector
 from fastmcp.experimental.utilities.openapi.models import (
@@ -148,12 +148,12 @@ class TestRequestDirector:
     @pytest.fixture
     def director(self, basic_openapi_30_spec):
         """Create a RequestDirector instance."""
-        spec = Spec.from_dict(basic_openapi_30_spec)
+        spec = SchemaPath.from_dict(basic_openapi_30_spec)
         return RequestDirector(spec)
 
     def test_director_initialization(self, basic_openapi_30_spec):
         """Test RequestDirector initialization."""
-        spec = Spec.from_dict(basic_openapi_30_spec)
+        spec = SchemaPath.from_dict(basic_openapi_30_spec)
         director = RequestDirector(spec)
 
         assert director._spec is not None
@@ -350,7 +350,7 @@ class TestRequestDirector:
         request = director.build(route, flat_args, "https://api.example.com")
 
         assert request.method == "POST"
-        # For non-JSON content, httpx uses 'data' parameter which becomes bytes
+        # For non-JSON content, httpx uses 'content' parameter which becomes bytes
         assert request.content == b"Hello, World!"
 
     def test_body_construction_multiple_properties_non_object_schema(self, director):
@@ -392,7 +392,7 @@ class TestRequestDirectorIntegration:
         assert len(routes) == 1
 
         route = routes[0]
-        spec = Spec.from_dict(basic_openapi_30_spec)
+        spec = SchemaPath.from_dict(basic_openapi_30_spec)
         director = RequestDirector(spec)
 
         flat_args = {"id": 42}
@@ -407,7 +407,7 @@ class TestRequestDirectorIntegration:
         assert len(routes) == 1
 
         route = routes[0]
-        spec = Spec.from_dict(collision_spec)
+        spec = SchemaPath.from_dict(collision_spec)
         director = RequestDirector(spec)
 
         # Use the parameter names from the actual parameter map
@@ -440,7 +440,7 @@ class TestRequestDirectorIntegration:
         assert len(routes) == 1
 
         route = routes[0]
-        spec = Spec.from_dict(deepobject_spec)
+        spec = SchemaPath.from_dict(deepobject_spec)
         director = RequestDirector(spec)
 
         # DeepObject parameters should be flattened in the parameter map

tests/experimental/utilities/openapi/test_nullable_fields.py

@@ -1,6 +1,8 @@
 """Tests for nullable field handling in OpenAPI schemas."""
 
-from fastmcp.experimental.utilities.openapi.schemas import _handle_nullable_fields
+from fastmcp.experimental.utilities.openapi.json_schema_converter import (
+    convert_openapi_schema_to_json_schema,
+)
 
 
 class TestHandleNullableFields:
@@ -10,21 +12,21 @@ class TestHandleNullableFields:
         """Test nullable string at root level."""
         input_schema = {"type": "string", "nullable": True}
         expected = {"type": ["string", "null"]}
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_root_level_nullable_integer(self):
         """Test nullable integer at root level."""
         input_schema = {"type": "integer", "nullable": True}
         expected = {"type": ["integer", "null"]}
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_root_level_nullable_boolean(self):
         """Test nullable boolean at root level."""
         input_schema = {"type": "boolean", "nullable": True}
         expected = {"type": ["boolean", "null"]}
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_property_level_nullable_fields(self):
@@ -47,7 +49,7 @@ class TestHandleNullableFields:
                 "active": {"type": ["boolean", "null"]},
             },
         }
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_mixed_nullable_and_non_nullable(self):
@@ -70,14 +72,14 @@ class TestHandleNullableFields:
             },
             "required": ["required_field"],
         }
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_nullable_false_ignored(self):
         """Test that nullable: false is ignored (removed but no type change)."""
         input_schema = {"type": "string", "nullable": False}
         expected = {"type": "string"}
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_no_nullable_field_unchanged(self):
@@ -87,14 +89,14 @@ class TestHandleNullableFields:
             "properties": {"name": {"type": "string"}},
         }
         expected = input_schema.copy()
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_nullable_without_type_removes_nullable(self):
         """Test that nullable field is removed even without type."""
         input_schema = {"nullable": True, "description": "Some field"}
         expected = {"description": "Some field"}
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_preserves_other_fields(self):
@@ -112,15 +114,15 @@ class TestHandleNullableFields:
             "example": "test",
             "format": "email",
         }
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_non_dict_input_unchanged(self):
         """Test that non-dict inputs are returned unchanged."""
-        assert _handle_nullable_fields("string") == "string"  # type: ignore[arg-type]
-        assert _handle_nullable_fields(123) == 123  # type: ignore[arg-type]
-        assert _handle_nullable_fields(None) is None  # type: ignore[arg-type]
-        assert _handle_nullable_fields([1, 2, 3]) == [1, 2, 3]  # type: ignore[arg-type]
+        assert convert_openapi_schema_to_json_schema("string", "3.0.0") == "string"  # type: ignore[arg-type]
+        assert convert_openapi_schema_to_json_schema(123, "3.0.0") == 123  # type: ignore[arg-type]
+        assert convert_openapi_schema_to_json_schema(None, "3.0.0") is None  # type: ignore[arg-type]
+        assert convert_openapi_schema_to_json_schema([1, 2, 3], "3.0.0") == [1, 2, 3]  # type: ignore[arg-type]
 
     def test_performance_optimization_no_copy_when_unchanged(self):
         """Test that schemas without nullable fields return the same object (no copy)."""
@@ -128,7 +130,7 @@ class TestHandleNullableFields:
             "type": "object",
             "properties": {"name": {"type": "string"}},
         }
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         # Should return the exact same object, not a copy
         assert result is input_schema
 
@@ -136,14 +138,14 @@ class TestHandleNullableFields:
         """Test nullable handling with existing union types (type as array)."""
         input_schema = {"type": ["string", "integer"], "nullable": True}
         expected = {"type": ["string", "integer", "null"]}
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_already_nullable_union_unchanged(self):
         """Test that union types already containing null are not modified."""
         input_schema = {"type": ["string", "null"], "nullable": True}
         expected = {"type": ["string", "null"]}
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_property_level_union_with_nullable(self):
@@ -156,19 +158,19 @@ class TestHandleNullableFields:
             "type": "object",
             "properties": {"value": {"type": ["string", "integer", "null"]}},
         }
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_complex_union_nullable_scenarios(self):
         """Test various complex union type scenarios."""
         # Already has null in different position
         input1 = {"type": ["null", "string", "integer"], "nullable": True}
-        result1 = _handle_nullable_fields(input1)
+        result1 = convert_openapi_schema_to_json_schema(input1, "3.0.0")
         assert result1 == {"type": ["null", "string", "integer"]}
 
         # Single item array
         input2 = {"type": ["string"], "nullable": True}
-        result2 = _handle_nullable_fields(input2)
+        result2 = convert_openapi_schema_to_json_schema(input2, "3.0.0")
         assert result2 == {"type": ["string", "null"]}
 
     def test_oneof_with_nullable(self):
@@ -180,7 +182,7 @@ class TestHandleNullableFields:
         expected = {
             "anyOf": [{"type": "string"}, {"type": "integer"}, {"type": "null"}]
         }
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_anyof_with_nullable(self):
@@ -192,7 +194,7 @@ class TestHandleNullableFields:
         expected = {
             "anyOf": [{"type": "string"}, {"type": "integer"}, {"type": "null"}]
         }
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_anyof_already_nullable(self):
@@ -202,7 +204,7 @@ class TestHandleNullableFields:
             "nullable": True,
         }
         expected = {"anyOf": [{"type": "string"}, {"type": "null"}]}
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_allof_with_nullable(self):
@@ -217,7 +219,7 @@ class TestHandleNullableFields:
                 {"type": "null"},
             ]
         }
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected
 
     def test_property_level_oneof_with_nullable(self):
@@ -239,5 +241,5 @@ class TestHandleNullableFields:
                 }
             },
         }
-        result = _handle_nullable_fields(input_schema)
+        result = convert_openapi_schema_to_json_schema(input_schema, "3.0.0")
         assert result == expected