crewAI/lib/crewai/src/crewai/utilities/pydantic_schema_utils.py

"""Dynamic Pydantic model creation from JSON schemas.

This module provides utilities for converting JSON schemas to Pydantic models at runtime.
The main function is `create_model_from_schema`, which takes a JSON schema and returns
a dynamically created Pydantic model class.

This is used by the A2A server to honor response schemas sent by clients, allowing
structured output from agent tasks.

Based on dydantic (https://github.com/zenbase-ai/dydantic).

This module provides functions for converting Pydantic models to JSON schemas
suitable for use with LLMs and tool definitions.
"""

from __future__ import annotations

from collections.abc import Callable
from copy import deepcopy
import datetime
import logging
from typing import (
    TYPE_CHECKING,
    Annotated,
    Any,
    Final,
    ForwardRef,
    Literal,
    Optional,
    TypedDict,
    Union,
    cast,
)
import uuid

import jsonref  # type: ignore[import-untyped]
from pydantic import (
    UUID1,
    UUID3,
    UUID4,
    UUID5,
    AnyUrl,
    BaseModel,
    ConfigDict,
    DirectoryPath,
    Field,
    FilePath,
    FileUrl,
    HttpUrl,
    Json,
    MongoDsn,
    NewPath,
    PostgresDsn,
    SecretBytes,
    SecretStr,
    StrictBytes,
    create_model as create_model_base,
)
from pydantic.networks import (  # type: ignore[attr-defined]
    IPv4Address,
    IPv6Address,
    IPvAnyAddress,
    IPvAnyInterface,
    IPvAnyNetwork,
)


logger = logging.getLogger(__name__)

if TYPE_CHECKING:
    from pydantic import EmailStr
    from pydantic.main import AnyClassMethod
else:
    try:
        from pydantic import EmailStr
    except ImportError:
        logger.warning(
            "EmailStr unavailable, using str fallback",
            extra={"missing_package": "email_validator"},
        )
        EmailStr = str


class JsonSchemaInfo(TypedDict):
    """Inner structure for JSON schema metadata."""

    name: str
    strict: Literal[True]
    schema: dict[str, Any]


class ModelDescription(TypedDict):
    """Return type for generate_model_description."""

    type: Literal["json_schema"]
    json_schema: JsonSchemaInfo


def resolve_refs(schema: dict[str, Any]) -> dict[str, Any]:
    """Recursively resolve all local $refs in the given JSON Schema using $defs as the source.

    This is needed because Pydantic generates $ref-based schemas that
    some consumers (e.g. LLMs, tool frameworks) don't handle well.

    Args:
        schema: JSON Schema dict that may contain "$refs" and "$defs".

    Returns:
        A new schema dictionary with all local $refs replaced by their definitions.
    """
    defs = schema.get("$defs", {})
    schema_copy = deepcopy(schema)
    expanding: set[str] = set()

    def _resolve(node: Any) -> Any:
        if isinstance(node, dict):
            ref = node.get("$ref")
            if isinstance(ref, str) and ref.startswith("#/$defs/"):
                def_name = ref.replace("#/$defs/", "")
                if def_name not in defs:
                    raise KeyError(f"Definition '{def_name}' not found in $defs.")
                if def_name in expanding:
                    def_schema = defs[def_name]
                    stub: dict[str, Any] = {"type": def_schema.get("type", "object")}
                    if "description" in def_schema:
                        stub["description"] = def_schema["description"]
                    return stub
                expanding.add(def_name)
                try:
                    return _resolve(deepcopy(defs[def_name]))
                finally:
                    expanding.discard(def_name)
            return {k: _resolve(v) for k, v in node.items()}

        if isinstance(node, list):
            return [_resolve(i) for i in node]

        return node

    return _resolve(schema_copy)  # type: ignore[no-any-return]


def add_key_in_dict_recursively(
    d: dict[str, Any],
    key: str,
    value: Any,
    criteria: Callable[[dict[str, Any]], bool],
    _seen: set[int] | None = None,
) -> dict[str, Any]:
    """Recursively adds a key/value pair to all nested dicts matching `criteria`.

    Args:
        d: The dictionary to modify.
        key: The key to add.
        value: The value to add.
        criteria: A function that returns True for dicts that should receive the key.
        _seen: Internal set of visited ``id()``s, used to guard cyclic schemas.

    Returns:
        The modified dictionary.
    """
    if _seen is None:
        _seen = set()
    if isinstance(d, dict):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        if criteria(d) and key not in d:
            d[key] = value
        for v in d.values():
            add_key_in_dict_recursively(v, key, value, criteria, _seen)
    elif isinstance(d, list):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        for i in d:
            add_key_in_dict_recursively(i, key, value, criteria, _seen)
    return d


