import json
import time
from collections.abc import Mapping, Sequence
from enum import Enum
from types import MappingProxyType
from typing import (
    TYPE_CHECKING,
    Any,
    Final,
    Literal,
    get_args,
)

import httpx
from openai._models import BaseModel as OpenAIObject
from openai.types.audio.transcription_create_params import (
    FileTypes as FileTypes,
)
from openai.types.chat.chat_completion import ChatCompletion as ChatCompletion
from openai.types.completion_usage import (
    CompletionTokensDetails,
    CompletionUsage,
    PromptTokensDetails,
)
from openai.types.moderation import Categories as Categories
from openai.types.moderation import (
    CategoryAppliedInputTypes as CategoryAppliedInputTypes,
)
from openai.types.moderation import CategoryScores as CategoryScores
from openai.types.moderation_create_response import Moderation as Moderation
from openai.types.moderation_create_response import (
    ModerationCreateResponse as ModerationCreateResponse,
)
from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    PrivateAttr,
    SkipValidation,
    field_serializer,
    field_validator,
)
from typing_extensions import NotRequired, ReadOnly, Required, TypedDict

from litellm._logging import verbose_logger
from litellm._uuid import uuid
from litellm.types.llms.base import (
    BaseLiteLLMOpenAIResponseObject,
    LiteLLMPydanticObjectBase,
)
from litellm.types.mcp import MCPServerCostInfo

from ..litellm_core_utils.core_helpers import map_finish_reason, process_response_headers
from .agents import LiteLLMSendMessageResponse
from .guardrails import GuardrailEventHooks
from .llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse
from .llms.base import HiddenParams
from .llms.openai import (
    AllMessageValues,
    Batch,
    ChatCompletionAnnotation,
    ChatCompletionReasoningItem,
    ChatCompletionRedactedThinkingBlock,
    ChatCompletionThinkingBlock,
    ChatCompletionToolCallChunk,
    ChatCompletionToolParam,
    ChatCompletionUsageBlock,
    FileSearchTool,
    FineTuningJob,
    ImageURLListItem,
    OpenAIChatCompletionChunk,
    OpenAIChatCompletionFinishReason,
    OpenAIFileObject,
    OpenAIRealtimeStreamList,
    ResponsesAPIResponse,
    WebSearchOptions,
)
from .rerank import RerankResponse as RerankResponse

if TYPE_CHECKING:
    from .vector_stores import VectorStoreSearchResponse
else:
    VectorStoreSearchResponse = Any


def _generate_id():  # private helper function
    return "chatcmpl-" + str(uuid.uuid4())


class SafeAttributeModel:
    """
    A base model that provides safe attribute access.
    """

    def __delattr__(self, name) -> None:
        # Dropping an unset optional field stored in __dict__ goes straight to
        # object.__delattr__, skipping pydantic's __delattr__ whose per-call
        # class getattr lookup and _check_frozen dominate response construction.
        try:
            if (
                name in type(self).__pydantic_fields__
                and name in self.__dict__
                and not type(self).model_config.get("frozen")
            ):
                object.__delattr__(self, name)
                return
            super().__delattr__(name)
        except AttributeError:
            # noop if attribute does not exist
            pass


class LiteLLMCommonStrings(Enum):
    redacted_by_litellm = "redacted by litellm. 'litellm.turn_off_message_logging=True'"
    llm_provider_not_provided = "Unmapped LLM provider for this endpoint. You passed model={model}, custom_llm_provider={custom_llm_provider}. Check supported provider and route: https://docs.litellm.ai/docs/providers"


SupportedCacheControls: Final = ["ttl", "s-maxage", "no-cache", "no-store"]


class CostPerToken(TypedDict, total=False):
    # Required base rates — kept under total=False so we can mark them
    # Required individually while leaving the cache rates NotRequired.
    input_cost_per_token: Required[float]
    output_cost_per_token: Required[float]
    cache_read_input_token_cost: float
    cache_creation_input_token_cost: float


class ProviderField(TypedDict):
    field_name: str
    field_type: Literal["string"]
    field_description: str
    field_value: str


class ProviderSpecificModelInfo(TypedDict, total=False):
    supports_system_messages: bool | None
    supports_response_schema: bool | None
    supports_vision: bool | None
    supports_function_calling: bool | None
    supports_tool_choice: bool | None
    supports_assistant_prefill: bool | None
    supports_prompt_caching: bool | None
    supports_prompt_cache_breakpoint: ReadOnly[bool | None]
    supports_computer_use: bool | None
    supports_audio_input: bool | None
    supports_embedding_image_input: bool | None
    supports_audio_output: bool | None
    supports_pdf_input: bool | None
    supports_native_streaming: bool | None
    supports_native_structured_output: bool | None
    supports_parallel_function_calling: bool | None
    supports_web_search: bool | None
    supports_reasoning: bool | None
    supports_adaptive_thinking: bool | None
    supports_legacy_thinking: ReadOnly[bool | None]
    thinking_always_on: ReadOnly[bool | None]
    supports_tool_search: bool | None
    supports_mid_conversation_system: bool | None
    supports_url_context: bool | None
    supports_none_reasoning_effort: bool | None
    supports_minimal_reasoning_effort: bool | None
    supports_low_reasoning_effort: bool | None
    supports_xhigh_reasoning_effort: bool | None
    supports_max_reasoning_effort: bool | None
    reasoning_effort_levels: ReadOnly[Sequence[str] | None]
    default_reasoning_effort: ReadOnly[Literal["none", "minimal", "low", "medium", "high", "xhigh"] | None]
    supports_output_config: bool | None
    supports_image_size: bool | None
    bedrock_output_config_effort_ceiling: Literal["low", "medium", "high", "max", "xhigh"] | None
    bedrock_converse_supports_strict_tools: bool | None


class SearchContextCostPerQuery(TypedDict, total=False):
    search_context_size_low: float
    search_context_size_medium: float
    search_context_size_high: float


class AgenticLoopParams(TypedDict, total=False):
    """
    Parameters passed to agentic loop hooks (e.g., WebSearch interception).

    Stored in logging_obj.model_call_details["agentic_loop_params"] to provide
    agentic hooks with the original request context needed for follow-up calls.
    """

    model: str
    """The model string with provider prefix (e.g., 'bedrock/invoke/...')"""

    custom_llm_provider: str
    """The LLM provider name (e.g., 'bedrock', 'anthropic')"""


class OffPeakWindow(TypedDict, total=False):
    """One off-peak rule: UTC time-of-day windows, optionally restricted to weekdays.

    hours_utc is a "HH:MM-HH:MM" string in UTC, or a list of them; a window may wrap past
    midnight and an equal-ended window covers the whole day. weekdays is a list of days the
    rule applies on, as ISO-8601 numbers (1 = Monday .. 7 = Sunday) or English day names;
    omitted means every day. The weekday is read on the calendar named by the block's
    weekday_timezone.
    """

    hours_utc: ReadOnly[str | Sequence[str]]
    weekdays: ReadOnly[Sequence[int | str]]


class OffPeakPricing(TypedDict, total=False):
    """Time-windowed off-peak rates for providers that discount by time of day (e.g. DeepSeek).

    hours_utc is a "HH:MM-HH:MM" string in UTC, or a list of them for multiple daily windows,
    applying on every day of the week; a window may wrap past midnight. windows adds
    day-of-week-qualified rules (e.g. weekend-only whole-day off-peak), matched as a union
    with hours_utc. weekday_timezone names the IANA calendar weekdays are read on, defaulting
    to UTC. Any rate left unset falls back to the standard rate.
    """

    hours_utc: ReadOnly[str | Sequence[str]]
    windows: ReadOnly[Sequence[OffPeakWindow]]
    weekday_timezone: ReadOnly[str]
    input_cost_per_token: ReadOnly[float]
    output_cost_per_token: ReadOnly[float]
    cache_read_input_token_cost: ReadOnly[float]


class ModelInfoBase(ProviderSpecificModelInfo, total=False):
    key: Required[str]  # the key in litellm.model_cost which is returned

    max_tokens: Required[int | None]
    max_input_tokens: Required[int | None]
    max_output_tokens: Required[int | None]
    input_cost_per_token: Required[float | None]
    input_cost_per_token_flex: float | None  # OpenAI flex service tier pricing
    input_cost_per_token_priority: float | None  # OpenAI priority service tier pricing
    input_cost_per_token_ultrafast: ReadOnly[float | None]  # OpenAI ultrafast service tier pricing
    cache_creation_input_token_cost: float | None
    cache_creation_input_token_cost_above_200k_tokens: float | None
    cache_creation_input_token_cost_above_272k_tokens: float | None
    cache_creation_input_token_cost_above_272k_tokens_priority: float | None
    cache_creation_input_token_cost_above_272k_tokens_flex: float | None
    cache_creation_input_token_cost_above_1hr: float | None
    cache_creation_input_token_cost_flex: float | None  # OpenAI flex service tier pricing
    cache_creation_input_token_cost_priority: float | None  # OpenAI priority service tier pricing
    cache_creation_input_token_cost_ultrafast: ReadOnly[float | None]  # OpenAI ultrafast service tier pricing
    cache_read_input_token_cost: float | None
    cache_read_input_token_cost_flex: float | None  # OpenAI flex service tier pricing
    cache_read_input_token_cost_priority: float | None  # OpenAI priority service tier pricing
    cache_read_input_token_cost_ultrafast: ReadOnly[float | None]  # OpenAI ultrafast service tier pricing
    cache_read_input_token_cost_above_200k_tokens: float | None
    cache_read_input_token_cost_above_200k_tokens_priority: float | None
    cache_read_input_token_cost_above_272k_tokens: float | None
    cache_read_input_token_cost_above_272k_tokens_priority: float | None
    cache_read_input_token_cost_above_272k_tokens_flex: float | None
    cache_read_input_token_cost_above_512k_tokens: float | None
    # Smallest prefix this model will actually cache, whatever caching mechanism its provider uses.
    # Absent means the provider-agnostic default applies; see MINIMUM_PROMPT_CACHE_TOKEN_COUNT.
    prompt_cache_min_tokens: int | None
    off_peak_pricing: ReadOnly[OffPeakPricing | None]  # time-windowed off-peak rates
    input_cost_per_character: float | None  # only for vertex ai models
    input_cost_per_audio_token: float | None
    input_cost_per_token_above_128k_tokens: float | None  # only for vertex ai models
    input_cost_per_token_above_200k_tokens: float | None  # only for vertex ai gemini-2.5-pro models
    input_cost_per_token_above_200k_tokens_priority: float | None
    input_cost_per_token_above_272k_tokens: float | None  # GPT-5.4/5.4-pro: prompts >272K priced at 2x input
    input_cost_per_token_above_272k_tokens_priority: float | None
    input_cost_per_token_above_272k_tokens_flex: float | None
    input_cost_per_token_above_512k_tokens: float | None  # MiniMax-M3: prompts >512K priced at 2x input
    input_cost_per_character_above_128k_tokens: float | None  # only for vertex ai models
    input_cost_per_query: float | None  # only for rerank models
    input_cost_per_image: float | None  # only for vertex ai models
    input_cost_per_image_token: float | None  # for gpt-image-1 and similar models
    input_cost_per_video_token: float | None  # for gemini omni models with video input
    input_cost_per_audio_per_second: float | None  # only for vertex ai models
    input_cost_per_video_per_second: float | None  # only for vertex ai models
    input_cost_per_second: float | None  # for OpenAI Speech models
    input_cost_per_token_batches: float | None
    output_cost_per_token_batches: float | None
    output_cost_per_token: Required[float | None]
    output_cost_per_token_flex: float | None  # OpenAI flex service tier pricing
    output_cost_per_token_priority: float | None  # OpenAI priority service tier pricing
    output_cost_per_token_ultrafast: ReadOnly[float | None]  # OpenAI ultrafast service tier pricing
    regional_processing_uplift_multiplier_eu: (
        float | None
    )  # OpenAI EU data-residency uplift multiplier applied to all token costs (e.g. 1.10 = +10%)
    regional_processing_uplift_multiplier_us: (
        float | None
    )  # OpenAI US data-residency uplift multiplier applied to all token costs (e.g. 1.10 = +10%)
    regional_endpoint_uplift_multiplier: ReadOnly[
        float | None
    ]  # Vertex AI non-global (regional) endpoint uplift multiplier applied to all token costs (e.g. 1.10 = +10%)
    output_cost_per_character: float | None  # only for vertex ai models
    output_cost_per_audio_token: float | None
    output_cost_per_token_above_128k_tokens: float | None  # only for vertex ai models
    output_cost_per_token_above_200k_tokens: float | None  # only for vertex ai gemini-2.5-pro models
    output_cost_per_token_above_200k_tokens_priority: float | None
    output_cost_per_token_above_272k_tokens: float | None  # GPT-5.4/5.4-pro: prompts >272K priced at 1.5x output
    output_cost_per_token_above_272k_tokens_priority: float | None
    output_cost_per_token_above_272k_tokens_flex: float | None
    output_cost_per_token_above_512k_tokens: float | None  # MiniMax-M3: prompts >512K priced at 2x output
    output_cost_per_character_above_128k_tokens: float | None  # only for vertex ai models
    output_cost_per_image: float | None
    output_cost_per_image_token: float | None
    output_cost_per_video_token: float | None  # for gemini omni models with video output
    output_vector_size: int | None
    output_cost_per_reasoning_token: float | None
    output_cost_per_reasoning_token_flex: float | None
    output_cost_per_reasoning_token_priority: float | None
    output_cost_per_video_per_second: float | None  # only for vertex ai models
    output_cost_per_audio_per_second: float | None  # only for vertex ai models
    output_cost_per_second: float | None  # for OpenAI Speech models
    output_cost_per_second_1080p: (
        float | None
    )  # video_generation tier: key output_cost_per_second_<resolution> (e.g. 1080p, 720p)
    output_cost_per_second_480p: ReadOnly[float | None]
    output_cost_per_second_4k: ReadOnly[float | None]
    ocr_cost_per_page: float | None  # for OCR models
    ocr_cost_per_credit: float | None  # for OCR models priced by credit
    annotation_cost_per_page: float | None  # for OCR models
    search_context_cost_per_query: SearchContextCostPerQuery | None  # Cost for using web search tool
    web_search_billing_unit: (
        Literal["per_query", "per_prompt"] | None
    )  # "per_query" (Gemini 3.x) or "per_prompt" (Gemini 2.x)
    google_maps_grounding_cost_per_query: ReadOnly[float | None]
    citation_cost_per_token: float | None  # Cost per citation token for Perplexity
    tiered_pricing: list[dict[str, Any]] | None  # Tiered pricing structure for models like Dashscope
    litellm_provider: Required[str]
    mode: Required[
        Literal[
            "completion",
            "embedding",
            "image_generation",
            "chat",
            "audio_transcription",
            "responses",
            "ocr",
            "realtime",
        ]
    ]
    supported_endpoints: list[str] | None
    use_openai_responses_path: bool | None
    tpm: int | None
    rpm: int | None
    provider_specific_entry: dict[str, float] | None
    uses_embed_content: bool | None


class ModelInfo(ModelInfoBase, total=False):
    """
    Model info for a given model, this is information found in litellm.model_prices_and_context_window.json
    """

    supported_openai_params: Required[list[str] | None]


class GenericStreamingChunk(TypedDict, total=False):
    text: Required[str]
    tool_use: ChatCompletionToolCallChunk | None
    is_finished: Required[bool]
    finish_reason: Required[str]
    usage: Required[ChatCompletionUsageBlock | None]
    index: int

    # use this dict if you want to return any provider specific fields in the response
    provider_specific_fields: dict[str, Any] | None


from enum import Enum


