[DNM] Handle allOf and oneOf in json_schema_to_type

fastmcp_slim/fastmcp/utilities/json_schema_type.py

@@ -403,6 +403,10 @@ def _object_schema_to_type(
     """
     has_properties = bool(schema.get("properties"))
     additional_props = schema.get("additionalProperties")
+    # Per JSON Schema, an empty schema {} means "accept anything" —
+    # equivalent to additionalProperties: true.
+    if isinstance(additional_props, dict) and not additional_props:
+        additional_props = True
     class_name = name if name is not None else schema.get("title")
 
     if not has_properties and additional_props:
@@ -453,7 +457,13 @@ def _schema_to_type(
     if not schema:
         return object
 
-    if "type" not in schema and "properties" in schema:
+    if (
+        "type" not in schema
+        and "properties" in schema
+        and "allOf" not in schema
+        and "oneOf" not in schema
+        and "anyOf" not in schema
+    ):
         return _create_dataclass(schema, schema.get("title", "<unknown>"), schemas)
 
     # Handle references first
@@ -493,6 +503,109 @@ def _schema_to_type(
             else:
                 return Union[tuple(types)]  # type: ignore # noqa: UP007
 
+    # Handle allOf (schema intersection / composition).
+    # Flatten all sub-schemas into a single merged object schema.
+    if "allOf" in schema:
+        merged: dict[str, Any] = {}
+        merged_properties: dict[str, Any] = {}
+        merged_required: list[str] = []
+        has_false = False
+
+        def _collect_allof(sub: Any) -> None:
+            """Recursively collect properties from a sub-schema."""
+            nonlocal has_false
+            if sub is False:
+                has_false = True
+                return
+            if sub is True:
+                return
+            if isinstance(sub, bool):
+                return
+            if "$ref" in sub:
+                resolved = _resolve_ref(sub["$ref"], schemas)
+                if isinstance(resolved, bool):
+                    if resolved is False:
+                        has_false = True
+                    return
+                sub = dict(resolved)
+            # Recurse into nested allOf
+            if "allOf" in sub:
+                for nested in sub["allOf"]:
+                    _collect_allof(nested)
+            merged_properties.update(sub.get("properties", {}))
+            merged_required.extend(sub.get("required", []))
+            for key in ("title", "description"):
+                if key in sub and key not in merged:
+                    merged[key] = sub[key]
+            # Intersect additionalProperties: false is most restrictive and wins.
+            if "additionalProperties" in sub:
+                existing = merged.get("additionalProperties")
+                if existing is None:
+                    merged["additionalProperties"] = sub["additionalProperties"]
+                elif sub["additionalProperties"] is False:
+                    merged["additionalProperties"] = False
+
+        # Collect from allOf children first, then overlay sibling
+        # properties so local definitions take precedence over inherited.
+        for sub in schema["allOf"]:
+            _collect_allof(sub)
+
+        merged_properties.update(schema.get("properties", {}))
+        merged_required.extend(schema.get("required", []))
+
+        if has_false:
+            return _UnsatisfiableType  # type: ignore[return-value]
+
+        if merged_properties:
+            merged["type"] = "object"
+            merged["properties"] = merged_properties
+            if merged_required:
+                merged["required"] = list(dict.fromkeys(merged_required))
+            # Preserve additionalProperties from the parent schema, not just
+            # from allOf children, so schemas like
+            # {"additionalProperties": true, "allOf": [...]} allow extra keys.
+            if (
+                "additionalProperties" not in merged
+                and "additionalProperties" in schema
+            ):
+                merged["additionalProperties"] = schema["additionalProperties"]
+            return _schema_to_type(merged, schemas)
+        # allOf with no mergeable properties — fall through to Any
+
+    # Handle oneOf (exactly-one-of union).
+    # Treat like anyOf for type construction — Pydantic's Union
+    # does "first match" which is a reasonable approximation.
+    if "oneOf" in schema:
+        types: list[type | Any] = []
+        for subschema in schema["oneOf"]:
+            # Same dict special-case as the anyOf handler: detect
+            # map-like objects so they become dict[str, X] instead
+            # of empty dataclasses that discard key/value data.
+            if (
+                isinstance(subschema, dict)
+                and subschema.get("type") == "object"
+                and not subschema.get("properties")
+                and "additionalProperties" in subschema
+            ):
+                additional_props = subschema["additionalProperties"]
+                if additional_props is True or additional_props == {}:
+                    types.append(dict[str, Any])
+                else:
+                    value_type = _schema_to_type(additional_props, schemas)
+                    types.append(dict[str, value_type])  # type: ignore
+            else:
+                types.append(_schema_to_type(subschema, schemas))
+        has_null = type(None) in types
+        types = [t for t in types if t is not type(None)]
+        if len(types) == 0:
+            return type(None)
+        elif len(types) == 1:
+            return types[0] | None if has_null else types[0]  # type: ignore
+        else:
+            if has_null:
+                return Union[(*types, type(None))]  # type: ignore
+            return Union[tuple(types)]  # type: ignore # noqa: UP007
+
     schema_type = schema.get("type")
     if not schema_type:
         return Any