def force_additional_properties_false(d: Any, _seen: set[int] | None = None) -> Any:
    """Force additionalProperties=false on all object-type dicts recursively.

    OpenAI strict mode requires all objects to have additionalProperties=false.
    This function overwrites any existing value to ensure compliance.

    Also ensures objects have properties and required arrays, even if empty,
    as OpenAI strict mode requires these for all object types.

    Args:
        d: The dictionary/list to modify.
        _seen: Internal set of visited ``id()``s, used to guard cyclic schemas.

    Returns:
        The modified dictionary/list.
    """
    if _seen is None:
        _seen = set()
    if isinstance(d, dict):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        if d.get("type") == "object":
            d["additionalProperties"] = False
            if "properties" not in d:
                d["properties"] = {}
            if "required" not in d:
                d["required"] = []
        for v in d.values():
            force_additional_properties_false(v, _seen)
    elif isinstance(d, list):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        for i in d:
            force_additional_properties_false(i, _seen)
    return d


OPENAI_SUPPORTED_FORMATS: Final[
    set[Literal["date-time", "date", "time", "duration"]]
] = {
    "date-time",
    "date",
    "time",
    "duration",
}


def strip_unsupported_formats(d: Any, _seen: set[int] | None = None) -> Any:
    """Remove format annotations that OpenAI strict mode doesn't support.

    OpenAI only supports: date-time, date, time, duration.
    Other formats like uri, email, uuid etc. cause validation errors.

    Args:
        d: The dictionary/list to modify.
        _seen: Internal set of visited ``id()``s, used to guard cyclic schemas.

    Returns:
        The modified dictionary/list.
    """
    if _seen is None:
        _seen = set()
    if isinstance(d, dict):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        format_value = d.get("format")
        if (
            isinstance(format_value, str)
            and format_value not in OPENAI_SUPPORTED_FORMATS
        ):
            del d["format"]
        for v in d.values():
            strip_unsupported_formats(v, _seen)
    elif isinstance(d, list):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        for i in d:
            strip_unsupported_formats(i, _seen)
    return d


def ensure_type_in_schemas(d: Any, _seen: set[int] | None = None) -> Any:
    """Ensure all schema objects in anyOf/oneOf have a 'type' key.

    OpenAI strict mode requires every schema to have a 'type' key.
    Empty schemas {} in anyOf/oneOf are converted to {"type": "object"}.

    Args:
        d: The dictionary/list to modify.
        _seen: Internal set of visited ``id()``s, used to guard cyclic schemas.

    Returns:
        The modified dictionary/list.
    """
    if _seen is None:
        _seen = set()
    if isinstance(d, dict):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        for key in ("anyOf", "oneOf"):
            if key in d:
                schema_list = d[key]
                for i, schema in enumerate(schema_list):
                    if isinstance(schema, dict) and schema == {}:
                        schema_list[i] = {"type": "object"}
                    else:
                        ensure_type_in_schemas(schema, _seen)
        for v in d.values():
            ensure_type_in_schemas(v, _seen)
    elif isinstance(d, list):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        for item in d:
            ensure_type_in_schemas(item, _seen)
    return d


def fix_discriminator_mappings(schema: dict[str, Any]) -> dict[str, Any]:
    """Replace '#/$defs/...' references in discriminator.mapping with just the model name.

    Args:
        schema: JSON schema dictionary.

    Returns:
        Modified schema with fixed discriminator mappings.
    """
    output = schema.get("properties", {}).get("output")
    if not output:
        return schema

    disc = output.get("discriminator")
    if not disc or "mapping" not in disc:
        return schema

    disc["mapping"] = {k: v.split("/")[-1] for k, v in disc["mapping"].items()}
    return schema


def add_const_to_oneof_variants(schema: dict[str, Any]) -> dict[str, Any]:
    """Add const fields to oneOf variants for discriminated unions.

    The json_schema_to_pydantic library requires each oneOf variant to have
    a const field for the discriminator property. This function adds those
    const fields based on the discriminator mapping.

    Args:
        schema: JSON Schema dict that may contain discriminated unions

    Returns:
        Modified schema with const fields added to oneOf variants
    """

    def _process_oneof(node: dict[str, Any]) -> dict[str, Any]:
        """Process a single node that might contain a oneOf with discriminator."""
        if not isinstance(node, dict):
            return node

        if "oneOf" in node and "discriminator" in node:
            discriminator = node["discriminator"]
            property_name = discriminator.get("propertyName")
            mapping = discriminator.get("mapping", {})

            if property_name and mapping:
                one_of_variants = node.get("oneOf", [])

                for variant in one_of_variants:
                    if isinstance(variant, dict) and "properties" in variant:
                        variant_title = variant.get("title", "")

                        matched_disc_value = None
                        for disc_value, schema_name in mapping.items():
                            if variant_title == schema_name or variant_title.endswith(
                                schema_name
                            ):
                                matched_disc_value = disc_value
                                break

                        if matched_disc_value is not None:
                            props = variant["properties"]
                            if property_name in props:
                                props[property_name]["const"] = matched_disc_value

        for key, value in node.items():
            if isinstance(value, dict):
                node[key] = _process_oneof(value)
            elif isinstance(value, list):
                node[key] = [
                    _process_oneof(item) if isinstance(item, dict) else item
                    for item in value
                ]

        return node

    return _process_oneof(deepcopy(schema))