class CallTypes(str, Enum):
    embedding = "embedding"
    aembedding = "aembedding"
    completion = "completion"
    acompletion = "acompletion"
    atext_completion = "atext_completion"
    text_completion = "text_completion"
    image_generation = "image_generation"
    aimage_generation = "aimage_generation"
    image_edit = "image_edit"
    aimage_edit = "aimage_edit"
    moderation = "moderation"
    amoderation = "amoderation"
    atranscription = "atranscription"
    transcription = "transcription"
    aspeech = "aspeech"
    speech = "speech"
    rerank = "rerank"
    arerank = "arerank"
    search = "search"
    asearch = "asearch"
    arealtime = "_arealtime"
    aresponses_websocket = "_aresponses_websocket"
    create_batch = "create_batch"
    acreate_batch = "acreate_batch"
    aretrieve_batch = "aretrieve_batch"
    retrieve_batch = "retrieve_batch"
    acancel_batch = "acancel_batch"
    cancel_batch = "cancel_batch"
    pass_through = "pass_through_endpoint"
    anthropic_messages = "anthropic_messages"
    aanthropic_messages = "aanthropic_messages"
    get_assistants = "get_assistants"
    aget_assistants = "aget_assistants"
    create_assistants = "create_assistants"
    acreate_assistants = "acreate_assistants"
    delete_assistant = "delete_assistant"
    adelete_assistant = "adelete_assistant"
    acreate_thread = "acreate_thread"
    create_thread = "create_thread"
    aget_thread = "aget_thread"
    get_thread = "get_thread"
    a_add_message = "a_add_message"
    add_message = "add_message"
    aget_messages = "aget_messages"
    get_messages = "get_messages"
    arun_thread = "arun_thread"
    run_thread = "run_thread"
    arun_thread_stream = "arun_thread_stream"
    run_thread_stream = "run_thread_stream"
    afile_retrieve = "afile_retrieve"
    file_retrieve = "file_retrieve"
    afile_delete = "afile_delete"
    file_delete = "file_delete"
    afile_list = "afile_list"
    file_list = "file_list"
    acreate_file = "acreate_file"
    create_file = "create_file"
    afile_content = "afile_content"
    file_content = "file_content"
    create_fine_tuning_job = "create_fine_tuning_job"
    acreate_fine_tuning_job = "acreate_fine_tuning_job"

    #########################################################
    # Video Generation Call Types
    #########################################################
    create_video = "create_video"
    acreate_video = "acreate_video"
    avideo_retrieve = "avideo_retrieve"
    video_retrieve = "video_retrieve"
    avideo_content = "avideo_content"
    video_content = "video_content"
    video_remix = "video_remix"
    avideo_remix = "avideo_remix"
    video_list = "video_list"
    avideo_list = "avideo_list"
    video_retrieve_job = "video_retrieve_job"
    avideo_retrieve_job = "avideo_retrieve_job"
    video_delete = "video_delete"
    avideo_delete = "avideo_delete"
    video_create_character = "video_create_character"
    avideo_create_character = "avideo_create_character"
    video_get_character = "video_get_character"
    avideo_get_character = "avideo_get_character"
    video_edit = "video_edit"
    avideo_edit = "avideo_edit"
    video_extension = "video_extension"
    avideo_extension = "avideo_extension"
    vector_store_file_create = "vector_store_file_create"
    avector_store_file_create = "avector_store_file_create"
    vector_store_file_list = "vector_store_file_list"
    avector_store_file_list = "avector_store_file_list"
    vector_store_file_retrieve = "vector_store_file_retrieve"
    avector_store_file_retrieve = "avector_store_file_retrieve"
    vector_store_file_content = "vector_store_file_content"
    avector_store_file_content = "avector_store_file_content"
    vector_store_file_update = "vector_store_file_update"
    avector_store_file_update = "avector_store_file_update"
    vector_store_file_delete = "vector_store_file_delete"
    avector_store_file_delete = "avector_store_file_delete"
    vector_store_create = "vector_store_create"
    avector_store_create = "avector_store_create"
    vector_store_search = "vector_store_search"
    avector_store_search = "avector_store_search"

    ingest = "ingest"
    aingest = "aingest"
    query = "query"
    aquery = "aquery"

    #########################################################
    # Google Interactions API Call Types
    #########################################################
    create_interaction = "create_interaction"
    acreate_interaction = "acreate_interaction"

    #########################################################
    # Container Call Types
    #########################################################
    create_container = "create_container"
    acreate_container = "acreate_container"
    list_containers = "list_containers"
    alist_containers = "alist_containers"
    retrieve_container = "retrieve_container"
    aretrieve_container = "aretrieve_container"
    delete_container = "delete_container"
    adelete_container = "adelete_container"
    list_container_files = "list_container_files"
    alist_container_files = "alist_container_files"
    upload_container_file = "upload_container_file"
    aupload_container_file = "aupload_container_file"
    create_sandbox = "create_sandbox"
    acreate_sandbox = "acreate_sandbox"
    delete_sandbox = "delete_sandbox"
    adelete_sandbox = "adelete_sandbox"
    run_code = "run_code"
    arun_code = "arun_code"
    code_interpreter_tool = "code_interpreter_tool"
    acode_interpreter_tool = "acode_interpreter_tool"

    acancel_fine_tuning_job = "acancel_fine_tuning_job"
    cancel_fine_tuning_job = "cancel_fine_tuning_job"
    alist_fine_tuning_jobs = "alist_fine_tuning_jobs"
    list_fine_tuning_jobs = "list_fine_tuning_jobs"
    aretrieve_fine_tuning_job = "aretrieve_fine_tuning_job"
    retrieve_fine_tuning_job = "retrieve_fine_tuning_job"
    responses = "responses"
    aresponses = "aresponses"
    alist_input_items = "alist_input_items"
    llm_passthrough_route = "llm_passthrough_route"
    allm_passthrough_route = "allm_passthrough_route"

    #########################################################
    # Google GenAI Native Call Types
    #########################################################
    generate_content = "generate_content"
    agenerate_content = "agenerate_content"
    generate_content_stream = "generate_content_stream"
    agenerate_content_stream = "agenerate_content_stream"

    #########################################################
    # OCR Call Types
    #########################################################
    ocr = "ocr"
    aocr = "aocr"

    #########################################################
    # MCP Call Types
    #########################################################
    call_mcp_tool = "call_mcp_tool"
    list_mcp_tools = "list_mcp_tools"

    #########################################################
    # A2A Call Types
    #########################################################
    asend_message = "asend_message"
    send_message = "send_message"

    #########################################################
    # Claude Code Call Types
    #########################################################
    acreate_skill = "acreate_skill"


CallTypesLiteral = Literal[
    "embedding",
    "aembedding",
    "completion",
    "acompletion",
    "atext_completion",
    "text_completion",
    "image_generation",
    "aimage_generation",
    "image_edit",
    "aimage_edit",
    "moderation",
    "amoderation",
    "atranscription",
    "transcription",
    "aspeech",
    "speech",
    "rerank",
    "arerank",
    "search",
    "asearch",
    "_arealtime",
    "create_batch",
    "acreate_batch",
    "create_file",
    "acreate_file",
    "pass_through_endpoint",
    "allm_passthrough_route",
    "anthropic_messages",
    "aanthropic_messages",
    "aretrieve_batch",
    "retrieve_batch",
    "generate_content",
    "agenerate_content",
    "generate_content_stream",
    "agenerate_content_stream",
    "ocr",
    "aocr",
    "vector_store_create",
    "avector_store_create",
    "vector_store_search",
    "avector_store_search",
    "vector_store_file_create",
    "avector_store_file_create",
    "vector_store_file_list",
    "avector_store_file_list",
    "vector_store_file_retrieve",
    "avector_store_file_retrieve",
    "vector_store_file_content",
    "avector_store_file_content",
    "vector_store_file_update",
    "avector_store_file_update",
    "vector_store_file_delete",
    "avector_store_file_delete",
    "call_mcp_tool",
    "list_mcp_tools",
    "asend_message",
    "send_message",
    "aresponses",
    "responses",
    "acreate_skill",
    "acreate_realtime_client_secret",
    "arealtime_calls",
    "acreate_realtime_transcription_session",
]

# Mapping of API routes to their corresponding call types
API_ROUTE_TO_CALL_TYPES: Final[Mapping[str, Sequence[CallTypes]]] = {
    # Chat Completions
    "/chat/completions": [CallTypes.acompletion, CallTypes.completion],
    "/v1/chat/completions": [CallTypes.acompletion, CallTypes.completion],
    "/engines/{model}/chat/completions": [CallTypes.acompletion, CallTypes.completion],
    "/openai/deployments/{model}/chat/completions": [
        CallTypes.acompletion,
        CallTypes.completion,
    ],
    # Text Completions
    "/completions": [CallTypes.atext_completion, CallTypes.text_completion],
    "/v1/completions": [CallTypes.atext_completion, CallTypes.text_completion],
    "/engines/{model}/completions": [
        CallTypes.atext_completion,
        CallTypes.text_completion,
    ],
    "/openai/deployments/{model}/completions": [
        CallTypes.atext_completion,
        CallTypes.text_completion,
    ],
    # Embeddings
    "/embeddings": [CallTypes.aembedding, CallTypes.embedding],
    "/v1/embeddings": [CallTypes.aembedding, CallTypes.embedding],
    "/engines/{model}/embeddings": [CallTypes.aembedding, CallTypes.embedding],
    "/openai/deployments/{model}/embeddings": [
        CallTypes.aembedding,
        CallTypes.embedding,
    ],
    # Image Generation
    "/images/generations": [CallTypes.aimage_generation, CallTypes.image_generation],
    "/v1/images/generations": [CallTypes.aimage_generation, CallTypes.image_generation],
    "/engines/{model}/images/generations": [
        CallTypes.aimage_generation,
        CallTypes.image_generation,
    ],
    "/openai/deployments/{model}/images/generations": [
        CallTypes.aimage_generation,
        CallTypes.image_generation,
    ],
    # Image Edits
    "/images/edits": [CallTypes.aimage_edit, CallTypes.image_edit],
    "/v1/images/edits": [CallTypes.aimage_edit, CallTypes.image_edit],
    # Audio Transcriptions
    "/audio/transcriptions": [CallTypes.atranscription, CallTypes.transcription],
    "/v1/audio/transcriptions": [CallTypes.atranscription, CallTypes.transcription],
    # Audio Speech
    "/audio/speech": [CallTypes.aspeech, CallTypes.speech],
    "/v1/audio/speech": [CallTypes.aspeech, CallTypes.speech],
    # Moderations
    "/moderations": [CallTypes.amoderation, CallTypes.moderation],
    "/v1/moderations": [CallTypes.amoderation, CallTypes.moderation],
    # Rerank
    "/rerank": [CallTypes.arerank, CallTypes.rerank],
    "/v1/rerank": [CallTypes.arerank, CallTypes.rerank],
    "/v2/rerank": [CallTypes.arerank, CallTypes.rerank],
    # Search
    "/search": [CallTypes.asearch, CallTypes.search],
    "/v1/search": [CallTypes.asearch, CallTypes.search],
    # Batches
    "/batches": [CallTypes.acreate_batch, CallTypes.create_batch],
    "/v1/batches": [CallTypes.acreate_batch, CallTypes.create_batch],
    "/batches/{batch_id}": [CallTypes.aretrieve_batch, CallTypes.retrieve_batch],
    "/v1/batches/{batch_id}": [CallTypes.aretrieve_batch, CallTypes.retrieve_batch],
    # Files
    "/files": [
        CallTypes.acreate_file,
        CallTypes.create_file,
        CallTypes.afile_list,
        CallTypes.file_list,
    ],
    "/v1/files": [
        CallTypes.acreate_file,
        CallTypes.create_file,
        CallTypes.afile_list,
        CallTypes.file_list,
    ],
    "/files/{file_id}": [
        CallTypes.afile_retrieve,
        CallTypes.file_retrieve,
        CallTypes.afile_delete,
        CallTypes.file_delete,
    ],
    "/v1/files/{file_id}": [
        CallTypes.afile_retrieve,
        CallTypes.file_retrieve,
        CallTypes.afile_delete,
        CallTypes.file_delete,
    ],
    "/files/{file_id}/content": [CallTypes.afile_content, CallTypes.file_content],
    "/v1/files/{file_id}/content": [CallTypes.afile_content, CallTypes.file_content],
    # Assistants
    "/assistants": [
        CallTypes.aget_assistants,
        CallTypes.get_assistants,
        CallTypes.acreate_assistants,
        CallTypes.create_assistants,
    ],
    "/v1/assistants": [
        CallTypes.aget_assistants,
        CallTypes.get_assistants,
        CallTypes.acreate_assistants,
        CallTypes.create_assistants,
    ],
    "/assistants/{assistant_id}": [
        CallTypes.adelete_assistant,
        CallTypes.delete_assistant,
    ],
    "/v1/assistants/{assistant_id}": [
        CallTypes.adelete_assistant,
        CallTypes.delete_assistant,
    ],
    # Threads
    "/threads": [CallTypes.acreate_thread, CallTypes.create_thread],
    "/v1/threads": [CallTypes.acreate_thread, CallTypes.create_thread],
    "/threads/{thread_id}": [CallTypes.aget_thread, CallTypes.get_thread],
    "/v1/threads/{thread_id}": [CallTypes.aget_thread, CallTypes.get_thread],
    # Thread Messages
    "/threads/{thread_id}/messages": [
        CallTypes.a_add_message,
        CallTypes.add_message,
        CallTypes.aget_messages,
        CallTypes.get_messages,
    ],
    "/v1/threads/{thread_id}/messages": [
        CallTypes.a_add_message,
        CallTypes.add_message,
        CallTypes.aget_messages,
        CallTypes.get_messages,
    ],
    # Thread Runs
    "/threads/{thread_id}/runs": [
        CallTypes.arun_thread,
        CallTypes.run_thread,
        CallTypes.arun_thread_stream,
        CallTypes.run_thread_stream,
    ],
    "/v1/threads/{thread_id}/runs": [
        CallTypes.arun_thread,
        CallTypes.run_thread,
        CallTypes.arun_thread_stream,
        CallTypes.run_thread_stream,
    ],
    # Fine-tuning Jobs
    "/fine_tuning/jobs": [
        CallTypes.acreate_fine_tuning_job,
        CallTypes.create_fine_tuning_job,
        CallTypes.alist_fine_tuning_jobs,
        CallTypes.list_fine_tuning_jobs,
    ],
    "/v1/fine_tuning/jobs": [
        CallTypes.acreate_fine_tuning_job,
        CallTypes.create_fine_tuning_job,
        CallTypes.alist_fine_tuning_jobs,
        CallTypes.list_fine_tuning_jobs,
    ],
    "/fine_tuning/jobs/{fine_tuning_job_id}": [
        CallTypes.aretrieve_fine_tuning_job,
        CallTypes.retrieve_fine_tuning_job,
    ],
    "/v1/fine_tuning/jobs/{fine_tuning_job_id}": [
        CallTypes.aretrieve_fine_tuning_job,
        CallTypes.retrieve_fine_tuning_job,
    ],
    "/fine_tuning/jobs/{fine_tuning_job_id}/cancel": [
        CallTypes.acancel_fine_tuning_job,
        CallTypes.cancel_fine_tuning_job,
    ],
    "/v1/fine_tuning/jobs/{fine_tuning_job_id}/cancel": [
        CallTypes.acancel_fine_tuning_job,
        CallTypes.cancel_fine_tuning_job,
    ],
    # Video Generation
    "/videos": [
        CallTypes.acreate_video,
        CallTypes.create_video,
        CallTypes.avideo_list,
        CallTypes.video_list,
    ],
    "/v1/videos": [
        CallTypes.acreate_video,
        CallTypes.create_video,
        CallTypes.avideo_list,
        CallTypes.video_list,
    ],
    "/videos/{video_id}": [
        CallTypes.avideo_retrieve,
        CallTypes.video_retrieve,
        CallTypes.avideo_delete,
        CallTypes.video_delete,
    ],
    "/v1/videos/{video_id}": [
        CallTypes.avideo_retrieve,
        CallTypes.video_retrieve,
        CallTypes.avideo_delete,
        CallTypes.video_delete,
    ],
    "/videos/{video_id}/content": [CallTypes.avideo_content, CallTypes.video_content],
    "/v1/videos/{video_id}/content": [
        CallTypes.avideo_content,
        CallTypes.video_content,
    ],
    "/videos/{video_id}/remix": [CallTypes.avideo_remix, CallTypes.video_remix],
    "/v1/videos/{video_id}/remix": [CallTypes.avideo_remix, CallTypes.video_remix],
    "/videos/characters": [
        CallTypes.avideo_create_character,
        CallTypes.video_create_character,
    ],
    "/v1/videos/characters": [
        CallTypes.avideo_create_character,
        CallTypes.video_create_character,
    ],
    "/videos/characters/{character_id}": [
        CallTypes.avideo_get_character,
        CallTypes.video_get_character,
    ],
    "/v1/videos/characters/{character_id}": [
        CallTypes.avideo_get_character,
        CallTypes.video_get_character,
    ],
    "/videos/edits": [CallTypes.avideo_edit, CallTypes.video_edit],
    "/v1/videos/edits": [CallTypes.avideo_edit, CallTypes.video_edit],
    "/videos/extensions": [CallTypes.avideo_extension, CallTypes.video_extension],
    "/v1/videos/extensions": [CallTypes.avideo_extension, CallTypes.video_extension],
    # Vector Stores
    "/vector_stores": [CallTypes.avector_store_create, CallTypes.vector_store_create],
    "/v1/vector_stores": [
        CallTypes.avector_store_create,
        CallTypes.vector_store_create,
    ],
    "/vector_stores/{vector_store_id}/search": [
        CallTypes.avector_store_search,
        CallTypes.vector_store_search,
    ],
    "/v1/vector_stores/{vector_store_id}/search": [
        CallTypes.avector_store_search,
        CallTypes.vector_store_search,
    ],
    "/vector_stores/{vector_store_id}/files": [
        CallTypes.avector_store_file_create,
        CallTypes.vector_store_file_create,
        CallTypes.avector_store_file_list,
        CallTypes.vector_store_file_list,
    ],
    "/v1/vector_stores/{vector_store_id}/files": [
        CallTypes.avector_store_file_create,
        CallTypes.vector_store_file_create,
        CallTypes.avector_store_file_list,
        CallTypes.vector_store_file_list,
    ],
    "/vector_stores/{vector_store_id}/files/{file_id}": [
        CallTypes.avector_store_file_retrieve,
        CallTypes.vector_store_file_retrieve,
        CallTypes.avector_store_file_delete,
        CallTypes.vector_store_file_delete,
    ],
    "/v1/vector_stores/{vector_store_id}/files/{file_id}": [
        CallTypes.avector_store_file_retrieve,
        CallTypes.vector_store_file_retrieve,
        CallTypes.avector_store_file_delete,
        CallTypes.vector_store_file_delete,
    ],
    "/vector_stores/{vector_store_id}/files/{file_id}/content": [
        CallTypes.avector_store_file_content,
        CallTypes.vector_store_file_content,
    ],
    "/v1/vector_stores/{vector_store_id}/files/{file_id}/content": [
        CallTypes.avector_store_file_content,
        CallTypes.vector_store_file_content,
    ],
    "/vector_stores/{vector_store_id}/files/{file_id}/update": [
        CallTypes.avector_store_file_update,
        CallTypes.vector_store_file_update,
    ],
    "/v1/vector_stores/{vector_store_id}/files/{file_id}/update": [
        CallTypes.avector_store_file_update,
        CallTypes.vector_store_file_update,
    ],
    # Containers
    "/containers": [
        CallTypes.acreate_container,
        CallTypes.create_container,
        CallTypes.alist_containers,
        CallTypes.list_containers,
    ],
    "/v1/containers": [
        CallTypes.acreate_container,
        CallTypes.create_container,
        CallTypes.alist_containers,
        CallTypes.list_containers,
    ],
    "/containers/{container_id}": [
        CallTypes.aretrieve_container,
        CallTypes.retrieve_container,
        CallTypes.adelete_container,
        CallTypes.delete_container,
    ],
    "/v1/containers/{container_id}": [
        CallTypes.aretrieve_container,
        CallTypes.retrieve_container,
        CallTypes.adelete_container,
        CallTypes.delete_container,
    ],
    # Responses API
    "/responses": (CallTypes.aresponses, CallTypes.responses),
    "/v1/responses": (CallTypes.aresponses, CallTypes.responses),
    "/openai/v1/responses": (CallTypes.aresponses, CallTypes.responses),
    "/responses/{response_id}": (CallTypes.aresponses, CallTypes.responses),
    "/v1/responses/{response_id}": (CallTypes.aresponses, CallTypes.responses),
    "/openai/v1/responses/{response_id}": (CallTypes.aresponses, CallTypes.responses),
    "/responses/{response_id}/input_items": (CallTypes.alist_input_items,),
    "/v1/responses/{response_id}/input_items": (CallTypes.alist_input_items,),
    "/openai/v1/responses/{response_id}/input_items": (CallTypes.alist_input_items,),
    # Realtime API
    "/realtime": [CallTypes.arealtime],
    "/v1/realtime": [CallTypes.arealtime],
    "/openai/v1/realtime": [CallTypes.arealtime],
    # Provider-specific routes
    "/anthropic/v1/messages": [CallTypes.anthropic_messages],
    # Google GenAI routes
    "/generate_content": [CallTypes.agenerate_content, CallTypes.generate_content],
    "/models/{model}:generateContent": [
        CallTypes.agenerate_content,
        CallTypes.generate_content,
    ],
    "/generate_content_stream": [
        CallTypes.agenerate_content_stream,
        CallTypes.generate_content_stream,
    ],
    "/models/{model}:streamGenerateContent": [
        CallTypes.agenerate_content_stream,
        CallTypes.generate_content_stream,
    ],
    # MCP (Model Context Protocol)
    "/mcp/call_tool": [CallTypes.call_mcp_tool],
    # A2A (Agent-to-Agent)
    "/a2a/{agent_id}": [CallTypes.asend_message, CallTypes.send_message],
    "/a2a/{agent_id}/message/send": [CallTypes.asend_message, CallTypes.send_message],
    # Passthrough endpoints
    "/llm_passthrough": [
        CallTypes.llm_passthrough_route,
        CallTypes.allm_passthrough_route,
    ],
    "/v1/llm_passthrough": [
        CallTypes.llm_passthrough_route,
        CallTypes.allm_passthrough_route,
    ],
    "/v1/messages": [CallTypes.anthropic_messages],
    # OCR
    "/ocr": [CallTypes.aocr, CallTypes.ocr],
    "/v1/ocr": [CallTypes.aocr, CallTypes.ocr],
}


