fix: replace oneOf with anyOf in OpenAPI schemas to handle overlapping unions (#1119)

src/fastmcp/utilities/openapi.py

@@ -1,6 +1,6 @@
 import json
 import logging
-from typing import Any, Generic, Literal, TypeVar
+from typing import Any, Generic, Literal, TypeVar, cast
 
 from openapi_pydantic import (
     OpenAPI,
@@ -1188,6 +1188,20 @@ def _combine_schemas(route: HTTPRoute) -> dict[str, Any]:
     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):
+        if "oneOf" in schema:
+            schema["anyOf"] = schema.pop("oneOf")
+        for k, v in schema.items():
+            schema[k] = _adjust_union_types(v)
+    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
 ) -> dict[str, Any] | None:
@@ -1276,4 +1290,7 @@ def extract_output_schema_from_responses(
     # Use compress_schema to remove unused definitions
     output_schema = compress_schema(output_schema)
 
+    # Adjust union types to handle overlapping unions
+    output_schema = cast(dict[str, Any], _adjust_union_types(output_schema))
+
     return output_schema

tests/utilities/openapi/test_openapi_output_schemas.py

@@ -2,6 +2,7 @@
 
 from fastmcp.utilities.openapi import (
     ResponseInfo,
+    _adjust_union_types,
     extract_output_schema_from_responses,
 )
 
@@ -234,3 +235,42 @@ class TestExtractOutputSchema:
         assert "User" in result["$defs"]
         assert result["properties"]["result"]["type"] == "array"
         assert result["properties"]["result"]["items"]["$ref"] == "#/$defs/User"
+
+
+def test_adjust_union_types():
+    """Test that oneOf is replaced with anyOf in schemas."""
+    schema = {"oneOf": [{"type": "string"}, {"type": "number"}]}
+    result = _adjust_union_types(schema)
+    assert isinstance(result, dict)
+    assert "anyOf" in result
+    assert "oneOf" not in result
+    assert len(result["anyOf"]) == 2
+    assert result["anyOf"][0] == {"type": "string"}
+    assert result["anyOf"][1] == {"type": "number"}
+
+
+def test_extract_output_schema_converts_oneOf_to_anyOf():
+    """Test that extracted schema converts oneOf to anyOf."""
+    responses = {
+        "200": ResponseInfo(
+            description="Success",
+            content_schema={
+                "application/json": {
+                    "type": "object",
+                    "properties": {
+                        "result": {
+                            "oneOf": [
+                                {"$ref": "#/$defs/TypeA"},
+                                {"$ref": "#/$defs/TypeB"},
+                            ]
+                        }
+                    },
+                }
+            },
+        )
+    }
+    schema_definitions = {"TypeA": {"type": "string"}, "TypeB": {"type": "number"}}
+    result = extract_output_schema_from_responses(responses, schema_definitions)
+    assert result is not None
+    assert "oneOf" not in str(result)  # Ensure no oneOf remains
+    assert "anyOf" in str(result)  # Ensure anyOf is present