def convert_oneof_to_anyof(
    schema: dict[str, Any], _seen: set[int] | None = None
) -> dict[str, Any]:
    """Convert oneOf to anyOf for OpenAI compatibility.

    OpenAI's Structured Outputs support anyOf better than oneOf.
    This recursively converts all oneOf occurrences to anyOf.

    Args:
        schema: JSON schema dictionary.
        _seen: Internal set of visited ``id()``s, used to guard cyclic schemas.

    Returns:
        Modified schema with anyOf instead of oneOf.
    """
    if _seen is None:
        _seen = set()
    if isinstance(schema, dict):
        if id(schema) in _seen:
            return schema
        _seen.add(id(schema))
        if "oneOf" in schema:
            schema["anyOf"] = schema.pop("oneOf")

        for value in schema.values():
            if isinstance(value, dict):
                convert_oneof_to_anyof(value, _seen)
            elif isinstance(value, list):
                if id(value) in _seen:
                    continue
                _seen.add(id(value))
                for item in value:
                    if isinstance(item, dict):
                        convert_oneof_to_anyof(item, _seen)

    return schema


def ensure_all_properties_required(
    schema: dict[str, Any], _seen: set[int] | None = None
) -> dict[str, Any]:
    """Ensure all properties are in the required array for OpenAI strict mode.

    OpenAI's strict structured outputs require all properties to be listed
    in the required array. This recursively updates all objects to include
    all their properties in required.

    Args:
        schema: JSON schema dictionary.
        _seen: Internal set of visited ``id()``s, used to guard cyclic schemas.

    Returns:
        Modified schema with all properties marked as required.
    """
    if _seen is None:
        _seen = set()
    if isinstance(schema, dict):
        if id(schema) in _seen:
            return schema
        _seen.add(id(schema))
        if schema.get("type") == "object" and "properties" in schema:
            properties = schema["properties"]
            if properties:
                schema["required"] = list(properties.keys())

        for value in schema.values():
            if isinstance(value, dict):
                ensure_all_properties_required(value, _seen)
            elif isinstance(value, list):
                if id(value) in _seen:
                    continue
                _seen.add(id(value))
                for item in value:
                    if isinstance(item, dict):
                        ensure_all_properties_required(item, _seen)

    return schema


def strip_null_from_types(
    schema: dict[str, Any], _seen: set[int] | None = None
) -> dict[str, Any]:
    """Remove null type from anyOf/type arrays.

    Pydantic generates `T | None` for optional fields, which creates schemas with
    null in the type. However, for MCP tools, optional fields should be omitted
    entirely rather than sent as null. This function strips null from types.

    Args:
        schema: JSON schema dictionary.
        _seen: Internal set of visited ``id()``s, used to guard cyclic schemas.

    Returns:
        Modified schema with null types removed.
    """
    if _seen is None:
        _seen = set()
    if isinstance(schema, dict):
        if id(schema) in _seen:
            return schema
        _seen.add(id(schema))
        if "anyOf" in schema:
            any_of = schema["anyOf"]
            non_null = [opt for opt in any_of if opt.get("type") != "null"]
            if len(non_null) == 1:
                schema.pop("anyOf")
                schema.update(non_null[0])
            elif len(non_null) > 1:
                schema["anyOf"] = non_null

        type_value = schema.get("type")
        if isinstance(type_value, list) and "null" in type_value:
            non_null_types = [t for t in type_value if t != "null"]
            if len(non_null_types) == 1:
                schema["type"] = non_null_types[0]
            elif len(non_null_types) > 1:
                schema["type"] = non_null_types

        for value in schema.values():
            if isinstance(value, dict):
                strip_null_from_types(value, _seen)
            elif isinstance(value, list):
                if id(value) in _seen:
                    continue
                _seen.add(id(value))
                for item in value:
                    if isinstance(item, dict):
                        strip_null_from_types(item, _seen)

    return schema


_STRICT_METADATA_KEYS: Final[tuple[str, ...]] = (
    "title",
    "default",
    "examples",
    "example",
    "$comment",
    "readOnly",
    "writeOnly",
    "deprecated",
)