class PassthroughCallTypes(Enum):
    passthrough_image_generation = "passthrough-image-generation"


class TopLogprob(OpenAIObject):
    token: str
    """The token."""

    bytes: list[int] | None = None
    """A list of integers representing the UTF-8 bytes representation of the token.

    Useful in instances where characters are represented by multiple tokens and
    their byte representations must be combined to generate the correct text
    representation. Can be `null` if there is no bytes representation for the token.
    """

    logprob: float
    """The log probability of this token, if it is within the top 20 most likely
    tokens.

    Otherwise, the value `-9999.0` is used to signify that the token is very
    unlikely.
    """


class ChatCompletionTokenLogprob(OpenAIObject):
    token: str
    """The token."""

    bytes: list[int] | None = None
    """A list of integers representing the UTF-8 bytes representation of the token.

    Useful in instances where characters are represented by multiple tokens and
    their byte representations must be combined to generate the correct text
    representation. Can be `null` if there is no bytes representation for the token.
    """

    logprob: float
    """The log probability of this token, if it is within the top 20 most likely
    tokens.

    Otherwise, the value `-9999.0` is used to signify that the token is very
    unlikely.
    """

    top_logprobs: list[TopLogprob]
    """List of the most likely tokens and their log probability, at this token
    position.

    In rare cases, there may be fewer than the number of requested `top_logprobs`
    returned.
    """

    # Some OpenAI-compatible providers return null for top_logprobs when
    # omitted; normalize to [] to preserve the typed List[TopLogprob] contract.
    @field_validator("top_logprobs", mode="before")
    @classmethod
    def ensure_top_logprobs_is_list(cls, v):
        """Normalize null top_logprobs to empty list.

        Some providers return null instead of [] when logprobs=true but
        top_logprobs is unset. The OpenAI spec requires an array.
        Fixes https://github.com/BerriAI/litellm/issues/21932
        """
        if v is None:
            return []
        return v

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)


class ChoiceLogprobs(OpenAIObject):
    content: list[ChatCompletionTokenLogprob] | None = None
    """A list of message content tokens with log probability information."""

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)


class FunctionCall(OpenAIObject):
    arguments: str
    name: str | None = None


class Function(OpenAIObject):
    arguments: str
    name: str | None  # can be None - openai e.g.: ChoiceDeltaToolCallFunction(arguments='{"', name=None), type=None)

    def __init__(
        self,
        arguments: dict | str | None = None,
        name: str | None = None,
        **params,
    ) -> None:
        if arguments is None:
            if params.get("parameters", None) is not None and isinstance(params["parameters"], dict):
                arguments = json.dumps(params["parameters"])
                params.pop("parameters")
            else:
                arguments = ""
        elif isinstance(arguments, dict):
            arguments = json.dumps(arguments)
        else:
            arguments = arguments

        name = name

        # Build a dictionary with the structure your BaseModel expects
        data: Final = {"arguments": arguments, "name": name}

        super().__init__(**data)

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


class ChatCompletionDeltaToolCall(OpenAIObject):
    id: str | None = None
    function: Function
    type: str | None = None
    index: int

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


class _CustomToolCallAccess(OpenAIObject):
    def __contains__(self, key) -> bool:
        return hasattr(self, key)

    def get(self, key, default=None):
        return getattr(self, key, default)

    def __getitem__(self, key):
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        setattr(self, key, value)


class ChatCompletionCustomToolCallPayload(_CustomToolCallAccess):
    name: str
    input: str


class ChatCompletionDeltaCustomToolCallPayload(_CustomToolCallAccess):
    name: str | None = None
    input: str | None = None


class ChatCompletionMessageCustomToolCall(_CustomToolCallAccess):
    id: str
    type: Literal["custom"] = "custom"
    custom: ChatCompletionCustomToolCallPayload


class ChatCompletionDeltaCustomToolCall(_CustomToolCallAccess):
    id: str | None = None
    type: str | None = None
    custom: ChatCompletionDeltaCustomToolCallPayload
    index: int


class ChatCompletionMessageToolCall(OpenAIObject):
    def __init__(
        self,
        function: dict | Function,
        id: str | None = None,
        type: str | None = None,
        **params,
    ) -> None:
        super().__init__(**params)
        if isinstance(function, dict):
            self.function = Function(**function)
        else:
            self.function = function

        if id is not None:
            self.id = id
        else:
            self.id = f"{uuid.uuid4()}"

        if type is not None:
            self.type = type
        else:
            self.type = "function"

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


def is_custom_tool_call_dict(tool_call: Mapping[str, Any]) -> bool:
    return tool_call.get("type") == "custom" or tool_call.get("custom") is not None


def chat_completion_tool_call_from_dict(
    tool_call: Mapping[str, Any],
) -> "ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall":
    if is_custom_tool_call_dict(tool_call):
        return ChatCompletionMessageCustomToolCall(
            **MappingProxyType({k: v for k, v in tool_call.items() if not (k in ("function", "type") and v is None)})
        )
    return ChatCompletionMessageToolCall(**tool_call)


from openai.types.chat.chat_completion_audio import ChatCompletionAudio


class ChatCompletionAudioResponse(ChatCompletionAudio):
    def __init__(
        self,
        data: str,
        expires_at: int,
        transcript: str,
        id: str | None = None,
        **params,
    ) -> None:
        if id is not None:
            id = id
        else:
            id = f"{uuid.uuid4()}"
        super().__init__(data=data, expires_at=expires_at, transcript=transcript, id=id, **params)

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


"""
Reference:
ChatCompletionMessage(content='This is a test', role='assistant', function_call=None, tool_calls=None))
"""


def add_provider_specific_fields(object: BaseModel, provider_specific_fields: dict[str, Any] | None) -> None:
    if not provider_specific_fields:  # set if provider_specific_fields is not empty
        return
    object.provider_specific_fields = provider_specific_fields  # rebind-ok: sets the field on the caller's model


class Message(SafeAttributeModel, OpenAIObject):
    content: str | None
    role: Literal["assistant", "user", "system", "tool", "function"]
    tool_calls: (
        list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall] | None
    )  # mutable-ok: public pydantic response field; only the union member is new
    function_call: FunctionCall | None
    audio: ChatCompletionAudioResponse | None = None
    images: list[ImageURLListItem] | None = None
    reasoning_content: str | None = None
    thinking_blocks: list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock] | None = None
    reasoning_items: list[ChatCompletionReasoningItem] | None = None
    provider_specific_fields: dict[str, Any] | None = Field(default=None)
    annotations: list[ChatCompletionAnnotation] | None = None

    def __init__(
        self,
        content: str | None = None,
        role: Literal["assistant", "user", "system", "tool", "function"] = "assistant",
        function_call=None,
        tool_calls: list | None = None,
        audio: ChatCompletionAudioResponse | None = None,
        images: list[ImageURLListItem] | None = None,
        provider_specific_fields: dict[str, Any] | None = None,
        reasoning_content: str | None = None,
        thinking_blocks: list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock] | None = None,
        reasoning_items: list[ChatCompletionReasoningItem] | None = None,
        annotations: list[ChatCompletionAnnotation] | None = None,
        **params,
    ) -> None:
        init_values: Final[dict[str, Any]] = {
            "content": content,
            "role": role or "assistant",  # handle null input
            "function_call": (FunctionCall(**function_call) if function_call is not None else None),
            "tool_calls": (
                [
                    (chat_completion_tool_call_from_dict(tool_call) if isinstance(tool_call, dict) else tool_call)
                    for tool_call in tool_calls
                ]
                if tool_calls is not None and len(tool_calls) > 0
                else None
            ),
        }

        if audio is not None:
            init_values["audio"] = audio

        if images is not None:
            init_values["images"] = images

        if thinking_blocks is not None:
            init_values["thinking_blocks"] = thinking_blocks

        if reasoning_items is not None:
            init_values["reasoning_items"] = reasoning_items

        if annotations is not None:
            init_values["annotations"] = annotations

        if reasoning_content is not None:
            init_values["reasoning_content"] = reasoning_content

        super().__init__(
            **init_values,
            **params,
        )

        if audio is None:
            # delete audio from self
            # OpenAI compatible APIs like mistral API will raise an error if audio is passed in
            if hasattr(self, "audio"):
                del self.audio

        if images is None and hasattr(self, "images"):
            del self.images

        if annotations is None:
            # ensure default response matches OpenAI spec
            # Some OpenAI compatible APIs raise an error if annotations are passed in
            if hasattr(self, "annotations"):
                del self.annotations

        if reasoning_content is None:
            # ensure default response matches OpenAI spec
            if hasattr(self, "reasoning_content"):
                del self.reasoning_content

        if thinking_blocks is None:
            # ensure default response matches OpenAI spec
            if hasattr(self, "thinking_blocks"):
                del self.thinking_blocks

        if reasoning_items is None:
            # ensure default response matches OpenAI spec
            if hasattr(self, "reasoning_items"):
                del self.reasoning_items

        add_provider_specific_fields(self, provider_specific_fields)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class Delta(SafeAttributeModel, OpenAIObject):
    if TYPE_CHECKING:
        # Stored in __pydantic_extra__ at runtime (extra='allow'), set directly in
        # __init__ rather than via self.<attr> = .... Declared here only so type
        # checkers still see them as attributes for consumers that read delta.content
        # etc.; the runtime branch is skipped so pydantic does not treat them as fields.
        content: str | None
        role: str | None
        function_call: FunctionCall | None
        tool_calls: (
            list[ChatCompletionDeltaToolCall | ChatCompletionDeltaCustomToolCall] | None
        )  # mutable-ok: public pydantic response field; only the union member is new
        audio: ChatCompletionAudioResponse | None
        images: list[ImageURLListItem] | None
        annotations: list[ChatCompletionAnnotation] | None

    reasoning_content: str | None = None
    thinking_blocks: list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock] | None = None
    reasoning_items: list[ChatCompletionReasoningItem] | None = None
    provider_specific_fields: dict[str, Any] | None = Field(default=None)

    def __init__(
        self,
        content=None,
        role=None,
        function_call=None,
        tool_calls=None,
        audio: ChatCompletionAudioResponse | None = None,
        images: list[ImageURLListItem] | None = None,
        reasoning_content: str | None = None,
        thinking_blocks: list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock] | None = None,
        reasoning_items: list[ChatCompletionReasoningItem] | None = None,
        annotations: list[ChatCompletionAnnotation] | None = None,
        **params,
    ) -> None:
        # Map 'reasoning' to 'reasoning_content' for providers that return
        # delta.reasoning (e.g., Cerebras, Groq gpt-oss models).
        # Must be done before super().__init__ to prevent 'reasoning' from
        # leaking as an extra attribute on the parent model.
        if reasoning_content is None and "reasoning" in params:
            reasoning_content = params.pop("reasoning", None)

        super().__init__(**params)
        add_provider_specific_fields(self, params.get("provider_specific_fields", {}))

        if function_call is not None and isinstance(function_call, dict):
            function_call = FunctionCall(**function_call)

        if tool_calls is not None and isinstance(tool_calls, (list, tuple)):
            coerced_tool_calls: list[
                ChatCompletionDeltaToolCall | ChatCompletionDeltaCustomToolCall
            ] = []  # mutable-ok: public Delta.tool_calls contract is a list
            current_index = 0
            for tool_call in tool_calls:
                if isinstance(tool_call, dict):
                    if tool_call.get("index", None) is None:
                        tool_call["index"] = current_index
                        current_index += 1
                    if is_custom_tool_call_dict(tool_call):
                        coerced_tool_calls.append(
                            ChatCompletionDeltaCustomToolCall(
                                **MappingProxyType(
                                    {k: v for k, v in tool_call.items() if not (k == "function" and v is None)}
                                )
                            )
                        )
                    else:
                        if tool_call.get("type", None) is None:
                            tool_call["type"] = "function"
                        coerced_tool_calls.append(ChatCompletionDeltaToolCall(**tool_call))
                elif isinstance(tool_call, (ChatCompletionDeltaToolCall, ChatCompletionDeltaCustomToolCall)):
                    coerced_tool_calls.append(tool_call)
            tool_calls = coerced_tool_calls

        # Build the per-chunk state directly instead of round-tripping every
        # field through pydantic's __setattr__/__delattr__ (the dominant
        # streaming cost). These keys are not declared model fields, so they
        # live in __pydantic_extra__; the slow path set each of content, role,
        # function_call, tool_calls, audio, images and annotations (marking them
        # in __pydantic_fields_set__) and then deleted the ones OpenAI omits.
        extra = self.__pydantic_extra__
        if extra is None:  # pragma: no cover - extra='allow' guarantees a dict
            extra = self.__pydantic_extra__ = {}
        fields_set: Final = self.__pydantic_fields_set__
        fields_set.update(
            (
                "content",
                "role",
                "function_call",
                "tool_calls",
                "audio",
                "images",
                "annotations",
            )
        )
        extra["content"] = content
        extra["role"] = role
        extra["function_call"] = function_call
        extra["tool_calls"] = tool_calls
        extra["audio"] = audio
        if images is not None and len(images) > 0:
            extra["images"] = images
        if annotations is not None:
            extra["annotations"] = annotations

        if reasoning_content is not None:
            self.reasoning_content = reasoning_content
        else:
            # ensure default response matches OpenAI spec
            del self.reasoning_content

        if thinking_blocks is not None:
            self.thinking_blocks = thinking_blocks
        else:
            # ensure default response matches OpenAI spec
            del self.thinking_blocks

        if reasoning_items is not None:
            self.reasoning_items = reasoning_items
        else:
            # ensure default response matches OpenAI spec
            if hasattr(self, "reasoning_items"):
                del self.reasoning_items

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


