Fix OpenAPI allOf handling at requestBody top level (#1378) (#1425)

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

@@ -237,6 +237,32 @@ def _combine_schemas_and_map_params(
             route.request_body.content_schema[content_type].copy(),
             route.request_body.description,
         )
+
+        # Handle allOf at the top level by merging all schemas
+        if "allOf" in body_schema and isinstance(body_schema["allOf"], list):
+            merged_props = {}
+            merged_required = []
+
+            for sub_schema in body_schema["allOf"]:
+                if isinstance(sub_schema, dict):
+                    # Merge properties
+                    if "properties" in sub_schema:
+                        merged_props.update(sub_schema["properties"])
+                    # Merge required fields
+                    if "required" in sub_schema:
+                        merged_required.extend(sub_schema["required"])
+
+            # Update body_schema with merged properties
+            body_schema["properties"] = merged_props
+            if merged_required:
+                # Remove duplicates while preserving order
+                seen = set()
+                body_schema["required"] = [
+                    x for x in merged_required if not (x in seen or seen.add(x))
+                ]
+            # Remove the allOf since we've merged it
+            body_schema.pop("allOf", None)
+
         body_props = body_schema.get("properties", {})
 
     # Detect collisions: parameters that exist in both body and path/query/header

tests/experimental/utilities/openapi/test_allof_requestbody.py

@@ -0,0 +1,227 @@
+"""Tests for allOf handling at requestBody top level."""
+
+from fastmcp.experimental.utilities.openapi.models import (
+    HTTPRoute,
+    RequestBodyInfo,
+)
+from fastmcp.experimental.utilities.openapi.schemas import _combine_schemas
+
+
+def test_allof_at_requestbody_top_level():
+    """Test that allOf schemas at requestBody top level are properly merged."""
+
+    # Create a route with allOf at the requestBody top level
+    route = HTTPRoute(
+        path="/test",
+        method="POST",
+        operation_id="testOperation",
+        parameters=[],
+        request_body=RequestBodyInfo(
+            required=True,
+            content_schema={
+                "application/json": {
+                    "allOf": [
+                        {
+                            "type": "object",
+                            "properties": {
+                                "name": {"type": "string"},
+                                "age": {"type": "integer"},
+                            },
+                            "required": ["name"],
+                        },
+                        {
+                            "type": "object",
+                            "properties": {
+                                "email": {"type": "string"},
+                                "phone": {"type": "string"},
+                            },
+                            "required": ["email"],
+                        },
+                    ]
+                }
+            },
+        ),
+        responses={},
+    )
+
+    # Combine schemas - this should merge allOf schemas
+    combined = _combine_schemas(route)
+
+    # Check that all properties from both allOf schemas are present
+    properties = combined.get("properties", {})
+    assert "name" in properties
+    assert "age" in properties
+    assert "email" in properties
+    assert "phone" in properties
+
+    # Check property types
+    assert properties["name"]["type"] == "string"
+    assert properties["age"]["type"] == "integer"
+    assert properties["email"]["type"] == "string"
+    assert properties["phone"]["type"] == "string"
+
+    # Check that required fields are merged correctly
+    required = set(combined.get("required", []))
+    assert "name" in required
+    assert "email" in required
+
+    # allOf should be removed after merging
+    assert "allOf" not in combined
+
+
+def test_allof_with_nested_properties():
+    """Test allOf with nested object properties."""
+
+    route = HTTPRoute(
+        path="/test",
+        method="POST",
+        operation_id="testNested",
+        parameters=[],
+        request_body=RequestBodyInfo(
+            required=True,
+            content_schema={
+                "application/json": {
+                    "allOf": [
+                        {
+                            "type": "object",
+                            "properties": {
+                                "user": {
+                                    "type": "object",
+                                    "properties": {
+                                        "id": {"type": "integer"},
+                                        "name": {"type": "string"},
+                                    },
+                                }
+                            },
+                            "required": ["user"],
+                        },
+                        {
+                            "type": "object",
+                            "properties": {
+                                "metadata": {
+                                    "type": "object",
+                                    "properties": {
+                                        "created": {"type": "string"},
+                                        "updated": {"type": "string"},
+                                    },
+                                }
+                            },
+                        },
+                    ]
+                }
+            },
+        ),
+        responses={},
+    )
+
+    combined = _combine_schemas(route)
+
+    # Check nested properties are preserved
+    properties = combined.get("properties", {})
+    assert "user" in properties
+    assert "metadata" in properties
+
+    # Check nested structure
+    assert properties["user"]["type"] == "object"
+    assert "id" in properties["user"]["properties"]
+    assert "name" in properties["user"]["properties"]
+
+    assert properties["metadata"]["type"] == "object"
+    assert "created" in properties["metadata"]["properties"]
+    assert "updated" in properties["metadata"]["properties"]
+
+    # Check required
+    required = set(combined.get("required", []))
+    assert "user" in required
+    assert "metadata" not in required  # Not in any required array
+
+
+def test_allof_with_overlapping_properties():
+    """Test allOf with overlapping property names (later schemas override)."""
+
+    route = HTTPRoute(
+        path="/test",
+        method="POST",
+        operation_id="testOverlap",
+        parameters=[],
+        request_body=RequestBodyInfo(
+            required=True,
+            content_schema={
+                "application/json": {
+                    "allOf": [
+                        {
+                            "type": "object",
+                            "properties": {
+                                "name": {"type": "string", "minLength": 1},
+                                "age": {"type": "integer"},
+                            },
+                            "required": ["name"],
+                        },
+                        {
+                            "type": "object",
+                            "properties": {
+                                "name": {"type": "string", "maxLength": 50},  # Override
+                                "email": {"type": "string"},
+                            },
+                            "required": ["email"],
+                        },
+                    ]
+                }
+            },
+        ),
+        responses={},
+    )
+
+    combined = _combine_schemas(route)
+
+    properties = combined.get("properties", {})
+
+    # Later schema should win for overlapping properties
+    assert "name" in properties
+    assert properties["name"]["type"] == "string"
+    assert "maxLength" in properties["name"]  # From second schema
+    assert properties["name"]["maxLength"] == 50
+
+    # Check other properties
+    assert "age" in properties
+    assert "email" in properties
+
+    # Both name and email should be required
+    required = set(combined.get("required", []))
+    assert "name" in required
+    assert "email" in required
+
+
+def test_no_allof_passthrough():
+    """Test that schemas without allOf pass through unchanged."""
+
+    route = HTTPRoute(
+        path="/test",
+        method="POST",
+        operation_id="testNoAllOf",
+        parameters=[],
+        request_body=RequestBodyInfo(
+            required=True,
+            content_schema={
+                "application/json": {
+                    "type": "object",
+                    "properties": {"simple": {"type": "string"}},
+                    "required": ["simple"],
+                }
+            },
+        ),
+        responses={},
+    )
+
+    combined = _combine_schemas(route)
+
+    # Should pass through unchanged
+    properties = combined.get("properties", {})
+    assert "simple" in properties
+    assert properties["simple"]["type"] == "string"
+
+    required = set(combined.get("required", []))
+    assert "simple" in required
+
+    # No allOf in original or result
+    assert "allOf" not in combined