_CLAUDE_STRICT_UNSUPPORTED: Final[tuple[str, ...]] = (
    "minimum",
    "maximum",
    "exclusiveMinimum",
    "exclusiveMaximum",
    "multipleOf",
    "minLength",
    "maxLength",
    "pattern",
    "minItems",
    "maxItems",
    "uniqueItems",
    "minContains",
    "maxContains",
    "minProperties",
    "maxProperties",
    "patternProperties",
    "propertyNames",
    "dependentRequired",
    "dependentSchemas",
)


def _strip_keys_recursive(
    d: Any, keys: tuple[str, ...], _seen: set[int] | None = None
) -> Any:
    """Recursively delete schema metadata keys without deleting property names.

    JSON Schema stores fields under a ``properties`` map. A user/tool argument can
    legitimately be named ``title``, ``default``, or another metadata key. Those
    entries must stay in the map; only metadata inside the schema nodes should be
    stripped.
    """
    if _seen is None:
        _seen = set()
    if isinstance(d, dict):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        for key in keys:
            d.pop(key, None)
        for key, value in d.items():
            if key == "properties" and isinstance(value, dict):
                for property_schema in value.values():
                    _strip_keys_recursive(property_schema, keys, _seen)
            else:
                _strip_keys_recursive(value, keys, _seen)
    elif isinstance(d, list):
        if id(d) in _seen:
            return d
        _seen.add(id(d))
        for i in d:
            _strip_keys_recursive(i, keys, _seen)
    return d


def lift_top_level_anyof(schema: dict[str, Any]) -> dict[str, Any]:
    """Unwrap a top-level anyOf/oneOf/allOf wrapping a single object variant.

    Anthropic's strict ``input_schema`` rejects top-level union keywords. When
    exactly one variant is an object schema, lift it so the root is a plain
    object; otherwise leave the schema alone.
    """
    for key in ("anyOf", "oneOf", "allOf"):
        variants = schema.get(key)
        if not isinstance(variants, list):
            continue
        object_variants = [
            v for v in variants if isinstance(v, dict) and v.get("type") == "object"
        ]
        if len(object_variants) == 1:
            lifted = deepcopy(object_variants[0])
            schema.pop(key)
            schema.update(lifted)
            break
    return schema


def _common_strict_pipeline(params: dict[str, Any]) -> dict[str, Any]:
    """Shared strict sanitization: inline refs, close objects, require all properties."""
    sanitized = resolve_refs(deepcopy(params))
    sanitized.pop("$defs", None)
    sanitized = convert_oneof_to_anyof(sanitized)
    sanitized = ensure_type_in_schemas(sanitized)
    sanitized = force_additional_properties_false(sanitized)
    sanitized = ensure_all_properties_required(sanitized)
    return cast(dict[str, Any], _strip_keys_recursive(sanitized, _STRICT_METADATA_KEYS))


def sanitize_tool_params_for_openai_strict(
    params: dict[str, Any],
) -> dict[str, Any]:
    """Sanitize a JSON schema for OpenAI strict function calling."""
    if not isinstance(params, dict):
        return params
    return cast(
        dict[str, Any], strip_unsupported_formats(_common_strict_pipeline(params))
    )


def sanitize_tool_params_for_anthropic_strict(
    params: dict[str, Any],
) -> dict[str, Any]:
    """Sanitize a JSON schema for Anthropic strict tool use."""
    if not isinstance(params, dict):
        return params
    sanitized = lift_top_level_anyof(_common_strict_pipeline(params))
    sanitized = _strip_keys_recursive(sanitized, _CLAUDE_STRICT_UNSUPPORTED)
    return cast(dict[str, Any], strip_unsupported_formats(sanitized))


def sanitize_tool_params_for_bedrock_strict(
    params: dict[str, Any],
) -> dict[str, Any]:
    """Sanitize a JSON schema for Bedrock Converse strict tool use.

    Bedrock Converse uses the same grammar compiler as the underlying Claude
    model, so the constraints match Anthropic's.
    """
    return sanitize_tool_params_for_anthropic_strict(params)


def generate_model_description(
    model: type[BaseModel],
    *,
    strip_null_types: bool = True,
) -> ModelDescription:
    """Generate JSON schema description of a Pydantic model.

    This function takes a Pydantic model class and returns its JSON schema,
    which includes full type information, discriminators, and all metadata.
    The schema is dereferenced to inline all $ref references for better LLM understanding.

    Args:
        model: A Pydantic model class.
        strip_null_types: When ``True`` (default), remove ``null`` from
            ``anyOf`` / ``type`` arrays.  Set to ``False`` to allow sending ``null`` for
            optional fields.

    Returns:
        A ModelDescription with JSON schema representation of the model.
    """
    json_schema = model.model_json_schema(ref_template="#/$defs/{model}")

    json_schema = force_additional_properties_false(json_schema)
    json_schema = strip_unsupported_formats(json_schema)
    json_schema = ensure_type_in_schemas(json_schema)

    json_schema = resolve_refs(json_schema)

    json_schema.pop("$defs", None)
    json_schema = fix_discriminator_mappings(json_schema)
    json_schema = convert_oneof_to_anyof(json_schema)
    json_schema = ensure_all_properties_required(json_schema)

    if strip_null_types:
        json_schema = strip_null_from_types(json_schema)

    return {
        "type": "json_schema",
        "json_schema": {
            "name": model.__name__,
            "strict": True,
            "schema": json_schema,
        },
    }