class Choices(SafeAttributeModel, OpenAIObject):
    finish_reason: OpenAIChatCompletionFinishReason
    index: int
    message: Message
    logprobs: ChoiceLogprobs | Any | None = None

    provider_specific_fields: dict[str, Any] | None = Field(default=None)

    def __init__(
        self,
        finish_reason=None,
        index=0,
        message: Message | dict | None = None,
        logprobs: ChoiceLogprobs | dict | Any | None = None,
        enhancements=None,
        provider_specific_fields: dict[str, Any] | None = None,
        **params,
    ) -> None:
        if finish_reason is not None:
            mapped: Final = map_finish_reason(finish_reason)
            params["finish_reason"] = mapped
            if finish_reason != mapped:
                provider_specific_fields = dict(provider_specific_fields) if provider_specific_fields else {}
                provider_specific_fields["native_finish_reason"] = finish_reason
        else:
            params["finish_reason"] = "stop"
        if index is not None:
            params["index"] = index
        else:
            params["index"] = 0
        if message is None:
            params["message"] = Message()
        else:
            if isinstance(message, Message):
                params["message"] = message
            elif isinstance(message, dict):
                params["message"] = Message(**message)
            elif isinstance(message, BaseModel):
                # Normalize provider/OpenAI SDK message models into LiteLLM's Message type.
                dump: Final = message.model_dump() if hasattr(message, "model_dump") else message.dict()
                params["message"] = Message(**dump)
        if logprobs is not None:
            if isinstance(logprobs, dict):
                params["logprobs"] = ChoiceLogprobs(**logprobs)
            else:
                params["logprobs"] = logprobs
        else:
            params["logprobs"] = None
        super().__init__(**params)

        if enhancements is not None:
            self.enhancements = enhancements

        self.provider_specific_fields = provider_specific_fields

        if self.logprobs is None:
            del self.logprobs
        if self.provider_specific_fields is None:
            del self.provider_specific_fields

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


class CompletionTokensDetailsWrapper(CompletionTokensDetails):  # wrapper for older openai versions
    text_tokens: int | None = None
    """Text tokens generated by the model."""

    image_tokens: int | None = None
    """Image tokens generated by the model."""

    video_tokens: int | None = None
    """Video tokens generated by the model."""


class CacheCreationTokenDetails(BaseModel):
    ephemeral_5m_input_tokens: int | None = None
    ephemeral_1h_input_tokens: int | None = None


class PromptTokensDetailsWrapper(
    SafeAttributeModel, PromptTokensDetails
):  # extends with image generation fields (text_tokens, image_tokens)
    text_tokens: int | None = None
    """Text tokens sent to the model."""

    image_tokens: int | None = None
    """Image tokens sent to the model."""

    video_tokens: int | None = None
    """Video tokens sent to the model."""

    web_search_requests: int | None = None
    """Number of web search requests made by the tool call. Used for Anthropic to calculate web search cost."""

    google_maps_grounding_requests: int | None = None
    """Number of Grounding with Google Maps requests made by the tool call. Used for Gemini to calculate Maps cost."""

    tool_use_tokens: int | None = None
    """Prompt tokens consumed by server-side tool use (e.g. Gemini grounding via googleSearch)."""

    character_count: int | None = None
    """Character count sent to the model. Used for Vertex AI multimodal embeddings."""

    image_count: int | None = None
    """Number of images sent to the model. Used for Vertex AI multimodal embeddings."""

    video_length_seconds: float | None = None
    """Length of videos sent to the model. Used for Vertex AI multimodal embeddings."""

    audio_length_seconds: float | None = None
    """Length of audio sent to the model. Used for multimodal embeddings priced per audio-second."""

    cache_write_tokens: int | None = None
    """Number of cache write (creation) tokens sent to the model. OpenAI naming (prompt_tokens_details.cache_write_tokens); this is the canonical field."""

    cache_creation_tokens: int | None = None
    """Number of cache creation tokens sent to the model. Anthropic/Bedrock naming; kept in sync with cache_write_tokens (assigning either mirrors to the other)."""

    cache_creation_token_details: CacheCreationTokenDetails | None = None
    """Details of cache creation tokens sent to the model. Used for tracking 5m/1h cache creation tokens for Anthropic prompt caching."""

    def __setattr__(self, name: str, value: object) -> None:
        super().__setattr__(name, value)
        if name == "cache_write_tokens":
            super().__setattr__("cache_creation_tokens", value)
        elif name == "cache_creation_tokens":
            super().__setattr__("cache_write_tokens", value)

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        extra_fields: Final = self.model_extra
        nested_cache_creation_input_tokens: Final = (
            extra_fields.get("cache_creation_input_tokens") if extra_fields is not None else None
        )
        self.cache_write_tokens = (
            self.cache_write_tokens
            if self.cache_write_tokens is not None
            else (
                self.cache_creation_tokens
                if self.cache_creation_tokens is not None
                else (
                    nested_cache_creation_input_tokens if isinstance(nested_cache_creation_input_tokens, int) else None
                )
            )
        )
        if self.character_count is None:
            del self.character_count
        if self.image_count is None:
            del self.image_count
        if self.video_length_seconds is None:
            del self.video_length_seconds
        if self.audio_length_seconds is None:
            del self.audio_length_seconds
        if self.web_search_requests is None:
            del self.web_search_requests
        if self.google_maps_grounding_requests is None:
            del self.google_maps_grounding_requests
        if self.tool_use_tokens is None:
            del self.tool_use_tokens
        if self.cache_write_tokens is None:
            del self.cache_write_tokens
        if self.cache_creation_tokens is None:
            del self.cache_creation_tokens
        if self.cache_creation_token_details is None:
            del self.cache_creation_token_details


class ServerToolUse(BaseModel):
    web_search_requests: int | None = None
    tool_search_requests: int | None = None
    browser_open_requests: int | None = None

    def __getitem__(self, key: str) -> int | None:
        if key not in self.__class__.model_fields:
            raise KeyError(key)
        return getattr(self, key)


class Usage(SafeAttributeModel, CompletionUsage):
    _cache_creation_input_tokens: int = PrivateAttr(
        0
    )  # hidden param for prompt caching. Might change, once openai introduces their equivalent.
    _cache_read_input_tokens: int = PrivateAttr(
        0
    )  # hidden param for prompt caching. Might change, once openai introduces their equivalent.

    server_tool_use: ServerToolUse | None = None
    cost: float | None = None

    completion_tokens_details: CompletionTokensDetailsWrapper | None = None
    """Breakdown of tokens used in a completion."""

    prompt_tokens_details: PromptTokensDetailsWrapper | None = None
    """Breakdown of tokens used in the prompt."""

    def __init__(
        self,
        prompt_tokens: int | None = None,
        completion_tokens: int | None = None,
        total_tokens: int | None = None,
        reasoning_tokens: int | None = None,
        prompt_tokens_details: PromptTokensDetailsWrapper | PromptTokensDetails | dict | None = None,
        completion_tokens_details: CompletionTokensDetailsWrapper | dict | None = None,
        server_tool_use: ServerToolUse | dict | None = None,
        cost: float | None = None,
        **params,
    ) -> None:
        # handle reasoning_tokens
        _completion_tokens_details: CompletionTokensDetailsWrapper | None = None

        # First, handle existing completion_tokens_details
        if completion_tokens_details:
            if isinstance(completion_tokens_details, dict):
                _completion_tokens_details = CompletionTokensDetailsWrapper(**completion_tokens_details)
            elif isinstance(completion_tokens_details, CompletionTokensDetails):
                _completion_tokens_details = completion_tokens_details

        # Handle reasoning_tokens and auto-calculate text_tokens if needed
        if reasoning_tokens:
            # Ensure we have a details object to work with
            if _completion_tokens_details is None:
                _completion_tokens_details = CompletionTokensDetailsWrapper()

            # Set reasoning_tokens if not already set by provider
            if _completion_tokens_details.reasoning_tokens is None:
                _completion_tokens_details.reasoning_tokens = reasoning_tokens

            # Auto-calculate text_tokens only if provider didn't set it explicitly
            # Formula: text_tokens = completion_tokens - reasoning_tokens - image_tokens - audio_tokens
            if _completion_tokens_details.text_tokens is None and completion_tokens is not None:
                calculated_text_tokens = completion_tokens - reasoning_tokens

                # Subtract other modality tokens if present
                if _completion_tokens_details.image_tokens:
                    calculated_text_tokens -= _completion_tokens_details.image_tokens
                if _completion_tokens_details.audio_tokens:
                    calculated_text_tokens -= _completion_tokens_details.audio_tokens

                # Prevent negative token counts from inconsistent data
                _completion_tokens_details.text_tokens = max(0, calculated_text_tokens)

        # handle prompt_tokens_details
        _prompt_tokens_details: PromptTokensDetailsWrapper | None = None

        # guarantee prompt_token_details is always a PromptTokensDetailsWrapper
        if prompt_tokens_details:
            if isinstance(prompt_tokens_details, dict):
                _prompt_tokens_details = PromptTokensDetailsWrapper(**prompt_tokens_details)
            elif isinstance(prompt_tokens_details, PromptTokensDetails):
                _prompt_tokens_details = PromptTokensDetailsWrapper(**prompt_tokens_details.model_dump())
            elif isinstance(prompt_tokens_details, PromptTokensDetailsWrapper):
                _prompt_tokens_details = prompt_tokens_details

        ## DEEPSEEK MAPPING ##
        if "prompt_cache_hit_tokens" in params and isinstance(params["prompt_cache_hit_tokens"], int):
            if _prompt_tokens_details is None:
                _prompt_tokens_details = PromptTokensDetailsWrapper(cached_tokens=params["prompt_cache_hit_tokens"])
            else:
                _prompt_tokens_details.cached_tokens = params["prompt_cache_hit_tokens"]

        ## ANTHROPIC MAPPING ##
        if "cache_read_input_tokens" in params and isinstance(params["cache_read_input_tokens"], int):
            if _prompt_tokens_details is None:
                _prompt_tokens_details = PromptTokensDetailsWrapper(cached_tokens=params["cache_read_input_tokens"])
            else:
                _prompt_tokens_details.cached_tokens = params["cache_read_input_tokens"]

        if "cache_creation_input_tokens" in params and isinstance(params["cache_creation_input_tokens"], int):
            if _prompt_tokens_details is None:
                _prompt_tokens_details = PromptTokensDetailsWrapper(
                    cache_write_tokens=params["cache_creation_input_tokens"]
                )
            else:
                _prompt_tokens_details.cache_write_tokens = params["cache_creation_input_tokens"]

        super().__init__(
            prompt_tokens=prompt_tokens or 0,
            completion_tokens=completion_tokens or 0,
            total_tokens=total_tokens or 0,
            completion_tokens_details=_completion_tokens_details or None,
            prompt_tokens_details=_prompt_tokens_details or None,
        )

        if isinstance(server_tool_use, dict):
            server_tool_use = ServerToolUse(**server_tool_use)

        if server_tool_use is not None:
            self.server_tool_use = server_tool_use
        else:  # maintain openai compatibility in usage object if possible
            del self.server_tool_use

        if cost is not None:
            self.cost = cost
        else:
            del self.cost

        ## ANTHROPIC MAPPING ##
        if "cache_creation_input_tokens" in params and isinstance(params["cache_creation_input_tokens"], int):
            self._cache_creation_input_tokens = params["cache_creation_input_tokens"]

        if "cache_read_input_tokens" in params and isinstance(params["cache_read_input_tokens"], int):
            self._cache_read_input_tokens = params["cache_read_input_tokens"]

        ## DEEPSEEK MAPPING ##
        if "prompt_cache_hit_tokens" in params and isinstance(params["prompt_cache_hit_tokens"], int):
            self._cache_read_input_tokens = params["prompt_cache_hit_tokens"]

        for k, v in params.items():
            setattr(self, k, v)

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


class StreamingChoices(OpenAIObject):
    def __init__(
        self,
        finish_reason=None,
        index=0,
        delta: Delta | None = None,
        logprobs=None,
        enhancements=None,
        **params,
    ) -> None:
        # Fix Perplexity return both delta and message cause OpenWebUI repect text
        # https://github.com/BerriAI/litellm/issues/8455
        params.pop("message", None)
        super().__init__(**params)
        if finish_reason:
            self.finish_reason = map_finish_reason(finish_reason)
        else:
            self.finish_reason = None
        self.index = index
        if delta is not None:
            if isinstance(delta, Delta):
                self.delta = delta
            elif isinstance(delta, dict):
                self.delta = Delta(**delta)
        else:
            self.delta = Delta()
        if enhancements is not None:
            self.enhancements = enhancements

        if logprobs is not None and isinstance(logprobs, dict):
            self.logprobs = ChoiceLogprobs(**logprobs)
        else:
            self.logprobs = logprobs

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


class StreamingChatCompletionChunk(OpenAIChatCompletionChunk):
    def __init__(self, **kwargs) -> None:
        new_choices: Final = []
        for choice in kwargs["choices"]:
            new_choice = StreamingChoices(**choice).model_dump()
            new_choices.append(new_choice)
        kwargs["choices"] = new_choices

        super().__init__(**kwargs)


class ModelResponseBase(OpenAIObject):
    id: str
    """A unique identifier for the completion."""

    created: int
    """The Unix timestamp (in seconds) of when the completion was created."""

    model: str | None = None
    """The model used for completion."""

    object: str
    """The object type, which is always "text_completion" """

    system_fingerprint: str | None = None
    """This fingerprint represents the backend configuration that the model runs with.

    Can be used in conjunction with the `seed` request parameter to understand when
    backend changes have been made that might impact determinism.
    """

    _hidden_params: dict = {}

    _response_headers: dict | None = None

    def set_provider_response_headers(self, headers: httpx.Headers) -> None:
        """Surface a provider's raw response headers to the caller as `llm_provider-*` headers."""
        self._hidden_params["additional_headers"] = process_response_headers(headers)

    def model_dump(self, **kwargs):
        """Default to exclude_unset to avoid Pydantic serializer warnings for OpenAIObject-derived types."""
        if "exclude_unset" not in kwargs and "exclude_none" not in kwargs:
            kwargs["exclude_unset"] = True
        return super().model_dump(**kwargs)


class ModelResponseStream(ModelResponseBase):
    choices: list[StreamingChoices]
    provider_specific_fields: dict[str, Any] | None = Field(default=None)

    def __init__(
        self,
        choices: list[StreamingChoices] | StreamingChoices | dict | BaseModel | None = None,
        id: str | None = None,
        created: int | None = None,
        provider_specific_fields: dict[str, Any] | None = None,
        **kwargs,
    ) -> None:
        if choices is not None and isinstance(choices, list):
            new_choices: Final = []
            for choice in choices:
                _new_choice = None
                if isinstance(choice, StreamingChoices):
                    _new_choice = choice
                elif isinstance(choice, dict):
                    _new_choice = StreamingChoices(**choice)
                elif isinstance(choice, BaseModel):
                    _new_choice = StreamingChoices(**choice.model_dump())
                new_choices.append(_new_choice)
            kwargs["choices"] = new_choices
        else:
            kwargs["choices"] = [StreamingChoices()]

        if id is None:
            id = _generate_id()
        else:
            id = id
        if created is None:
            created = int(time.time())
        else:
            created = created

        usage_to_set = None
        if "usage" in kwargs and kwargs["usage"] is not None:
            if isinstance(kwargs["usage"], dict):
                usage_to_set = Usage(**kwargs["usage"])
                kwargs["usage"] = usage_to_set
            elif isinstance(kwargs["usage"], BaseModel):
                dump: Final = (
                    kwargs["usage"].model_dump() if hasattr(kwargs["usage"], "model_dump") else kwargs["usage"].dict()
                )
                usage_to_set = Usage(**dump)
                kwargs["usage"] = usage_to_set

        kwargs["id"] = id
        kwargs["created"] = created
        kwargs["object"] = "chat.completion.chunk"
        kwargs["provider_specific_fields"] = provider_specific_fields

        super().__init__(**kwargs)

        if usage_to_set is not None:
            self.usage = usage_to_set

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class ModelResponse(ModelResponseBase):
    choices: list[Choices]
    """The list of completion choices the model generated for the input prompt."""

    def __init__(
        self,
        id=None,
        choices=None,
        created=None,
        model=None,
        object=None,
        system_fingerprint=None,
        usage=None,
        stream=None,
        stream_options=None,
        response_ms=None,
        hidden_params=None,
        _response_headers=None,
        **params,
    ) -> None:
        object = "chat.completion"
        if choices is not None and isinstance(choices, list):
            new_choices: Final = []
            for choice in choices:
                if isinstance(choice, Choices):
                    _new_choice = choice
                elif isinstance(choice, dict):
                    _new_choice = Choices(**choice)
                elif isinstance(choice, BaseModel):
                    dump = choice.model_dump() if hasattr(choice, "model_dump") else choice.dict()
                    _new_choice = Choices(**dump)
                else:
                    _new_choice = choice
                new_choices.append(_new_choice)
            choices = new_choices
        else:
            choices = [Choices()]
        if id is None:
            id = _generate_id()
        else:
            id = id
        if created is None:
            created = int(time.time())
        else:
            created = created
        model = model
        if usage is not None:
            if isinstance(usage, dict):
                usage = Usage(**usage)
            elif isinstance(usage, BaseModel):
                dump = usage.model_dump() if hasattr(usage, "model_dump") else usage.dict()
                usage = Usage(**dump)
            else:
                usage = usage
        elif stream is None or stream is False:
            usage = None  # avoid constructing throwaway Usage; set by convert_to_model_response_object
        if hidden_params:
            self._hidden_params = hidden_params

        if _response_headers:
            self._response_headers = _response_headers

        init_values: Final = {
            "id": id,
            "choices": choices,
            "created": created,
            "model": model,
            "object": object,
            "system_fingerprint": system_fingerprint,
        }

        if usage is not None:
            init_values["usage"] = usage

        super().__init__(
            **init_values,
            **params,
        )

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class Embedding(OpenAIObject):
    embedding: list | str = []
    index: int
    object: Literal["embedding"]

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


class EmbeddingResponse(OpenAIObject):
    model: str | None = None
    """The model used for embedding."""

    data: list
    """The actual embedding value"""

    object: Literal["list"]
    """The object type, which is always "list" """

    usage: Usage | None = None
    """Usage statistics for the embedding request."""

    _hidden_params: dict = {}
    _response_headers: dict | None = None
    _response_ms: float | None = None

    def __init__(
        self,
        model: str | None = None,
        usage: Usage | None = None,
        response_ms=None,
        data: list | list[Embedding] | None = None,
        hidden_params=None,
        _response_headers=None,
        **params,
    ) -> None:
        object: Final = "list"
        if response_ms:
            _response_ms = response_ms
        else:
            _response_ms = None
        if data:
            data = data
        else:
            data = []

        if usage:
            usage = usage
        else:
            usage = Usage()

        if _response_headers:
            self._response_headers = _response_headers

        model = model
        super().__init__(model=model, object=object, data=data, usage=usage)

        if hidden_params:
            self._hidden_params = hidden_params

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class Logprobs(OpenAIObject):
    text_offset: list[int] | None
    token_logprobs: list[float | None] | None
    tokens: list[str] | None
    top_logprobs: list[dict[str, float] | None] | None


class TextChoices(OpenAIObject):
    def __init__(self, finish_reason=None, index=0, text=None, logprobs=None, **params) -> None:
        super().__init__(**params)
        if finish_reason:
            self.finish_reason = map_finish_reason(finish_reason)
        else:
            self.finish_reason = None
        self.index = index
        if text is not None:
            self.text = text
        else:
            self.text = None
        if logprobs is None:
            self.logprobs = None
        else:
            if isinstance(logprobs, dict):
                self.logprobs = Logprobs(**logprobs)
            else:
                self.logprobs = logprobs

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class TextCompletionResponse(OpenAIObject):
    """
    {
        "id": response["id"],
        "object": "text_completion",
        "created": response["created"],
        "model": response["model"],
        "choices": [
        {
            "text": response["choices"][0]["message"]["content"],
            "index": response["choices"][0]["index"],
            "logprobs": transformed_logprobs,
            "finish_reason": response["choices"][0]["finish_reason"]
        }
        ],
        "usage": response["usage"]
    }
    """

    id: str
    object: str
    created: int
    model: str | None
    choices: list[TextChoices]
    usage: Usage | None
    _response_ms: int | None = None
    _hidden_params: HiddenParams

    def __init__(
        self,
        id=None,
        choices=None,
        created=None,
        model=None,
        usage=None,
        stream=False,
        response_ms=None,
        object=None,
        **params,
    ) -> None:
        if stream:
            object = "text_completion.chunk"
            choices = [TextChoices()]
        else:
            object = "text_completion"
            if choices is not None and isinstance(choices, list):
                new_choices: Final = []
                for choice in choices:
                    _new_choice = None
                    if isinstance(choice, TextChoices):
                        _new_choice = choice
                    elif isinstance(choice, dict):
                        _new_choice = TextChoices(**choice)
                    new_choices.append(_new_choice)
                choices = new_choices
            else:
                choices = [TextChoices()]
        if object is not None:
            object = object
        if id is None:
            id = _generate_id()
        else:
            id = id
        if created is None:
            created = int(time.time())
        else:
            created = created

        model = model
        if usage:
            usage = usage
        else:
            usage = Usage()

        super().__init__(
            id=id,
            object=object,
            created=created,
            model=model,
            choices=choices,
            usage=usage,
            **params,
        )

        if response_ms:
            self._response_ms = response_ms
        else:
            self._response_ms = None
        self._hidden_params = HiddenParams()

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)


from openai.types.images_response import Image as OpenAIImage


class ImageObject(OpenAIImage):
    """
    Represents the url or the content of an image generated by the OpenAI API.

    Attributes:
    b64_json: The base64-encoded JSON of the generated image, if response_format is b64_json.
    url: The URL of the generated image, if response_format is url (default).
    revised_prompt: The prompt that was used to generate the image, if there was any revision to the prompt.
    provider_specific_fields: Provider-specific fields not part of OpenAI spec.

    https://platform.openai.com/docs/api-reference/images/object
    """

    b64_json: str | None = None
    url: str | None = None
    revised_prompt: str | None = None
    provider_specific_fields: dict[str, Any] | None = None

    def __init__(
        self,
        b64_json=None,
        url=None,
        revised_prompt=None,
        provider_specific_fields=None,
        **kwargs,
    ) -> None:
        super().__init__(b64_json=b64_json, url=url, revised_prompt=revised_prompt)
        if provider_specific_fields:
            self.provider_specific_fields = provider_specific_fields

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class ImageUsageInputTokensDetails(BaseLiteLLMOpenAIResponseObject):
    image_tokens: int
    """The number of image tokens in the input prompt."""

    text_tokens: int
    """The number of text tokens in the input prompt."""


class ImageUsage(BaseLiteLLMOpenAIResponseObject):
    input_tokens: int
    """The number of tokens (images and text) in the input prompt."""

    input_tokens_details: ImageUsageInputTokensDetails
    """The input tokens detailed information for the image generation."""

    output_tokens: int
    """The number of image tokens in the output image."""

    total_tokens: int
    """The total number of tokens (images and text) used for the image generation."""


from openai.types.images_response import ImagesResponse as OpenAIImageResponse


class ImageResponse(OpenAIImageResponse, BaseLiteLLMOpenAIResponseObject):
    _hidden_params: dict = {}

    usage: ImageUsage | None = None
    """
    Users might use litellm with older python versions, we don't want this to break for them.
    Happens when their OpenAIImageResponse has the old OpenAI usage class.
    """

    model_config = ConfigDict(extra="allow", protected_namespaces=())

    def __init__(
        self,
        created: int | None = None,
        data: list[ImageObject] | None = None,
        response_ms=None,
        usage: ImageUsage | None = None,
        hidden_params: dict | None = None,
        **kwargs,
    ) -> None:
        if response_ms:
            _response_ms = response_ms
        else:
            _response_ms = None
        if data:
            data = data
        else:
            data = []

        if created:
            created = created
        else:
            created = int(time.time())

        _data: Final[list[OpenAIImage]] = []
        for d in data:
            if isinstance(d, dict):
                _data.append(ImageObject(**d))
            elif isinstance(d, BaseModel):
                _data.append(ImageObject(**d.model_dump()))

        _usage: Final = usage or ImageUsage(
            input_tokens=0,
            input_tokens_details=ImageUsageInputTokensDetails(
                image_tokens=0,
                text_tokens=0,
            ),
            output_tokens=0,
            total_tokens=0,
        )
        super().__init__(created=created, data=_data, usage=_usage)

        self.quality = kwargs.get("quality", None)
        self.output_format = kwargs.get("output_format", None)
        self.size = kwargs.get("size", None)
        self._hidden_params = hidden_params or {}

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class TranscriptionUsageDurationObject(BaseModel):
    type: Literal["duration"]
    seconds: float


class TranscriptionUsageInputTokenDetailsObject(BaseModel):
    audio_tokens: int
    text_tokens: int


class TranscriptionUsageTokensObject(BaseModel):
    type: Literal["tokens"]
    input_tokens: int
    output_tokens: int
    total_tokens: int
    input_token_details: TranscriptionUsageInputTokenDetailsObject


class TranscriptionResponse(OpenAIObject):
    text: str | None = None
    usage: TranscriptionUsageDurationObject | TranscriptionUsageTokensObject | None = None

    _hidden_params: dict = {}
    _response_headers: dict | None = None

    def __init__(self, text=None) -> None:
        super().__init__(text=text)

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def __setitem__(self, key, value) -> None:
        # Allow dictionary-style assignment of attributes
        setattr(self, key, value)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class GenericImageParsingChunk(TypedDict):
    type: str
    media_type: str
    data: str


class ResponseFormatChunk(TypedDict, total=False):
    type: Required[Literal["json_object", "text"]]
    response_schema: dict


class LoggedLiteLLMParams(TypedDict, total=False):
    force_timeout: float | None
    custom_llm_provider: str | None
    api_base: str | None
    litellm_call_id: str | None
    model_alias_map: dict | None
    metadata: dict | None
    litellm_metadata: dict | None
    model_info: dict | None
    proxy_server_request: dict | None
    acompletion: bool | None
    preset_cache_key: str | None
    no_log: bool | None
    input_cost_per_second: float | None
    input_cost_per_token: float | None
    output_cost_per_token: float | None
    output_cost_per_second: float | None
    cooldown_time: float | None


class AdapterCompletionStreamWrapper:
    def __init__(self, completion_stream) -> None:
        self.completion_stream = completion_stream

    def __iter__(self):
        return self

    def __aiter__(self):
        return self

    def __next__(self):
        try:
            for chunk in self.completion_stream:
                if chunk == "None" or chunk is None:
                    raise Exception
                return chunk
            raise StopIteration
        except StopIteration:
            raise StopIteration
        except Exception as e:
            verbose_logger.debug("AdapterCompletionStreamWrapper - %s", e)

    async def __anext__(self):
        try:
            async for chunk in self.completion_stream:
                if chunk == "None" or chunk is None:
                    raise Exception
                return chunk
            raise StopIteration
        except StopIteration:
            raise StopAsyncIteration


class StandardLoggingUserAPIKeyMetadata(TypedDict):
    user_api_key_hash: str | None  # hash of the litellm virtual key used
    user_api_key_alias: str | None
    user_api_key_spend: float | None
    user_api_key_max_budget: float | None
    user_api_key_budget_reset_at: str | None
    user_api_key_user_spend: float | None
    user_api_key_user_max_budget: float | None
    user_api_key_team_spend: float | None
    user_api_key_team_max_budget: float | None
    user_api_key_org_id: str | None
    user_api_key_org_alias: str | None
    user_api_key_team_id: str | None
    user_api_key_project_id: str | None
    user_api_key_project_alias: str | None
    user_api_key_user_id: str | None
    user_api_key_user_email: str | None
    user_api_key_team_alias: str | None
    user_api_key_end_user_id: str | None
    user_api_key_request_route: str | None
    user_api_key_auth_metadata: dict[str, str] | None


class StandardLoggingMCPToolCall(TypedDict, total=False):
    name: str
    """
    Name of the tool to call
    """
    arguments: dict
    """
    Arguments to pass to the tool
    """
    result: dict
    """
    Result of the tool call
    """

    mcp_server_name: str | None
    """
    Name of the MCP server that the tool call was made to
    """

    mcp_server_logo_url: str | None
    """
    Optional logo URL of the MCP server that the tool call was made to

    (this is to render the logo on the logs page on litellm ui)
    """

    namespaced_tool_name: str | None
    """
    Namespaced tool name of the MCP tool that the tool call was made to

    Includes the server name prefix if it exists - eg. `deepwiki-mcp/get_page_content`
    """

    mcp_server_cost_info: MCPServerCostInfo | None
    """
    Cost per query for the MCP server tool call
    """

    mcp_session_id: str | None
    """
    The MCP `mcp-session-id` of the stateful session this tool call ran in, when
    the client is driving a stateful session. Absent for stateless calls.
    """

    mcp_auth_mode: str | None
    """
    The server's auth_type for this call (e.g. `true_passthrough`, `oauth_delegate`,
    `oauth2`). For the client-forwarded token modes this records that the caller's own
    upstream token was relayed, so an audit can attribute a relayed request to its mode
    without logging any credential.
    """

    mcp_server_resource: str | None
    """
    The origin (scheme + host + port) of the upstream MCP server the tool call was forwarded
    to. Redacted for logging: userinfo, the path, the query string, and the fragment are all
    stripped, because hosted MCP servers routinely embed the credential in the URL path and
    this value is readable by callers via request logs.
    Records which upstream received a relayed request; never a credential.
    """


class StandardLoggingVectorStoreRequest(TypedDict, total=False):
    """
    Logging information for a vector store request/payload
    """

    vector_store_id: str | None
    """
    ID of the vector store
    """

    custom_llm_provider: str | None
    """
    Custom LLM provider the vector store is associated with eg. bedrock, openai, anthropic, etc.
    """

    query: str | None
    """
    Query to the vector store
    """

    vector_store_search_response: VectorStoreSearchResponse | None
    """
    OpenAI format vector store search response
    """

    start_time: float | None
    """
    Start time of the vector store request
    """

    end_time: float | None
    """
    End time of the vector store request
    """


class StandardBuiltInToolsParams(TypedDict, total=False):
    """
    Standard built-in OpenAItools parameters

    This is used to calculate the cost of built-in tools, insert any standard built-in tools parameters here

    OpenAI charges users based on the `web_search_options` parameter
    """

    web_search_options: WebSearchOptions | None
    file_search: FileSearchTool | None


class StandardLoggingPromptManagementMetadata(TypedDict):
    prompt_id: str
    prompt_variables: dict | None
    prompt_integration: str


class StandardLoggingRoutingDecisionTierBoundaries(TypedDict):
    """Snapshot of the complexity scorer's tier boundaries at decision time, so a
    historical spend log row stays explainable after the router config changes."""

    simple_medium: float
    medium_complex: float
    complex_reasoning: float