FORMAT_TYPE_MAP: dict[str, type[Any]] = {
    "base64": Annotated[bytes, Field(json_schema_extra={"format": "base64"})],  # type: ignore[dict-item]
    "binary": StrictBytes,
    "date": datetime.date,
    "time": datetime.time,
    "date-time": datetime.datetime,
    "duration": datetime.timedelta,
    "directory-path": DirectoryPath,
    "email": EmailStr,
    "file-path": FilePath,
    "ipv4": IPv4Address,
    "ipv6": IPv6Address,
    "ipvanyaddress": IPvAnyAddress,  # type: ignore[dict-item]
    "ipvanyinterface": IPvAnyInterface,  # type: ignore[dict-item]
    "ipvanynetwork": IPvAnyNetwork,  # type: ignore[dict-item]
    "json-string": Json,
    "multi-host-uri": PostgresDsn | MongoDsn,  # type: ignore[dict-item]
    "password": SecretStr,
    "path": NewPath,
    "uri": AnyUrl,
    "uuid": uuid.UUID,
    "uuid1": UUID1,
    "uuid3": UUID3,
    "uuid4": UUID4,
    "uuid5": UUID5,
}


def build_rich_field_description(prop_schema: dict[str, Any]) -> str:
    """Build a comprehensive field description including constraints.

    Embeds format, enum, pattern, min/max, and example constraints into the
    description text so that LLMs can understand tool parameter requirements
    without inspecting the raw JSON Schema.

    Args:
        prop_schema: Property schema with description and constraints.

    Returns:
        Enhanced description with format, enum, and other constraints.
    """
    parts: list[str] = []

    description = prop_schema.get("description", "")
    if description:
        parts.append(description)

    format_type = prop_schema.get("format")
    if format_type:
        parts.append(f"Format: {format_type}")

    enum_values = prop_schema.get("enum")
    if enum_values:
        enum_str = ", ".join(repr(v) for v in enum_values)
        parts.append(f"Allowed values: [{enum_str}]")

    pattern = prop_schema.get("pattern")
    if pattern:
        parts.append(f"Pattern: {pattern}")

    minimum = prop_schema.get("minimum")
    maximum = prop_schema.get("maximum")
    if minimum is not None:
        parts.append(f"Minimum: {minimum}")
    if maximum is not None:
        parts.append(f"Maximum: {maximum}")

    min_length = prop_schema.get("minLength")
    max_length = prop_schema.get("maxLength")
    if min_length is not None:
        parts.append(f"Min length: {min_length}")
    if max_length is not None:
        parts.append(f"Max length: {max_length}")

    examples = prop_schema.get("examples")
    if examples:
        examples_str = ", ".join(repr(e) for e in examples[:3])
        parts.append(f"Examples: {examples_str}")

    return ". ".join(parts) if parts else ""


def _inline_top_level_ref(schema: dict[str, Any]) -> dict[str, Any]:
    """Resolve only the top-level ``$ref``, preserving ``$defs`` for lazy inner resolution.

    Used as a fallback when ``jsonref.replace_refs`` fails on circular schemas.
    Inner ``$ref`` pointers are left intact so that :func:`_resolve_ref` can
    resolve them during model construction, with cycle detection via ``in_progress``.
    """
    schema = deepcopy(schema)
    ref = schema.get("$ref")
    if isinstance(ref, str) and ref.startswith("#/$defs/"):
        def_name = ref[len("#/$defs/") :]
        defs = schema.get("$defs", {})
        if def_name in defs:
            resolved: dict[str, Any] = deepcopy(defs[def_name])
            resolved.setdefault("$defs", defs)
            return resolved
    return schema


def serialize_model_class(value: Any) -> Any:
    """Serialize a ``type[BaseModel]`` field value as its JSON schema.

    Args:
        value: A ``type[BaseModel]`` subclass, ``None``, or another union member.

    Returns:
        ``value.model_json_schema()`` for model classes, ``value`` otherwise.
    """
    if isinstance(value, type) and issubclass(value, BaseModel):
        return value.model_json_schema()
    return value