RoutingDecisionCause = Literal[
    "heuristic_scorer",
    # The scorer found 2+ reasoning markers and forced REASONING regardless of score.
    # A distinct cause rather than a marker inside `signals`, because it is the fact
    # that tells a reader the score did NOT choose the tier; encoding it as free text
    # meant anything that filtered `signals` silently changed what the row claimed.
    "reasoning_override",
    "llm_classifier",
    # classifier_type 'heuristic_first': the local scorer produced at least one signal and landed at
    # or below heuristic_first_max_tier, so it decided the tier and the LLM classifier was never
    # called. Distinct from "heuristic_scorer", which is a router whose only classifier IS the
    # scorer, and from "classifier_fallback", which is the scorer running because a call failed:
    # only this cause means an LLM classifier was configured, reachable, and deliberately skipped.
    "heuristic_first_short_circuit",
    # The operator's classifier plugin (classifier_type 'custom') decided the tier.
    "classifier_plugin",
    # The LLM classifier or classifier plugin failed on a router with an operator-defined
    # tier set, so the request routed to the configured fallback_tier without being classified.
    "classifier_fallback",
    # The LLM classifier or classifier plugin failed and classifier_fallback is
    # 'default_model', so the request went to default_model without being classified.
    # Distinct from "default_fallback",
    # which is a tier having no model configured rather than classification not happening.
    "default_model_fallback",
    "literal_keyword_match",
    "semantic_keyword_match",
    # A plan-mode sentinel (Claude Code / Copilot plan mode) was detected on the request and
    # plan_mode_min_tier decided the tier: either it raised what the pipeline chose (classifier,
    # keyword rule, or session pin), or the floor was already the top configured tier and the
    # classifier was skipped. The matched sentinel rides in matched_keyword.
    "plan_mode",
    # A client housekeeping sentinel (a coding agent's conversation-title prompt) was detected on
    # the newest ask, so the request routed to the cheapest configured tier and the classifier was
    # never called. The matched sentinel rides in matched_keyword. Distinct from the keyword causes,
    # which are operator-authored rules; these sentinels ship with the router.
    "housekeeping",
    # modality_routing replaced the decided placement: the request carries an image and the
    # routed model does not accept image input, so the nearest higher capable tier or
    # default_model served instead. The displaced placement rides in signals.
    "modality_escalation",
    "session_affinity_pin",
    "session_affinity_escalation",
    # classification_mode 'user_turn': the request is an agent loop's continuation turn (no new
    # human ask), so the session's held routing decision was replayed and the classifier was never
    # called. Distinct from "session_affinity_pin", which reports the session_affinity flag pinning
    # every turn including new asks; this cause only appears when session_affinity is off.
    "user_turn_continuation",
    "default_fallback",
    "keyword",
    "quality_tier",
    "bandit",
]


InternalCallOrigin = Literal[
    "autorouter_classifier",
    "shadow_eval_router",
    "shadow_eval_judge",
    "background_response_cost_poll",
]
"""Which internal litellm feature originated a billed sub-call, so a spend log row
records that it is not traffic the caller sent."""

AUTOROUTER_CLASSIFIER_CALL_ORIGIN: Final[InternalCallOrigin] = "autorouter_classifier"
SHADOW_EVAL_ROUTER_CALL_ORIGIN: Final[InternalCallOrigin] = "shadow_eval_router"
SHADOW_EVAL_JUDGE_CALL_ORIGIN: Final[InternalCallOrigin] = "shadow_eval_judge"
BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN: Final[InternalCallOrigin] = "background_response_cost_poll"


class StandardLoggingRoutingDecision(TypedDict, total=False):
    """Per-request provenance for a pre-routing strategy (auto-router) decision."""

    router_model_name: str
    router_type: Literal["complexity", "adaptive", "quality"]
    routed_model: str
    cause: RoutingDecisionCause
    tier: str
    tier_label: str
    request_type: str
    score: float
    signals: Sequence[str]
    matched_keyword: str
    escalation_keyword: str
    classifier_model: str
    classifier_cost: float
    escalated: bool
    context_escalated: bool  # writable-ok: Pydantic warns on ReadOnly TypedDict fields
    context_escalation_original_tier: str  # writable-ok: Pydantic warns on ReadOnly TypedDict fields
    tier_boundaries: StandardLoggingRoutingDecisionTierBoundaries
    reasoning_override_min_score: float  # writable-ok: Pydantic warns on ReadOnly TypedDict fields
    conversation_continuing: bool
    savings_baseline_model: str
    savings_baseline_deployment_id: str
    tier_litellm_params: Mapping[str, object]  # writable-ok: Pydantic warns on ReadOnly TypedDict fields


# Fields whose values quote the caller's prompt. Dropped when an operator turns message
# logging off. Every other field aggregates the prompt without reproducing it and is kept,
# so a redacted row stays explainable. `test_every_routing_decision_field_is_classified`
# fails if a field is added to the record without being placed in one set or the other.
PROMPT_QUOTING_ROUTING_DECISION_FIELDS: frozenset[str] = frozenset({"signals", "matched_keyword", "escalation_keyword"})
DERIVED_ROUTING_DECISION_FIELDS: Final[frozenset[str]] = frozenset(
    {
        "router_model_name",
        "router_type",
        "routed_model",
        "cause",
        "tier",
        "tier_label",
        "request_type",
        "score",
        "classifier_model",
        "classifier_cost",
        "escalated",
        "context_escalated",
        "context_escalation_original_tier",
        "tier_boundaries",
        "reasoning_override_min_score",
        "conversation_continuing",
        "savings_baseline_model",
        "savings_baseline_deployment_id",
        "tier_litellm_params",
    }
)


class StandardLoggingMetadata(StandardLoggingUserAPIKeyMetadata):
    """
    Specific metadata k,v pairs logged to integration for easier cost tracking and prompt management
    """

    spend_logs_metadata: dict | None  # special param to log k,v pairs to spendlogs for a call
    requester_ip_address: str | None
    user_agent: str | None
    requester_metadata: dict | None
    requester_custom_headers: dict[str, str] | None  # Log any custom (`x-`) headers sent by the client to the proxy.
    prompt_management_metadata: StandardLoggingPromptManagementMetadata | None
    mcp_tool_call_metadata: StandardLoggingMCPToolCall | None
    vector_store_request_metadata: list[StandardLoggingVectorStoreRequest] | None
    routing_decision: StandardLoggingRoutingDecision | None
    applied_guardrails: list[str] | None
    usage_object: dict | None
    cold_storage_object_key: str | None  # S3/GCS object key for cold storage retrieval
    team_alias: str | None
    team_id: str | None


class StandardLoggingAdditionalHeaders(TypedDict, total=False):
    x_ratelimit_limit_requests: int
    x_ratelimit_limit_tokens: int
    x_ratelimit_remaining_requests: int
    x_ratelimit_remaining_tokens: int
    x_ratelimit_reset_requests: str
    x_ratelimit_reset_tokens: str


class StandardLoggingHiddenParams(TypedDict):
    model_id: (
        str | None
    )  # id of the model in the router, separates multiple models with the same name but different credentials
    cache_key: str | None
    api_base: str | None
    response_cost: str | float | None
    litellm_overhead_time_ms: float | None
    additional_headers: StandardLoggingAdditionalHeaders | None
    batch_models: list[str] | None
    batch_successful_requests: ReadOnly[int | None]
    batch_failed_requests: ReadOnly[int | None]
    litellm_model_name: str | None  # the model name sent to the provider by litellm
    usage_object: dict | None


class StandardLoggingModelInformation(TypedDict):
    model_map_key: str
    model_map_value: ModelInfo | None


class StandardLoggingModelCostFailureDebugInformation(TypedDict, total=False):
    """
    Debug information, if cost tracking fails.

    Avoid logging sensitive information like response or optional params
    """

    error_str: Required[str]
    traceback_str: Required[str]
    model: str
    cache_hit: bool | None
    custom_llm_provider: str | None
    base_model: str | None
    call_type: str
    custom_pricing: bool | None


class StandardLoggingPayloadErrorInformation(TypedDict, total=False):
    error_code: str | None
    error_class: str | None
    llm_provider: str | None
    traceback: str | None
    error_message: str | None
    # error_rate_limit_category:
    #   For 429 / rate-limit errors, the source of the rate limit. One of the
    #   string values defined by `litellm.exceptions.RateLimitErrorCategory`
    #   (vendor_rate_limit, vendor_batch_rate_limit, litellm_rate_limit,
    #   litellm_batch_rate_limit). None for non-rate-limit exceptions.
    #   Surfaced here so custom callbacks / metrics consumers can switch on
    #   the rate-limit source without reaching for the raw exception.
    error_rate_limit_category: str | None
    # error_rate_limit_type:
    #   For 429 / rate-limit errors, the dimension that was exceeded. One of
    #   the string values defined by `litellm.exceptions.RateLimitType`
    #   (requests, tokens, concurrent_requests, budget, max_iterations).
    #   None for non-rate-limit exceptions and for rate-limit exceptions that
    #   did not classify the failure (e.g. legacy vendor 429 with no header
    #   hints). Lets dashboards split rate-limit failures by cause without
    #   parsing free-text error messages.
    error_rate_limit_type: str | None
    error_budget_entity_type: str | None
    error_budget_entity_id: str | None
    error_budget_limit: float | None
    error_budget_spend: float | None


class GuardrailMode(TypedDict, total=False):
    tags: dict[str, str | list[str]] | None
    default: str | list[str] | None


GuardrailStatus = Literal["success", "guardrail_intervened", "guardrail_failed_to_respond", "not_run"]


class StandardLoggingGuardrailInformation(TypedDict, total=False):
    guardrail_name: str | None
    guardrail_provider: str | None
    guardrail_mode: GuardrailEventHooks | list[GuardrailEventHooks] | GuardrailMode | None
    guardrail_request: str | dict | None
    guardrail_response: dict | str | list[dict] | None
    guardrail_status: GuardrailStatus
    start_time: float | None
    end_time: float | None
    duration: float | None
    """
    Duration of the guardrail in seconds
    """

    masked_entity_count: dict[str, int] | None
    """
    Count of masked entities
    {
        "CREDIT_CARD": 2,
        "PHONE": 1
    }
    """

    guardrail_id: str | None
    """Unique identifier for the guardrail configuration, e.g. 'gd-eu-pii-001'"""

    policy_template: str | None
    """Name of the policy template this guardrail belongs to, e.g. 'EU AI Act Article 5'"""

    detection_method: str | None
    """How detection was performed: 'regex', 'keyword', 'llm-judge', 'presidio', etc."""

    confidence_score: float | None
    """For LLM-judge guardrails: confidence score 0.0-1.0"""

    classification: str | dict | None
    """For LLM-judge guardrails: structured classification output"""

    match_details: str | list[dict] | None
    """Detailed match information for each detected pattern"""

    patterns_checked: int | None
    """Total number of patterns evaluated by this guardrail"""

    alert_recipients: list[str] | None
    """Email addresses that were notified"""

    risk_score: float | None
    """Risk score 0-10 indicating how risky the request was (higher = riskier). Computed by the guardrail provider."""

    violation_categories: list[str] | None
    """Names of the policy items that intervened on this request (e.g. Bedrock
    topic-policy topic names, content-policy filter types, PII entity types).
    Populated by the provider hook before redaction so downstream loggers
    (OTEL, Langfuse, ...) can filter by violation category without parsing
    the raw guardrail_response blob. Empty/absent when the guardrail allowed
    the request through."""

    guardrail_action: str | None
    """Provider's raw top-level action string (e.g. Bedrock's ``GUARDRAIL_INTERVENED``
    or ``NONE``). Populated by the provider hook so the OTEL integration can
    surface it as a queryable span attribute without parsing the raw
    guardrail_response blob."""

    guardrail_usage: ReadOnly[Mapping[str, int] | None]
    """Provider-reported billable usage counters for this invocation, keyed by the
    provider's counter name (e.g. Bedrock's ``contentPolicyUnits``). Kept as a
    sibling of guardrail_response so spend-log prompt redaction never drops it."""

    guardrail_cost: ReadOnly[float | None]
    """USD cost of this guardrail invocation, priced from ``guardrail_usage`` by the
    provider hook. Summed into the request's ``response_cost`` so it counts against
    spend and budgets like token cost, unless ``guardrail_cost_in_spend`` is False."""

    guardrail_cost_in_spend: ReadOnly[bool | None]
    """Whether ``guardrail_cost`` participates in the request's ``response_cost`` and
    the spend/budget aggregates built from it. Absent, None, or True keeps the default
    (cost counts against spend, the Bedrock behavior); False reports the cost on
    logs, OTEL spans, and the UI while every spend and budget total ignores it."""


class EvalVerdict(TypedDict, total=False):
    criterion_name: str
    score: float  # 0-100
    reasoning: str
    passed: bool
    weight: int  # criterion weight (0-100) as configured in the guardrail


class StandardLoggingEvalInformation(TypedDict, total=False):
    eval_id: str | None
    eval_name: str
    overall_score: float
    passed: bool
    judge_model: str
    iteration: int
    eval_error: str | None
    start_time: str
    end_time: str
    duration: float
    verdicts: list[Any]
    threshold: float | None


class GuardrailTracingDetail(TypedDict, total=False):
    """
    Typed fields for guardrail tracing metadata.

    Passed to add_standard_logging_guardrail_information_to_request_data()
    to enrich the StandardLoggingGuardrailInformation with provider-specific details.
    """

    guardrail_id: str | None
    policy_template: str | None
    detection_method: str | None
    confidence_score: float | None
    classification: dict | None
    match_details: list[dict] | None
    patterns_checked: int | None
    alert_recipients: list[str] | None
    risk_score: float | None
    violation_categories: list[str] | None
    guardrail_action: str | None
    guardrail_usage: ReadOnly[Mapping[str, int] | None]
    guardrail_cost: ReadOnly[float | None]
    guardrail_cost_in_spend: ReadOnly[bool | None]


StandardLoggingPayloadStatus = Literal["success", "failure"]


class CachingDetails(TypedDict):
    """
    Track all caching related metrics, fields for a given request
    """

    cache_hit: bool | None
    """
    Whether the request hit the cache
    """
    cache_duration_ms: float | None
    """
    Duration for reading from cache
    """


class CostBreakdown(TypedDict, total=False):
    """
    Detailed cost breakdown for a request.

    ``service_tier``, ``data_residency``, and ``vertex_location`` record the pricing
    basis the cost was computed on, not what the caller asked for. A consumer that has
    to price a counterfactual against this request (what another model would have
    charged for it) needs the same basis to compare like with like, and re-deriving it
    from the request is not possible after the fact: the tier the biller used comes
    from ``optional_params``, which no log record carries.
    """

    service_tier: str | None
    data_residency: str | None
    vertex_location: ReadOnly[str | None]
    input_cost: float  # Cost of raw (non-cached) input tokens only
    cache_read_cost: float  # Cost of cache-read tokens (discounted rate)
    cache_creation_cost: float  # Cost of cache-write tokens (premium rate)
    output_cost: float  # Cost of output/completion tokens (includes reasoning if applicable)
    reasoning_cost: float  # Cost of reasoning tokens (subset of output_cost)
    total_cost: ReadOnly[float]  # Total cost (input + output + tool usage + guardrail)
    tool_usage_cost: float  # Cost of usage of built-in tools
    guardrail_cost: ReadOnly[float]  # Cost counted in spend; report-only (guardrail_cost_in_spend=False) is excluded
    additional_costs: dict[str, float]  # Free-form additional costs (e.g., {"azure_model_router_flat_cost": 0.00014})
    original_cost: float  # Cost before discount (optional)
    discount_percent: float  # Discount percentage applied (e.g., 0.05 = 5%) (optional)
    discount_amount: float  # Discount amount in USD (optional)
    margin_percent: float  # Margin percentage applied (e.g., 0.10 = 10%) (optional)
    margin_fixed_amount: float  # Fixed margin amount in USD (optional)
    margin_total_amount: float  # Total margin added in USD (optional)


class StandardLoggingPayloadStatusFields(TypedDict, total=False):
    """Status fields for easy filtering and analytics"""

    llm_api_status: StandardLoggingPayloadStatus
    """Status of the LLM API call - 'success' if completed, 'failure' if errored"""
    guardrail_status: GuardrailStatus
    """
    Status of guardrail execution:
    - 'success': Guardrail ran and allowed content through
    - 'guardrail_intervened': Guardrail blocked or modified content
    - 'guardrail_failed_to_respond': Guardrail had technical failure
    - 'not_run': No guardrail was run
    """


class StandardAuditLogPayload(TypedDict):
    """Payload for audit log events dispatched to external callbacks."""

    id: str
    updated_at: str  # ISO-8601
    changed_by: str
    changed_by_api_key: str
    action: str  # "created" | "updated" | "deleted" | "blocked" | "rotated"
    table_name: str
    object_id: str
    before_value: str | None
    updated_values: str | None


class StandardLoggingPayload(TypedDict):
    id: str
    trace_id: str  # Trace multiple LLM calls belonging to same overall request (e.g. fallbacks/retries)
    session_id: str  # End-user/conversation session id (litellm_session_id), independent of trace_id
    litellm_call_id: str | None  # UUID returned in x-litellm-call-id response header
    call_type: str
    stream: bool | None
    response_cost: float
    cost_breakdown: CostBreakdown | None  # Detailed cost breakdown
    autorouter_savings: ReadOnly[float | None]  # None = not an auto-routed caller request; 0.0 is a real figure
    response_cost_failure_debug_info: StandardLoggingModelCostFailureDebugInformation | None
    status: StandardLoggingPayloadStatus
    status_fields: StandardLoggingPayloadStatusFields
    custom_llm_provider: str | None
    total_tokens: int
    prompt_tokens: int
    completion_tokens: int
    startTime: float  # Note: making this camelCase was a mistake, everything should be snake case
    endTime: float
    completionStartTime: float
    response_time: float
    model_map_information: StandardLoggingModelInformation
    model: str
    model_id: str | None
    model_group: str | None
    api_base: str
    metadata: StandardLoggingMetadata
    cache_hit: bool | None
    cache_key: str | None
    saved_cache_cost: float
    request_tags: list
    request_model_access_groups: NotRequired[ReadOnly[Sequence[str]]]
    end_user: str | None
    requester_ip_address: str | None
    user_agent: str | None
    messages: str | list | dict | None
    response: str | list | dict | None
    error_str: str | None
    error_information: StandardLoggingPayloadErrorInformation | None
    model_parameters: dict
    hidden_params: StandardLoggingHiddenParams
    guardrail_information: list[StandardLoggingGuardrailInformation] | None
    standard_built_in_tools_params: StandardBuiltInToolsParams | None


from collections.abc import AsyncIterator, Iterator


class CustomStreamingDecoder:
    async def aiter_bytes(
        self, iterator: AsyncIterator[bytes]
    ) -> AsyncIterator[GenericStreamingChunk | StreamingChatCompletionChunk | None]:
        raise NotImplementedError

    def iter_bytes(
        self, iterator: Iterator[bytes]
    ) -> Iterator[GenericStreamingChunk | StreamingChatCompletionChunk | None]:
        raise NotImplementedError


class StandardPassThroughResponseObject(TypedDict):
    response: str | dict


OPENAI_RESPONSE_HEADERS: Final = [
    "x-ratelimit-remaining-requests",
    "x-ratelimit-remaining-tokens",
    "x-ratelimit-limit-requests",
    "x-ratelimit-limit-tokens",
    "x-ratelimit-reset-requests",
    "x-ratelimit-reset-tokens",
]


class StandardCallbackDynamicParams(TypedDict, total=False):
    # Langfuse dynamic params
    langfuse_public_key: str | None
    langfuse_secret: str | None
    langfuse_secret_key: str | None
    langfuse_host: str | None
    langfuse_environment: ReadOnly[str | None]

    # Langfuse prompt version
    langfuse_prompt_version: int | None

    # GCS dynamic params
    gcs_bucket_name: str | None
    gcs_path_service_account: str | None

    # Langsmith dynamic params
    langsmith_api_key: str | None
    langsmith_project: str | None
    langsmith_base_url: str | None
    langsmith_sampling_rate: float | None
    langsmith_tenant_id: str | None

    # Humanloop dynamic params
    humanloop_api_key: str | None

    # Arize dynamic params
    arize_api_key: str | None
    arize_space_key: str | None
    arize_space_id: str | None

    # PostHog dynamic params
    posthog_api_key: str | None
    posthog_api_url: str | None

    # Weave (W&B) dynamic params
    wandb_api_key: str | None
    weave_project_id: str | None

    # Datadog dynamic params
    dd_api_key: str | None
    dd_site: str | None
    dd_agent_host: str | None
    dd_agent_port: str | None

    # New Relic dynamic params (proxy-stamped team/key callback vars only;
    # request-supplied values are blocked)
    newrelic_api_key: str | None  # writable-ok: initialize_standard_callback_dynamic_params assigns into the dict
    newrelic_region: str | None  # writable-ok: initialize_standard_callback_dynamic_params assigns into the dict

    # Logging settings
    turn_off_message_logging: bool | None  # when true will not log messages
    litellm_disabled_callbacks: list[str] | None


class MirroredPricingParams(BaseModel):
    """Pricing overrides that ``Deployment.__init__`` mirrors from ``litellm_params``
    onto ``model_info``, so both blobs hold the same rate.

    Declared once and inherited by both sides of that mirror, so the two can't drift.
    """

    input_cost_per_token: float | None = None
    output_cost_per_token: float | None = None
    input_cost_per_character: float | None = None
    output_cost_per_character: float | None = None
    cache_read_input_token_cost: float | None = None
    cache_creation_input_token_cost: float | None = None
    tiered_pricing: list[dict[str, Any]] | None = None


class CustomPricingLiteLLMParams(MirroredPricingParams):
    ## CUSTOM PRICING ##
    input_cost_per_second: float | None = None
    output_cost_per_second: float | None = None
    output_cost_per_second_1080p: float | None = None
    output_cost_per_second_480p: float | None = None
    output_cost_per_second_4k: float | None = None
    input_cost_per_pixel: float | None = None
    output_cost_per_pixel: float | None = None

    # Include all ModelInfoBase fields as optional
    # This allows any model_info parameter to be set in litellm_params
    input_cost_per_token_flex: float | None = None
    input_cost_per_token_priority: float | None = None
    input_cost_per_token_ultrafast: float | None = None
    cache_creation_input_token_cost_above_1hr: float | None = None
    cache_creation_input_token_cost_above_200k_tokens: float | None = None
    cache_creation_input_token_cost_above_272k_tokens: float | None = None
    cache_creation_input_token_cost_above_272k_tokens_priority: float | None = None
    cache_creation_input_token_cost_above_272k_tokens_flex: float | None = None
    cache_creation_input_token_cost_flex: float | None = None
    cache_creation_input_token_cost_priority: float | None = None
    cache_creation_input_token_cost_ultrafast: float | None = None
    cache_creation_input_audio_token_cost: float | None = None
    cache_read_input_token_cost_flex: float | None = None
    cache_read_input_token_cost_priority: float | None = None
    cache_read_input_token_cost_ultrafast: float | None = None
    cache_read_input_token_cost_above_200k_tokens: float | None = None
    cache_read_input_token_cost_above_200k_tokens_priority: float | None = None
    cache_read_input_token_cost_above_272k_tokens_priority: float | None = None
    cache_read_input_token_cost_above_272k_tokens_flex: float | None = None
    cache_read_input_audio_token_cost: float | None = None
    input_cost_per_character_above_128k_tokens: float | None = None
    input_cost_per_audio_token: float | None = None
    input_cost_per_token_cache_hit: float | None = None
    input_cost_per_token_above_128k_tokens: float | None = None
    input_cost_per_token_above_200k_tokens: float | None = None
    input_cost_per_token_above_200k_tokens_priority: float | None = None
    input_cost_per_token_above_272k_tokens_priority: float | None = None
    input_cost_per_token_above_272k_tokens_flex: float | None = None
    input_cost_per_query: float | None = None
    input_cost_per_image: float | None = None
    input_cost_per_image_above_128k_tokens: float | None = None
    input_cost_per_audio_per_second: float | None = None
    input_cost_per_audio_per_second_above_128k_tokens: float | None = None
    input_cost_per_video_per_second: float | None = None
    input_cost_per_video_per_second_above_128k_tokens: float | None = None
    input_cost_per_video_per_second_above_15s_interval: float | None = None
    input_cost_per_video_per_second_above_8s_interval: float | None = None
    input_cost_per_token_batches: float | None = None
    output_cost_per_token_batches: float | None = None
    output_cost_per_token_flex: float | None = None
    output_cost_per_token_priority: float | None = None
    output_cost_per_token_ultrafast: float | None = None
    output_cost_per_audio_token: float | None = None
    output_cost_per_token_above_128k_tokens: float | None = None
    output_cost_per_token_above_200k_tokens: float | None = None
    output_cost_per_token_above_200k_tokens_priority: float | None = None
    output_cost_per_token_above_272k_tokens_priority: float | None = None
    output_cost_per_token_above_272k_tokens_flex: float | None = None
    output_cost_per_character_above_128k_tokens: float | None = None
    output_cost_per_image: float | None = None
    output_cost_per_image_token: float | None = None
    output_cost_per_video_token: float | None = None
    output_cost_per_reasoning_token: float | None = None
    output_cost_per_reasoning_token_flex: float | None = None
    output_cost_per_reasoning_token_priority: float | None = None
    output_cost_per_video_per_second: float | None = None
    output_cost_per_audio_per_second: float | None = None
    search_context_cost_per_query: dict[str, Any] | None = None
    google_maps_grounding_cost_per_query: float | None = None
    citation_cost_per_token: float | None = None
    cache_read_input_token_cost_above_272k_tokens: float | None = None
    cache_read_input_token_cost_above_512k_tokens: float | None = None
    input_cost_per_image_token: float | None = None
    input_cost_per_video_token: float | None = None
    input_cost_per_token_above_272k_tokens: float | None = None
    input_cost_per_token_above_512k_tokens: float | None = None
    output_cost_per_token_above_272k_tokens: float | None = None
    output_cost_per_token_above_512k_tokens: float | None = None
    output_vector_size: int | None = None
    ocr_cost_per_page: float | None = None
    ocr_cost_per_credit: float | None = None
    annotation_cost_per_page: float | None = None
    regional_processing_uplift_multiplier_eu: float | None = None
    regional_processing_uplift_multiplier_us: float | None = None
    regional_endpoint_uplift_multiplier: float | None = None

    @classmethod
    def strip_custom_pricing_fields(cls, model_info: dict[str, Any]) -> dict[str, Any]:
        """Return a copy of ``model_info`` without per-deployment custom pricing fields.

        Used when registering a deployment's info under the shared
        ``{provider}/{model}`` key in ``litellm.model_cost``, so one deployment's
        pricing overrides don't pollute sibling deployments that share the same
        backend model. Full pricing stays under the deployment's unique model id.
        """
        return {k: v for k, v in model_info.items() if k not in cls.model_fields}


DEPLOYMENT_SCOPED_PRICING_FIELDS: Final[frozenset[str]] = frozenset({"off_peak_pricing"})

SHARED_BACKEND_MODEL_INFO_FIELDS: Final[frozenset[str]] = (
    frozenset(ModelInfoBase.__required_keys__ | ModelInfoBase.__optional_keys__)
    - frozenset(CustomPricingLiteLLMParams.model_fields)
    - DEPLOYMENT_SCOPED_PRICING_FIELDS
)


def shared_backend_model_info(model_info: dict[str, Any]) -> dict[str, Any]:
    """Return only the fields safe to register under a shared ``{provider}/{model}``
    key in ``litellm.model_cost``: cost-map schema fields (``ModelInfoBase``) minus
    per-deployment pricing overrides and deployment-scoped pricing blocks such as
    ``off_peak_pricing``. Per-deployment metadata (``id``, ``access_via_team_ids``,
    arbitrary custom keys) never belongs on the shared key; it stays under the
    deployment's unique model id.
    """
    return {k: v for k, v in model_info.items() if k in SHARED_BACKEND_MODEL_INFO_FIELDS}


# Server-controlled fields that bound or drive an interceptor's agentic loop
# (depth, cycle fingerprints, ceiling, code-interpreter sandbox state). Listed
# in all_litellm_params so they are treated as LiteLLM-level and excluded from
# get_non_default_completion_params; otherwise the OpenAI param builder sweeps
# any unrecognized top-level key into extra_body and leaks them to the provider.
# This is what lets the loop carry state across rerun calls without a provider
# scrubber.
agentic_loop_internal_litellm_params: Final = [
    "_agentic_loop_depth",
    "_agentic_loop_fingerprints",
    "_agentic_loop_api_surface",
    "max_agentic_loops",
    "_code_interpreter_interception_active",
    "_code_interpreter_interception_sandbox_key",
    "_code_interpreter_interception_session_scoped",
    "_code_interpreter_interception_converted_stream",
    "_websearch_interception_emit_native_blocks",
    "_websearch_interception_converted_stream",
    "_headroom_interception_converted_stream",
]

# Proxy-owned callback credentials, stamped from admin-configured team/key callback
# settings. Listed in all_litellm_params for the same reason as the agentic-loop
# fields above: an unrecognized top-level key is swept into extra_body and sent to
# the provider.
TRUSTED_CALLBACK_VARS_FIELD: Final = "litellm_trusted_callback_vars"

# Bedrock managed-batch deployment config, read from litellm_params by the batch and
# files transformations. Listed for the same reason as the fields above: these sit on
# a deployment that also serves chat, so leaking them into extra_body makes Bedrock
# reject every non-batch request to that deployment.
bedrock_batch_litellm_params: Final = (
    "aws_batch_role_arn",
    "s3_bucket_name",
    "s3_region_name",
    "s3_output_bucket_name",
    "bedrock_tags",
)

all_litellm_params = (
    agentic_loop_internal_litellm_params
    + [TRUSTED_CALLBACK_VARS_FIELD, *bedrock_batch_litellm_params]
    + [
        "metadata",
        "litellm_metadata",
        "keepalive_seconds",
        "allow_client_keepalive_override",
        "litellm_trace_id",
        "litellm_request_debug",
        "guardrails",
        "tags",
        "acompletion",
        "aimg_generation",
        "atext_completion",
        "text_completion",
        "caching",
        "mock_response",
        "mock_timeout",
        "disable_add_transform_inline_image_block",
        "api_key",
        "api_version",
        "prompt_id",
        "prompt_variables",
        "litellm_system_prompt",
        "provider_specific_header",
        "prompt_version",
        "api_base",
        "force_timeout",
        "logger_fn",
        "verbose",
        "custom_llm_provider",
        "model_file_id_mapping",
        "litellm_logging_obj",
        "litellm_call_id",
        "_litellm_strip_stream_usage",
        "use_client",
        "id",
        "fallbacks",
        "routing_strategy",
        "azure",
        "headers",
        "model_list",
        "num_retries",
        "context_window_fallback_dict",
        "retry_policy",
        "retry_strategy",
        "roles",
        "final_prompt_value",
        "bos_token",
        "eos_token",
        "request_timeout",
        "client_side_timeout",
        "complete_response",
        "self",
        "client",
        "rpm",
        "tpm",
        "default_api_key_rpm_limit",
        "default_api_key_tpm_limit",
        "itpm",
        "otpm",
        "max_parallel_requests",
        "input_cost_per_token",
        "output_cost_per_token",
        "input_cost_per_second",
        "output_cost_per_second",
        "hf_model_name",
        "model_info",
        "proxy_server_request",
        "secret_fields",
        "preset_cache_key",
        "caching_groups",
        "ttl",
        "cache",
        "enable_prompt_caching",
        "no-log",
        "base_model",
        "stream_timeout",
        "supports_system_message",
        "region_name",
        "allowed_model_region",
        "model_config",
        "fastest_response",
        "cooldown_time",
        "cache_key",
        "max_retries",
        "azure_ad_token_provider",
        "tenant_id",
        "client_id",
        "azure_username",
        "azure_password",
        "azure_scope",
        "client_secret",
        "user_continue_message",
        "configurable_clientside_auth_params",
        "weight",
        "ensure_alternating_roles",
        "assistant_continue_message",
        "user_continue_message",
        "fallback_depth",
        "max_fallbacks",
        "attempted_targets",
        "max_budget",
        "budget_duration",
        "use_in_pass_through",
        "merge_reasoning_content_in_choices",
        "litellm_credential_name",
        "allowed_openai_params",
        "litellm_session_id",
        "use_litellm_proxy",
        "use_chat_completions_api",
        "rust",
        "prompt_label",
        "shared_session",
        "search_tool_name",
        "order",
        "enable_tag_filtering",
        "enable_json_schema_validation",
        "use_xai_oauth",
        "auto_router_config_path",
        "auto_router_config",
        "auto_router_default_model",
        "auto_router_embedding_model",
        "auto_router_max_input_chars",
        "complexity_router_config",
        "complexity_router_default_model",
        "adaptive_router_config",
        "adaptive_router_default_model",
        "quality_router_config",
        "quality_router_default_model",
    ]
    + list(StandardCallbackDynamicParams.__annotations__.keys())
    + list(CustomPricingLiteLLMParams.model_fields.keys())
)


class KeyGenerationConfig(TypedDict, total=False):
    required_params: list[str]  # specify params that must be present in the key generation request


class TeamUIKeyGenerationConfig(KeyGenerationConfig):
    allowed_team_member_roles: list[str]


class PersonalUIKeyGenerationConfig(KeyGenerationConfig):
    allowed_user_roles: list[str]


class StandardKeyGenerationConfig(TypedDict, total=False):
    team_key_generation: TeamUIKeyGenerationConfig
    personal_key_generation: PersonalUIKeyGenerationConfig


class BudgetConfig(BaseModel):
    max_budget: float | None = None
    budget_duration: str | None = None
    tpm_limit: int | None = None
    rpm_limit: int | None = None

    def __init__(self, **data: Any) -> None:
        # Map time_period to budget_duration if present
        if "time_period" in data:
            data["budget_duration"] = data.pop("time_period")

        # Map budget_limit to max_budget if present
        if "budget_limit" in data:
            data["max_budget"] = data.pop("budget_limit")

        super().__init__(**data)


GenericBudgetConfigType = dict[str, BudgetConfig]