def create_model_from_schema(  # type: ignore[no-any-unimported]
    json_schema: dict[str, Any],
    *,
    root_schema: dict[str, Any] | None = None,
    model_name: str | None = None,
    enrich_descriptions: bool = False,
    __config__: ConfigDict | None = None,
    __base__: type[BaseModel] | None = None,
    __module__: str = __name__,
    __validators__: dict[str, AnyClassMethod] | None = None,
    __cls_kwargs__: dict[str, Any] | None = None,
) -> type[BaseModel]:
    """Create a Pydantic model from a JSON schema.

    This function takes a JSON schema as input and dynamically creates a Pydantic
    model class based on the schema. It supports various JSON schema features such
    as nested objects, referenced definitions ($ref), arrays with typed items,
    union types (anyOf/oneOf), and string formats.

    Args:
        json_schema: A dictionary representing the JSON schema.
        root_schema: The root schema containing $defs. If not provided, the
            current schema is treated as the root schema.
        model_name: Override for the model name. If not provided, the schema
            ``title`` field is used, falling back to ``"DynamicModel"``.
        enrich_descriptions: When True, augment field descriptions with
            constraint info (format, enum, pattern, min/max, examples) via
            :func:`build_rich_field_description`.  Useful for LLM-facing tool
            schemas where constraints in the description help the model
            understand parameter requirements.
        __config__: Pydantic configuration for the generated model.
        __base__: Base class for the generated model. Defaults to BaseModel.
        __module__: Module name for the generated model class.
        __validators__: A dictionary of custom validators for the generated model.
        __cls_kwargs__: Additional keyword arguments for the generated model class.

    Returns:
        A dynamically created Pydantic model class based on the provided JSON schema.

    Example:
        >>> schema = {
        ...     "title": "Person",
        ...     "type": "object",
        ...     "properties": {
        ...         "name": {"type": "string"},
        ...         "age": {"type": "integer"},
        ...     },
        ...     "required": ["name"],
        ... }
        >>> Person = create_model_from_schema(schema)
        >>> person = Person(name="John", age=30)
        >>> person.name
        'John'
    """
    try:
        json_schema = dict(jsonref.replace_refs(json_schema, proxies=False))
    except (jsonref.JsonRefError, RecursionError):
        json_schema = _inline_top_level_ref(json_schema)

    effective_root = root_schema or json_schema

    json_schema = force_additional_properties_false(json_schema)
    effective_root = force_additional_properties_false(effective_root)

    in_progress: dict[int, Any] = {}
    model = _build_model_from_schema(
        json_schema,
        effective_root,
        model_name=model_name,
        enrich_descriptions=enrich_descriptions,
        in_progress=in_progress,
        __config__=__config__,
        __base__=__base__,
        __module__=__module__,
        __validators__=__validators__,
        __cls_kwargs__=__cls_kwargs__,
    )

    types_namespace: dict[str, Any] = {
        entry.__name__: entry
        for entry in in_progress.values()
        if isinstance(entry, type) and issubclass(entry, BaseModel)
    }
    for entry in in_progress.values():
        if (
            isinstance(entry, type)
            and issubclass(entry, BaseModel)
            and not getattr(entry, "__pydantic_complete__", True)
        ):
            try:
                entry.model_rebuild(_types_namespace=types_namespace)
            except Exception as e:
                logger.debug("model_rebuild failed for %s: %s", entry.__name__, e)
    return model


def _build_model_from_schema(  # type: ignore[no-any-unimported]
    json_schema: dict[str, Any],
    effective_root: dict[str, Any],
    *,
    model_name: str | None,
    enrich_descriptions: bool,
    in_progress: dict[int, Any],
    __config__: ConfigDict | None = None,
    __base__: type[BaseModel] | None = None,
    __module__: str = __name__,
    __validators__: dict[str, AnyClassMethod] | None = None,
    __cls_kwargs__: dict[str, Any] | None = None,
) -> type[BaseModel]:
    """Inner builder shared by the public entry point and recursive nested-object creation.

    Preprocessing via ``jsonref.replace_refs`` and the sanitization walkers is
    run once by the public entry; this helper walks the already-normalized
    schema and emits Pydantic models. ``in_progress`` maps ``id(schema)`` to
    the model being built for that schema, so a cyclic ``$ref`` graph
    degrades to a ``ForwardRef`` back-edge instead of blowing the stack.
    """
    original_id = id(json_schema)
    if "allOf" in json_schema:
        json_schema = _merge_all_of_schemas(json_schema["allOf"], effective_root)

    effective_name = model_name or json_schema.get("title") or "DynamicModel"

    schema_id = id(json_schema)
    in_progress[original_id] = effective_name
    if schema_id != original_id:
        in_progress[schema_id] = effective_name

    field_definitions = {
        name: _json_schema_to_pydantic_field(
            name,
            prop,
            json_schema.get("required", []),
            effective_root,
            enrich_descriptions=enrich_descriptions,
            in_progress=in_progress,
        )
        for name, prop in (json_schema.get("properties", {}) or {}).items()
    }

    effective_config = __config__ or ConfigDict(extra="forbid")

    model = create_model_base(
        effective_name,
        __config__=effective_config,
        __base__=__base__,
        __module__=__module__,
        __validators__=__validators__,
        __cls_kwargs__=__cls_kwargs__,
        **field_definitions,
    )
    in_progress[original_id] = model
    if schema_id != original_id:
        in_progress[schema_id] = model
    return model