class LlmProviders(str, Enum):
    OPENAI = "openai"
    CHATGPT = "chatgpt"
    OPENAI_LIKE = "openai_like"  # embedding only
    JINA_AI = "jina_ai"
    XAI = "xai"
    ZAI = "zai"
    CUSTOM_OPENAI = "custom_openai"
    TEXT_COMPLETION_OPENAI = "text-completion-openai"
    COHERE = "cohere"
    COHERE_CHAT = "cohere_chat"
    CLARIFAI = "clarifai"
    ANTHROPIC = "anthropic"
    ANTHROPIC_TEXT = "anthropic_text"
    BYTEZ = "bytez"
    REPLICATE = "replicate"
    REDUCTO = "reducto"
    RUNWAYML = "runwayml"
    AWS_POLLY = "aws_polly"
    HUGGINGFACE = "huggingface"
    TOGETHER_AI = "together_ai"
    OPENROUTER = "openrouter"
    DATAROBOT = "datarobot"
    VERTEX_AI = "vertex_ai"
    VERTEX_AI_BETA = "vertex_ai_beta"
    GEMINI = "gemini"
    AI21 = "ai21"
    BASETEN = "baseten"
    BLACK_FOREST_LABS = "black_forest_labs"
    AZURE = "azure"
    AZURE_TEXT = "azure_text"
    AZURE_AI = "azure_ai"
    SAGEMAKER = "sagemaker"
    SAGEMAKER_CHAT = "sagemaker_chat"
    SAGEMAKER_NOVA = "sagemaker_nova"
    BEDROCK = "bedrock"
    VLLM = "vllm"
    NLP_CLOUD = "nlp_cloud"
    PETALS = "petals"
    OOBABOOGA = "oobabooga"
    OLLAMA = "ollama"
    OLLAMA_CHAT = "ollama_chat"
    DEEPINFRA = "deepinfra"
    PERPLEXITY = "perplexity"
    MISTRAL = "mistral"
    MILVUS = "milvus"
    GROQ = "groq"
    A2A = "a2a"
    GIGACHAT = "gigachat"
    NVIDIA_NIM = "nvidia_nim"
    NVIDIA_RIVA = "nvidia_riva"
    SONIOX = "soniox"
    CEREBRAS = "cerebras"
    AI21_CHAT = "ai21_chat"
    VOLCENGINE = "volcengine"
    CODESTRAL = "codestral"
    TEXT_COMPLETION_CODESTRAL = "text-completion-codestral"
    DASHSCOPE = "dashscope"
    QWENCLOUD = "qwencloud"
    QWEN_AI_PLATFORM = "qwen_ai_platform"
    MODELSCOPE = "modelscope"
    MOONSHOT = "moonshot"
    PUBLICAI = "publicai"
    V0 = "v0"
    MORPH = "morph"
    LAMBDA_AI = "lambda_ai"
    INCEPTION = "inception"
    TEXT_COMPLETION_INCEPTION = "text-completion-inception"
    DEEPSEEK = "deepseek"
    SAMBANOVA = "sambanova"
    MARITALK = "maritalk"
    VOYAGE = "voyage"
    CLOUDFLARE = "cloudflare"
    XINFERENCE = "xinference"
    FIREWORKS_AI = "fireworks_ai"
    FRIENDLIAI = "friendliai"
    FEATHERLESS_AI = "featherless_ai"
    WATSONX = "watsonx"
    WATSONX_TEXT = "watsonx_text"
    TRITON = "triton"
    PREDIBASE = "predibase"
    DATABRICKS = "databricks"
    EMPOWER = "empower"
    GITHUB = "github"
    RAGFLOW = "ragflow"
    COMPACTIFAI = "compactifai"
    DOCKER_MODEL_RUNNER = "docker_model_runner"
    CUSTOM = "custom"
    LITELLM_PROXY = "litellm_proxy"
    HOSTED_VLLM = "hosted_vllm"
    TENCENT = "tencent"
    LLAMAFILE = "llamafile"
    LM_STUDIO = "lm_studio"
    GALADRIEL = "galadriel"
    NEBIUS = "nebius"
    INFINITY = "infinity"
    DEEPGRAM = "deepgram"
    ELEVENLABS = "elevenlabs"
    NOVITA = "novita"
    AIOHTTP_OPENAI = "aiohttp_openai"
    LANGFUSE = "langfuse"
    HUMANLOOP = "humanloop"
    TOPAZ = "topaz"
    SAP_GENERATIVE_AI_HUB = "sap"
    ASSEMBLYAI = "assemblyai"
    CHARITY_ENGINE = "charity_engine"
    GITHUB_COPILOT = "github_copilot"
    SNOWFLAKE = "snowflake"
    GRADIENT_AI = "gradient_ai"
    LLAMA = "meta_llama"
    NSCALE = "nscale"
    PG_VECTOR = "pg_vector"
    S3_VECTORS = "s3_vectors"
    VALKEY = "valkey"
    HELICONE = "helicone"
    HYPERBOLIC = "hyperbolic"
    RECRAFT = "recraft"
    FAL_AI = "fal_ai"
    STABILITY = "stability"
    HEROKU = "heroku"
    AIML = "aiml"
    COMETAPI = "cometapi"
    OCI = "oci"
    AUTO_ROUTER = "auto_router"
    VERCEL_AI_GATEWAY = "vercel_ai_gateway"
    DOTPROMPT = "dotprompt"
    MANUS = "manus"
    WANDB = "wandb"
    OVHCLOUD = "ovhcloud"
    SCALEWAY = "scaleway"
    LEMONADE = "lemonade"
    AMAZON_NOVA = "amazon_nova"
    A2A_AGENT = "a2a_agent"
    LANGGRAPH = "langgraph"
    LANGFLOW = "langflow"
    MINIMAX = "minimax"
    SYNTHETIC = "synthetic"
    APERTIS = "apertis"
    NANOGPT = "nano-gpt"
    POE = "poe"
    CHUTES = "chutes"
    NEOSANTARA = "neosantara"
    PARASAIL = "parasail"
    XIAOMI_MIMO = "xiaomi_mimo"
    TENSORMESH = "tensormesh"
    LIBERTAI = "libertai"
    PINSTRIPES = "pinstripes"
    COGNITION = "cognition"
    SCX_AI = "scx-ai"
    DARKBLOOM = "darkbloom"
    META = "meta"
    LITELLM_AGENT = "litellm_agent"
    CURSOR = "cursor"
    BEDROCK_MANTLE = "bedrock_mantle"
    GDC = "gdc"


# Create a set of all provider values for quick lookup
LlmProvidersSet: Final = {provider.value for provider in LlmProviders}

# File and Batch API providers that are OpenAI-compatible
OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS: set[str] = {
    LlmProviders.OPENAI.value,
    LlmProviders.HOSTED_VLLM.value,
    LlmProviders.LITELLM_PROXY.value,
}

ListBatchesSupportedProvider = Literal["openai", "azure", "hosted_vllm", "litellm_proxy", "vertex_ai"]

LIST_BATCHES_SUPPORTED_PROVIDERS: Final[frozenset[str]] = frozenset(get_args(ListBatchesSupportedProvider))


class SearchProviders(str, Enum):
    """
    Enum for search provider types.
    Separate from LlmProviders for semantic clarity.
    """

    PERPLEXITY = "perplexity"
    TAVILY = "tavily"
    PARALLEL_AI = "parallel_ai"
    EXA_AI = "exa_ai"
    BRAVE = "brave"
    GOOGLE_PSE = "google_pse"
    DATAFORSEO = "dataforseo"
    FIRECRAWL = "firecrawl"
    FASTCRW = "fastcrw"
    SEARXNG = "searxng"
    LINKUP = "linkup"
    DUCKDUCKGO = "duckduckgo"
    SEARCHAPI = "searchapi"
    SERPER = "serper"
    YOU_COM = "you_com"
    APISERPENT = "apiserpent"
    TINYFISH = "tinyfish"
    AGENTCORE = "agentcore"
    NIMBLE = "nimble"
    BING_GROUNDING = "bing_grounding"


# Create a set of all search provider values for quick lookup
SearchProvidersSet: Final = {provider.value for provider in SearchProviders}


class SandboxProviders(str, Enum):
    """
    Enum for code execution sandbox provider types.
    Separate from LlmProviders for semantic clarity.
    """

    E2B = "e2b"
    OPENSANDBOX = "opensandbox"


class LiteLLMLoggingBaseClass:
    """
    Base class for logging pre and post call

    Meant to simplify type checking for logging obj.
    """

    def pre_call(self, input, api_key, model=None, additional_args=None) -> None:
        pass

    def post_call(self, original_response, input=None, api_key=None, additional_args=None) -> None:
        pass


class TokenCountResponse(LiteLLMPydanticObjectBase):
    total_tokens: int
    request_model: str
    model_used: str
    tokenizer_type: str
    original_response: dict | None = None
    """
    Original Response from upstream API call - if an API call was made for token counting
    """
    error: bool = False
    error_message: str | None = None
    """
    HTTP status code from the token counting API (e.g., 200 for success, 429 for rate limit, 400 for bad request)
    """
    status_code: int | None = None


class CustomHuggingfaceTokenizer(TypedDict):
    identifier: str
    revision: str  # usually 'main'
    auth_token: str | None


class LITELLM_IMAGE_VARIATION_PROVIDERS(Enum):
    """
    Try using an enum for endpoints. This should make it easier to track what provider is supported for what endpoint.
    """

    OPENAI = LlmProviders.OPENAI.value
    TOPAZ = LlmProviders.TOPAZ.value


class HttpHandlerRequestFields(TypedDict, total=False):
    data: dict  # request body
    params: dict  # query params
    files: dict  # file uploads
    content: Any  # raw content


class ProviderSpecificHeader(TypedDict):
    custom_llm_provider: str
    extra_headers: dict


class SelectTokenizerResponse(TypedDict):
    type: Literal["openai_tokenizer", "huggingface_tokenizer"]
    tokenizer: Any


class LiteLLMFineTuningJob(FineTuningJob):
    _hidden_params: dict = {}
    seed: int | None = None

    def __init__(self, **kwargs) -> None:
        if "error" in kwargs and kwargs["error"] is not None:
            # check if error is all None - if so, set error to None
            if all(value is None for value in kwargs["error"].values()):
                kwargs["error"] = None
        super().__init__(**kwargs)
        self._hidden_params = kwargs.get("_hidden_params", {})


class LiteLLMBatch(Batch):
    _hidden_params: dict = {}
    usage: Usage | None = None

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class LiteLLMRealtimeStreamLoggingObject(LiteLLMPydanticObjectBase):
    # Events are already well-formed provider dicts. Validating them against the
    # OpenAIRealtimeEvents union makes Pydantic try every member per event, which
    # floods thousands of ValidationErrors for events outside the union (e.g.
    # rate_limits.updated), blocks the event loop, and discards the session usage.
    results: SkipValidation[OpenAIRealtimeStreamList]
    usage: Usage
    _hidden_params: dict = {}

    @field_serializer("results")
    def _serialize_results(self, results: OpenAIRealtimeStreamList) -> list[dict[str, Any]]:
        return [dict(event) for event in results]

    def __contains__(self, key) -> bool:
        # Define custom behavior for the 'in' operator
        return hasattr(self, key)

    def get(self, key, default=None):
        # Custom .get() method to access attributes with a default value if the attribute doesn't exist
        return getattr(self, key, default)

    def __getitem__(self, key):
        # Allow dictionary-style access to attributes
        return getattr(self, key)

    def json(self, **kwargs):
        try:
            return self.model_dump()
        except Exception:
            # if using pydantic v1
            return self.dict()


class RawRequestTypedDict(TypedDict, total=False):
    raw_request_api_base: str | None
    raw_request_body: dict | None
    raw_request_headers: dict | None
    error: str | None


from litellm.models.credentials import (  # noqa: E402
    CreateCredentialItem as CreateCredentialItem,
)
from litellm.models.credentials import CredentialBase as CredentialBase  # noqa: E402
from litellm.models.credentials import CredentialItem as CredentialItem  # noqa: E402


class ExtractedFileData(TypedDict):
    """
    TypedDict for storing processed file data

    Attributes:
        filename: Name of the file if provided
        content: The file content in bytes
        content_type: MIME type of the file
        headers: Any additional headers for the file
    """

    filename: str | None
    content: bytes
    content_type: str | None
    headers: Mapping[str, str]


class SpecialEnums(Enum):
    LITELM_MANAGED_FILE_ID_PREFIX = "litellm_proxy"
    LITELLM_MANAGED_FILE_COMPLETE_STR = (
        "litellm_proxy:{};unified_id,{};target_model_names,{};llm_output_file_id,{};llm_output_file_model_id,{}"
    )

    LITELLM_MANAGED_RESPONSE_COMPLETE_STR = "litellm:custom_llm_provider:{};model_id:{};response_id:{}"

    LITELLM_MANAGED_BATCH_COMPLETE_STR = "litellm_proxy;model_id:{};llm_batch_id:{}"

    LITELLM_MANAGED_RESPONSE_API_RESPONSE_ID_COMPLETE_STR = (
        "litellm_proxy:responses_api:response_id:{};user_id:{};team_id:{}"
    )

    LITELLM_MANAGED_GENERIC_RESPONSE_COMPLETE_STR = "litellm_proxy;model_id:{};generic_response_id:{}"  # generic implementation of 'managed batches' - used for finetuning and any future work.

    LITELLM_MANAGED_VIDEO_COMPLETE_STR = "litellm:custom_llm_provider:{};model_id:{};video_id:{}"

    LITELLM_PASSTHROUGH_MANAGED_ID_COMPLETE_STR = "litellm_proxy:passthrough;provider:{};unified_id,{};raw_id,{}"


class ServiceTier(Enum):
    """Enum for service tier types used in cost calculations."""

    AUTO = "auto"
    FLEX = "flex"
    PRIORITY = "priority"
    FAST = "fast"
    ULTRAFAST = "ultrafast"


class DataResidency(Enum):
    """
    OpenAI data-residency / regional-processing regions.

    Inferred from the OpenAI api_base host (eu.api.openai.com -> EU,
    us.api.openai.com -> US). Used to apply the regional-processing
    cost uplift (see ``regional_processing_uplift_multiplier_<region>``
    on ModelInfo).
    """

    US = "us"
    EU = "eu"


LLMResponseTypes = (
    ModelResponse
    | EmbeddingResponse
    | ImageResponse
    | OpenAIFileObject
    | LiteLLMBatch
    | LiteLLMFineTuningJob
    | AnthropicMessagesResponse
    | ResponsesAPIResponse
    | LiteLLMSendMessageResponse
)


class DynamicPromptManagementParamLiteral(str, Enum):
    """
    If any of these params are passed, the user is trying to use dynamic prompt management
    """

    CACHE_CONTROL_INJECTION_POINTS = "cache_control_injection_points"
    KNOWLEDGE_BASES = "knowledge_bases"
    VECTOR_STORE_IDS = "vector_store_ids"

    @classmethod
    def list_all_params(cls):
        return [param.value for param in cls]


class CallbacksByType(TypedDict):
    success: list[str]
    failure: list[str]
    success_and_failure: list[str]


CostResponseTypes = ModelResponse | TextCompletionResponse | EmbeddingResponse | ImageResponse | TranscriptionResponse


class PriorityReservationDict(TypedDict, total=False):
    """
    Dictionary format for priority reservation values.

    Used in litellm.priority_reservation to specify how much capacity to reserve
    for each priority level. Supports three formats:
    1. Percentage-based: {"type": "percent", "value": 0.9} -> 90% of capacity
    2. RPM-based: {"type": "rpm", "value": 900} -> 900 requests per minute
    3. TPM-based: {"type": "tpm", "value": 900000} -> 900,000 tokens per minute

    Attributes:
        type: The type of value - "percent", "rpm", or "tpm". Defaults to "percent".
        value: The numeric value. For percent (0.0-1.0), for rpm/tpm (absolute value).
    """

    type: Literal["percent", "rpm", "tpm"]
    value: float


class PriorityReservationSettings(BaseModel):
    """
    Settings for priority-based rate limiting reservation.

    Defines what priority to assign to keys without explicit priority metadata.
    The priority_reservation mapping is configured separately via litellm.priority_reservation.
    """

    default_priority: float = Field(
        default=0.25,
        description="Priority level to assign to API keys without explicit priority metadata. Should match a key in litellm.priority_reservation.",
    )

    saturation_threshold: float = Field(
        default=0.50,
        description="Saturation threshold (0.0-1.0) at which strict priority enforcement begins. Below this threshold, generous mode allows priority borrowing. Above this threshold, strict mode enforces normalized priority limits.",
    )

    saturation_check_cache_ttl: int = Field(
        default=60,
        description="TTL in seconds for local cache when reading saturation check values from Redis.",
    )

    model_config = ConfigDict(protected_namespaces=())


class GenericGuardrailAPIInputs(TypedDict, total=False):
    texts: list[str]  # extracted text from the LLM response - for basic text guardrails
    images: list[str]  # extracted images from the LLM response - for image guardrails
    tools: list[ChatCompletionToolParam]  # tools sent to the LLM
    tool_calls: list[ChatCompletionToolCallChunk] | list[ChatCompletionMessageToolCall]  # tool calls sent from the LLM
    structured_messages: list[
        AllMessageValues
    ]  # structured messages sent to the LLM - indicates if text is from system or user
    model: str | None  # the model being used for the LLM call
    stream_holdback_chars: list[
        int
    ]  # trailing chars to withhold from streaming emission per text (word-boundary safety)