def _json_schema_to_pydantic_field(
    name: str,
    json_schema: dict[str, Any],
    required: list[str],
    root_schema: dict[str, Any],
    *,
    enrich_descriptions: bool = False,
    in_progress: dict[int, Any] | None = None,
) -> Any:
    """Convert a JSON schema property to a Pydantic field definition.

    Args:
        name: The field name.
        json_schema: The JSON schema for this field.
        required: List of required field names.
        root_schema: The root schema for resolving $ref.
        enrich_descriptions: When True, embed constraints in the description.

    Returns:
        A tuple of (type, Field) for use with create_model.
    """
    type_ = _json_schema_to_pydantic_type(
        json_schema,
        root_schema,
        name_=name.title(),
        enrich_descriptions=enrich_descriptions,
        in_progress=in_progress,
    )
    is_required = name in required

    field_params: dict[str, Any] = {}
    schema_extra: dict[str, Any] = {}

    if enrich_descriptions:
        rich_desc = build_rich_field_description(json_schema)
        if rich_desc:
            field_params["description"] = rich_desc
    else:
        description = json_schema.get("description")
        if description:
            field_params["description"] = description

    examples = json_schema.get("examples")
    if examples:
        schema_extra["examples"] = examples

    default = (
        json_schema["default"]
        if "default" in json_schema
        else (... if is_required else None)
    )

    if isinstance(type_, type) and issubclass(type_, (int, float)):
        if "minimum" in json_schema:
            field_params["ge"] = json_schema["minimum"]
        if "exclusiveMinimum" in json_schema:
            field_params["gt"] = json_schema["exclusiveMinimum"]
        if "maximum" in json_schema:
            field_params["le"] = json_schema["maximum"]
        if "exclusiveMaximum" in json_schema:
            field_params["lt"] = json_schema["exclusiveMaximum"]
        if "multipleOf" in json_schema:
            field_params["multiple_of"] = json_schema["multipleOf"]

    format_ = json_schema.get("format")
    if format_ in FORMAT_TYPE_MAP:
        pydantic_type = FORMAT_TYPE_MAP[format_]

        if format_ == "password":
            if json_schema.get("writeOnly"):
                pydantic_type = SecretBytes
        elif format_ == "uri":
            allowed_schemes = json_schema.get("scheme")
            if allowed_schemes:
                if len(allowed_schemes) == 1 and allowed_schemes[0] == "http":
                    pydantic_type = HttpUrl
                elif len(allowed_schemes) == 1 and allowed_schemes[0] == "file":
                    pydantic_type = FileUrl

        type_ = pydantic_type

    if isinstance(type_, type) and issubclass(type_, str):
        if "minLength" in json_schema:
            field_params["min_length"] = json_schema["minLength"]
        if "maxLength" in json_schema:
            field_params["max_length"] = json_schema["maxLength"]
        if "pattern" in json_schema:
            field_params["pattern"] = json_schema["pattern"]

    if not is_required:
        type_ = Optional[type_]  # noqa: UP045 - ForwardRef does not support `|`

    if schema_extra:
        field_params["json_schema_extra"] = schema_extra

    return type_, Field(default, **field_params)


def _resolve_ref(ref: str, root_schema: dict[str, Any]) -> dict[str, Any]:
    """Resolve a $ref to its actual schema.

    Args:
        ref: The $ref string (e.g., "#/$defs/MyType").
        root_schema: The root schema containing $defs.

    Returns:
        The resolved schema dict.
    """
    from typing import cast

    ref_path = ref.split("/")
    if ref.startswith("#/$defs/"):
        ref_schema: dict[str, Any] = root_schema["$defs"]
        start_idx = 2
    else:
        ref_schema = root_schema
        start_idx = 1
    for path in ref_path[start_idx:]:
        ref_schema = cast(dict[str, Any], ref_schema[path])
    return ref_schema


def _merge_all_of_schemas(
    schemas: list[dict[str, Any]],
    root_schema: dict[str, Any],
) -> dict[str, Any]:
    """Merge multiple allOf schemas into a single schema.

    Combines properties and required fields from all schemas.

    Args:
        schemas: List of schemas to merge.
        root_schema: The root schema for resolving $ref.

    Returns:
        Merged schema with combined properties and required fields.
    """
    merged: dict[str, Any] = {"type": "object", "properties": {}, "required": []}

    for schema in schemas:
        if "$ref" in schema:
            schema = _resolve_ref(schema["$ref"], root_schema)

        if "properties" in schema:
            merged["properties"].update(schema["properties"])

        if "required" in schema:
            for field in schema["required"]:
                if field not in merged["required"]:
                    merged["required"].append(field)

        if "title" in schema and "title" not in merged:
            merged["title"] = schema["title"]

    return merged


def _json_schema_to_pydantic_type(
    json_schema: dict[str, Any],
    root_schema: dict[str, Any],
    *,
    name_: str | None = None,
    enrich_descriptions: bool = False,
    in_progress: dict[int, Any] | None = None,
) -> Any:
    """Convert a JSON schema to a Python/Pydantic type.

    Args:
        json_schema: The JSON schema to convert.
        root_schema: The root schema for resolving $ref.
        name_: Optional name for nested models.
        enrich_descriptions: Propagated to nested model creation.
        in_progress: Map of ``id(schema_dict)`` to the Pydantic model
            currently being built for that schema, or to a placeholder name
            as a plain ``str`` while the model is still being constructed.
            Populated by :func:`_build_model_from_schema`. Enables cycle
            detection so a self-referential ``$ref`` graph resolves to a
            :class:`ForwardRef` back-edge rather than recursing forever.

    Returns:
        A Python type corresponding to the JSON schema.
    """
    if in_progress is not None:
        cached = in_progress.get(id(json_schema))
        if isinstance(cached, str):
            return ForwardRef(cached)
        if cached is not None:
            return cached

    ref = json_schema.get("$ref")
    if ref:
        ref_schema = _resolve_ref(ref, root_schema)
        return _json_schema_to_pydantic_type(
            ref_schema,
            root_schema,
            name_=name_,
            enrich_descriptions=enrich_descriptions,
            in_progress=in_progress,
        )

    enum_values = json_schema.get("enum")
    if enum_values:
        return Literal[tuple(enum_values)]

    if "const" in json_schema:
        return Literal[json_schema["const"]]

    any_of_schemas = []
    if "anyOf" in json_schema or "oneOf" in json_schema:
        any_of_schemas = json_schema.get("anyOf", []) + json_schema.get("oneOf", [])
    if any_of_schemas:
        any_of_types = [
            _json_schema_to_pydantic_type(
                schema,
                root_schema,
                name_=f"{name_ or 'Union'}Option{i}",
                enrich_descriptions=enrich_descriptions,
                in_progress=in_progress,
            )
            for i, schema in enumerate(any_of_schemas)
        ]
        return Union[tuple(any_of_types)]  # noqa: UP007

    all_of_schemas = json_schema.get("allOf")
    if all_of_schemas:
        if len(all_of_schemas) == 1:
            return _json_schema_to_pydantic_type(
                all_of_schemas[0],
                root_schema,
                name_=name_,
                enrich_descriptions=enrich_descriptions,
                in_progress=in_progress,
            )
        if in_progress is not None:
            return _build_model_from_schema(
                json_schema,
                root_schema,
                model_name=name_,
                enrich_descriptions=enrich_descriptions,
                in_progress=in_progress,
            )
        merged = _merge_all_of_schemas(all_of_schemas, root_schema)
        return _json_schema_to_pydantic_type(
            merged,
            root_schema,
            name_=name_,
            enrich_descriptions=enrich_descriptions,
            in_progress=in_progress,
        )

    type_ = json_schema.get("type")

    if type_ == "string":
        return str
    if type_ == "integer":
        return int
    if type_ == "number":
        return float
    if type_ == "boolean":
        return bool
    if type_ == "array":
        items_schema = json_schema.get("items")
        if items_schema:
            item_type = _json_schema_to_pydantic_type(
                items_schema,
                root_schema,
                name_=name_,
                enrich_descriptions=enrich_descriptions,
                in_progress=in_progress,
            )
            return list[item_type]  # type: ignore[valid-type]
        return list
    if type_ == "object":
        properties = json_schema.get("properties")
        if properties:
            if in_progress is not None:
                return _build_model_from_schema(
                    json_schema,
                    root_schema,
                    model_name=name_,
                    enrich_descriptions=enrich_descriptions,
                    in_progress=in_progress,
                )
            json_schema_ = json_schema.copy()
            if json_schema_.get("title") is None:
                json_schema_["title"] = name_ or "DynamicModel"
            return create_model_from_schema(
                json_schema_,
                root_schema=root_schema,
                enrich_descriptions=enrich_descriptions,
            )
        return dict
    if type_ == "null":
        return None
    if type_ is None:
        return Any
    raise ValueError(f"Unsupported JSON schema type: {type_} from {json_schema}")