# What is this? ## Common Utility file for Logging handler # Logging function -> log the exact model details + what's being sent | Non-Blocking import copy import datetime import json import os import re import subprocess import sys import time import traceback from collections.abc import Callable, Mapping, Sequence from datetime import datetime as dt_object from functools import lru_cache from types import MappingProxyType, TracebackType from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Union, cast from httpx import Response from pydantic import BaseModel import litellm from litellm import ( _custom_logger_compatible_callbacks_literal, json_logs, log_raw_request_response, turn_off_message_logging, ) from litellm._logging import ( _is_debugging_on, _redact_string, session_id_var, set_session_id, set_trace_id, trace_id_var, verbose_logger, ) from litellm._uuid import uuid from litellm.batches.batch_utils import _handle_completed_batch from litellm.caching.caching import DualCache, InMemoryCache from litellm.caching.caching_handler import LLMCachingHandler from litellm.constants import ( DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT, DEFAULT_MOCK_RESPONSE_PROMPT_TOKEN_COUNT, SENTRY_DENYLIST, SENTRY_PII_DENYLIST, ) from litellm.cost_calculator import ( RealtimeAPITokenUsageProcessor, _select_model_name_for_cost_calc, ) from litellm.exceptions import ( BudgetExceededError, validate_rate_limit_category, validate_rate_limit_type, ) from litellm.integrations.agentops import AgentOps from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook from litellm.integrations.arize.arize import ArizeLogger from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.integrations.custom_logger import CustomLogger from litellm.integrations.deepeval.deepeval import DeepEvalLogger from litellm.integrations.mlflow import MlflowLogger from litellm.integrations.sqs import SQSLogger from litellm.litellm_core_utils.core_helpers import is_expected_client_error, reconstruct_model_name from litellm.litellm_core_utils.get_litellm_params import get_litellm_params from litellm.litellm_core_utils.internal_call_metadata import ( MODEL_ACCESS_GROUP_METADATA_KEY, is_unbilled_non_inference_call, ) from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import ( cost_breakdown_with_guardrail, guardrail_information_cost, ) from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import ( StandardBuiltInToolCostTracking, ) from litellm.litellm_core_utils.llm_cost_calc.usage_object_transformation import ( InteractionsUsageObjectTransformation, ) from litellm.litellm_core_utils.logging_utils import truncate_base64_in_messages from litellm.litellm_core_utils.model_param_helper import ModelParamHelper from litellm.litellm_core_utils.redact_messages import ( redact_message_input_output_from_custom_logger, redact_message_input_output_from_logging, redact_streaming_responses_for_custom_logger, ) from litellm.llms.base_llm.ocr.transformation import OCRResponse from litellm.llms.base_llm.search.transformation import SearchResponse from litellm.responses.utils import ResponseAPILoggingUtils from litellm.types.agents import LiteLLMSendMessageResponse from litellm.types.containers.main import ContainerObject from litellm.types.interactions import ( InteractionsAPIResponse, InteractionsAPIStreamingResponse, ) from litellm.types.llms.openai import ( AllMessageValues, Batch, FineTuningJob, HttpxBinaryResponseContent, OpenAIFileObject, OpenAIModerationResponse, ResponseAPIUsage, ResponseCompletedEvent, ResponseFailedEvent, ResponseIncompleteEvent, ResponsesAPIResponse, ) from litellm.types.mcp import MCPPostCallResponseObject from litellm.types.prompts.init_prompts import PromptSpec from litellm.types.rerank import RerankResponse from litellm.types.utils import ( CachingDetails, CallTypes, CostBreakdown, CostResponseTypes, CustomPricingLiteLLMParams, DynamicPromptManagementParamLiteral, EmbeddingResponse, GuardrailStatus, ImageResponse, LiteLLMBatch, LiteLLMLoggingBaseClass, LiteLLMRealtimeStreamLoggingObject, ModelInfo, ModelResponse, ModelResponseStream, RawRequestTypedDict, StandardBuiltInToolsParams, StandardCallbackDynamicParams, StandardLoggingAdditionalHeaders, StandardLoggingHiddenParams, StandardLoggingMCPToolCall, StandardLoggingMetadata, StandardLoggingModelCostFailureDebugInformation, StandardLoggingModelInformation, StandardLoggingPayload, StandardLoggingPayloadErrorInformation, StandardLoggingPayloadStatus, StandardLoggingPayloadStatusFields, StandardLoggingPromptManagementMetadata, StandardLoggingVectorStoreRequest, TextCompletionResponse, TranscriptionResponse, Usage, ) from litellm.types.videos.main import VideoObject from litellm.utils import _get_base_model_from_metadata, executor, print_verbose from ..integrations.argilla import ArgillaLogger from ..integrations.arize.arize_phoenix import ArizePhoenixLogger from ..integrations.athina import AthinaLogger from ..integrations.azure_sentinel.azure_sentinel import AzureSentinelLogger from ..integrations.azure_storage.azure_storage import AzureBlobStorageLogger from ..integrations.custom_prompt_management import CustomPromptManagement from ..integrations.datadog.datadog import DataDogLogger from ..integrations.datadog.datadog_llm_obs import DataDogLLMObsLogger from ..integrations.datadog.datadog_metrics import DatadogMetricsLogger from ..integrations.dotprompt import DotpromptManager from ..integrations.dynamodb import DyanmoDBLogger from ..integrations.galileo import GalileoObserve from ..integrations.gcs_bucket.gcs_bucket import GCSBucketLogger from ..integrations.gcs_pubsub.pub_sub import GcsPubSubLogger from ..integrations.greenscale import GreenscaleLogger from ..integrations.helicone import HeliconeLogger from ..integrations.humanloop import HumanloopLogger from ..integrations.lago import LagoLogger from ..integrations.langfuse.langfuse import LangFuseLogger from ..integrations.langfuse.langfuse_handler import LangFuseHandler from ..integrations.langfuse.langfuse_prompt_management import LangfusePromptManagement from ..integrations.langsmith import LangsmithLogger from ..integrations.litellm_agent import LiteLLMAgentModelResolver from ..integrations.literal_ai import LiteralAILogger from ..integrations.logfire_logger import LogfireLevel, LogfireLogger from ..integrations.lunary import LunaryLogger from ..integrations.newrelic import NewRelicLogger from ..integrations.openmeter import OpenMeterLogger from ..integrations.opik.opik import OpikLogger from ..integrations.posthog import PostHogLogger from ..integrations.prompt_layer import PromptLayerLogger from ..integrations.s3 import S3Logger from ..integrations.s3_v2 import S3Logger as S3V2Logger from ..integrations.supabase import Supabase from ..integrations.traceloop import TraceloopLogger from .exception_mapping_utils import _get_response_headers from .initialize_dynamic_callback_params import ( get_trusted_callback_params, ) from .initialize_dynamic_callback_params import ( initialize_standard_callback_dynamic_params as _initialize_standard_callback_dynamic_params, ) from .specialty_caches.dynamic_logging_cache import DynamicLoggingCache if TYPE_CHECKING: from mcp.types import EmbeddedResource, ImageContent, TextContent from litellm.integrations.otel.logger import OpenTelemetryV2 from litellm.llms.base_llm.passthrough.transformation import BasePassthroughConfig try: from litellm_enterprise.enterprise_callbacks.callback_controls import ( EnterpriseCallbackControls, ) from litellm_enterprise.enterprise_callbacks.pagerduty.pagerduty import ( PagerDutyAlerting, ) from litellm_enterprise.enterprise_callbacks.send_emails.resend_email import ( ResendEmailLogger, ) from litellm_enterprise.enterprise_callbacks.send_emails.sendgrid_email import ( SendGridEmailLogger, ) from litellm_enterprise.enterprise_callbacks.send_emails.smtp_email import ( SMTPEmailLogger, ) from litellm_enterprise.litellm_core_utils.litellm_logging import ( StandardLoggingPayloadSetup as EnterpriseStandardLoggingPayloadSetup, ) from litellm.integrations.generic_api.generic_api_callback import GenericAPILogger EnterpriseStandardLoggingPayloadSetupVAR: type[EnterpriseStandardLoggingPayloadSetup] | None = ( EnterpriseStandardLoggingPayloadSetup ) except Exception as e: verbose_logger.debug("[Non-Blocking] Unable to import GenericAPILogger - LiteLLM Enterprise Feature - %s", e) GenericAPILogger = CustomLogger ResendEmailLogger = CustomLogger SendGridEmailLogger = CustomLogger SMTPEmailLogger = CustomLogger PagerDutyAlerting = CustomLogger EnterpriseCallbackControls = None EnterpriseStandardLoggingPayloadSetupVAR = None if TYPE_CHECKING: from litellm.integrations.generic_api.generic_api_callback import ( GenericAPILogger as _GenericAPILoggerCls, ) _GENERIC_API_LOGGER_CLS: Final = _GenericAPILoggerCls _RESEND_EMAIL_LOGGER_FACTORY: Final = CustomLogger _SENDGRID_EMAIL_LOGGER_FACTORY: Final = CustomLogger _SMTP_EMAIL_LOGGER_FACTORY: Final = CustomLogger _PAGERDUTY_ALERTING_FACTORY: Final = CustomLogger else: _GENERIC_API_LOGGER_CLS: Final = GenericAPILogger _RESEND_EMAIL_LOGGER_FACTORY: Final = ResendEmailLogger _SENDGRID_EMAIL_LOGGER_FACTORY: Final = SendGridEmailLogger _SMTP_EMAIL_LOGGER_FACTORY: Final = SMTPEmailLogger _PAGERDUTY_ALERTING_FACTORY: Final = PagerDutyAlerting _in_memory_loggers: Final[list[CustomLogger]] = [] _STANDARD_LOGGING_METADATA_KEYS: Final[frozenset[str]] = frozenset(StandardLoggingMetadata.__annotations__.keys()) ### GLOBAL VARIABLES ### # Cache custom pricing keys as frozenset for O(1) lookups instead of looping through 49 keys _CUSTOM_PRICING_KEYS: Final[frozenset[str]] = frozenset(CustomPricingLiteLLMParams.model_fields.keys()) sentry_sdk_instance = None capture_exception = None add_breadcrumb = None slack_app = None alerts_channel = None heliconeLogger = None athinaLogger = None promptLayerLogger = None logfireLogger = None weightsBiasesLogger = None customLogger = None langFuseLogger = None openMeterLogger = None lagoLogger: Final = None dataDogLogger = None prometheusLogger = None dynamoLogger = None s3Logger = None greenscaleLogger = None lunaryLogger = None supabaseClient = None deepevalLogger = None callback_list: Final[list[str] | None] = [] user_logger_fn: Final = None additional_details: Final[dict[str, str] | None] = {} local_cache: Final[dict[str, str] | None] = {} last_fetched_at: Final = None last_fetched_at_keys: Final = None #### class ServiceTraceIDCache: def __init__(self) -> None: self.cache = InMemoryCache() def get_cache(self, litellm_call_id: str, service_name: str) -> str | None: key_name: Final = f"{service_name}:{litellm_call_id}" response: Final = self.cache.get_cache(key=key_name) return response def set_cache(self, litellm_call_id: str, service_name: str, trace_id: str) -> None: key_name: Final = f"{service_name}:{litellm_call_id}" self.cache.set_cache(key=key_name, value=trace_id) in_memory_trace_id_cache: Final = ServiceTraceIDCache() in_memory_dynamic_logger_cache: Final = DynamicLoggingCache() # Cached lazy import for PrometheusLogger # Module-level cache to avoid repeated imports while preserving memory benefits _PrometheusLogger = None def _get_cached_prometheus_logger(): """ Get cached PrometheusLogger class. Lazy imports on first call to avoid loading prometheus.py and utils.py at import time (60MB saved). Subsequent calls use cached class for better performance. """ global _PrometheusLogger if _PrometheusLogger is None: from litellm.integrations.prometheus import PrometheusLogger _PrometheusLogger = PrometheusLogger return _PrometheusLogger _DEPLOYMENT_PRICING_KEYS: Final = ( "input_cost_per_token", "output_cost_per_token", "input_cost_per_token_batches", "output_cost_per_token_batches", ) def deployment_pricing_model_info(model_id: str | None, deployment_model: str | None) -> ModelInfo | None: """Pricing the router registered under this deployment's model_info.id. Returns None when the deployment declares no pricing of its own, so the caller falls back to the global cost map. The raw registration is what decides that: the router registers an entry for every deployment, and get_model_info fills absent costs with 0, so asking it directly cannot tell "configured as free" apart from "no pricing configured". A deployment may declare only one side of its pricing, so the side it leaves out keeps the model's published rates instead of billing as zero. Ownership is per token direction: declaring either rate for a direction takes that whole direction, so a published batch rate can never displace a standard rate the deployment configured itself. """ if model_id is None: return None registered: Final = litellm.model_cost.get(model_id) if not isinstance(registered, dict) or not any(registered.get(key) is not None for key in _DEPLOYMENT_PRICING_KEYS): return None try: merged: Final = litellm.get_model_info(model=model_id).copy() except Exception: # noqa: BLE001 # get_model_info raises for ids it cannot resolve a provider for return None published: Final = _published_pricing(deployment_model) if published is None: return merged declares_input: Final = ( registered.get("input_cost_per_token") is not None or registered.get("input_cost_per_token_batches") is not None ) declares_output: Final = ( registered.get("output_cost_per_token") is not None or registered.get("output_cost_per_token_batches") is not None ) if not declares_input: merged["input_cost_per_token"] = published.get("input_cost_per_token") merged["input_cost_per_token_batches"] = published.get("input_cost_per_token_batches") if not declares_output: merged["output_cost_per_token"] = published.get("output_cost_per_token") merged["output_cost_per_token_batches"] = published.get("output_cost_per_token_batches") return merged def _published_pricing(deployment_model: str | None) -> ModelInfo | None: """The cost map's own entry for the deployment's model, when it resolves.""" if deployment_model is None: return None try: return litellm.get_model_info(model=deployment_model) except Exception: # noqa: BLE001 # no published entry to layer the declared rates over return None def _resolve_vertex_location_for_cost( custom_llm_provider: str | None, litellm_params: Mapping[str, object] | None, optional_params: Mapping[str, object] | None, model: str, ) -> str | None: """ The Vertex AI location a request was served from, resolved the same way dispatch resolves it, so regional deployments price with the regional-endpoint uplift. None for non-Vertex providers. Chat dispatch reads the location from request kwargs, which reach this logging object through optional_params: on the proxy the logging object is created before the router picks a deployment, so the deployment's location never lands in litellm_params. """ if custom_llm_provider is None or not custom_llm_provider.startswith("vertex_ai"): return None from litellm.llms.vertex_ai.vertex_llm_base import VertexBase empty: Final[Mapping[str, object]] = MappingProxyType({}) configured_location: Final = ( VertexBase.explicit_vertex_ai_location(optional_params or empty) or VertexBase.explicit_vertex_ai_location(litellm_params or empty) or VertexBase.safe_get_vertex_ai_location(empty) ) return VertexBase.get_vertex_region(configured_location, model) class Logging(LiteLLMLoggingBaseClass): global \ supabaseClient, \ promptLayerLogger, \ weightsBiasesLogger, \ logfireLogger, \ capture_exception, \ add_breadcrumb, \ lunaryLogger, \ logfireLogger, \ prometheusLogger, \ slack_app custom_pricing: bool = False stream_options = None litellm_request_debug: bool = False def __init__( self, model: str, messages, stream, call_type, start_time, litellm_call_id: str, function_id: str, litellm_trace_id: str | None = None, dynamic_input_callbacks: list[str | Callable | CustomLogger] | None = None, dynamic_success_callbacks: list[str | Callable | CustomLogger] | None = None, dynamic_async_success_callbacks: list[str | Callable | CustomLogger] | None = None, dynamic_failure_callbacks: list[str | Callable | CustomLogger] | None = None, dynamic_async_failure_callbacks: list[str | Callable | CustomLogger] | None = None, applied_guardrails: list[str] | None = None, kwargs: dict | None = None, log_raw_request_response: bool = False, supports_correlation_logging: bool = True, ): _input: Final[str | None] = messages # save original value of messages if messages is not None: if isinstance(messages, str): messages = [ {"role": "user", "content": messages} ] # convert text completion input to the chat completion format elif isinstance(messages, list) and len(messages) > 0 and isinstance(messages[0], str): new_messages: Final = [] for m in messages: new_messages.append({"role": "user", "content": m}) messages = new_messages self.model = model # Shallow copy of the outer list only (inner message dicts are shared). # Safe because the logging layer does not mutate individual message dicts. _copy_start: Final = time.time() self.messages = copy.copy(messages) if messages is not None else None self.message_copy_duration_ms: float = (time.time() - _copy_start) * 1000 self.callback_duration_ms: float = 0.0 self.stream = stream self.start_time = start_time # log the call start time self.call_type = call_type self.litellm_call_id = litellm_call_id self.litellm_trace_id: str = litellm_trace_id if litellm_trace_id else str(uuid.uuid4()) # Capture the pre-call *value* (not a contextvars.Token) so restoration works # even if this attempt's own logging ends up dispatched onto a different # asyncio Task/context (e.g. via asyncio.create_task or the logging worker) - # a Token can only be reset in the exact Context where it was created. self._pre_call_trace_id: str = trace_id_var.get() self._pre_call_session_id: str = session_id_var.get() _sid: Final = kwargs.get("litellm_session_id") if kwargs else None self.litellm_session_id: str = str(_sid) if _sid else "" # supports_correlation_logging is False for calls originating from the # sync client entry point (wrapper() in utils.py): a plain OS thread # has no per-call context isolation the way an asyncio Task does, and # a thread pool's worker threads are recycled across unrelated # requests, so stamping trace_id/session_id there risks one request's # ids leaking into a different, later request on the same thread. Sync # support is deferred to a follow-up PR with its own safe-restore # mechanism; async calls (the proxy's only call path) are unaffected. if supports_correlation_logging: set_trace_id(self.litellm_trace_id) set_session_id(self.litellm_session_id) # set_trace_id()/set_session_id() sanitize (strip control chars, bound # length) before storing, so the contextvar's actual value can differ # from self.litellm_trace_id/litellm_session_id. Capture what was # really stored - _restore_correlation_context_if_unclaimed() must # compare against this, not the raw ids, or a caller-supplied id # containing control characters/oversized input would never match # and cleanup would be skipped forever. self._own_trace_id: str = trace_id_var.get() self._own_session_id: str = session_id_var.get() self.function_id = function_id self.streaming_chunks: list[Any] = [] # for generating complete stream response self.sync_streaming_chunks: list[Any] = [] # for generating complete stream response self.log_raw_request_response = log_raw_request_response # Initialize dynamic callbacks self.dynamic_input_callbacks: list[str | Callable | CustomLogger] | None = dynamic_input_callbacks self.dynamic_success_callbacks: list[str | Callable | CustomLogger] | None = dynamic_success_callbacks self.dynamic_async_success_callbacks: list[str | Callable | CustomLogger] | None = ( dynamic_async_success_callbacks ) self.dynamic_failure_callbacks: list[str | Callable | CustomLogger] | None = dynamic_failure_callbacks self.dynamic_async_failure_callbacks: list[str | Callable | CustomLogger] | None = ( dynamic_async_failure_callbacks ) ## DYNAMIC LANGFUSE / GCS / logging callback KEYS ## self.standard_callback_dynamic_params: StandardCallbackDynamicParams = ( self.initialize_standard_callback_dynamic_params(kwargs) ) self._trusted_callback_vars: tuple[tuple[str, str], ...] = get_trusted_callback_params(kwargs) # Process dynamic callbacks (after standard_callback_dynamic_params is initialized, # so team-scoped credentials are available for callback initialization) self.process_dynamic_callbacks() self.standard_built_in_tools_params: StandardBuiltInToolsParams = ( self.initialize_standard_built_in_tools_params(kwargs) ) ## TIME TO FIRST TOKEN LOGGING ## self.completion_start_time: datetime.datetime | None = None self._llm_caching_handler: LLMCachingHandler | None = None # INITIAL LITELLM_PARAMS litellm_params = {} if kwargs is not None: litellm_params = get_litellm_params(**kwargs) litellm_params = scrub_sensitive_keys_in_metadata(litellm_params) self.litellm_params = litellm_params # Initialize cost breakdown field self.cost_breakdown: CostBreakdown | None = None # Init Caching related details self.caching_details: CachingDetails | None = None # Passthrough endpoint guardrails config for field targeting self.passthrough_guardrails_config: dict[str, Any] | None = None self.model_call_details: dict[str, Any] = { "litellm_trace_id": self.litellm_trace_id, "litellm_call_id": litellm_call_id, "input": _input, "litellm_params": litellm_params, "applied_guardrails": applied_guardrails, "model": model, } # Set by proxy request handlers to defer spend-log fire until after # post_call guardrails have run; the @client decorator then stores the # enqueue closure here instead of firing it immediately. self._defer_async_logging: bool = False self._enqueue_deferred_logging: Callable[[], None] | None = None def process_dynamic_callbacks(self): """ Initializes CustomLogger compatible callbacks in self.dynamic_* callbacks If a callback is in litellm._known_custom_logger_compatible_callbacks, it needs to be intialized and added to the respective dynamic_* callback list. """ # Process input callbacks self.dynamic_input_callbacks = self._process_dynamic_callback_list( self.dynamic_input_callbacks, dynamic_callbacks_type="input" ) # Process failure callbacks self.dynamic_failure_callbacks = self._process_dynamic_callback_list( self.dynamic_failure_callbacks, dynamic_callbacks_type="failure" ) # Process async failure callbacks self.dynamic_async_failure_callbacks = self._process_dynamic_callback_list( self.dynamic_async_failure_callbacks, dynamic_callbacks_type="async_failure" ) # Process success callbacks self.dynamic_success_callbacks = self._process_dynamic_callback_list( self.dynamic_success_callbacks, dynamic_callbacks_type="success" ) # Process async success callbacks self.dynamic_async_success_callbacks = self._process_dynamic_callback_list( self.dynamic_async_success_callbacks, dynamic_callbacks_type="async_success" ) def _process_dynamic_callback_list( self, callback_list: list[str | Callable | CustomLogger] | None, dynamic_callbacks_type: Literal["input", "success", "failure", "async_success", "async_failure"], ) -> list[str | Callable | CustomLogger] | None: """ Helper function to initialize CustomLogger compatible callbacks in self.dynamic_* callbacks - If a callback is in litellm._known_custom_logger_compatible_callbacks, replace the string with the initialized callback class. - If dynamic callback is a "success" callback that is a known_custom_logger_compatible_callbacks then add it to dynamic_async_success_callbacks - If dynamic callback is a "failure" callback that is a known_custom_logger_compatible_callbacks then add it to dynamic_failure_callbacks """ if callback_list is None: return None processed_list: Final[list[str | Callable | CustomLogger]] = [] for callback in callback_list: if isinstance(callback, str) and callback in litellm._known_custom_logger_compatible_callbacks: for callback_instance in self._resolve_dynamic_callback_string(callback): processed_list.append(callback_instance) # If processing dynamic_success_callbacks, add to dynamic_async_success_callbacks if dynamic_callbacks_type == "success": if self.dynamic_async_success_callbacks is None: self.dynamic_async_success_callbacks = [] self.dynamic_async_success_callbacks.append(callback_instance) elif dynamic_callbacks_type == "failure": if self.dynamic_async_failure_callbacks is None: self.dynamic_async_failure_callbacks = [] self.dynamic_async_failure_callbacks.append(callback_instance) else: processed_list.append(callback) return processed_list def _resolve_dynamic_callback_string(self, callback: str) -> "tuple[CustomLogger, ...]": """ Resolve a known callback name to the logger instance(s) it dispatches to. For callbacks that support team-scoped credentials (datadog, newrelic), only the proxy-stamped team/key callback vars are passed as custom_logger_init_args: dd_*/newrelic_* params are blocked from standard_callback_dynamic_params (request-level security), so the trusted-vars channel is the only way credentials reach a per-team logger. """ _trusted_var_prefix: Final = "dd_" if callback == "datadog" else "newrelic_" if callback == "newrelic" else None _custom_logger_init_args: Final[dict | None] = ( {k: v for k, v in self._trusted_callback_vars if k.startswith(_trusted_var_prefix)} if _trusted_var_prefix is not None else None ) callback_class: Final = _init_custom_logger_compatible_class( callback, internal_usage_cache=None, llm_router=None, custom_logger_init_args=_custom_logger_init_args, ) if callback_class is None: return () # With team creds, "newrelic" resolves to the per-team METRICS logger; # resolve the name again without creds so the trace logger (OTel v2 / # legacy agent) keeps receiving this request. _newrelic_trace_class: Final = ( _init_custom_logger_compatible_class(callback, internal_usage_cache=None, llm_router=None) if callback == "newrelic" and _custom_logger_init_args and _custom_logger_init_args.get("newrelic_api_key") else None ) if _newrelic_trace_class is not None and _newrelic_trace_class is not callback_class: return (callback_class, _newrelic_trace_class) return (callback_class,) def initialize_standard_callback_dynamic_params(self, kwargs: dict | None = None) -> StandardCallbackDynamicParams: """ Initialize the standard callback dynamic params from the kwargs checks if langfuse_secret_key, gcs_bucket_name in kwargs and sets the corresponding attributes in StandardCallbackDynamicParams """ return _initialize_standard_callback_dynamic_params(kwargs) def initialize_standard_built_in_tools_params(self, kwargs: dict | None = None) -> StandardBuiltInToolsParams: """ Initialize the standard built-in tools params from the kwargs checks if web_search_options in kwargs or tools and sets the corresponding attribute in StandardBuiltInToolsParams """ return StandardBuiltInToolsParams( web_search_options=StandardBuiltInToolCostTracking._get_web_search_options(kwargs or {}), file_search=StandardBuiltInToolCostTracking._get_file_search_tool_call(kwargs or {}), ) def get_router_model_id(self) -> str | None: """Extract the router deployment model_id from litellm_params. Checks both litellm_metadata and metadata for model_info.id. Used by cost calculators to look up custom pricing registered under the deployment's model_info.id in litellm.model_cost. """ if not hasattr(self, "litellm_params"): return None for key in ("litellm_metadata", "metadata"): meta = self.litellm_params.get(key, {}) or {} info = meta.get("model_info", {}) or {} model_id = info.get("id") if model_id is not None: return model_id return None def get_deployment_model_for_cost(self) -> str | None: """The provider-qualified model to price against. On a batch retrieve both self.model and litellm_params["model"] can be unset, and self.model can otherwise carry the router's model_group alias, which no cost map resolves. model_call_details holds the deployment's own provider-qualified model, so it is preferred. """ candidates: Final = ( (self.model_call_details or {}).get("model") if hasattr(self, "model_call_details") else None, self.litellm_params.get("model") if hasattr(self, "litellm_params") else None, self.model, ) return next((candidate for candidate in candidates if isinstance(candidate, str) and candidate), None) def get_router_deployment_model_info(self) -> ModelInfo | None: """See deployment_pricing_model_info; None means fall back to the global cost map.""" return deployment_pricing_model_info( model_id=self.get_router_model_id(), deployment_model=self.get_deployment_model_for_cost(), ) def update_environment_variables( self, litellm_params: dict, optional_params: dict, model: str | None = None, user: str | None = None, **additional_params, ): self.optional_params = optional_params if model is not None: self.model = model self.user = user self.litellm_params = { **self.litellm_params, **scrub_sensitive_keys_in_metadata(litellm_params), } self.litellm_request_debug = litellm_params.get("litellm_request_debug", False) self.logger_fn = litellm_params.get("logger_fn", None) if _is_debugging_on() or self.litellm_request_debug: verbose_logger.debug("self.optional_params: %s", self.optional_params) self.model_call_details.update( { "model": self.model, "messages": self.messages, "optional_params": self.optional_params, "litellm_params": self.litellm_params, "start_time": self.start_time, "stream": self.stream, "user": user, "call_type": str(self.call_type), "litellm_call_id": self.litellm_call_id, "completion_start_time": self.completion_start_time, "standard_callback_dynamic_params": self.standard_callback_dynamic_params, **self.optional_params, **additional_params, } ) ## check if stream options is set ## - used by CustomStreamWrapper for easy instrumentation if "stream_options" in additional_params: self.stream_options = additional_params["stream_options"] ## check if custom pricing set ## if any(litellm_params.get(key) is not None for key in _CUSTOM_PRICING_KEYS & litellm_params.keys()): self.custom_pricing = True if "custom_llm_provider" in self.model_call_details: self.custom_llm_provider = self.model_call_details["custom_llm_provider"] def update_from_kwargs( self, kwargs: dict, litellm_params: dict | None = None, optional_params: dict | None = None, model: str | None = None, user: str | None = None, **additional_params, ): """ Convenience wrapper around update_environment_variables that automatically extracts metadata/litellm_metadata from kwargs, so callers don't need to manually plumb them into litellm_params. """ base_litellm_params: Final[dict[str, Any]] = {} if isinstance(kwargs.get("metadata"), dict): base_litellm_params["metadata"] = kwargs["metadata"].copy() if "litellm_metadata" in kwargs and isinstance(kwargs["litellm_metadata"], dict): base_litellm_params["litellm_metadata"] = kwargs["litellm_metadata"] if "metadata" not in base_litellm_params: base_litellm_params["metadata"] = kwargs["litellm_metadata"].copy() if litellm_params: # Merge metadata carefully — don't overwrite the merged metadata # from kwargs/litellm_metadata with the caller's litellm_params metadata. # e.g. anthropic_messages passes Anthropic's native metadata ({user_id: ...}) # in litellm_params, which would overwrite proxy key-auth fields. lp_metadata: Final = litellm_params.pop("metadata", None) base_litellm_params.update(litellm_params) if lp_metadata and isinstance(lp_metadata, dict): base_litellm_params.setdefault("metadata", {}) for k, v in lp_metadata.items(): if k not in base_litellm_params["metadata"]: base_litellm_params["metadata"][k] = v self.update_environment_variables( litellm_params=base_litellm_params, optional_params=optional_params or {}, model=model, user=user, **additional_params, ) def update_messages(self, messages: list[AllMessageValues]): """ Update the logged value of the messages in the model_call_details Allows pre-call hooks to update the messages before the call is made """ self.messages = messages self.model_call_details["messages"] = messages def should_run_prompt_management_hooks( self, non_default_params: dict, prompt_id: str | None = None, tools: list[dict] | None = None, ) -> bool: """ Return True if prompt management hooks should be run """ if prompt_id: return True # Check if model uses litellm_agent prefix (model replacement without prompt_id) model: Final = non_default_params.get("model", "") if isinstance(model, str) and model.startswith("litellm_agent/"): return True if self._should_run_prompt_management_hooks_without_prompt_id( non_default_params=non_default_params, tools=tools, ): return True return False def _should_run_prompt_management_hooks_without_prompt_id( self, non_default_params: dict, tools: list[dict] | None = None, ) -> bool: """ Certain prompt management hooks don't need a `prompt_id` to be passed in, they are triggered by dynamic params eg. AnthropicCacheControlHook and BedrockKnowledgeBaseHook both don't require a `prompt_id` to be passed in, they are triggered by dynamic params """ for param in DynamicPromptManagementParamLiteral.list_all_params(): if non_default_params.get(param): return True ############################################################################# # Check if Vector Store / Knowledge Base hooks should be applied to the prompt ############################################################################# if litellm.vector_store_registry is not None: if litellm.vector_store_registry.get_vector_store_to_run( non_default_params=non_default_params, tools=tools ): return True return False def get_chat_completion_prompt( self, model: str, messages: list[AllMessageValues], non_default_params: dict, prompt_variables: dict | None, prompt_id: str | None = None, prompt_spec: PromptSpec | None = None, prompt_management_logger: CustomLogger | None = None, prompt_label: str | None = None, prompt_version: int | None = None, request_kwargs: dict[str, object] | None = None, # mutable-ok: marker stamped into live request kwargs ) -> tuple[str, list[AllMessageValues], dict]: from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook custom_logger: Final = prompt_management_logger or self.get_custom_logger_for_prompt_management( model=model, non_default_params=non_default_params, prompt_id=prompt_id, prompt_spec=prompt_spec, dynamic_callback_params=self.standard_callback_dynamic_params, ) if custom_logger: breakpoints_before: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(messages) ( model, messages, non_default_params, ) = custom_logger.get_chat_completion_prompt( model=model, messages=messages, non_default_params=non_default_params or {}, prompt_id=prompt_id, prompt_spec=prompt_spec, prompt_variables=prompt_variables, dynamic_callback_params=self.standard_callback_dynamic_params, prompt_label=prompt_label, prompt_version=prompt_version, ) if request_kwargs is not None: AnthropicCacheControlHook.record_gateway_injection( request_kwargs, AnthropicCacheControlHook.count_request_cache_breakpoints(messages) - breakpoints_before, ) self.messages = messages return model, messages, non_default_params async def async_get_chat_completion_prompt( self, model: str, messages: list[AllMessageValues], non_default_params: dict, prompt_variables: dict | None, prompt_id: str | None = None, prompt_spec: PromptSpec | None = None, prompt_management_logger: CustomLogger | None = None, tools: list[dict] | None = None, prompt_label: str | None = None, prompt_version: int | None = None, request_kwargs: dict[str, object] | None = None, # mutable-ok: marker stamped into live request kwargs ) -> tuple[str, list[AllMessageValues], dict]: from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook custom_logger: Final = prompt_management_logger or self.get_custom_logger_for_prompt_management( model=model, tools=tools, non_default_params=non_default_params, prompt_id=prompt_id, prompt_spec=prompt_spec, dynamic_callback_params=self.standard_callback_dynamic_params, ) if custom_logger: breakpoints_before: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(messages) ( model, messages, non_default_params, ) = await custom_logger.async_get_chat_completion_prompt( model=model, messages=messages, non_default_params=non_default_params or {}, prompt_id=prompt_id, prompt_spec=prompt_spec, prompt_variables=prompt_variables, dynamic_callback_params=self.standard_callback_dynamic_params, litellm_logging_obj=self, tools=tools, prompt_label=prompt_label, prompt_version=prompt_version, ) if request_kwargs is not None: AnthropicCacheControlHook.record_gateway_injection( request_kwargs, AnthropicCacheControlHook.count_request_cache_breakpoints(messages) - breakpoints_before, ) self.messages = messages return model, messages, non_default_params def _auto_detect_prompt_management_logger( self, prompt_id: str, prompt_spec: PromptSpec | None, dynamic_callback_params: StandardCallbackDynamicParams, ) -> CustomLogger | None: """ Auto-detect which prompt management system owns the given prompt_id. This allows a user to just pass prompt_id in the completion call and it will be auto-detected which system owns this prompt. Args: prompt_id: The prompt ID to check dynamic_callback_params: Dynamic callback parameters for should_run_prompt_management checks Returns: A CustomLogger instance if a matching prompt management system is found, None otherwise """ prompt_management_loggers: Final = litellm.logging_callback_manager.get_custom_loggers_for_type( callback_type=CustomPromptManagement ) for logger in prompt_management_loggers: if isinstance(logger, CustomPromptManagement): try: if logger.should_run_prompt_management( prompt_id=prompt_id, prompt_spec=prompt_spec, dynamic_callback_params=dynamic_callback_params, ): self.model_call_details["prompt_integration"] = logger.__class__.__name__ return logger except Exception: # If check fails, continue to next logger continue return None @staticmethod def _prompt_manager_runs_without_prompt_id( logger: CustomLogger, prompt_spec: PromptSpec | None, dynamic_callback_params: StandardCallbackDynamicParams | None, ) -> bool: if not isinstance(logger, CustomPromptManagement): return False try: return logger.should_run_prompt_management( prompt_id=None, prompt_spec=prompt_spec, dynamic_callback_params=dynamic_callback_params or StandardCallbackDynamicParams(), ) except Exception: return False def get_custom_logger_for_prompt_management( self, model: str, non_default_params: dict, tools: list[dict] | None = None, prompt_id: str | None = None, prompt_spec: PromptSpec | None = None, dynamic_callback_params: StandardCallbackDynamicParams | None = None, ) -> CustomLogger | None: """ Get a custom logger for prompt management based on model name or available callbacks. Args: model: The model name to check for prompt management integration non_default_params: Non-default parameters passed to the completion call tools: Optional tools passed to the completion call prompt_id: Optional prompt ID to auto-detect which system owns this prompt dynamic_callback_params: Dynamic callback parameters for should_run_prompt_management checks Returns: A CustomLogger instance if one is found, None otherwise """ # First check if model starts with a known custom logger compatible callback # This takes precedence for backward compatibility for callback_name in litellm._known_custom_logger_compatible_callbacks: if model.startswith(callback_name): custom_logger = _init_custom_logger_compatible_class( logging_integration=callback_name, internal_usage_cache=None, llm_router=None, ) if custom_logger is not None: self.model_call_details["prompt_integration"] = model.split("/")[0] return custom_logger # If prompt_id is provided, try to auto-detect which system has this prompt if prompt_id and dynamic_callback_params is not None: auto_detected_logger: Final = self._auto_detect_prompt_management_logger( prompt_id=prompt_id, prompt_spec=prompt_spec, dynamic_callback_params=dynamic_callback_params, ) if auto_detected_logger is not None: return auto_detected_logger # Then check for any registered CustomPromptManagement loggers (fallback) prompt_management_loggers: Final = litellm.logging_callback_manager.get_custom_loggers_for_type( callback_type=CustomPromptManagement ) for logger in prompt_management_loggers: if prompt_id is None and not self._prompt_manager_runs_without_prompt_id( logger=logger, prompt_spec=prompt_spec, dynamic_callback_params=dynamic_callback_params, ): continue self.model_call_details["prompt_integration"] = logger.__class__.__name__ return logger if ( anthropic_cache_control_logger := AnthropicCacheControlHook.get_custom_logger_for_anthropic_cache_control_hook(non_default_params) ): self.model_call_details["prompt_integration"] = anthropic_cache_control_logger.__class__.__name__ return anthropic_cache_control_logger ######################################################### # Vector Store / Knowledge Base hooks ######################################################### if litellm.vector_store_registry is not None: vector_store_custom_logger: Final = _init_custom_logger_compatible_class( logging_integration="vector_store_pre_call_hook", internal_usage_cache=None, llm_router=None, ) self.model_call_details["prompt_integration"] = vector_store_custom_logger.__class__.__name__ # Add to global callbacks so post-call hooks are invoked if vector_store_custom_logger and vector_store_custom_logger not in litellm.callbacks: litellm.logging_callback_manager.add_litellm_callback(vector_store_custom_logger) return vector_store_custom_logger return None def get_custom_logger_for_anthropic_cache_control_hook(self, non_default_params: dict) -> CustomLogger | None: if non_default_params.get("cache_control_injection_points", None): custom_logger: Final = _init_custom_logger_compatible_class( logging_integration="anthropic_cache_control_hook", internal_usage_cache=None, llm_router=None, ) return custom_logger return None def _get_raw_request_body(self, data: dict | str | None) -> dict: if data is None: return {"error": "Received empty dictionary for raw request body"} if isinstance(data, str): try: return json.loads(data) except Exception: return {"error": f"Unable to parse raw request body. Got - {data}"} return data def _get_masked_api_base(self, api_base: str) -> str: if "key=" in api_base: # Find the position of "key=" in the string key_index: Final = api_base.find("key=") + 4 # Mask the last 5 characters after "key=" masked_api_base = api_base[:key_index] + "*" * 5 + api_base[-4:] else: masked_api_base = api_base return str(masked_api_base) def _pre_call(self, input, api_key, model=None, additional_args={}): """ Common helper function across the sync + async pre-call function """ self.model_call_details["input"] = input self.model_call_details["api_key"] = api_key self.model_call_details["additional_args"] = additional_args self.model_call_details["log_event_type"] = "pre_api_call" if model: # if model name was changes pre-call, overwrite the initial model call name with the new one self.model_call_details["model"] = model self.model_call_details["litellm_params"]["api_base"] = self._get_masked_api_base( additional_args.get("api_base", "") ) def pre_call(self, input, api_key, model=None, additional_args={}): # Log the exact input to the LLM API try: self._pre_call( input=input, api_key=api_key, model=model, additional_args=additional_args, ) # User Logging -> if you pass in a custom logging function self._print_llm_call_debugging_log( api_base=additional_args.get("api_base", ""), headers=additional_args.get("headers", {}), additional_args=additional_args, ) # log raw request to provider (like LangFuse) -- if opted in. if self.log_raw_request_response is True or log_raw_request_response is True: _litellm_params: Final = self.model_call_details.get("litellm_params", {}) _metadata: Final = _litellm_params.get("metadata", {}) or {} try: # [Non-blocking Extra Debug Information in metadata] if turn_off_message_logging is True: _metadata["raw_request"] = "redacted by litellm. \ 'litellm.turn_off_message_logging=True'" else: curl_command: Final = self._get_request_curl_command( api_base=additional_args.get("api_base", ""), headers=additional_args.get("headers", {}), additional_args=additional_args, data=additional_args.get("complete_input_dict", {}), ) _metadata["raw_request"] = _redact_string(str(curl_command)) # split up, so it's easier to parse in the UI self.model_call_details["raw_request_typed_dict"] = RawRequestTypedDict( raw_request_api_base=self._get_masked_api_base(str(additional_args.get("api_base") or "")), raw_request_body=self._get_raw_request_body(additional_args.get("complete_input_dict", {})), # NOTE: setting ignore_sensitive_headers to True will cause # the Authorization header to be leaked when calls to the health # endpoint are made and fail. raw_request_headers=self._get_masked_headers( additional_args.get("headers", {}) or {}, ), error=None, ) except Exception as e: self.model_call_details["raw_request_typed_dict"] = RawRequestTypedDict( error=str(e), ) _metadata["raw_request"] = _redact_string( f"Unable to Log \ raw request: {e}" ) if getattr(self, "logger_fn", None) and callable(self.logger_fn): try: self.logger_fn( self.model_call_details ) # Expectation: any logger function passed in by the user should accept a dict object except Exception as e: verbose_logger.exception( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while logging %s", e ) self.model_call_details["api_call_start_time"] = datetime.datetime.now() # Set-once first provider-handoff instant. api_call_start_time # is overwritten on every retry, so it can't measure one-time # preprocessing; pinning the first attempt excludes retry loops # + backoff. Logging object only — must NOT go into # litellm_params["metadata"] (caller request metadata, typed # Dict[str, str], echoed downstream; a datetime breaks it). if self.model_call_details.get("first_api_call_start_time") is None: self.model_call_details["first_api_call_start_time"] = self.model_call_details["api_call_start_time"] # Input Integration Logging -> If you want to log the fact that an attempt to call the model was made callbacks: Final = litellm.input_callback + (self.dynamic_input_callbacks or []) for callback in callbacks: try: if callback == "supabase" and supabaseClient is not None: verbose_logger.debug("reaches supabase for logging!") model = self.model_call_details["model"] messages = self.model_call_details["input"] verbose_logger.debug("supabaseClient: %s", supabaseClient) supabaseClient.input_log_event( model=model, messages=messages, end_user=self.model_call_details.get("user", "default"), litellm_call_id=self.litellm_params["litellm_call_id"], print_verbose=print_verbose, ) elif callback == "sentry" and add_breadcrumb: try: details_to_log = copy.deepcopy(self.model_call_details) except Exception: details_to_log = self.model_call_details if litellm.turn_off_message_logging: # make a copy of the _model_Call_details and log it details_to_log.pop("messages", None) details_to_log.pop("input", None) details_to_log.pop("prompt", None) add_breadcrumb( category="litellm.llm_call", message=f"Model Call Details pre-call: {details_to_log}", level="info", ) elif isinstance(callback, CustomLogger): # custom logger class callback.log_pre_api_call( model=self.model, messages=self.messages, kwargs=self.model_call_details, ) elif callable(callback) and customLogger is not None: # custom logger functions customLogger.log_input_event( model=self.model, messages=self.messages, kwargs=self.model_call_details, print_verbose=print_verbose, callback_func=callback, ) except Exception as e: verbose_logger.exception("litellm.Logging.pre_call(): Exception occured - %s", e) verbose_logger.debug( "LiteLLM.Logging: is sentry capture exception initialized %s", capture_exception ) if capture_exception: # log this error to sentry for debugging capture_exception(e) except Exception as e: verbose_logger.exception("LiteLLM.LoggingError: [Non-Blocking] Exception occurred while logging %s", e) verbose_logger.error("LiteLLM.Logging: is sentry capture exception initialized %s", capture_exception) if capture_exception: # log this error to sentry for debugging capture_exception(e) def _print_llm_call_debugging_log( self, api_base: str, headers: dict, additional_args: dict, ): """ Internal debugging helper function Prints the RAW curl command sent from LiteLLM """ if _is_debugging_on() or self.litellm_request_debug: if json_logs: masked_headers: Final = self._get_masked_headers(headers) masked_api_base: Final = self._get_masked_api_base(str(api_base or "")) if self.litellm_request_debug: verbose_logger.warning( # .warning ensures this shows up in all environments "POST Request Sent from LiteLLM", extra={"api_base": {masked_api_base}, **masked_headers}, ) else: verbose_logger.debug( "POST Request Sent from LiteLLM", extra={"api_base": {masked_api_base}, **masked_headers}, ) else: headers = additional_args.get("headers", {}) if headers is None: headers = {} data: Final = additional_args.get("complete_input_dict", {}) api_base = str(additional_args.get("api_base", "")) curl_command: Final = self._get_request_curl_command( api_base=api_base, headers=headers, additional_args=additional_args, data=data, ) if self.litellm_request_debug: verbose_logger.warning( "\x1b[92m%s\x1b[0m\n", curl_command ) # .warning ensures this shows up in all environments else: verbose_logger.debug("\x1b[92m%s\x1b[0m\n", curl_command) def _get_request_body(self, data: dict) -> str: return str(data) def _get_request_curl_command(self, api_base: str, headers: dict | None, additional_args: dict, data: dict) -> str: masked_api_base: Final = self._get_masked_api_base(api_base) if headers is None: headers = {} curl_command = "\n\nPOST Request Sent from LiteLLM:\n" curl_command += "curl -X POST \\\n" curl_command += f"{masked_api_base} \\\n" masked_headers: Final = self._get_masked_headers(headers) formatted_headers: Final = " ".join([f"-H '{k}: {v}'" for k, v in masked_headers.items()]) curl_command += f"{formatted_headers} \\\n" if formatted_headers.strip() != "" else "" curl_command += f"-d '{self._get_request_body(data)}'\n" if additional_args.get("request_str", None) is not None: # print the sagemaker / bedrock client request curl_command = "\nRequest Sent from LiteLLM:\n" request_str: Final = additional_args.get("request_str", "") curl_command += request_str return curl_command def _get_masked_headers(self, headers: dict, ignore_sensitive_headers: bool = False) -> dict: """ Internal debugging helper function Masks the headers of the request sent from LiteLLM """ return _get_masked_values(headers, ignore_sensitive_values=ignore_sensitive_headers) def post_call(self, original_response, input=None, api_key=None, additional_args={}): # Log the exact result from the LLM API, for streaming - log the type of response received if isinstance(original_response, dict): original_response = json.dumps(original_response, default=str) try: self.model_call_details["input"] = input self.model_call_details["api_key"] = api_key self.model_call_details["original_response"] = original_response self.model_call_details["additional_args"] = additional_args self.model_call_details["log_event_type"] = "post_api_call" attr: Literal["warning", "debug"] if self.litellm_request_debug: attr = "warning" else: attr = "debug" if json_logs: callattr = getattr(verbose_logger, attr) callattr( "RAW RESPONSE:\n{}\n\n".format( self.model_call_details.get("original_response", self.model_call_details) ), ) else: callattr = getattr(verbose_logger, attr) callattr( "RAW RESPONSE:\n{}\n\n".format( self.model_call_details.get("original_response", self.model_call_details) ) ) if getattr(self, "logger_fn", None) and callable(self.logger_fn): try: self.logger_fn( self.model_call_details ) # Expectation: any logger function passed in by the user should accept a dict object except Exception as e: verbose_logger.exception( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while logging %s", e ) original_response = redact_message_input_output_from_logging( model_call_details=(self.model_call_details if hasattr(self, "model_call_details") else {}), result=original_response, ) # Input Integration Logging -> If you want to log the fact that an attempt to call the model was made callbacks: Final = litellm.input_callback + (self.dynamic_input_callbacks or []) for callback in callbacks: try: if callback == "sentry" and add_breadcrumb: verbose_logger.debug("reaches sentry breadcrumbing") try: details_to_log = copy.deepcopy(self.model_call_details) except Exception: details_to_log = self.model_call_details if litellm.turn_off_message_logging: # make a copy of the _model_Call_details and log it details_to_log.pop("messages", None) details_to_log.pop("input", None) details_to_log.pop("prompt", None) add_breadcrumb( category="litellm.llm_call", message=f"Model Call Details post-call: {details_to_log}", level="info", ) elif isinstance(callback, CustomLogger): # custom logger class callback.log_post_api_call( kwargs=self.model_call_details, response_obj=None, start_time=self.start_time, end_time=None, ) except Exception as e: verbose_logger.exception( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while post-call logging with integrations %s", e, ) verbose_logger.debug( "LiteLLM.Logging: is sentry capture exception initialized %s", capture_exception ) if capture_exception: # log this error to sentry for debugging capture_exception(e) except Exception as e: verbose_logger.exception("LiteLLM.LoggingError: [Non-Blocking] Exception occurred while logging %s", e) async def async_post_mcp_tool_call_hook( self, kwargs: dict, response_obj: Any, start_time: datetime.datetime, end_time: datetime.datetime, ): """ Post MCP Tool Call Hook Use this to modify the MCP tool call response before it is returned to the user. """ from litellm.types.llms.base import HiddenParams from litellm.types.mcp import MCPPostCallResponseObject callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_success_callbacks, global_callbacks=litellm.success_callback, ) post_mcp_tool_call_response_obj: Final[MCPPostCallResponseObject] = MCPPostCallResponseObject( mcp_tool_call_response=response_obj, hidden_params=HiddenParams() ) for callback in callbacks: try: if isinstance(callback, CustomLogger): response: MCPPostCallResponseObject | None = await callback.async_post_mcp_tool_call_hook( kwargs=kwargs, response_obj=post_mcp_tool_call_response_obj, start_time=start_time, end_time=end_time, ) ###################################################################### # if any of the callbacks modify the response, use the modified response # current implementation returns the first modified response ###################################################################### if response is not None: response_obj = self._parse_post_mcp_call_hook_response(response=response) except Exception as e: verbose_logger.exception("LiteLLM.LoggingError: [Non-Blocking] Exception occurred while logging %s", e) return response_obj def _parse_post_mcp_call_hook_response( self, response: MCPPostCallResponseObject | None ) -> "Sequence[TextContent | ImageContent | EmbeddedResource] | None": """ Parse the response from the post_mcp_tool_call_hook 1. Unpack the mcp_tool_call_response 2. save the updated response_cost to the model_call_details """ if response is None: return None self.model_call_details["response_cost"] = response.hidden_params.response_cost return response.mcp_tool_call_response def get_response_ms(self) -> float: return ( self.model_call_details.get("end_time", datetime.datetime.now()) - self.model_call_details.get("start_time", datetime.datetime.now()) ).total_seconds() * 1000 def set_cost_breakdown( self, input_cost: float, output_cost: float, total_cost: float, cost_for_built_in_tools_cost_usd_dollar: float, additional_costs: dict | None = None, original_cost: float | None = None, discount_percent: float | None = None, discount_amount: float | None = None, margin_percent: float | None = None, margin_fixed_amount: float | None = None, margin_total_amount: float | None = None, cache_read_cost: float | None = None, cache_creation_cost: float | None = None, reasoning_cost: float | None = None, service_tier: str | None = None, data_residency: str | None = None, vertex_location: str | None = None, ) -> None: """ Helper method to store cost breakdown in the logging object. Args: input_cost: Cost of input/prompt tokens output_cost: Cost of output/completion tokens cost_for_built_in_tools_cost_usd_dollar: Cost of built-in tools total_cost: Total cost of request additional_costs: Free-form additional costs dict (e.g., {"azure_model_router_flat_cost": 0.00014}) original_cost: Cost before discount discount_percent: Discount percentage (0.05 = 5%) discount_amount: Discount amount in USD margin_percent: Margin percentage applied (0.10 = 10%) margin_fixed_amount: Fixed margin amount in USD margin_total_amount: Total margin added in USD service_tier: Tier the costs above were priced on, already resolved data_residency: Region uplift the costs above were priced on, already resolved vertex_location: Vertex AI location the costs above were priced on, already resolved """ self.cost_breakdown = CostBreakdown( input_cost=input_cost, output_cost=output_cost, total_cost=total_cost, tool_usage_cost=cost_for_built_in_tools_cost_usd_dollar, service_tier=service_tier, data_residency=data_residency, vertex_location=vertex_location, ) if cache_read_cost is not None and cache_read_cost > 0: self.cost_breakdown["cache_read_cost"] = cache_read_cost if cache_creation_cost is not None and cache_creation_cost > 0: self.cost_breakdown["cache_creation_cost"] = cache_creation_cost if reasoning_cost is not None and reasoning_cost > 0: self.cost_breakdown["reasoning_cost"] = reasoning_cost # Store additional costs if provided (free-form dict for extensibility) if additional_costs and isinstance(additional_costs, dict) and len(additional_costs) > 0: self.cost_breakdown["additional_costs"] = additional_costs # Store discount information if provided if original_cost is not None: self.cost_breakdown["original_cost"] = original_cost if discount_percent is not None: self.cost_breakdown["discount_percent"] = discount_percent if discount_amount is not None: self.cost_breakdown["discount_amount"] = discount_amount # Store margin information if provided if margin_percent is not None: self.cost_breakdown["margin_percent"] = margin_percent if margin_fixed_amount is not None: self.cost_breakdown["margin_fixed_amount"] = margin_fixed_amount if margin_total_amount is not None: self.cost_breakdown["margin_total_amount"] = margin_total_amount def _response_cost_calculator( self, result: Union[ ModelResponse, ModelResponseStream, EmbeddingResponse, ImageResponse, TranscriptionResponse, TextCompletionResponse, HttpxBinaryResponseContent, RerankResponse, Batch, FineTuningJob, ResponsesAPIResponse, ResponseCompletedEvent, OpenAIFileObject, LiteLLMRealtimeStreamLoggingObject, OpenAIModerationResponse, "SearchResponse", dict, list, ], cache_hit: bool | None = None, litellm_model_name: str | None = None, router_model_id: str | None = None, ) -> float | None: """ Calculate response cost using result + logging object variables. used for consistent cost calculation across response headers + logging integrations. """ if cache_hit is None: cache_hit = self.model_call_details.get("cache_hit", False) if cache_hit is True: return 0.0 if is_unbilled_non_inference_call( self.call_type, StandardLoggingPayloadSetup.merge_litellm_metadata(self.litellm_params), result ): return 0.0 transformed_result: Final = self._generate_content_result_as_model_response(result) if transformed_result is not None: result = transformed_result if isinstance(result, (BaseModel, HttpxBinaryResponseContent)) and hasattr(result, "_hidden_params"): hidden_params: Final = getattr(result, "_hidden_params", {}) if ( "response_cost" in hidden_params and hidden_params["response_cost"] is not None ): # use cost if already calculated return hidden_params["response_cost"] elif router_model_id is None and "model_id" in hidden_params: # use model_id if not already set router_model_id = hidden_params["model_id"] # Fallback: extract router_model_id from litellm_params when not available # from the result object. ResponsesAPIResponse objects (used by /v1/responses # streaming) don't carry _hidden_params["model_id"] like ModelResponse does. if router_model_id is None: router_model_id = self.get_router_model_id() ## RESPONSE COST ## custom_pricing: Final = use_custom_pricing_for_model( litellm_params=(self.litellm_params if hasattr(self, "litellm_params") else None) ) prompt = self._prompt_for_cost_calculation() if cache_hit is None: cache_hit = self.model_call_details.get("cache_hit", False) try: response_cost_calculator_kwargs: Final = { "response_object": result, "model": litellm_model_name or self.model, "cache_hit": cache_hit, "custom_llm_provider": self.model_call_details.get("custom_llm_provider", None), "base_model": _get_base_model_from_metadata(model_call_details=self.model_call_details), "call_type": self.call_type, "optional_params": self.optional_params, "custom_pricing": custom_pricing, "prompt": prompt, "standard_built_in_tools_params": self.standard_built_in_tools_params, "router_model_id": router_model_id, "litellm_logging_obj": self, "service_tier": (self.optional_params.get("service_tier") if self.optional_params else None), "data_residency": ( self.litellm_params.get("data_residency") if hasattr(self, "litellm_params") and self.litellm_params else None ), "vertex_location": _resolve_vertex_location_for_cost( custom_llm_provider=self.model_call_details.get("custom_llm_provider", None), litellm_params=(self.litellm_params if hasattr(self, "litellm_params") else None), optional_params=self.optional_params, model=litellm_model_name or self.model, ), } except Exception as e: # error creating kwargs for cost calculation debug_info = StandardLoggingModelCostFailureDebugInformation( error_str=str(e), traceback_str=_get_traceback_str_for_error(str(e)), ) verbose_logger.debug("response_cost_failure_debug_information: %s", debug_info) self.model_call_details["response_cost_failure_debug_information"] = debug_info return None try: response_cost: Final = litellm.response_cost_calculator(**response_cost_calculator_kwargs) verbose_logger.debug("response_cost: %s", response_cost) additional_response_cost: Final[object] = self.model_call_details.get("additional_response_cost") if isinstance(additional_response_cost, (int, float)) and additional_response_cost > 0: return (response_cost or 0.0) + additional_response_cost return response_cost except Exception as e: # error calculating cost debug_info = StandardLoggingModelCostFailureDebugInformation( error_str=str(e), traceback_str=_get_traceback_str_for_error(str(e)), model=response_cost_calculator_kwargs["model"], cache_hit=response_cost_calculator_kwargs["cache_hit"], custom_llm_provider=response_cost_calculator_kwargs["custom_llm_provider"], base_model=response_cost_calculator_kwargs["base_model"], call_type=response_cost_calculator_kwargs["call_type"], custom_pricing=response_cost_calculator_kwargs["custom_pricing"], ) verbose_logger.debug("response_cost_failure_debug_information: %s", debug_info) self.model_call_details["response_cost_failure_debug_information"] = debug_info return None def _prompt_for_cost_calculation(self) -> str: """ The raw input string is only priced directly for text-to-speech, which bills per character. Every other call type gets its billable units from the response usage object, and call types that carry no usage at all (file content retrieval, and anything else `function_setup` cannot build messages for) only have the ``"default-message-value"`` placeholder here, so passing the input along would token-price that placeholder. """ if self.call_type not in (CallTypes.speech.value, CallTypes.aspeech.value): return "" _input = self.model_call_details.get("input", None) return _input if isinstance(_input, str) else "" def _generate_content_result_as_model_response(self, result: object) -> ModelResponse | None: """ Native Google :generateContent bodies report token usage under ``usageMetadata``, which the cost calculator does not read, so a raw body always costs 0. The async success path already transforms it into a ``ModelResponse`` before costing; do the same transformation here so the synchronously-built ``x-litellm-response-cost`` header carries the real cost. Returns ``None`` (leaving the original result untouched) for other call types, for already-transformed ``ModelResponse`` results, and on any transformation failure. """ if self.call_type not in ( CallTypes.generate_content.value, CallTypes.agenerate_content.value, ): return None if isinstance(result, ModelResponse) or not isinstance(result, (BaseModel, dict)): return None try: import httpx completion_response = result.model_dump(by_alias=True) if isinstance(result, BaseModel) else dict(result) return litellm.VertexGeminiConfig()._transform_google_generate_content_to_openai_model_response( completion_response=completion_response, model_response=ModelResponse(), model=self.model or "", logging_obj=self, raw_response=httpx.Response(status_code=200, headers={}), ) except Exception as e: # noqa: BLE001 - cost normalization must never break the response path verbose_logger.debug("generate_content response cost normalization failed: %s", e) return None async def _response_cost_calculator_async( self, result: ModelResponse | ModelResponseStream | EmbeddingResponse | ImageResponse | TranscriptionResponse | TextCompletionResponse | HttpxBinaryResponseContent | RerankResponse | Batch | FineTuningJob, cache_hit: bool | None = None, ) -> float | None: return self._response_cost_calculator(result=result, cache_hit=cache_hit) @staticmethod def _is_sync_litellm_request(litellm_params: dict) -> bool: """True for sync SDK entrypoints (``completion``), false for async (``acompletion``, etc.).""" return ( litellm_params.get(CallTypes.acompletion.value, False) is not True and litellm_params.get(CallTypes.aresponses.value, False) is not True and litellm_params.get(CallTypes.aembedding.value, False) is not True and litellm_params.get(CallTypes.aimage_generation.value, False) is not True and litellm_params.get(CallTypes.atranscription.value, False) is not True and litellm_params.get(CallTypes.allm_passthrough_route.value, False) is not True and litellm_params.get(CallTypes.aanthropic_messages.value, False) is not True and litellm_params.get(CallTypes.agenerate_content.value, False) is not True and litellm_params.get(CallTypes.agenerate_content_stream.value, False) is not True ) def _is_assembled_stream_success(self, result=None) -> bool: """Final assembled stream export (not a per-chunk success call). Per-chunk callers pass a ``ModelResponseStream`` (or ``None``); the final assembled response is any other non-``None`` value (typically a ``ModelResponse``). Treating a chunk as the assembled response would prematurely set the ``has_dispatched_final_stream_success`` dedup guard and silently suppress the real final stream log. """ if self.stream is not True: return False if result is not None and not isinstance(result, ModelResponseStream): return True return ( "async_complete_streaming_response" in self.model_call_details or self.model_call_details.get("complete_streaming_response") is not None ) async def dispatch_success_handlers( self, result=None, start_time=None, end_time=None, cache_hit=None, prefer_async_handlers: bool = False, **kwargs, ) -> None: """Route success logging to async and/or sync handlers for this request. ``prefer_async_handlers`` only bypasses the sync-SDK-only shortcut (e.g. ``async for`` on a stream from ``completion()``). Legacy string callbacks still run via ``executor.submit(success_handler)`` when configured. """ from litellm.litellm_core_utils.thread_pool_executor import executor if self._is_assembled_stream_success(result): if self.model_call_details.get("has_dispatched_final_stream_success"): return self.model_call_details["has_dispatched_final_stream_success"] = True litellm_params: Final = self.model_call_details.get("litellm_params", {}) or {} sync_sdk: Final = self._is_sync_litellm_request(litellm_params) passthrough: Final = self.call_type == CallTypes.pass_through.value if sync_sdk and not prefer_async_handlers and not passthrough: self.success_handler( result, start_time=start_time, end_time=end_time, cache_hit=cache_hit, **kwargs, ) return await self.async_success_handler( result, start_time=start_time, end_time=end_time, cache_hit=cache_hit, **kwargs, ) if not self._should_run_sync_callbacks_for_async_calls(): return executor.submit( self.success_handler, result, start_time=start_time, end_time=end_time, cache_hit=cache_hit, **kwargs, ) async def dispatch_failure_handlers( self, exception: Exception, traceback_exception: str, prefer_async_handlers: bool = False, ) -> None: """Route failure logging to async and/or sync handlers for this request. Mirrors ``dispatch_success_handlers``: the sync ``failure_handler`` never runs concurrently with ``async_failure_handler`` on the shared logging object, so the two paths cannot mutate it at the same time. ``prefer_async_handlers`` only bypasses the sync-SDK-only shortcut (e.g. ``async for`` on a stream from ``completion()``); legacy string callbacks still run via ``executor.submit(failure_handler)`` when configured. """ litellm_params: Final = self.model_call_details.get("litellm_params", {}) or {} sync_sdk: Final = self._is_sync_litellm_request(litellm_params) passthrough: Final = self.call_type == CallTypes.pass_through.value if sync_sdk and not prefer_async_handlers and not passthrough: self.failure_handler(exception, traceback_exception) return await self.async_failure_handler(exception, traceback_exception) if not self._should_run_sync_failure_callbacks_for_async_calls(): return executor.submit(self.failure_handler, exception, traceback_exception) def should_run_logging( self, event_type: Literal["async_success", "sync_success", "async_failure", "sync_failure"], stream: bool = False, ) -> bool: try: if self.model_call_details.get(f"has_logged_{event_type}", False) is True: return False return True except Exception: return True def has_run_logging( self, event_type: Literal["async_success", "sync_success", "async_failure", "sync_failure"], ) -> None: if self.stream is not None and self.stream is True: """ Ignore check on stream, as there can be multiple chunks """ return self.model_call_details[f"has_logged_{event_type}"] = True return def should_run_callback(self, callback: litellm.CALLBACK_TYPES, litellm_params: dict, event_hook: str) -> bool: if litellm.global_disable_no_log_param: return True if litellm_params.get("no-log", False) is True: # proxy cost tracking cal backs should run if not (isinstance(callback, CustomLogger) and "_PROXY_" in callback.__class__.__name__): verbose_logger.debug("no-log request, skipping logging for %s event", event_hook) return False # Check for dynamically disabled callbacks via headers if EnterpriseCallbackControls is not None and EnterpriseCallbackControls.is_callback_disabled_dynamically( callback=callback, litellm_params=litellm_params, standard_callback_dynamic_params=self.standard_callback_dynamic_params, ): verbose_logger.debug( "Callback %s disabled via x-litellm-disable-callbacks header for %s event", callback, event_hook ) return False return True def _update_completion_start_time(self, completion_start_time: datetime.datetime): self.completion_start_time = completion_start_time self.model_call_details["completion_start_time"] = self.completion_start_time def normalize_logging_result(self, result: Any) -> object: """ Some endpoints return a different type of result than what is expected by the logging system. This function is used to normalize the result to the expected type. """ logging_result = result if self.call_type == CallTypes.arealtime.value and isinstance(result, list): combined_usage_object: Final = ( RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(results=result) ) logging_result = RealtimeAPITokenUsageProcessor.create_logging_realtime_object( usage=combined_usage_object, results=result, ) elif ( self.call_type == CallTypes.llm_passthrough_route.value or self.call_type == CallTypes.allm_passthrough_route.value ) and isinstance(result, Response): from litellm.utils import ProviderConfigManager provider_config: Final = ProviderConfigManager.get_provider_passthrough_config( provider=self.model_call_details.get("custom_llm_provider", ""), model=self.model, ) if provider_config is not None: logging_result = provider_config.logging_non_streaming_response( model=self.model, custom_llm_provider=self.model_call_details.get("custom_llm_provider", ""), httpx_response=result, request_data=self.model_call_details.get("request_data", {}), logging_obj=self, endpoint=self.model_call_details.get("endpoint", ""), ) return logging_result def _merge_hidden_params_from_response_into_metadata(self, logging_result: object) -> None: """ Copy response._hidden_params into litellm_params.metadata['hidden_params']. Non-streaming success uses _process_hidden_params_and_response_cost (skipped when stream=True). Streaming assembles the full response later; without this merge, OTEL/callbacks that read metadata.hidden_params miss cost-related fields. """ if logging_result is None: return hidden_params: Final = getattr(logging_result, "_hidden_params", None) if not hidden_params: return if self.model_call_details.get("litellm_params") is None: return metadata_hidden_params: Final = hidden_params.copy() response_cost: Final[object] = self.model_call_details.get("response_cost") if metadata_hidden_params.get("response_cost") is None and response_cost is not None: metadata_hidden_params["response_cost"] = response_cost litellm_params: Final = self.model_call_details["litellm_params"] metadata: Final = litellm_params.get("metadata") or {} litellm_params["metadata"] = metadata metadata["hidden_params"] = metadata_hidden_params def _process_hidden_params_and_response_cost( self, logging_result, start_time, end_time, ): """Resolve hidden params, compute response cost, and emit the standard logging payload.""" hidden_params: Final = getattr(logging_result, "_hidden_params", {}) if hidden_params: if self.model_call_details.get("litellm_params") is not None: self.model_call_details["litellm_params"].setdefault("metadata", {}) if self.model_call_details["litellm_params"]["metadata"] is None: self.model_call_details["litellm_params"]["metadata"] = {} self.model_call_details["litellm_params"]["metadata"]["hidden_params"] = getattr( logging_result, "_hidden_params", {} ) if self.model_call_details.get("cache_hit") is True: self.model_call_details["response_cost"] = 0.0 elif "response_cost" in hidden_params: self.model_call_details["response_cost"] = hidden_params["response_cost"] elif (existing_cost := self.model_call_details.get("response_cost")) is not None and existing_cost != 0: # Preserve response_cost if already calculated (e.g., by pass-through # handlers like Gemini/Vertex which call completion_cost directly). # Do not preserve 0 from failure_handler on intermediate router retries. pass else: self.model_call_details["response_cost"] = self._response_cost_calculator(result=logging_result) self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload( logging_result, start_time, end_time ) standard_logging_payload: Final[StandardLoggingPayload | None] = self.model_call_details.get( "standard_logging_object" ) if standard_logging_payload is not None: emit_standard_logging_payload(standard_logging_payload) def _build_standard_logging_payload( self, init_response_obj: object, start_time: Any, end_time: Any ) -> StandardLoggingPayload | None: """Build StandardLoggingPayload and accumulate its construction time.""" _start: Final = time.time() payload: Final = get_standard_logging_object_payload( kwargs=self.model_call_details, init_response_obj=init_response_obj, start_time=start_time, end_time=end_time, logging_obj=self, status="success", standard_built_in_tools_params=self.standard_built_in_tools_params, ) self.callback_duration_ms += (time.time() - _start) * 1000 return payload def _transform_usage_objects(self, result): if isinstance(result, ResponsesAPIResponse): result = result.model_copy() transformed_usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(result.usage) setattr(result, "usage", transformed_usage) if (standard_logging_payload := self.model_call_details.get("standard_logging_object")) is not None: response_dict: Final = result.model_dump() if hasattr(result, "model_dump") else dict(result) # Ensure usage is properly included with transformed chat format if transformed_usage is not None: response_dict["usage"] = ( transformed_usage.model_dump() if hasattr(transformed_usage, "model_dump") else dict(transformed_usage) ) standard_logging_payload["response"] = response_dict elif isinstance(result, TranscriptionResponse): from litellm.litellm_core_utils.llm_cost_calc.usage_object_transformation import ( TranscriptionUsageObjectTransformation, ) result = result.model_copy() transformed_usage = TranscriptionUsageObjectTransformation.transform_transcription_usage_object( result.usage ) setattr(result, "usage", transformed_usage) return result def _success_handler_helper_fn( self, result=None, start_time=None, end_time=None, cache_hit=None, standard_logging_object: StandardLoggingPayload | None = None, ): try: if start_time is None: start_time = self.start_time if end_time is None: end_time = datetime.datetime.now() if self.completion_start_time is None: self.completion_start_time = end_time self.model_call_details["completion_start_time"] = self.completion_start_time self.model_call_details["log_event_type"] = "successful_api_call" self.model_call_details["end_time"] = end_time self.model_call_details["cache_hit"] = cache_hit if self.call_type == CallTypes.anthropic_messages.value: result = self._handle_anthropic_messages_response_logging(result=result) elif ( self.call_type == CallTypes.generate_content.value or self.call_type == CallTypes.agenerate_content.value ): result = self._handle_non_streaming_google_genai_generate_content_response_logging(result=result) elif self.call_type == CallTypes.asend_message.value or self.call_type == CallTypes.send_message.value: result = self._handle_a2a_response_logging(result=result) logging_result: Final = self.normalize_logging_result(result=result) if standard_logging_object is None and result is not None and self.stream is not True: if self._is_recognized_call_type_for_logging(logging_result=logging_result) or isinstance( logging_result, (dict, list) ): self._process_hidden_params_and_response_cost( logging_result=logging_result, start_time=start_time, end_time=end_time, ) elif standard_logging_object is not None: self.model_call_details["standard_logging_object"] = standard_logging_object else: # Streaming reaches here before its cost is known, so the cost # is seeded to None, but only when nothing has already # established one. A stream that assembles into a response # object recomputes the cost right after this; a pass-through # stream cannot (its body is opaque) and carries the cost its # upstream reported in the response headers, which an # unconditional reset would discard. self.model_call_details.setdefault("response_cost", None) result = self._transform_usage_objects(result=result) if ( litellm.max_budget and self.stream is False and result is not None and isinstance(result, dict) and "content" in result ): time_diff: Final = (end_time - start_time).total_seconds() float_diff: Final = float(time_diff) litellm._current_cost += litellm.completion_cost( model=self.model, prompt="", completion=getattr(result, "content", ""), total_time=float_diff, standard_built_in_tools_params=self.standard_built_in_tools_params, ) return start_time, end_time, result except Exception as e: raise Exception(f"[Non-Blocking] LiteLLM.Success_Call Error: {e}") def _is_recognized_call_type_for_logging( self, logging_result: object, ): """ Returns True if the call type is recognized for logging (eg. ModelResponse, ModelResponseStream, etc.) """ if ( isinstance(logging_result, ModelResponse) or isinstance(logging_result, ModelResponseStream) or isinstance(logging_result, EmbeddingResponse) or isinstance(logging_result, ImageResponse) or isinstance(logging_result, TranscriptionResponse) or isinstance(logging_result, TextCompletionResponse) or isinstance(logging_result, HttpxBinaryResponseContent) # tts or isinstance(logging_result, RerankResponse) or isinstance(logging_result, FineTuningJob) or isinstance(logging_result, LiteLLMBatch) or isinstance(logging_result, ResponsesAPIResponse) or isinstance(logging_result, OpenAIFileObject) or isinstance(logging_result, LiteLLMRealtimeStreamLoggingObject) or isinstance(logging_result, OpenAIModerationResponse) or isinstance(logging_result, OCRResponse) # OCR or isinstance(logging_result, SearchResponse) # Search API or ( isinstance(logging_result, InteractionsAPIResponse) and logging_result.usage is not None and self._is_interactions_create_call_type() ) or isinstance(logging_result, dict) and logging_result.get("object") == "vector_store.search_results.page" or isinstance(logging_result, dict) and logging_result.get("object") == "search" # Search API (dict format) or isinstance(logging_result, VideoObject) or isinstance(logging_result, ContainerObject) or isinstance(logging_result, LiteLLMSendMessageResponse) # A2A or (self.call_type == CallTypes.call_mcp_tool.value) ): return True return False def _is_interactions_create_call_type(self) -> bool: """ Only interaction creation is billable. GET polls, deletes, and cancels also return an ``InteractionsAPIResponse`` (with usage once completed), so recognizing those would write spend on every poll of a background interaction. The proxy sets ``call_type`` from its route_type (``create_interaction``/``acreate_interaction``); the SDK sets it from the decorated function name (``create``/``acreate``). Recognition additionally requires a usage block (checked at the call site): a ``background=true`` create returns ``in_progress`` without usage, and billing it would write a $0 spend log under the interaction id that collides with the row the background poll task writes once the interaction completes (see ``litellm.interactions.background_cost_polling``). """ return self.call_type in ( CallTypes.create_interaction.value, CallTypes.acreate_interaction.value, "create", "acreate", ) async def async_log_background_interaction_completion( self, result: InteractionsAPIResponse, ) -> None: """ Log the terminal result of a background interaction as a fresh success event. The create request already ran success logging for its ``in_progress`` response (no usage, so no cost was tracked); clearing the dedup flags lets the completed result flow through cost calculation and spend tracking exactly once, spanning create to completion. The poll fetched this body through its own client call, which priced it against a throwaway logging object holding none of this request's deployment context: no ``model_info``, no router ``model_id``, no deployment ``litellm_params``. Keeping that price would bill a custom-priced deployment at the wrong rate, and it would also satisfy the "already calculated" shortcut and skip repricing here, leaving the cost breakdown at the zeros the usage-less create stamped and writing those zeros to the spend log. Dropping it makes this event price the settled body itself, against the deployment that served the create. The same throwaway call stamped the deployment identity that travels with the price, so ``model_id`` and ``litellm_model_name`` go with it. Left in place they overwrite the create's real deployment with the poll's empty one in the payload every logging integration reads. """ settled_hidden_params: Final = getattr(result, "_hidden_params", None) if isinstance(settled_hidden_params, dict): for poll_scoped_key in ("response_cost", "model_id", "litellm_model_name"): settled_hidden_params.pop(poll_scoped_key, None) self._reset_success_emission_dedupe() await self.async_success_handler(result=result) def _reset_success_emission_dedupe(self) -> None: """ Success callbacks dedupe per request, because the sync and async handlers both fire on some paths and would otherwise report one call twice. A settled background interaction is a genuinely second success event on the same request, so every such marker has to be cleared or the completion, the only event that carries usage and cost, is discarded as a duplicate of the in-progress create. """ self.model_call_details.pop("has_logged_async_success", None) litellm_params = self.model_call_details.get("litellm_params") if not isinstance(litellm_params, dict): return metadata = litellm_params.get("metadata") if not isinstance(metadata, dict): return otel_internal = metadata.get("_otel_internal") if not isinstance(otel_internal, dict): return spans_logged = otel_internal.get("spans_logged") if not isinstance(spans_logged, dict): return for scope in [key for key in spans_logged if isinstance(key, tuple) and key[-1:] == ("success",)]: del spans_logged[scope] def _flush_passthrough_collected_chunks_helper( self, raw_bytes: list[bytes], provider_config: "BasePassthroughConfig", ) -> Optional["CostResponseTypes"]: all_chunks: Final = provider_config._convert_raw_bytes_to_str_lines(raw_bytes) complete_streaming_response: Final = provider_config.handle_logging_collected_chunks( all_chunks=all_chunks, litellm_logging_obj=self, model=self.model, custom_llm_provider=self.model_call_details.get("custom_llm_provider", ""), endpoint=self.model_call_details.get("endpoint", ""), ) return complete_streaming_response def flush_passthrough_collected_chunks( self, raw_bytes: list[bytes], provider_config: "BasePassthroughConfig", ): """ Flush collected chunks from the logging object This is used to log the collected chunks once streaming is done on passthrough endpoints 1. Decode the raw bytes to string lines 2. Get the complete streaming response from the provider config 3. Log the complete streaming response (trigger success handler) This is used for passthrough endpoints """ complete_streaming_response: Final = self._flush_passthrough_collected_chunks_helper( raw_bytes=raw_bytes, provider_config=provider_config, ) if complete_streaming_response is not None: self.success_handler(result=complete_streaming_response) async def async_flush_passthrough_collected_chunks( self, raw_bytes: list[bytes], provider_config: "BasePassthroughConfig", ): complete_streaming_response: Final = self._flush_passthrough_collected_chunks_helper( raw_bytes=raw_bytes, provider_config=provider_config, ) if complete_streaming_response is not None: await self.async_success_handler(result=complete_streaming_response) def _restore_correlation_context(self) -> None: """Restore trace_id/session_id contextvars to their pre-call value. Without this, a nested LiteLLM call sharing the same asyncio Task as an outer request (e.g. a guardrail's own LLM-as-judge call, an MCP sampling call) would leave the outer request's subsequent log lines stamped with the nested call's trace_id/session_id instead of its own. Uses a plain set() of the captured pre-call value rather than contextvars.Token-based reset(), since this can end up called from a different asyncio Task/context than __init__ ran in (e.g. the request task's own wrapper() finally block, plus async_success_handler dispatched separately via asyncio.create_task/the logging worker) - reset() only works in the exact Context a Token was created in and raises otherwise. Deliberately NOT idempotent/guarded: each distinct Task that calls this needs its own restore to actually take effect in that Task's view of the contextvars, so calling it multiple times (once per Task involved in this attempt) is required, not just safe. """ set_trace_id(self._pre_call_trace_id) set_session_id(self._pre_call_session_id) def _restore_correlation_context_if_unclaimed(self) -> None: """Guarded variant for __del__-triggered cleanup only. __del__ can fire arbitrarily late (delayed by cyclic GC, possibly after the consuming Task/thread has already moved on to a different, still-active call). Unconditionally restoring in that case would stomp the active call's trace_id/session_id with this abandoned stream's stale pre-call snapshot. Only restore if the contextvars still hold the ids *this* call set - i.e. nothing has claimed them since - so an unrelated active call is never overwritten. """ if trace_id_var.get() == self._own_trace_id and session_id_var.get() == self._own_session_id: self._restore_correlation_context() def success_handler( self, result: Any = None, # heterogeneous response object; varies by call type (ANN401 ignored, see ruff-strict.toml) start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, cache_hit: bool | None = None, **kwargs: Any, # kwargs-ok: forwarded to _success_handler_body ) -> None: """Restores trace_id/session_id contextvars once this attempt's own success logging (including any nested calls its callbacks trigger) is fully done.""" try: return self._success_handler_body( result=result, start_time=start_time, end_time=end_time, cache_hit=cache_hit, **kwargs ) finally: self._restore_correlation_context() def _success_handler_body( self, result: object = None, start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, cache_hit: bool | None = None, **kwargs: Any, # kwargs-ok: forwarded from success_handler ) -> None: verbose_logger.debug("Logging Details LiteLLM-Success Call: Cache_hit=%s", cache_hit) if not self.should_run_logging(event_type="sync_success"): # prevent double logging return start_time, end_time, result = self._success_handler_helper_fn( start_time=start_time, end_time=end_time, result=result, cache_hit=cache_hit, standard_logging_object=kwargs.get("standard_logging_object", None), ) litellm_params = self.model_call_details.get("litellm_params", {}) is_sync_request: Final = self._is_sync_litellm_request(litellm_params) try: ## BUILD COMPLETE STREAMED RESPONSE complete_streaming_response: ( ModelResponse | TextCompletionResponse | ResponsesAPIResponse | InteractionsAPIResponse | None ) = None if "complete_streaming_response" in self.model_call_details: return # break out of this. complete_streaming_response = self._get_assembled_streaming_response( result=result, start_time=start_time, end_time=end_time, is_async=False, streaming_chunks=self.sync_streaming_chunks, ) if complete_streaming_response is not None: verbose_logger.debug("Logging Details LiteLLM-Success Call streaming complete") self.model_call_details["complete_streaming_response"] = complete_streaming_response self.model_call_details["response_cost"] = self._response_cost_calculator( result=complete_streaming_response ) self._merge_hidden_params_from_response_into_metadata(complete_streaming_response) ## STANDARDIZED LOGGING PAYLOAD self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload( complete_streaming_response, start_time, end_time ) standard_logging_payload: Final[StandardLoggingPayload | None] = self.model_call_details.get( "standard_logging_object" ) if standard_logging_payload is not None: # Only emit for sync requests (async_success_handler handles async) if is_sync_request: emit_standard_logging_payload(standard_logging_payload) callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_success_callbacks, global_callbacks=litellm.success_callback, ) ## REDACT MESSAGES ## result = redact_message_input_output_from_logging( model_call_details=(self.model_call_details if hasattr(self, "model_call_details") else {}), result=result, ) ## LOGGING HOOK ## for callback in callbacks: if isinstance(callback, CustomGuardrail): from litellm.types.guardrails import GuardrailEventHooks if ( callback.should_run_guardrail( data=self.model_call_details, event_type=GuardrailEventHooks.logging_only, ) is not True ): continue self.model_call_details, result = callback.logging_hook( kwargs=self.model_call_details, result=result, call_type=self.call_type, ) elif isinstance(callback, CustomLogger): self.model_call_details, result = callback.logging_hook( kwargs=self.model_call_details, result=result, call_type=self.call_type, ) self.has_run_logging(event_type="sync_success") for callback in callbacks: try: should_run = self.should_run_callback( callback=callback, litellm_params=litellm_params, event_hook="success_handler", ) if not should_run: continue if callback == "promptlayer" and promptLayerLogger is not None: print_verbose("reaches promptlayer for logging!") promptLayerLogger.log_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, ) if callback == "supabase" and supabaseClient is not None: print_verbose("reaches supabase for logging!") kwargs = self.model_call_details # this only logs streaming once, complete_streaming_response exists i.e when stream ends if self.stream: if "complete_streaming_response" not in kwargs: continue else: print_verbose("reaches supabase for streaming logging!") result = kwargs["complete_streaming_response"] model = kwargs["model"] messages = kwargs["messages"] optional_params = kwargs.get("optional_params", {}) litellm_params = kwargs.get("litellm_params", {}) supabaseClient.log_event( model=model, messages=messages, end_user=optional_params.get("user", "default"), response_obj=result, start_time=start_time, end_time=end_time, litellm_call_id=( current_call_id if (current_call_id := litellm_params.get("litellm_call_id")) is not None else str(uuid.uuid4()) ), print_verbose=print_verbose, ) if callback == "wandb" and weightsBiasesLogger is not None: print_verbose("reaches wandb for logging!") weightsBiasesLogger.log_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, ) if callback == "logfire" and logfireLogger is not None: verbose_logger.debug("reaches logfire for success logging!") kwargs = {} for k, v in self.model_call_details.items(): if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine kwargs[k] = v # this only logs streaming once, complete_streaming_response exists i.e when stream ends if self.stream: if "complete_streaming_response" not in kwargs: continue else: print_verbose("reaches logfire for streaming logging!") result = kwargs["complete_streaming_response"] logfireLogger.log_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, level=LogfireLevel.INFO.value, ) if callback == "lunary" and lunaryLogger is not None: print_verbose("reaches lunary for logging!") model = self.model kwargs = self.model_call_details input = kwargs.get("messages", kwargs.get("input", None)) type = "embed" if self.call_type == CallTypes.embedding.value else "llm" # this only logs streaming once, complete_streaming_response exists i.e when stream ends if self.stream: if "complete_streaming_response" not in kwargs: continue else: result = kwargs["complete_streaming_response"] lunaryLogger.log_event( type=type, kwargs=kwargs, event="end", model=model, input=input, user_id=kwargs.get("user", None), # user_props=self.model_call_details.get("user_props", None), extra=kwargs.get("optional_params", {}), response_obj=result, start_time=start_time, end_time=end_time, run_id=self.litellm_call_id, print_verbose=print_verbose, ) if callback == "helicone" and heliconeLogger is not None: print_verbose("reaches helicone for logging!") model = self.model messages = self.model_call_details["input"] kwargs = self.model_call_details # this only logs streaming once, complete_streaming_response exists i.e when stream ends if self.stream: if "complete_streaming_response" not in kwargs: continue else: print_verbose("reaches helicone for streaming logging!") result = kwargs["complete_streaming_response"] heliconeLogger.log_success( model=model, messages=messages, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, kwargs=kwargs, ) if callback == "langfuse": global langFuseLogger print_verbose("reaches langfuse for success logging!") kwargs = {} for k, v in self.model_call_details.items(): if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine kwargs[k] = v # this only logs streaming once, complete_streaming_response exists i.e when stream ends if self.stream: verbose_logger.debug( "is complete_streaming_response in kwargs: %s", kwargs.get("complete_streaming_response", None), ) if complete_streaming_response is None: continue else: print_verbose("reaches langfuse for streaming logging!") result = kwargs["complete_streaming_response"] langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request( globalLangfuseLogger=langFuseLogger, standard_callback_dynamic_params=self.standard_callback_dynamic_params, in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache, ) if langfuse_logger_to_use is not None: _response = langfuse_logger_to_use.log_event_on_langfuse( kwargs=kwargs, response_obj=result, start_time=start_time, end_time=end_time, user_id=kwargs.get("user", None), ) if _response is not None and isinstance(_response, dict): _trace_id = _response.get("trace_id", None) if _trace_id is not None: in_memory_trace_id_cache.set_cache( litellm_call_id=self.litellm_call_id, service_name="langfuse", trace_id=_trace_id, ) if callback == "greenscale" and greenscaleLogger is not None: kwargs = {} for k, v in self.model_call_details.items(): if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine kwargs[k] = v # this only logs streaming once, complete_streaming_response exists i.e when stream ends if self.stream: verbose_logger.debug( "is complete_streaming_response in kwargs: %s", kwargs.get("complete_streaming_response", None), ) if complete_streaming_response is None: continue else: print_verbose("reaches greenscale for streaming logging!") result = kwargs["complete_streaming_response"] greenscaleLogger.log_event( kwargs=kwargs, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, ) if callback == "athina" and athinaLogger is not None: deep_copy = {} for k, v in self.model_call_details.items(): deep_copy[k] = v athinaLogger.log_event( kwargs=deep_copy, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, ) if callback == "traceloop": deep_copy = {} for k, v in self.model_call_details.items(): if k != "original_response": deep_copy[k] = v traceloopLogger.log_event( kwargs=deep_copy, response_obj=result, start_time=start_time, end_time=end_time, user_id=kwargs.get("user", None), print_verbose=print_verbose, ) if callback == "s3": global s3Logger if s3Logger is None: s3Logger = S3Logger() if self.stream: if "complete_streaming_response" in self.model_call_details: print_verbose("S3Logger Logger: Got Stream Event - Completed Stream Response") s3Logger.log_event( kwargs=self.model_call_details, response_obj=self.model_call_details["complete_streaming_response"], start_time=start_time, end_time=end_time, print_verbose=print_verbose, ) else: print_verbose("S3Logger Logger: Got Stream Event - No complete stream response as yet") else: s3Logger.log_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, ) if callback == "openmeter" and is_sync_request: global openMeterLogger if openMeterLogger is None: print_verbose("Instantiates openmeter client") openMeterLogger = OpenMeterLogger() if self.stream and complete_streaming_response is None: openMeterLogger.log_stream_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, ) else: if self.stream and complete_streaming_response: self.model_call_details["complete_response"] = self.model_call_details.get( "complete_streaming_response", {} ) result = self.model_call_details["complete_response"] openMeterLogger.log_success_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, ) if ( isinstance(callback, CustomLogger) and is_sync_request and self.call_type != CallTypes.pass_through.value # pass-through endpoints call async_log_success_event ): # custom logger class if self.stream and complete_streaming_response is None: callback.log_stream_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, ) else: if self.stream and complete_streaming_response: self.model_call_details["complete_response"] = self.model_call_details.get( "complete_streaming_response", {} ) result = self.model_call_details["complete_response"] callback.log_success_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, ) if ( callable(callback) is True and is_sync_request and customLogger is not None ): # custom logger functions print_verbose(f"success callbacks: Running Custom Callback Function - {callback}") customLogger.log_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, callback_func=callback, ) except Exception as e: print_verbose( f"LiteLLM.LoggingError: [Non-Blocking] Exception occurred while success logging with integrations {traceback.format_exc()}" ) print_verbose(f"LiteLLM.Logging: is sentry capture exception initialized {capture_exception}") if capture_exception: # log this error to sentry for debugging capture_exception(e) # Track callback logging failures in Prometheus try: self._handle_callback_failure(callback=callback) except Exception: pass except Exception as e: verbose_logger.exception( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while success logging %s", e, ) async def async_success_handler( self, result: Any = None, # heterogeneous response object; varies by call type (ANN401 ignored, see ruff-strict.toml) start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, cache_hit: bool | None = None, **kwargs: Any, # kwargs-ok: forwarded to _async_success_handler_body ) -> None: """Restores trace_id/session_id contextvars once this attempt's own success logging (including any nested calls its callbacks trigger) is fully done.""" try: return await self._async_success_handler_body( result=result, start_time=start_time, end_time=end_time, cache_hit=cache_hit, **kwargs ) finally: self._restore_correlation_context() async def _async_success_handler_body( self, result: Any = None, # heterogeneous response object; varies by call type (ANN401 ignored, see ruff-strict.toml) start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, cache_hit: bool | None = None, **kwargs: Any, # kwargs-ok: forwarded from async_success_handler ) -> None: """ Implementing async callbacks, to handle asyncio event loop issues when custom integrations need to use async functions. """ print_verbose(f"Logging Details LiteLLM-Async Success Call, cache_hit={cache_hit}") if not self._is_assembled_stream_success(result) and not self.should_run_logging( event_type="async_success" ): # prevent double logging (non-streaming) return ## CALCULATE COST FOR BATCH JOBS if self.call_type == CallTypes.aretrieve_batch.value and isinstance(result, LiteLLMBatch): litellm_params = self.litellm_params or {} litellm_metadata: Final = litellm_params.get("litellm_metadata") or {} if ( litellm_metadata.get("batch_ignore_default_logging", False) is True ): # polling job will query these frequently, don't spam db logs return from litellm.proxy.openai_files_endpoints.common_utils import ( _is_base64_encoded_unified_file_id, ) # check if file id is a unified file id is_base64_unified_file_id: Final = _is_base64_encoded_unified_file_id(result.id) batch_cost: Final = kwargs.get("batch_cost", None) batch_usage = kwargs.get("batch_usage", None) batch_models = kwargs.get("batch_models", None) batch_successful_requests: Final = kwargs.get("batch_successful_requests", None) batch_failed_requests: Final = kwargs.get("batch_failed_requests", None) has_explicit_batch_data: Final = all(x is not None for x in (batch_cost, batch_usage, batch_models)) should_compute_batch_data: Final = ( not is_base64_unified_file_id or not has_explicit_batch_data and result.status == "completed" ) if has_explicit_batch_data: result._hidden_params["response_cost"] = batch_cost result._hidden_params["batch_models"] = batch_models result._hidden_params["batch_successful_requests"] = batch_successful_requests # pyright: ignore[reportPrivateUsage] # rebind-ok: same result._hidden_params pattern as response_cost/batch_models above result._hidden_params["batch_failed_requests"] = batch_failed_requests # pyright: ignore[reportPrivateUsage] # rebind-ok: same pattern as above result.usage = batch_usage elif should_compute_batch_data: batch_result: Final = await _handle_completed_batch( batch=result, custom_llm_provider=self.custom_llm_provider, model_name=self.get_deployment_model_for_cost(), litellm_params=self.litellm_params, model_info=self.get_router_deployment_model_info(), ) result._hidden_params["response_cost"] = batch_result.cost result._hidden_params["batch_models"] = batch_result.models result._hidden_params["batch_successful_requests"] = batch_result.successful_requests # pyright: ignore[reportPrivateUsage] # rebind-ok: same pattern as above result._hidden_params["batch_failed_requests"] = batch_result.failed_requests # pyright: ignore[reportPrivateUsage] # rebind-ok: same pattern as above result.usage = batch_result.usage start_time, end_time, result = self._success_handler_helper_fn( start_time=start_time, end_time=end_time, result=result, cache_hit=cache_hit, standard_logging_object=kwargs.get("standard_logging_object", None), ) ## BUILD COMPLETE STREAMED RESPONSE if "async_complete_streaming_response" in self.model_call_details: return # break out of this. complete_streaming_response: Final[ ModelResponse | TextCompletionResponse | ResponsesAPIResponse | InteractionsAPIResponse | None ] = self._get_assembled_streaming_response( result=result, start_time=start_time, end_time=end_time, is_async=True, streaming_chunks=self.streaming_chunks, ) if complete_streaming_response is not None: print_verbose("Async success callbacks: Got a complete streaming response") self.model_call_details["async_complete_streaming_response"] = complete_streaming_response try: if self.model_call_details.get("cache_hit", False) is True: self.model_call_details["response_cost"] = 0.0 else: # check if base_model set on azure _get_base_model_from_metadata(model_call_details=self.model_call_details) # base_model defaults to None if not set on model_info self.model_call_details["response_cost"] = self._response_cost_calculator( result=complete_streaming_response ) verbose_logger.debug("Model=%s; cost=%s", self.model, self.model_call_details["response_cost"]) except litellm.NotFoundError: verbose_logger.warning( "Model=%s not found in completion cost map. Setting 'response_cost' to None", self.model ) self.model_call_details["response_cost"] = None self._merge_hidden_params_from_response_into_metadata(complete_streaming_response) ## STANDARDIZED LOGGING PAYLOAD self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload( complete_streaming_response, start_time, end_time ) # print standard logging payload if (standard_logging_payload := self.model_call_details.get("standard_logging_object")) is not None: emit_standard_logging_payload(standard_logging_payload) elif self.call_type == "pass_through_endpoint": print_verbose("Async success callbacks: Got a pass-through endpoint response") self.model_call_details["async_complete_streaming_response"] = result # Only set response_cost to None if not already calculated by # pass-through handlers (e.g. Gemini/Vertex handlers already # compute cost via completion_cost) if self.model_call_details.get("response_cost") is None: self.model_call_details["response_cost"] = None # Only build standard_logging_object if not already built by # _success_handler_helper_fn if self.model_call_details.get("standard_logging_object") is None: ## STANDARDIZED LOGGING PAYLOAD self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload( result, start_time, end_time ) # print standard logging payload if (standard_logging_payload := self.model_call_details.get("standard_logging_object")) is not None: emit_standard_logging_payload(standard_logging_payload) callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_async_success_callbacks, global_callbacks=litellm._async_success_callback, ) result = redact_message_input_output_from_logging( model_call_details=(self.model_call_details if hasattr(self, "model_call_details") else {}), result=result, ) ## LOGGING HOOK ## for callback in callbacks: if isinstance(callback, CustomGuardrail): from litellm.types.guardrails import GuardrailEventHooks if ( callback.should_run_guardrail( data=self.model_call_details, event_type=GuardrailEventHooks.logging_only, ) is not True ): continue self.model_call_details, result = await callback.async_logging_hook( kwargs=self.model_call_details, result=result, call_type=self.call_type, ) elif isinstance(callback, CustomLogger): result = redact_message_input_output_from_custom_logger( result=result, litellm_logging_obj=self, custom_logger=callback ) self.model_call_details, result = await callback.async_logging_hook( kwargs=self.model_call_details, result=result, call_type=self.call_type, ) self.has_run_logging(event_type="async_success") for callback in callbacks: # check if callback can run for this request litellm_params = self.model_call_details.get("litellm_params", {}) should_run = self.should_run_callback( callback=callback, litellm_params=litellm_params, event_hook="async_success_handler", ) if not should_run: continue try: if callback == "openmeter" and openMeterLogger is not None: if self.stream is True: if "async_complete_streaming_response" in self.model_call_details: await openMeterLogger.async_log_success_event( kwargs=self.model_call_details, response_obj=self.model_call_details["async_complete_streaming_response"], start_time=start_time, end_time=end_time, ) else: await openMeterLogger.async_log_stream_event( # [TODO]: move this to being an async log stream event function kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, ) else: await openMeterLogger.async_log_success_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, ) if isinstance(callback, CustomLogger): # custom logger class model_call_details: dict = self.model_call_details ################################## # call redaction hook for custom logger model_call_details = callback.redact_standard_logging_payload_from_model_call_details( model_call_details=model_call_details ) model_call_details = redact_streaming_responses_for_custom_logger( model_call_details=model_call_details, custom_logger=callback ) ################################## if self.stream is True: if "async_complete_streaming_response" in model_call_details: await callback.async_log_success_event( kwargs=model_call_details, response_obj=model_call_details["async_complete_streaming_response"], start_time=start_time, end_time=end_time, ) else: await callback.async_log_stream_event( # [TODO]: move this to being an async log stream event function kwargs=model_call_details, response_obj=result, start_time=start_time, end_time=end_time, ) else: await callback.async_log_success_event( kwargs=model_call_details, response_obj=result, start_time=start_time, end_time=end_time, ) if callable(callback): # custom logger functions global customLogger if customLogger is None: customLogger = CustomLogger() if self.stream: if "async_complete_streaming_response" in self.model_call_details: await customLogger.async_log_event( kwargs=self.model_call_details, response_obj=self.model_call_details["async_complete_streaming_response"], start_time=start_time, end_time=end_time, print_verbose=print_verbose, callback_func=callback, ) else: await customLogger.async_log_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, callback_func=callback, ) if callback == "dynamodb": global dynamoLogger if dynamoLogger is None: dynamoLogger = DyanmoDBLogger() if self.stream: if "async_complete_streaming_response" in self.model_call_details: print_verbose("DynamoDB Logger: Got Stream Event - Completed Stream Response") await dynamoLogger._async_log_event( kwargs=self.model_call_details, response_obj=self.model_call_details["async_complete_streaming_response"], start_time=start_time, end_time=end_time, print_verbose=print_verbose, ) else: print_verbose("DynamoDB Logger: Got Stream Event - No complete stream response as yet") else: await dynamoLogger._async_log_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, ) except Exception: verbose_logger.error( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while success logging %s", traceback.format_exc(), ) self._handle_callback_failure(callback=callback) def _handle_callback_failure(self, callback: Any): """ Handle callback logging failures by incrementing Prometheus metrics. Works for both sync and async contexts since Prometheus counter increment is synchronous. Args: callback: The callback that failed """ try: callback_name: Final = self._get_callback_name(callback) all_callbacks: Final = litellm.logging_callback_manager._get_all_callbacks() for callback_obj in all_callbacks: if hasattr(callback_obj, "increment_callback_logging_failure"): callback_obj.increment_callback_logging_failure(callback_name=callback_name) break # Only increment once except Exception as e: verbose_logger.debug("Error in _handle_callback_failure: %s", e) def _failure_handler_helper_fn(self, exception, traceback_exception, start_time=None, end_time=None): if start_time is None: start_time = self.start_time if end_time is None: end_time = datetime.datetime.now() # on some exceptions, model_call_details is not always initialized, this ensures that we still log those exceptions if not hasattr(self, "model_call_details"): self.model_call_details = {} if ( self.model_call_details.get("log_event_type") == "failed_api_call" and self.model_call_details.get("exception") is exception and self.model_call_details.get("standard_logging_object") is not None ): return start_time, self.model_call_details["end_time"] self.model_call_details["log_event_type"] = "failed_api_call" self.model_call_details["exception"] = exception self.model_call_details["traceback_exception"] = ( _redact_string(traceback_exception) if isinstance(traceback_exception, str) else traceback_exception ) self.model_call_details["end_time"] = end_time self.model_call_details.setdefault("original_response", None) # A stream interrupted mid-flight still billed the provider for the # chunks already delivered; the router stashes that recovered usage as # ``combined_usage_object`` and pre-computes its cost, so preserve it # here instead of zeroing the spend on an otherwise-failed request. if self.model_call_details.get("combined_usage_object") is None: self.model_call_details["response_cost"] = 0 if hasattr(exception, "headers") and isinstance(exception.headers, dict): self.model_call_details.setdefault("litellm_params", {}) metadata: Final = self.model_call_details["litellm_params"].get("metadata", {}) or {} metadata.update(exception.headers) ## STANDARDIZED LOGGING PAYLOAD self.model_call_details["standard_logging_object"] = get_standard_logging_object_payload( kwargs=self.model_call_details, init_response_obj={}, start_time=start_time, end_time=end_time, logging_obj=self, status="failure", error_str=_redact_string(str(exception)), original_exception=exception, standard_built_in_tools_params=self.standard_built_in_tools_params, ) return start_time, end_time async def special_failure_handlers(self, exception: Exception): """ Custom events, emitted for specific failures. Currently just for router model group rate limit error """ from litellm.types.router import RouterErrors litellm_params: Final[dict] = self.model_call_details.get("litellm_params") or {} metadata: Final = litellm_params.get("metadata") or {} ## BASE CASE ## check if rate limit error for model group size 1 is_base_case = False if metadata.get("model_group_size") is not None: model_group_size: Final = metadata.get("model_group_size") if isinstance(model_group_size, int) and model_group_size == 1: is_base_case = True ## check if special error ## if RouterErrors.no_deployments_available.value not in str(exception) and is_base_case is False: return ## get original model group ## model_group: Final = metadata.get("model_group") or None for callback in litellm._async_failure_callback: if isinstance(callback, CustomLogger): # custom logger class await callback.log_model_group_rate_limit_error( exception=exception, original_model_group=model_group, kwargs=self.model_call_details, ) def failure_handler( self, exception: Exception, traceback_exception: str, start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, ) -> None: """Restores trace_id/session_id contextvars once this attempt's own failure logging (including any nested calls its callbacks trigger) is fully done.""" try: return self._failure_handler_body( exception=exception, traceback_exception=traceback_exception, start_time=start_time, end_time=end_time, ) finally: self._restore_correlation_context() def _failure_handler_body( self, exception: Exception, traceback_exception: str, start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, ) -> None: verbose_logger.debug("Logging Details LiteLLM-Failure Call: %s", litellm.failure_callback) if not self.should_run_logging(event_type="sync_failure"): # prevent double logging return litellm_params: Final = self.model_call_details.get("litellm_params", {}) is_sync_request: Final = self._is_sync_litellm_request(litellm_params) try: start_time, end_time = self._failure_handler_helper_fn( exception=exception, traceback_exception=traceback_exception, start_time=start_time, end_time=end_time, ) callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_failure_callbacks, global_callbacks=litellm.failure_callback, ) result: object = None # result sent to all loggers, init this to None incase it's not created result = redact_message_input_output_from_logging( model_call_details=(self.model_call_details if hasattr(self, "model_call_details") else {}), result=result, ) self.has_run_logging(event_type="sync_failure") for callback in callbacks: try: should_run = self.should_run_callback( callback=callback, litellm_params=litellm_params, event_hook="failure_handler", ) if not should_run: continue if callback == "lunary" and lunaryLogger is not None: print_verbose("reaches lunary for logging error!") model = self.model input = self.model_call_details["input"] _type = "embed" if self.call_type == CallTypes.embedding.value else "llm" lunaryLogger.log_event( kwargs=self.model_call_details, type=_type, event="error", user_id=self.model_call_details.get("user", "default"), model=model, input=input, error=traceback_exception, run_id=self.litellm_call_id, start_time=start_time, end_time=end_time, print_verbose=print_verbose, ) if callback == "sentry": print_verbose("sending exception to sentry") if capture_exception: capture_exception(exception) else: print_verbose(f"capture exception not initialized: {capture_exception}") elif callback == "supabase" and supabaseClient is not None: print_verbose("reaches supabase for logging!") print_verbose(f"supabaseClient: {supabaseClient}") supabaseClient.log_event( model=self.model if hasattr(self, "model") else "", messages=self.messages, end_user=self.model_call_details.get("user", "default"), response_obj=result, start_time=start_time, end_time=end_time, litellm_call_id=self.model_call_details["litellm_call_id"], print_verbose=print_verbose, ) if callable(callback): # custom logger functions global customLogger if customLogger is None: customLogger = CustomLogger() customLogger.log_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, callback_func=callback, ) if ( isinstance(callback, CustomLogger) and is_sync_request and self.call_type != CallTypes.pass_through.value ): # custom logger class callback.log_failure_event( start_time=start_time, end_time=end_time, response_obj=result, kwargs=self.model_call_details, ) if callback == "langfuse": global langFuseLogger verbose_logger.debug("reaches langfuse for logging failure") kwargs = {} for k, v in self.model_call_details.items(): if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine kwargs[k] = v # this only logs streaming once, complete_streaming_response exists i.e when stream ends langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request( globalLangfuseLogger=langFuseLogger, standard_callback_dynamic_params=self.standard_callback_dynamic_params, in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache, ) _response = langfuse_logger_to_use.log_event_on_langfuse( start_time=start_time, end_time=end_time, response_obj=None, user_id=kwargs.get("user", None), status_message=str(exception), level="ERROR", kwargs=kwargs, ) if _response is not None and isinstance(_response, dict): _trace_id = _response.get("trace_id", None) if _trace_id is not None: in_memory_trace_id_cache.set_cache( litellm_call_id=self.litellm_call_id, service_name="langfuse", trace_id=_trace_id, ) if callback == "traceloop": traceloopLogger.log_event( start_time=start_time, end_time=end_time, response_obj=None, user_id=self.model_call_details.get("user", None), print_verbose=print_verbose, status_message=str(exception), level="ERROR", kwargs=self.model_call_details, ) if callback == "logfire" and logfireLogger is not None: verbose_logger.debug("reaches logfire for failure logging!") kwargs = {} for k, v in self.model_call_details.items(): if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine kwargs[k] = v kwargs["exception"] = exception logfireLogger.log_event( kwargs=kwargs, response_obj=result, start_time=start_time, end_time=end_time, level=LogfireLevel.ERROR.value, print_verbose=print_verbose, ) except Exception as e: print_verbose( f"LiteLLM.LoggingError: [Non-Blocking] Exception occurred while failure logging with integrations {e}" ) print_verbose(f"LiteLLM.Logging: is sentry capture exception initialized {capture_exception}") if capture_exception: # log this error to sentry for debugging capture_exception(e) except Exception as e: verbose_logger.exception( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while failure logging %s", e ) async def async_failure_handler( self, exception: Exception, traceback_exception: str, start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, ) -> None: """Restores trace_id/session_id contextvars once this attempt's own failure logging (including any nested calls its callbacks trigger) is fully done.""" try: return await self._async_failure_handler_body( exception=exception, traceback_exception=traceback_exception, start_time=start_time, end_time=end_time, ) finally: self._restore_correlation_context() async def _async_failure_handler_body( self, exception: Exception, traceback_exception: str, start_time: datetime.datetime | None = None, end_time: datetime.datetime | None = None, ) -> None: """ Implementing async callbacks, to handle asyncio event loop issues when custom integrations need to use async functions. """ await self.special_failure_handlers(exception=exception) if not self.should_run_logging(event_type="async_failure"): # prevent double logging return start_time, end_time = self._failure_handler_helper_fn( exception=exception, traceback_exception=traceback_exception, start_time=start_time, end_time=end_time, ) callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_async_failure_callbacks, global_callbacks=litellm._async_failure_callback, ) result: Final = None # result sent to all loggers, init this to None incase it's not created self.has_run_logging(event_type="async_failure") for callback in callbacks: try: litellm_params = self.model_call_details.get("litellm_params", {}) should_run = self.should_run_callback( callback=callback, litellm_params=litellm_params, event_hook="async_failure_handler", ) if not should_run: continue if isinstance(callback, CustomLogger): # custom logger class await callback.async_log_failure_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, ) if callable(callback): # custom logger functions global customLogger if customLogger is None: customLogger = CustomLogger() await customLogger.async_log_event( kwargs=self.model_call_details, response_obj=result, start_time=start_time, end_time=end_time, print_verbose=print_verbose, callback_func=callback, ) except Exception as e: verbose_logger.exception( "LiteLLM.LoggingError: [Non-Blocking] Exception occurred while failure logging %s\nCallback=%s", e, callback, ) # Track callback logging failures in Prometheus self._handle_callback_failure(callback=callback) def _get_trace_id(self, service_name: Literal["langfuse"]) -> str | None: """ For the given service (e.g. langfuse), return the trace_id actually logged. Used for constructing the url in slack alerting. Returns: - str: The logged trace id - None: If trace id not yet emitted. """ trace_id: str | None = None if service_name == "langfuse": trace_id = in_memory_trace_id_cache.get_cache( litellm_call_id=self.litellm_call_id, service_name=service_name ) return trace_id def _get_callback_object(self, service_name: Literal["langfuse"]) -> Any | None: """ Return dynamic callback object. Meant to solve issue when doing key-based/team-based logging """ global langFuseLogger if service_name == "langfuse": if langFuseLogger is None or ( ( self.standard_callback_dynamic_params.get("langfuse_public_key") is not None and self.standard_callback_dynamic_params.get("langfuse_public_key") != langFuseLogger.public_key ) or ( self.standard_callback_dynamic_params.get("langfuse_public_key") is not None and self.standard_callback_dynamic_params.get("langfuse_public_key") != langFuseLogger.public_key ) or ( self.standard_callback_dynamic_params.get("langfuse_host") is not None and self.standard_callback_dynamic_params.get("langfuse_host") != langFuseLogger.langfuse_host ) ): return LangFuseLogger( langfuse_public_key=self.standard_callback_dynamic_params.get("langfuse_public_key"), langfuse_secret=self.standard_callback_dynamic_params.get("langfuse_secret") or self.standard_callback_dynamic_params.get("langfuse_secret_key"), langfuse_host=self.standard_callback_dynamic_params.get("langfuse_host"), allow_env_credentials=self.standard_callback_dynamic_params.get("langfuse_host") is None, ) return langFuseLogger return None def handle_sync_success_callbacks_for_async_calls( self, result: Any, start_time: datetime.datetime, end_time: datetime.datetime, cache_hit: Any | None = None, ) -> None: """ Handles calling success callbacks for Async calls. Why: Some callbacks - `langfuse`, `s3` are sync callbacks. We need to call them in the executor. """ if self._should_run_sync_callbacks_for_async_calls() is False: return executor.submit( self.success_handler, result, start_time, end_time, cache_hit, ) def _should_run_sync_callbacks_for_async_calls(self) -> bool: """ Returns: - bool: True if sync callbacks should be run for async calls. eg. `langfuse`, `s3` """ _combined_sync_callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_success_callbacks, global_callbacks=litellm.success_callback, ) _filtered_success_callbacks = self._remove_internal_custom_logger_callbacks(_combined_sync_callbacks) _filtered_success_callbacks = self._remove_internal_litellm_callbacks(_filtered_success_callbacks) return len(_filtered_success_callbacks) > 0 def _should_run_sync_failure_callbacks_for_async_calls(self) -> bool: """ Returns: - bool: True if sync failure callbacks should be run for async calls. eg. `langfuse`, `s3` Mirrors ``_should_run_sync_callbacks_for_async_calls`` but reads the failure callback lists. Gating the legacy sync ``failure_handler`` on the success lists would drop sync failure callbacks for any caller that configures only failure callbacks, so streaming errors would be logged nowhere. """ _combined_sync_callbacks: Final = self.get_combined_callback_list( dynamic_success_callbacks=self.dynamic_failure_callbacks, global_callbacks=litellm.failure_callback, ) _filtered_failure_callbacks = self._remove_internal_custom_logger_callbacks(_combined_sync_callbacks) _filtered_failure_callbacks = self._remove_internal_litellm_callbacks(_filtered_failure_callbacks) return len(_filtered_failure_callbacks) > 0 def get_combined_callback_list(self, dynamic_success_callbacks: list | None, global_callbacks: list) -> list: if dynamic_success_callbacks is None: return list(global_callbacks) return list(dict.fromkeys(dynamic_success_callbacks + global_callbacks)) def _remove_internal_litellm_callbacks(self, callbacks: list) -> list: """ Creates a filtered list of callbacks, excluding internal LiteLLM callbacks. Args: callbacks: List of callback functions/strings to filter Returns: List of filtered callbacks with internal ones removed """ filtered: Final = [cb for cb in callbacks if not self._is_internal_litellm_proxy_callback(cb)] verbose_logger.debug("Filtered callbacks: %s", filtered) return filtered def _get_callback_name(self, cb) -> str: """ Helper to get the name of a callback function Args: cb: The callback object/function/string to get the name of Returns: The name of the callback """ if isinstance(cb, str): return cb if hasattr(cb, "__name__"): return cb.__name__ if hasattr(cb, "__func__"): return cb.__func__.__name__ if hasattr(cb, "__class__"): return cb.__class__.__name__ return str(cb) def _is_internal_litellm_proxy_callback(self, cb) -> bool: """Helper to check if a callback is internal""" INTERNAL_PREFIXES: Final = [ "_PROXY", "_service_logger.ServiceLogging", "sync_deployment_callback_on_success", ] if isinstance(cb, str): return False if not callable(cb): return True cb_name: Final = self._get_callback_name(cb) return any(prefix in cb_name for prefix in INTERNAL_PREFIXES) def _remove_internal_custom_logger_callbacks(self, callbacks: list) -> list: """ Removes internal custom logger callbacks from the list. """ _new_callbacks: Final = [] for _c in callbacks: if ( isinstance(_c, CustomLogger) or isinstance(_c, str) and _c in litellm._known_custom_logger_compatible_callbacks ): continue _new_callbacks.append(_c) return _new_callbacks def _get_assembled_streaming_response( self, result: ModelResponse | TextCompletionResponse | ModelResponseStream | ResponseCompletedEvent | object, start_time: datetime.datetime, end_time: datetime.datetime, is_async: bool, streaming_chunks: list[object], ) -> ModelResponse | TextCompletionResponse | ResponsesAPIResponse | InteractionsAPIResponse | None: if self.stream is not True: return None if isinstance(result, ModelResponse) or isinstance(result, TextCompletionResponse): return result elif isinstance( result, (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent), ): ## return unified Usage object if isinstance(result.response.usage, ResponseAPIUsage): transformed_usage: Final = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage( result.response.usage ) # Set as dict instead of Usage object so model_dump() serializes it correctly setattr( result.response, "usage", ( transformed_usage.model_dump() if hasattr(transformed_usage, "model_dump") else dict(transformed_usage) ), ) return result.response elif isinstance(result, InteractionsAPIStreamingResponse): return self._assemble_completed_interaction_response(result) else: return None @staticmethod def _assemble_completed_interaction_response( result: InteractionsAPIStreamingResponse, ) -> InteractionsAPIResponse | None: """ The Interactions API streaming iterator hands the terminal event to the success handlers: the new schema (Api-Revision: 2026-05-20) emits ``interaction.completed`` carrying the full interaction object, the legacy schema (2026-05-07) emits a chunk with ``status="completed"`` and usage on the chunk itself. Build the equivalent non-streaming response so cost calculation and spend tracking see one shape. """ if result.event_type == "interaction.completed" and result.interaction is not None: return InteractionsAPIResponse(**result.interaction) if result.status == "completed": return InteractionsAPIResponse( **result.model_dump( exclude={ # mutable-ok: pydantic types exclude as set[str], which a frozenset does not satisfy "event_type", "delta", "index", "step", "interaction_id", "interaction", } ) ) return None def _handle_anthropic_messages_response_logging(self, result: Any) -> ModelResponse: """ Handles logging for Anthropic messages responses. Args: result: The response object from the model call Returns: The the response object from the model call - For Non-streaming responses, we need to transform the response to a ModelResponse object. - For streaming responses, anthropic_messages handler calls success_handler with a assembled ModelResponse. """ import httpx if self.stream and isinstance(result, ModelResponse) or isinstance(result, ModelResponse): return result if isinstance( result, (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent), ): result = result.response if isinstance(result, ResponsesAPIResponse): return self._translate_responses_api_response_to_model_response(result) httpx_response: Final = self.model_call_details.get("httpx_response", None) if httpx_response and isinstance(httpx_response, httpx.Response): result = litellm.AnthropicConfig().transform_response( raw_response=httpx_response, model_response=litellm.ModelResponse(), model=self.model, messages=[], logging_obj=self, optional_params=self.optional_params or {}, api_key="", request_data={}, encoding=litellm.encoding, json_mode=False, litellm_params={}, ) else: from litellm.types.llms.anthropic import AnthropicResponse pydantic_result: Final = AnthropicResponse.model_validate(result) import httpx result = litellm.AnthropicConfig().transform_parsed_response( completion_response=pydantic_result.model_dump(), raw_response=httpx.Response( status_code=200, headers={}, ), model_response=litellm.ModelResponse(), json_mode=None, speed=self.optional_params.get("speed") if self.optional_params else None, ) return result def _translate_responses_api_response_to_model_response(self, result: ResponsesAPIResponse) -> ModelResponse: """ Convert a Responses API response into a ModelResponse for spend_logs. The proxy UI parses spend_log rows expecting chat-completion shape (response.choices[0].message); a raw ResponsesAPIResponse dump (output[...]) would render as empty in the Logs tab. Translation also yields full choices/message detail downstream consumers can rely on. """ from litellm.completion_extras.litellm_responses_transformation.transformation import ( LiteLLMResponsesTransformationHandler, ) try: return LiteLLMResponsesTransformationHandler().transform_response( model=self.model, raw_response=result, model_response=litellm.ModelResponse(), logging_obj=self, request_data={}, messages=[], optional_params={}, litellm_params={}, encoding=litellm.encoding, ) except Exception as e: verbose_logger.debug( "Responses API -> ModelResponse translation failed for " "anthropic_messages logging (%s); falling back to minimal " "usage-only ModelResponse to keep the spend_logs row.", str(e), ) model_response: Final = litellm.ModelResponse() model_response.model = self.model usage: Final = getattr(result, "usage", None) if usage is not None and ResponseAPILoggingUtils._is_response_api_usage(usage): setattr( model_response, "usage", ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(usage), ) return model_response def _handle_non_streaming_google_genai_generate_content_response_logging(self, result: Any) -> ModelResponse: """ Handles logging for Google GenAI generate content responses. """ import httpx httpx_response: Final = self.model_call_details.get("httpx_response", None) if httpx_response is None: raise ValueError("Google GenAI Generate Content: httpx_response is None") dict_result: Final = httpx_response.json() result = litellm.VertexGeminiConfig()._transform_google_generate_content_to_openai_model_response( completion_response=dict_result, model_response=litellm.ModelResponse(), model=self.model, logging_obj=self, raw_response=httpx.Response( status_code=200, headers={}, ), ) return result def _handle_a2a_response_logging(self, result: Any) -> Any: """ Handles logging for A2A (Agent-to-Agent) responses. Adds usage from model_call_details to the result if available. Uses Pydantic's model_copy to avoid modifying the original response. Args: result: The LiteLLMSendMessageResponse from the A2A call Returns: The response object with usage added if available """ # Get usage from model_call_details (set by asend_message) usage: Final = self.model_call_details.get("usage") if usage is None: return result # Deep copy result and add usage result_copy: Final = result.model_copy(deep=True) result_copy.usage = usage.model_dump() if hasattr(usage, "model_dump") else dict(usage) return result_copy def _get_masked_values( sensitive_object: dict, ignore_sensitive_values: bool = False, mask_all_values: bool = False, unmasked_length: int = 4, number_of_asterisks: int | None = 4, _depth: int = 0, _max_depth: int = 20, ) -> dict: """ Internal debugging helper function Masks the headers of the request sent from LiteLLM Args: masked_length: Optional length for the masked portion (number of *). If set, will use exactly this many * regardless of original string length. The total length will be unmasked_length + masked_length. """ sensitive_keywords: Final = [ "authorization", "token", "key", "secret", "vertex_credentials", "credentials", "password", "passwd", ] def _mask_value(v: Any) -> Any: if isinstance(v, dict): if _depth >= _max_depth: return v return _get_masked_values( v, ignore_sensitive_values=ignore_sensitive_values, mask_all_values=mask_all_values, unmasked_length=unmasked_length, number_of_asterisks=number_of_asterisks, _depth=_depth + 1, _max_depth=_max_depth, ) if not isinstance(v, str): return v if len(v) <= unmasked_length: return "*****" if number_of_asterisks is not None: return v[: unmasked_length // 2] + "*" * number_of_asterisks + v[-unmasked_length // 2 :] return v[: unmasked_length // 2] + "*" * (len(v) - unmasked_length) + v[-unmasked_length // 2 :] return { k: ( v if ignore_sensitive_values or not any(sensitive_keyword in k.lower() for sensitive_keyword in sensitive_keywords) else _mask_value(v) ) for k, v in sensitive_object.items() } def set_callbacks(callback_list, function_id=None): """ Globally sets the callback client """ global \ sentry_sdk_instance, \ capture_exception, \ add_breadcrumb, \ slack_app, \ alerts_channel, \ traceloopLogger, \ athinaLogger, \ heliconeLogger, \ supabaseClient, \ lunaryLogger, \ promptLayerLogger, \ langFuseLogger, \ customLogger, \ weightsBiasesLogger, \ logfireLogger, \ dynamoLogger, \ s3Logger, \ dataDogLogger, \ prometheusLogger, \ greenscaleLogger, \ openMeterLogger, \ deepevalLogger try: for callback in callback_list: if callback == "sentry": try: import sentry_sdk except ImportError: print_verbose("Package 'sentry_sdk' is missing. Installing it...") subprocess.check_call([sys.executable, "-m", "pip", "install", "sentry_sdk"]) import sentry_sdk from sentry_sdk.scrubber import EventScrubber sentry_sdk_instance = sentry_sdk sentry_trace_rate = os.environ.get("SENTRY_API_TRACE_RATE", "1.0") sentry_sample_rate = ( os.environ.get("SENTRY_API_SAMPLE_RATE") if "SENTRY_API_SAMPLE_RATE" in os.environ else "1.0" ) sentry_sdk_instance.init( dsn=os.environ.get("SENTRY_DSN"), traces_sample_rate=float(sentry_trace_rate), sample_rate=float(sentry_sample_rate if sentry_sample_rate else 1.0), send_default_pii=False, # Prevent sending Personal Identifiable Information event_scrubber=EventScrubber(denylist=SENTRY_DENYLIST, pii_denylist=SENTRY_PII_DENYLIST), environment=os.environ.get("SENTRY_ENVIRONMENT", "production"), ) capture_exception = sentry_sdk_instance.capture_exception add_breadcrumb = sentry_sdk_instance.add_breadcrumb elif callback == "slack": try: from slack_bolt import App except ImportError: print_verbose("Package 'slack_bolt' is missing. Installing it...") subprocess.check_call([sys.executable, "-m", "pip", "install", "slack_bolt"]) from slack_bolt import App slack_app = App( token=os.environ.get("SLACK_API_TOKEN"), signing_secret=os.environ.get("SLACK_API_SECRET"), ) alerts_channel = os.environ["SLACK_API_CHANNEL"] print_verbose(f"Initialized Slack App: {slack_app}") elif callback == "traceloop": traceloopLogger = TraceloopLogger() elif callback == "athina": athinaLogger = AthinaLogger() print_verbose("Initialized Athina Logger") elif callback == "helicone": heliconeLogger = HeliconeLogger() elif callback == "lunary": lunaryLogger = LunaryLogger() elif callback == "promptlayer": promptLayerLogger = PromptLayerLogger() elif callback == "langfuse": langFuseLogger = LangFuseLogger(langfuse_public_key=None, langfuse_secret=None, langfuse_host=None) elif callback == "openmeter": openMeterLogger = OpenMeterLogger() elif callback == "datadog": dataDogLogger = DataDogLogger() elif callback == "dynamodb": dynamoLogger = DyanmoDBLogger() elif callback == "s3": s3Logger = S3Logger() elif callback == "wandb": from litellm.integrations.weights_biases import WeightsBiasesLogger weightsBiasesLogger = WeightsBiasesLogger() elif callback == "logfire": logfireLogger = LogfireLogger() elif callback == "supabase": print_verbose("instantiating supabase") supabaseClient = Supabase() elif callback == "greenscale": greenscaleLogger = GreenscaleLogger() print_verbose("Initialized Greenscale Logger") elif callable(callback): customLogger = CustomLogger() except Exception as e: raise e def _init_custom_logger_compatible_class( logging_integration: _custom_logger_compatible_callbacks_literal, internal_usage_cache: DualCache | None, llm_router: Any | None, # expect litellm.Router, but typing errors due to circular import custom_logger_init_args: dict | None = {}, ) -> CustomLogger | None: """ Initialize a custom logger compatible class """ try: custom_logger_init_args = custom_logger_init_args or {} if logging_integration == "agentops": # Add AgentOps initialization _v2 = _maybe_construct_otel_v2("agentops", _in_memory_loggers) if _v2 is not None: return _v2 for callback in _in_memory_loggers: if isinstance(callback, AgentOps): return callback agentops_logger: Final = AgentOps() _in_memory_loggers.append(agentops_logger) return agentops_logger elif logging_integration == "lago": for callback in _in_memory_loggers: if isinstance(callback, LagoLogger): return callback lago_logger: Final = LagoLogger() _in_memory_loggers.append(lago_logger) return lago_logger elif logging_integration == "openmeter": for callback in _in_memory_loggers: if isinstance(callback, OpenMeterLogger): return callback _openmeter_logger: Final = OpenMeterLogger() _in_memory_loggers.append(_openmeter_logger) return _openmeter_logger elif logging_integration == "posthog": for callback in _in_memory_loggers: if isinstance(callback, PostHogLogger): return callback _posthog_logger: Final = PostHogLogger() _in_memory_loggers.append(_posthog_logger) return _posthog_logger elif logging_integration == "braintrust": from litellm.integrations.braintrust_logging import BraintrustLogger for callback in _in_memory_loggers: if isinstance(callback, BraintrustLogger): return callback braintrust_logger: Final = BraintrustLogger() _in_memory_loggers.append(braintrust_logger) return braintrust_logger elif logging_integration == "langsmith": for callback in _in_memory_loggers: if isinstance(callback, LangsmithLogger): return callback _langsmith_logger: Final = LangsmithLogger() _in_memory_loggers.append(_langsmith_logger) return _langsmith_logger elif logging_integration == "argilla": for callback in _in_memory_loggers: if isinstance(callback, ArgillaLogger): return callback _argilla_logger: Final = ArgillaLogger() _in_memory_loggers.append(_argilla_logger) return _argilla_logger elif logging_integration == "literalai": for callback in _in_memory_loggers: if isinstance(callback, LiteralAILogger): return callback _literalai_logger: Final = LiteralAILogger() _in_memory_loggers.append(_literalai_logger) return _literalai_logger elif logging_integration == "litellm_agent": for callback in _in_memory_loggers: if isinstance(callback, LiteLLMAgentModelResolver): return callback _litellm_agent_resolver: Final = LiteLLMAgentModelResolver() _in_memory_loggers.append(_litellm_agent_resolver) return _litellm_agent_resolver elif logging_integration == "prometheus": PrometheusLogger: Final = _get_cached_prometheus_logger() for callback in _in_memory_loggers: if isinstance(callback, PrometheusLogger): return callback _prometheus_logger: Final = PrometheusLogger() _in_memory_loggers.append(_prometheus_logger) return _prometheus_logger elif logging_integration == "datadog": # Check if team-scoped credentials are provided _dd_api_key: Final = custom_logger_init_args.get("dd_api_key") _dd_site: Final = custom_logger_init_args.get("dd_site") _dd_agent_host: Final = custom_logger_init_args.get("dd_agent_host") _dd_agent_port: Final = custom_logger_init_args.get("dd_agent_port") if _dd_api_key or _dd_site or _dd_agent_host: # Team-scoped credentials: use DynamicLoggingCache for per-credential isolation from litellm.integrations.datadog.datadog_team_handler import ( DataDogHandler, ) return DataDogHandler.get_datadog_logger_for_request( standard_callback_dynamic_params=custom_logger_init_args, in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache, ) # Global (env-var based): reuse cached instance for callback in _in_memory_loggers: if isinstance(callback, DataDogLogger): return callback _datadog_logger: Final = DataDogLogger() _in_memory_loggers.append(_datadog_logger) return _datadog_logger elif logging_integration == "datadog_metrics": for callback in _in_memory_loggers: if isinstance(callback, DatadogMetricsLogger): return callback _datadog_metrics_logger: Final = DatadogMetricsLogger() _in_memory_loggers.append(_datadog_metrics_logger) return _datadog_metrics_logger elif logging_integration == "datadog_llm_observability": _datadog_llm_obs_logger: Final = DataDogLLMObsLogger() _in_memory_loggers.append(_datadog_llm_obs_logger) return _datadog_llm_obs_logger elif logging_integration == "azure_sentinel": for callback in _in_memory_loggers: if isinstance(callback, AzureSentinelLogger): return callback _azure_sentinel_logger: Final = AzureSentinelLogger() _in_memory_loggers.append(_azure_sentinel_logger) return _azure_sentinel_logger elif logging_integration == "gcs_bucket": for callback in _in_memory_loggers: if isinstance(callback, GCSBucketLogger): return callback _gcs_bucket_logger: Final = GCSBucketLogger() _in_memory_loggers.append(_gcs_bucket_logger) return _gcs_bucket_logger elif logging_integration == "s3_v2": for callback in _in_memory_loggers: if isinstance(callback, S3V2Logger): return callback _s3_v2_logger: Final = S3V2Logger() _in_memory_loggers.append(_s3_v2_logger) return _s3_v2_logger elif logging_integration == "aws_sqs": for callback in _in_memory_loggers: if isinstance(callback, SQSLogger): return callback _aws_sqs_logger: Final = SQSLogger() _in_memory_loggers.append(_aws_sqs_logger) return _aws_sqs_logger elif logging_integration == "azure_storage": for callback in _in_memory_loggers: if isinstance(callback, AzureBlobStorageLogger): return callback _azure_storage_logger: Final = AzureBlobStorageLogger() _in_memory_loggers.append(_azure_storage_logger) return _azure_storage_logger elif logging_integration == "opik": for callback in _in_memory_loggers: if isinstance(callback, OpikLogger): return callback _opik_logger: Final = OpikLogger() _in_memory_loggers.append(_opik_logger) return _opik_logger elif logging_integration == "arize": _v2 = _maybe_construct_otel_v2("arize", _in_memory_loggers) if _v2 is not None: return _v2 from litellm.integrations.opentelemetry import ( OpenTelemetry, OpenTelemetryConfig, ) arize_config: Final = ArizeLogger.get_arize_config() if arize_config.endpoint is None: raise ValueError( "No valid endpoint found for Arize, please set 'ARIZE_ENDPOINT' to your GRPC endpoint or 'ARIZE_HTTP_ENDPOINT' to your HTTP endpoint" ) otel_config = OpenTelemetryConfig( exporter=arize_config.protocol, endpoint=arize_config.endpoint, service_name=arize_config.project_name, ) os.environ["OTEL_EXPORTER_OTLP_TRACES_HEADERS"] = ( f"space_id={arize_config.space_key or arize_config.space_id},api_key={arize_config.api_key}" ) for callback in _in_memory_loggers: if isinstance(callback, ArizeLogger) and callback.callback_name == "arize": return callback _arize_otel_logger: Final = ArizeLogger(config=otel_config, callback_name="arize") _in_memory_loggers.append(_arize_otel_logger) return _arize_otel_logger elif logging_integration == "arize_phoenix": _v2 = _maybe_construct_otel_v2("arize_phoenix", _in_memory_loggers) if _v2 is not None: return _v2 from litellm.integrations.opentelemetry import ( OpenTelemetry, OpenTelemetryConfig, ) arize_phoenix_config: Final = ArizePhoenixLogger.get_arize_phoenix_config() otel_config = OpenTelemetryConfig( exporter=arize_phoenix_config.protocol, endpoint=arize_phoenix_config.endpoint, headers=arize_phoenix_config.otlp_auth_headers, ) # auth can be disabled on local deployments of arize phoenix if arize_phoenix_config.otlp_auth_headers is not None: os.environ["OTEL_EXPORTER_OTLP_TRACES_HEADERS"] = arize_phoenix_config.otlp_auth_headers for callback in _in_memory_loggers: if isinstance(callback, ArizePhoenixLogger) and callback.callback_name == "arize_phoenix": return callback _arize_phoenix_otel_logger: Final = ArizePhoenixLogger(config=otel_config, callback_name="arize_phoenix") _in_memory_loggers.append(_arize_phoenix_otel_logger) return _arize_phoenix_otel_logger elif logging_integration == "levo": _v2 = _maybe_construct_otel_v2("levo", _in_memory_loggers) if _v2 is not None: return _v2 from litellm.integrations.levo.levo import LevoLogger from litellm.integrations.opentelemetry import ( OpenTelemetry, OpenTelemetryConfig, ) levo_config: Final = LevoLogger.get_levo_config() otel_config = OpenTelemetryConfig( exporter=levo_config.protocol, endpoint=levo_config.endpoint, headers=levo_config.otlp_auth_headers, ) # Check if LevoLogger instance already exists for callback in _in_memory_loggers: if isinstance(callback, LevoLogger) and callback.callback_name == "levo": return callback _levo_otel_logger: Final = LevoLogger(config=otel_config, callback_name="levo") _in_memory_loggers.append(_levo_otel_logger) return _levo_otel_logger elif logging_integration == "otel": # Gate the new typed V2 adapter behind LITELLM_OTEL_V2. When off, # the legacy 3,227-line god-class is used unchanged. The two are # never registered simultaneously — the dedup loop below treats # any module under ``litellm.integrations.otel`` or # ``litellm.integrations.opentelemetry`` as "the OTel callback". from litellm.integrations.otel.model.config import is_otel_v2_enabled if is_otel_v2_enabled(): from litellm.integrations.otel.logger import OpenTelemetryV2 for callback in _in_memory_loggers: if type(callback) is OpenTelemetryV2: return callback otel_logger_v2: Final = OpenTelemetryV2( **_get_custom_logger_settings_from_proxy_server(callback_name=logging_integration) ) _in_memory_loggers.append(otel_logger_v2) _maybe_auto_initialize_arize_phoenix(_in_memory_loggers) return otel_logger_v2 from litellm.integrations.opentelemetry import OpenTelemetry for callback in _in_memory_loggers: if type(callback) is OpenTelemetry: return callback otel_logger: Final = OpenTelemetry( **_get_custom_logger_settings_from_proxy_server(callback_name=logging_integration) ) _in_memory_loggers.append(otel_logger) # Auto-initialize Arize Phoenix if Phoenix env vars are configured # This allows users to get nested traces in both OTEL and Phoenix # by only specifying "otel" in callbacks _maybe_auto_initialize_arize_phoenix(_in_memory_loggers) return otel_logger elif logging_integration == "galileo": for callback in _in_memory_loggers: if isinstance(callback, GalileoObserve): return callback galileo_logger: Final = GalileoObserve() _in_memory_loggers.append(galileo_logger) return galileo_logger elif logging_integration == "cloudzero": from litellm.integrations.cloudzero.cloudzero import CloudZeroLogger for callback in _in_memory_loggers: if isinstance(callback, CloudZeroLogger): return callback cloudzero_logger: Final = CloudZeroLogger() _in_memory_loggers.append(cloudzero_logger) return cloudzero_logger elif logging_integration == "focus": from litellm.integrations.focus.focus_logger import FocusLogger for callback in _in_memory_loggers: if type(callback) is FocusLogger: # exact match; exclude subclasses like VantageLogger return callback focus_logger: Final = FocusLogger() _in_memory_loggers.append(focus_logger) return focus_logger elif logging_integration == "mavvrik": from litellm.integrations.mavvrik_focus.mavvrik_focus_logger import ( MavvrikFocusLogger, ) for callback in _in_memory_loggers: if type(callback) is MavvrikFocusLogger: return callback mavvrik_focus_logger: Final = MavvrikFocusLogger() _in_memory_loggers.append(mavvrik_focus_logger) return mavvrik_focus_logger elif logging_integration == "vantage": from litellm.integrations.vantage.vantage_logger import VantageLogger for callback in _in_memory_loggers: if isinstance(callback, VantageLogger): return callback vantage_logger: Final = VantageLogger() _in_memory_loggers.append(vantage_logger) return vantage_logger elif logging_integration == "deepeval": for callback in _in_memory_loggers: if isinstance(callback, DeepEvalLogger): return callback deepeval_logger: Final = DeepEvalLogger() _in_memory_loggers.append(deepeval_logger) return deepeval_logger elif logging_integration == "logfire": if "LOGFIRE_TOKEN" not in os.environ: raise ValueError("LOGFIRE_TOKEN not found in environment variables") from litellm.integrations.opentelemetry import ( OpenTelemetry, OpenTelemetryConfig, ) logfire_base_url: Final = os.getenv("LOGFIRE_BASE_URL", "https://logfire-api.pydantic.dev") otel_config = OpenTelemetryConfig( exporter="otlp_http", endpoint=f"{logfire_base_url.rstrip('/')}/v1/traces", headers=f"Authorization={os.getenv('LOGFIRE_TOKEN')}", ) for callback in _in_memory_loggers: # Use exact type check to avoid matching ArizePhoenixLogger (subclass) if type(callback) is OpenTelemetry: return callback _otel_logger = OpenTelemetry(config=otel_config) _in_memory_loggers.append(_otel_logger) return _otel_logger elif logging_integration == "dynamic_rate_limiter": from litellm.proxy.hooks.dynamic_rate_limiter import ( _PROXY_DynamicRateLimitHandler, ) for callback in _in_memory_loggers: if isinstance(callback, _PROXY_DynamicRateLimitHandler): return callback if internal_usage_cache is None: raise Exception(f"Internal Error: Cache cannot be empty - internal_usage_cache={internal_usage_cache}") dynamic_rate_limiter_obj: Final = _PROXY_DynamicRateLimitHandler(internal_usage_cache=internal_usage_cache) if llm_router is not None and isinstance(llm_router, litellm.Router): dynamic_rate_limiter_obj.update_variables(llm_router=llm_router) _in_memory_loggers.append(dynamic_rate_limiter_obj) return dynamic_rate_limiter_obj elif logging_integration == "dynamic_rate_limiter_v3": from litellm.proxy.hooks.dynamic_rate_limiter_v3 import ( _PROXY_DynamicRateLimitHandlerV3, ) for callback in _in_memory_loggers: if isinstance(callback, _PROXY_DynamicRateLimitHandlerV3): return callback if internal_usage_cache is None: raise Exception(f"Internal Error: Cache cannot be empty - internal_usage_cache={internal_usage_cache}") dynamic_rate_limiter_obj_v3 = _PROXY_DynamicRateLimitHandlerV3(internal_usage_cache=internal_usage_cache) if llm_router is not None and isinstance(llm_router, litellm.Router): dynamic_rate_limiter_obj_v3.update_variables(llm_router=llm_router) _in_memory_loggers.append(dynamic_rate_limiter_obj_v3) return dynamic_rate_limiter_obj_v3 elif logging_integration == "langtrace": if "LANGTRACE_API_KEY" not in os.environ: raise ValueError("LANGTRACE_API_KEY not found in environment variables") _v2 = _maybe_construct_otel_v2("langtrace", _in_memory_loggers) if _v2 is not None: return _v2 from litellm.integrations.opentelemetry import ( OpenTelemetry, OpenTelemetryConfig, ) otel_config = OpenTelemetryConfig( exporter="otlp_http", endpoint="https://langtrace.ai/api/trace", ) os.environ["OTEL_EXPORTER_OTLP_TRACES_HEADERS"] = f"api_key={os.getenv('LANGTRACE_API_KEY')}" for callback in _in_memory_loggers: if isinstance(callback, OpenTelemetry) and callback.callback_name == "langtrace": return callback _otel_logger = OpenTelemetry(config=otel_config, callback_name="langtrace") _in_memory_loggers.append(_otel_logger) return _otel_logger elif logging_integration == "mlflow": for callback in _in_memory_loggers: if isinstance(callback, MlflowLogger): return callback _mlflow_logger: Final = MlflowLogger() _in_memory_loggers.append(_mlflow_logger) return _mlflow_logger elif logging_integration == "langfuse": for callback in _in_memory_loggers: if isinstance(callback, LangfusePromptManagement): return callback langfuse_logger: Final = LangfusePromptManagement() _in_memory_loggers.append(langfuse_logger) return langfuse_logger elif logging_integration == "langfuse_otel": _v2 = _maybe_construct_otel_v2("langfuse_otel", _in_memory_loggers) if _v2 is not None: return _v2 from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger for callback in _in_memory_loggers: if isinstance(callback, LangfuseOtelLogger) and callback.callback_name == "langfuse_otel": return callback # Allow LangfuseOtelLogger to initialize its own config safely # This prevents startup crashes if LANGFUSE keys are not in env (e.g. for dynamic usage) _otel_logger = LangfuseOtelLogger(config=None, callback_name="langfuse_otel") _in_memory_loggers.append(_otel_logger) return _otel_logger elif logging_integration == "weave_otel": _v2 = _maybe_construct_otel_v2("weave_otel", _in_memory_loggers) if _v2 is not None: return _v2 from litellm.integrations.opentelemetry import OpenTelemetryConfig from litellm.integrations.weave.weave_otel import ( WeaveOtelLogger, get_weave_otel_config, ) weave_otel_config: Final = get_weave_otel_config() otel_config = OpenTelemetryConfig( exporter=weave_otel_config.protocol, endpoint=weave_otel_config.endpoint, headers=weave_otel_config.otlp_auth_headers, ) for callback in _in_memory_loggers: if isinstance(callback, WeaveOtelLogger) and callback.callback_name == "weave_otel": return callback _otel_logger = WeaveOtelLogger(config=otel_config, callback_name="weave_otel") _in_memory_loggers.append(_otel_logger) return _otel_logger elif logging_integration == "pagerduty": for callback in _in_memory_loggers: if isinstance(callback, PagerDutyAlerting): return callback pagerduty_logger: Final = _PAGERDUTY_ALERTING_FACTORY(**custom_logger_init_args) _in_memory_loggers.append(pagerduty_logger) return pagerduty_logger elif logging_integration == "anthropic_cache_control_hook": for callback in _in_memory_loggers: if isinstance(callback, AnthropicCacheControlHook): return callback anthropic_cache_control_hook: Final = AnthropicCacheControlHook() _in_memory_loggers.append(anthropic_cache_control_hook) return anthropic_cache_control_hook elif logging_integration == "vector_store_pre_call_hook": from litellm.integrations.vector_store_integrations.vector_store_pre_call_hook import ( VectorStorePreCallHook, ) for callback in _in_memory_loggers: if isinstance(callback, VectorStorePreCallHook): return callback vector_store_pre_call_hook: Final = VectorStorePreCallHook() _in_memory_loggers.append(vector_store_pre_call_hook) return vector_store_pre_call_hook elif logging_integration == "gcs_pubsub": for callback in _in_memory_loggers: if isinstance(callback, GcsPubSubLogger): return callback _gcs_pubsub_logger: Final = GcsPubSubLogger() _in_memory_loggers.append(_gcs_pubsub_logger) return _gcs_pubsub_logger elif logging_integration == "generic_api": for callback in _in_memory_loggers: if isinstance(callback, _GENERIC_API_LOGGER_CLS): return callback generic_api_logger: Final = GenericAPILogger() _in_memory_loggers.append(generic_api_logger) return generic_api_logger elif logging_integration == "resend_email": for callback in _in_memory_loggers: if isinstance(callback, ResendEmailLogger): return callback resend_email_logger: Final = _RESEND_EMAIL_LOGGER_FACTORY() _in_memory_loggers.append(resend_email_logger) return resend_email_logger elif logging_integration == "sendgrid_email": for callback in _in_memory_loggers: if isinstance(callback, SendGridEmailLogger): return callback sendgrid_email_logger: Final = _SENDGRID_EMAIL_LOGGER_FACTORY() _in_memory_loggers.append(sendgrid_email_logger) return sendgrid_email_logger elif logging_integration == "smtp_email": for callback in _in_memory_loggers: if isinstance(callback, SMTPEmailLogger): return callback smtp_email_logger: Final = _SMTP_EMAIL_LOGGER_FACTORY() _in_memory_loggers.append(smtp_email_logger) return smtp_email_logger elif logging_integration == "humanloop": for callback in _in_memory_loggers: if isinstance(callback, HumanloopLogger): return callback humanloop_logger: Final = HumanloopLogger() _in_memory_loggers.append(humanloop_logger) return humanloop_logger elif logging_integration == "dotprompt": for callback in _in_memory_loggers: if isinstance(callback, DotpromptManager): return callback dotprompt_logger: Final = DotpromptManager() _in_memory_loggers.append(dotprompt_logger) return dotprompt_logger elif logging_integration == "bitbucket": from litellm.integrations.bitbucket.bitbucket_prompt_manager import ( BitBucketPromptManager, ) for callback in _in_memory_loggers: if isinstance(callback, BitBucketPromptManager): return callback # Get global BitBucket config bitbucket_config: Final = getattr(litellm, "global_bitbucket_config", None) if bitbucket_config is None: raise ValueError("BitBucket configuration not found. Please set litellm.global_bitbucket_config first.") bitbucket_logger: Final = BitBucketPromptManager(bitbucket_config=bitbucket_config) _in_memory_loggers.append(bitbucket_logger) return bitbucket_logger elif logging_integration == "gitlab": from litellm.integrations.gitlab.gitlab_prompt_manager import ( GitLabPromptManager, ) for callback in _in_memory_loggers: if isinstance(callback, GitLabPromptManager): return callback # Get global BitBucket config gitlab_config: Final = getattr(litellm, "global_gitlab_config", None) if gitlab_config is None: raise ValueError("Gitlab configuration not found. Please set litellm.global_gitlab_config first.") gitlab_logger: Final = GitLabPromptManager(gitlab_config=gitlab_config) _in_memory_loggers.append(gitlab_logger) return gitlab_logger elif logging_integration == "newrelic": if custom_logger_init_args.get("newrelic_api_key"): # Team-scoped credentials: per-team METRICS logger, isolated per # credential set via DynamicLoggingCache. The trace logger for # this name stays on the global path below. from litellm.integrations.newrelic.newrelic_team_handler import ( NewRelicHandler, ) return NewRelicHandler.get_newrelic_logger_for_request( standard_callback_dynamic_params=custom_logger_init_args, in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache, ) _v2 = _maybe_construct_otel_v2("newrelic", _in_memory_loggers) if _v2 is not None: return _v2 for callback in _in_memory_loggers: if isinstance(callback, NewRelicLogger): return callback newrelic_logger: Final = NewRelicLogger() _in_memory_loggers.append(newrelic_logger) return newrelic_logger return None except Exception as e: verbose_logger.exception("[Non-Blocking Error] Error initializing custom logger: %s", e) return None return None def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[CustomLogger]) -> "OpenTelemetryV2 | None": """If ``LITELLM_OTEL_V2`` is on, build (or reuse) a single ``OpenTelemetryV2`` instance configured via the preset for ``callback_name``. Returns ``None`` when V2 is off OR when there's no preset registered for ``callback_name`` — callers should then fall through to the legacy path. """ from litellm.integrations.otel.model.config import is_otel_v2_enabled if not is_otel_v2_enabled(): return None from litellm.integrations.otel.logger import OpenTelemetryV2 from litellm.integrations.otel.presets import PRESET_BY_CALLBACK preset_fn: Final = PRESET_BY_CALLBACK.get(callback_name) if preset_fn is None: return None for callback in _in_memory_loggers: if isinstance(callback, OpenTelemetryV2) and getattr(callback, "callback_name", None) == callback_name: return callback try: config: Final = preset_fn() except Exception: # If env vars are missing or the preset raises, defer to the legacy path # so customers get the same error story they had before V2 landed. return None v2_logger: Final = OpenTelemetryV2(config=config, callback_name=callback_name) _in_memory_loggers.append(v2_logger) return v2_logger def _maybe_auto_initialize_arize_phoenix(_in_memory_loggers: list[CustomLogger]) -> None: """ Auto-initialize ArizePhoenixLogger when Phoenix env vars are detected. Called during ``otel`` callback setup so that users get nested traces in both their OTEL collector *and* Arize Phoenix by only listing ``"otel"`` in ``callbacks``. If no Phoenix env vars are set, this is a no-op. """ phoenix_env_vars: Final = ( "PHOENIX_API_KEY", "PHOENIX_COLLECTOR_HTTP_ENDPOINT", "PHOENIX_COLLECTOR_ENDPOINT", ) if not any(os.environ.get(v) for v in phoenix_env_vars): return # Already registered — nothing to do if any(isinstance(cb, ArizePhoenixLogger) and cb.callback_name == "arize_phoenix" for cb in _in_memory_loggers): return try: from litellm.integrations.opentelemetry import OpenTelemetryConfig arize_phoenix_config: Final = ArizePhoenixLogger.get_arize_phoenix_config() otel_config: Final = OpenTelemetryConfig( exporter=arize_phoenix_config.protocol, endpoint=arize_phoenix_config.endpoint, headers=arize_phoenix_config.otlp_auth_headers, ) phoenix_logger: Final = ArizePhoenixLogger(config=otel_config, callback_name="arize_phoenix") _in_memory_loggers.append(phoenix_logger) # Register as a litellm callback so it receives success/failure events litellm.logging_callback_manager.add_litellm_callback(phoenix_logger) verbose_logger.info( "Auto-initialized Arize Phoenix logger alongside otel (endpoint=%s)", arize_phoenix_config.endpoint, ) except Exception as e: verbose_logger.warning("Failed to auto-initialize Arize Phoenix logger: %s", str(e)) def get_custom_logger_compatible_class( logging_integration: _custom_logger_compatible_callbacks_literal, ) -> CustomLogger | None: try: if logging_integration == "lago": for callback in _in_memory_loggers: if isinstance(callback, LagoLogger): return callback elif logging_integration == "openmeter": for callback in _in_memory_loggers: if isinstance(callback, OpenMeterLogger): return callback elif logging_integration == "braintrust": from litellm.integrations.braintrust_logging import BraintrustLogger for callback in _in_memory_loggers: if isinstance(callback, BraintrustLogger): return callback elif logging_integration == "galileo": for callback in _in_memory_loggers: if isinstance(callback, GalileoObserve): return callback elif logging_integration == "cloudzero": from litellm.integrations.cloudzero.cloudzero import CloudZeroLogger for callback in _in_memory_loggers: if isinstance(callback, CloudZeroLogger): return callback elif logging_integration == "focus": from litellm.integrations.focus.focus_logger import FocusLogger for callback in _in_memory_loggers: if type(callback) is FocusLogger: # exact match; exclude subclasses like VantageLogger return callback elif logging_integration == "vantage": from litellm.integrations.vantage.vantage_logger import VantageLogger for callback in _in_memory_loggers: if isinstance(callback, VantageLogger): return callback elif logging_integration == "deepeval": for callback in _in_memory_loggers: if isinstance(callback, DeepEvalLogger): return callback elif logging_integration == "langsmith": for callback in _in_memory_loggers: if isinstance(callback, LangsmithLogger): return callback elif logging_integration == "argilla": for callback in _in_memory_loggers: if isinstance(callback, ArgillaLogger): return callback elif logging_integration == "literalai": for callback in _in_memory_loggers: if isinstance(callback, LiteralAILogger): return callback elif logging_integration == "litellm_agent": for callback in _in_memory_loggers: if isinstance(callback, LiteLLMAgentModelResolver): return callback elif logging_integration == "prometheus": PrometheusLogger: Final = _get_cached_prometheus_logger() for callback in _in_memory_loggers: if isinstance(callback, PrometheusLogger): return callback elif logging_integration == "datadog": for callback in _in_memory_loggers: if isinstance(callback, DataDogLogger): return callback elif logging_integration == "datadog_metrics": for callback in _in_memory_loggers: if isinstance(callback, DatadogMetricsLogger): return callback elif logging_integration == "datadog_llm_observability": for callback in _in_memory_loggers: if isinstance(callback, DataDogLLMObsLogger): return callback elif logging_integration == "azure_sentinel": for callback in _in_memory_loggers: if isinstance(callback, AzureSentinelLogger): return callback elif logging_integration == "gcs_bucket": for callback in _in_memory_loggers: if isinstance(callback, GCSBucketLogger): return callback elif logging_integration == "s3_v2": for callback in _in_memory_loggers: if isinstance(callback, S3V2Logger): return callback elif logging_integration == "aws_sqs": for callback in _in_memory_loggers: if isinstance(callback, SQSLogger): return callback _aws_sqs_logger: Final = SQSLogger() _in_memory_loggers.append(_aws_sqs_logger) return _aws_sqs_logger elif logging_integration == "azure_storage": for callback in _in_memory_loggers: if isinstance(callback, AzureBlobStorageLogger): return callback elif logging_integration == "opik": for callback in _in_memory_loggers: if isinstance(callback, OpikLogger): return callback elif logging_integration == "langfuse": for callback in _in_memory_loggers: if isinstance(callback, LangfusePromptManagement): return callback elif logging_integration == "otel": from litellm.integrations.opentelemetry import OpenTelemetry for callback in _in_memory_loggers: # Use exact type check to avoid matching ArizePhoenixLogger (subclass) if type(callback) is OpenTelemetry: return callback elif logging_integration == "arize": if "ARIZE_API_KEY" not in os.environ: raise ValueError("ARIZE_API_KEY not found in environment variables") for callback in _in_memory_loggers: if isinstance(callback, ArizeLogger) and callback.callback_name == "arize": return callback elif logging_integration == "logfire": if "LOGFIRE_TOKEN" not in os.environ: raise ValueError("LOGFIRE_TOKEN not found in environment variables") from litellm.integrations.opentelemetry import OpenTelemetry for callback in _in_memory_loggers: # Use exact type check to avoid matching ArizePhoenixLogger (subclass) if type(callback) is OpenTelemetry: return callback elif logging_integration == "dynamic_rate_limiter": from litellm.proxy.hooks.dynamic_rate_limiter import ( _PROXY_DynamicRateLimitHandler, ) for callback in _in_memory_loggers: if isinstance(callback, _PROXY_DynamicRateLimitHandler): return callback elif logging_integration == "dynamic_rate_limiter_v3": from litellm.proxy.hooks.dynamic_rate_limiter_v3 import ( _PROXY_DynamicRateLimitHandlerV3, ) for callback in _in_memory_loggers: if isinstance(callback, _PROXY_DynamicRateLimitHandlerV3): return callback elif logging_integration == "langtrace": from litellm.integrations.opentelemetry import OpenTelemetry if "LANGTRACE_API_KEY" not in os.environ: raise ValueError("LANGTRACE_API_KEY not found in environment variables") for callback in _in_memory_loggers: if isinstance(callback, OpenTelemetry) and callback.callback_name == "langtrace": return callback elif logging_integration == "mlflow": for callback in _in_memory_loggers: if isinstance(callback, MlflowLogger): return callback elif logging_integration == "pagerduty": for callback in _in_memory_loggers: if isinstance(callback, PagerDutyAlerting): return callback elif logging_integration == "anthropic_cache_control_hook": for callback in _in_memory_loggers: if isinstance(callback, AnthropicCacheControlHook): return callback elif logging_integration == "vector_store_pre_call_hook": from litellm.integrations.vector_store_integrations.vector_store_pre_call_hook import ( VectorStorePreCallHook, ) for callback in _in_memory_loggers: if isinstance(callback, VectorStorePreCallHook): return callback elif logging_integration == "gcs_pubsub": for callback in _in_memory_loggers: if isinstance(callback, GcsPubSubLogger): return callback elif logging_integration == "generic_api": for callback in _in_memory_loggers: if isinstance(callback, _GENERIC_API_LOGGER_CLS): return callback elif logging_integration == "resend_email": for callback in _in_memory_loggers: if isinstance(callback, ResendEmailLogger): return callback elif logging_integration == "sendgrid_email": for callback in _in_memory_loggers: if isinstance(callback, SendGridEmailLogger): return callback elif logging_integration == "smtp_email": for callback in _in_memory_loggers: if isinstance(callback, SMTPEmailLogger): return callback elif logging_integration == "newrelic": from litellm.integrations.otel.logger import OpenTelemetryV2 for callback in _in_memory_loggers: if isinstance(callback, OpenTelemetryV2) and callback.callback_name == "newrelic": return callback if isinstance(callback, NewRelicLogger): return callback return None except Exception as e: verbose_logger.exception("[Non-Blocking Error] Error getting custom logger: %s", e) return None def _get_custom_logger_settings_from_proxy_server(callback_name: str) -> dict: """ Get the settings for a custom logger from the proxy server config.yaml Proxy server config.yaml defines callback_settings as: callback_settings: otel: message_logging: False """ if litellm.callback_settings: return dict(litellm.callback_settings.get(callback_name, {})) return {} def use_custom_pricing_for_model(litellm_params: dict | None) -> bool: """ Check if the model uses custom pricing Returns True if any of `SPECIAL_MODEL_INFO_PARAMS` are present in `litellm_params` or `model_info` """ if litellm_params is None: return False # Check litellm_params using set intersection (only check keys that exist in both) matching_keys = _CUSTOM_PRICING_KEYS & litellm_params.keys() for key in matching_keys: if litellm_params.get(key) is not None: return True # Check model_info from metadata or litellm_metadata (generic_api_call routes # like /responses and /messages store model_info under litellm_metadata) for metadata_key in ("metadata", "litellm_metadata"): metadata: dict = litellm_params.get(metadata_key, {}) or {} model_info: dict = metadata.get("model_info", {}) or {} if model_info: matching_keys = _CUSTOM_PRICING_KEYS & model_info.keys() for key in matching_keys: if model_info.get(key) is not None: return True return False def is_valid_sha256_hash(value: str) -> bool: # Check if the value is a valid SHA-256 hash (64 hexadecimal characters) return bool(re.fullmatch(r"[a-fA-F0-9]{64}", value)) def coerce_model_access_groups(value: object) -> tuple[str, ...]: """Model access group names out of untrusted request metadata, deduped and order preserving.""" if not isinstance(value, (list, tuple)): return () return tuple(dict.fromkeys(group for group in value if isinstance(group, str) and group)) def _model_access_groups_on_auth_object(user_api_key_auth: object) -> object: if isinstance(user_api_key_auth, Mapping): return user_api_key_auth.get("matched_model_access_groups") return getattr(user_api_key_auth, "matched_model_access_groups", None) def _model_access_groups_from_metadata(metadata: Mapping[str, object]) -> tuple[str, ...]: stamped: Final = coerce_model_access_groups(metadata.get(MODEL_ACCESS_GROUP_METADATA_KEY)) if stamped: return stamped return coerce_model_access_groups(_model_access_groups_on_auth_object(metadata.get("user_api_key_auth"))) def request_model_access_groups_from_litellm_params(litellm_params: Mapping[str, object]) -> tuple[str, ...]: """Access groups the auth layer stamped onto this request, from whichever metadata field carries them. Detached internal sub-calls only inherit the identity keys, so the auth object is the fallback there, exactly as _get_budget_reservation_from_metadata does for reservations. """ for metadata_variable_name in ("metadata", "litellm_metadata"): metadata = litellm_params.get(metadata_variable_name) if not isinstance(metadata, Mapping): continue model_access_groups = _model_access_groups_from_metadata(metadata) if model_access_groups: return model_access_groups return () class StandardLoggingPayloadSetup: @staticmethod def cleanup_timestamps( start_time: dt_object | float, end_time: dt_object | float, completion_start_time: dt_object | float, ) -> tuple[float, float, float]: """ Convert datetime objects to floats Args: start_time: Union[dt_object, float] end_time: Union[dt_object, float] completion_start_time: Union[dt_object, float] Returns: Tuple[float, float, float]: A tuple containing the start time, end time, and completion start time as floats. """ if isinstance(start_time, datetime.datetime): start_time_float = start_time.timestamp() elif isinstance(start_time, float): start_time_float = start_time else: raise ValueError(f"start_time is required, got={start_time} of type {type(start_time)}") if isinstance(end_time, datetime.datetime): end_time_float = end_time.timestamp() elif isinstance(end_time, float): end_time_float = end_time else: raise ValueError(f"end_time is required, got={end_time} of type {type(end_time)}") if isinstance(completion_start_time, datetime.datetime): completion_start_time_float = completion_start_time.timestamp() elif isinstance(completion_start_time, float): completion_start_time_float = completion_start_time else: completion_start_time_float = end_time_float return start_time_float, end_time_float, completion_start_time_float @staticmethod def append_system_prompt_messages(kwargs: dict | None = None, messages: Any | None = None): """ Append system prompt messages to the messages """ if kwargs is not None: if kwargs.get("system") is not None and isinstance(kwargs.get("system"), str): if messages is None: return [{"role": "system", "content": kwargs.get("system")}] elif isinstance(messages, list): if len(messages) == 0: return [{"role": "system", "content": kwargs.get("system")}] # check for duplicates if messages[0].get("role") == "system" and messages[0].get("content") == kwargs.get("system"): return messages messages = [{"role": "system", "content": kwargs.get("system")}] + messages elif isinstance(messages, str): messages = [ {"role": "system", "content": kwargs.get("system")}, {"role": "user", "content": messages}, ] return messages return messages @staticmethod def merge_litellm_metadata(litellm_params: Mapping[str, object]) -> dict: """ Merge both litellm_metadata and metadata from litellm_params. litellm_metadata contains model-related fields, metadata contains user API key fields. We need both for complete standard logging payload. Args: litellm_params: Dictionary containing metadata and litellm_metadata Returns: dict: Merged metadata with user API key fields taking precedence """ merged_metadata: Final[dict] = {} # Start with metadata (user API key fields) - but skip non-serializable objects if litellm_params.get("metadata") and isinstance(litellm_params.get("metadata"), dict): for key, value in litellm_params["metadata"].items(): # Skip non-serializable objects like UserAPIKeyAuth if key in {"user_api_key_auth", "user_api_key_budget_reservation"}: continue merged_metadata[key] = value # Then merge litellm_metadata (model-related fields) - this will NOT overwrite existing keys if litellm_params.get("litellm_metadata") and isinstance(litellm_params.get("litellm_metadata"), dict): for key, value in litellm_params["litellm_metadata"].items(): if key not in merged_metadata: # Don't overwrite existing keys from metadata merged_metadata[key] = value return merged_metadata @staticmethod def get_standard_logging_metadata( metadata: dict[str, Any] | None, litellm_params: dict | None = None, prompt_integration: str | None = None, applied_guardrails: list[str] | None = None, mcp_tool_call_metadata: StandardLoggingMCPToolCall | None = None, vector_store_request_metadata: list[StandardLoggingVectorStoreRequest] | None = None, usage_object: dict | None = None, proxy_server_request: dict | None = None, start_time: dt_object | None = None, response_id: str | None = None, ) -> StandardLoggingMetadata: """ Clean and filter the metadata dictionary to include only the specified keys in StandardLoggingMetadata. Args: metadata (Optional[Dict[str, Any]]): The original metadata dictionary. Returns: StandardLoggingMetadata: A StandardLoggingMetadata object containing the cleaned metadata. Note: - If the input metadata is None or not a dictionary, an empty StandardLoggingMetadata object is returned. - If 'user_api_key' is present in metadata and is a valid SHA256 hash, it's stored as 'user_api_key_hash'. """ prompt_management_metadata: StandardLoggingPromptManagementMetadata | None = None if litellm_params is not None: prompt_id: Final = cast(str | None, litellm_params.get("prompt_id", None)) prompt_variables: Final = cast(dict | None, litellm_params.get("prompt_variables", None)) if prompt_id is not None and prompt_integration is not None: prompt_management_metadata = StandardLoggingPromptManagementMetadata( prompt_id=prompt_id, prompt_variables=prompt_variables, prompt_integration=prompt_integration, ) # Initialize with default values clean_metadata = StandardLoggingMetadata( user_api_key_hash=None, user_api_key_alias=None, user_api_key_spend=None, user_api_key_max_budget=None, user_api_key_budget_reset_at=None, user_api_key_user_spend=None, user_api_key_user_max_budget=None, user_api_key_team_spend=None, user_api_key_team_max_budget=None, user_api_key_team_id=None, user_api_key_org_id=None, user_api_key_org_alias=None, user_api_key_project_id=None, user_api_key_project_alias=None, user_api_key_user_id=None, user_api_key_team_alias=None, user_api_key_user_email=None, user_api_key_end_user_id=None, user_api_key_request_route=None, spend_logs_metadata=None, requester_ip_address=None, user_agent=None, requester_metadata=None, prompt_management_metadata=prompt_management_metadata, applied_guardrails=applied_guardrails, mcp_tool_call_metadata=mcp_tool_call_metadata, vector_store_request_metadata=vector_store_request_metadata, routing_decision=None, usage_object=usage_object, requester_custom_headers=None, cold_storage_object_key=None, user_api_key_auth_metadata=None, team_alias=None, team_id=None, ) if isinstance(metadata, dict): for key in metadata.keys() & _STANDARD_LOGGING_METADATA_KEYS: clean_metadata[key] = metadata[key] user_api_key: Final = metadata.get("user_api_key") if user_api_key and isinstance(user_api_key, str) and is_valid_sha256_hash(user_api_key): clean_metadata["user_api_key_hash"] = user_api_key _potential_requester_metadata: Final = metadata.get( "metadata", None ) # check if user passed metadata in the sdk request - e.g. metadata for langsmith logging - https://docs.litellm.ai/docs/observability/langsmith_integration#set-langsmith-fields if ( clean_metadata["requester_metadata"] is None and _potential_requester_metadata is not None and isinstance(_potential_requester_metadata, dict) ): clean_metadata["requester_metadata"] = _potential_requester_metadata if EnterpriseStandardLoggingPayloadSetupVAR and proxy_server_request is not None: clean_metadata = EnterpriseStandardLoggingPayloadSetupVAR.apply_enterprise_specific_metadata( standard_logging_metadata=clean_metadata, proxy_server_request=proxy_server_request, ) # Generate cold storage object key if cold storage is configured if start_time is not None and response_id is not None: cold_storage_object_key: Final = StandardLoggingPayloadSetup._generate_cold_storage_object_key( start_time=start_time, response_id=response_id, team_alias=clean_metadata.get("user_api_key_team_alias"), ) if cold_storage_object_key: clean_metadata["cold_storage_object_key"] = cold_storage_object_key return clean_metadata @staticmethod def get_usage_from_response_obj(response_obj: dict | None, combined_usage_object: Usage | None = None) -> Usage: ## BASE CASE ## if combined_usage_object is not None: return combined_usage_object if response_obj is None: return Usage( prompt_tokens=0, completion_tokens=0, total_tokens=0, ) usage: Final = response_obj.get("usage", None) or {} if usage is None or (not isinstance(usage, dict) and not isinstance(usage, Usage)): return Usage( prompt_tokens=0, completion_tokens=0, total_tokens=0, ) elif isinstance(usage, Usage): return usage elif isinstance(usage, ResponseAPIUsage): return ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(usage) elif isinstance(usage, dict): if ResponseAPILoggingUtils._is_response_api_usage(usage): return ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(usage) if InteractionsUsageObjectTransformation.is_interactions_usage_object(usage): return InteractionsUsageObjectTransformation.transform_interactions_usage_object(usage) return Usage(**usage) raise ValueError(f"usage is required, got={usage} of type {type(usage)}") @staticmethod def get_usage_as_dict( response_obj: dict | None, combined_usage_object: Usage | None = None, ) -> dict: """ Like get_usage_from_response_obj but returns a plain dict, skipping the Pydantic Usage construction on the hot path. """ _empty: Final[dict] = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0} if combined_usage_object is not None: return combined_usage_object.model_dump() if not response_obj: return _empty _raw: Final = response_obj.get("usage", None) if _raw is None: return _empty if isinstance(_raw, ResponseAPIUsage): return ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(_raw).model_dump() if isinstance(_raw, dict): if ResponseAPILoggingUtils._is_response_api_usage(_raw): return ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(_raw).model_dump() if InteractionsUsageObjectTransformation.is_interactions_usage_object(_raw): return InteractionsUsageObjectTransformation.transform_interactions_usage_object(_raw).model_dump() return _raw if isinstance(_raw, Usage): return _raw.model_dump() return _empty @staticmethod def get_model_cost_information( base_model: str | None, custom_pricing: bool | None, custom_llm_provider: str | None, init_response_obj: Any | BaseModel | dict, api_base: str | None = None, ) -> StandardLoggingModelInformation: model_cost_name: Final = _select_model_name_for_cost_calc( model=base_model if custom_pricing else None, completion_response=init_response_obj, base_model=base_model, custom_pricing=custom_pricing, ) if model_cost_name is None: model_cost_information = StandardLoggingModelInformation(model_map_key="", model_map_value=None) else: try: _model_cost_information: Final = litellm.get_model_info( model=model_cost_name, custom_llm_provider=custom_llm_provider, api_base=api_base, ) model_cost_information = StandardLoggingModelInformation( model_map_key=model_cost_name, model_map_value=_model_cost_information, ) except Exception: verbose_logger.debug( # keep in debug otherwise it will trigger on every call "Model=%s is not mapped in model cost map. Defaulting to None model_cost_information for standard_logging_payload", model_cost_name, ) model_cost_information = StandardLoggingModelInformation( model_map_key=model_cost_name, model_map_value=None ) return model_cost_information @staticmethod def get_final_response_obj( response_obj: dict, init_response_obj: Any | BaseModel | dict, kwargs: dict ) -> dict | str | list | None: """ Get final response object after redacting the message input/output from logging """ if response_obj: final_response_obj: dict | str | list | None = response_obj elif isinstance(init_response_obj, list) or isinstance(init_response_obj, str): final_response_obj = init_response_obj else: final_response_obj = {} modified_final_response_obj: Final = redact_message_input_output_from_logging( model_call_details=kwargs, result=final_response_obj, ) if modified_final_response_obj is not None and isinstance(modified_final_response_obj, BaseModel): final_response_obj = modified_final_response_obj.model_dump() else: final_response_obj = modified_final_response_obj return final_response_obj @staticmethod def get_additional_headers( additiona_headers: dict | None, ) -> StandardLoggingAdditionalHeaders | None: if additiona_headers is None: return None additional_logging_headers: Final[StandardLoggingAdditionalHeaders] = {} # Populate well-known typed fields with int/str coercion where needed typed_keys: Final[dict] = {} for key in StandardLoggingAdditionalHeaders.__annotations__: _key = key.lower().replace("_", "-") typed_keys[_key] = key if _key in additiona_headers: try: additional_logging_headers[key] = int(additiona_headers[_key]) except (ValueError, TypeError): additional_logging_headers[key] = additiona_headers[_key] # Preserve all remaining headers verbatim (e.g. llm_provider-x-request-id) for k, v in additiona_headers.items(): if k.lower() not in typed_keys: additional_logging_headers[k] = v return additional_logging_headers @staticmethod def get_hidden_params( hidden_params: dict | None, ) -> StandardLoggingHiddenParams: clean_hidden_params: Final = StandardLoggingHiddenParams( model_id=None, cache_key=None, api_base=None, response_cost=None, additional_headers=None, litellm_overhead_time_ms=None, batch_models=None, batch_successful_requests=None, batch_failed_requests=None, litellm_model_name=None, usage_object=None, ) if hidden_params is not None: for key in StandardLoggingHiddenParams.__annotations__: if key in hidden_params: if key == "additional_headers": clean_hidden_params["additional_headers"] = StandardLoggingPayloadSetup.get_additional_headers( hidden_params[key] ) else: clean_hidden_params[key] = hidden_params[key] return clean_hidden_params @staticmethod def strip_trailing_slash(api_base: str | None) -> str | None: if api_base: if api_base.endswith("//"): return api_base.rstrip("/") if api_base[-1] == "/": return api_base[:-1] return api_base @staticmethod def _generate_cold_storage_object_key( start_time: dt_object, response_id: str, team_alias: str | None = None, ) -> str | None: """ Generate cold storage object key in the same format as S3Logger. Args: start_time: The start time of the request response_id: The response ID team_alias: Optional team alias for team-based prefixing Returns: Optional[str]: The generated object key or None if cold storage not configured """ # Generate object key in same format as S3Logger from litellm.integrations.s3 import get_s3_object_key # Only generate object key if cold storage is configured cold_storage_custom_logger: Final = litellm.cold_storage_custom_logger if cold_storage_custom_logger is None: return None try: # Generate file name in same format as litellm.utils.get_logging_id s3_file_name: Final = f"time-{start_time.strftime('%H-%M-%S-%f')}_{response_id}" # Get the actual s3_path from the configured cold storage logger instance s3_path = "" # default value # Try to get the actual logger instance from the logger name try: custom_logger: Final = litellm.logging_callback_manager.get_active_custom_logger_for_callback_name( cold_storage_custom_logger ) if custom_logger and hasattr(custom_logger, "s3_path") and getattr(custom_logger, "s3_path"): s3_path = getattr(custom_logger, "s3_path") except Exception: # If any error occurs in getting the logger instance, use default empty s3_path pass s3_object_key: Final = get_s3_object_key( s3_path=s3_path, # Use actual s3_path from logger configuration prefix="", # Don't split by team alias for cold storage start_time=start_time, s3_file_name=s3_file_name, ) return s3_object_key except Exception: # If any error occurs in generating the key, return None return None @staticmethod def get_error_information( original_exception: Exception | None, traceback_str: str | None = None, ) -> StandardLoggingPayloadErrorInformation: from litellm.constants import MAXIMUM_TRACEBACK_LINES_TO_LOG # ProxyException uses .code, LiteLLM exceptions use .status_code, # httpx.HTTPStatusError exposes status only as .response.status_code. # Stringified for Prisma JSON compatibility. error_code_attr: Final[object] = getattr(original_exception, "code", None) if error_code_attr is not None and str(error_code_attr) not in ("", "None"): error_status: str = str(error_code_attr) else: status_code_attr: object = getattr(original_exception, "status_code", None) if status_code_attr is None: response_attr: Final[object] = getattr(original_exception, "response", None) status_code_attr = getattr(response_attr, "status_code", None) error_status = str(status_code_attr) if status_code_attr is not None else "" error_class: Final[str] = str(original_exception.__class__.__name__) if original_exception else "" _llm_provider_in_exception: Final = getattr(original_exception, "llm_provider", "") traceback_info = traceback_str or "" if original_exception and ( litellm.log_client_error_tracebacks or not is_expected_client_error(original_exception) ): tb: Final[TracebackType | None] = getattr(original_exception, "__traceback__", None) if tb: tb_lines: Final = traceback.format_tb(tb) traceback_info += "".join(tb_lines[:MAXIMUM_TRACEBACK_LINES_TO_LOG]) # Limit to first 100 lines # Prefer the `.message` attribute (set by ProxyException and every # litellm.exceptions.* class) over str(exc); ProxyException does not # call super().__init__() nor define __str__, so str() on it returns # an empty string, which used to silently strip the human-readable # message from spend_logs.metadata.error_information. # Use isinstance, not truthiness: an explicit empty string on # `.message` is a deliberate value and must not be replaced by # `str(exc)`. explicit_message: Final = getattr(original_exception, "message", None) if isinstance(explicit_message, str): error_message = explicit_message else: error_message = str(original_exception) if original_exception else "" rate_limit_category: Final = validate_rate_limit_category(getattr(original_exception, "category", None)) rate_limit_type: Final = validate_rate_limit_type(getattr(original_exception, "rate_limit_type", None)) budget_error: Final = original_exception if isinstance(original_exception, BudgetExceededError) else None return StandardLoggingPayloadErrorInformation( error_code=error_status, error_class=error_class, llm_provider=_llm_provider_in_exception, traceback=traceback_info, error_message=error_message, error_rate_limit_category=rate_limit_category, error_rate_limit_type=rate_limit_type, error_budget_entity_type=budget_error.entity_type if budget_error else None, error_budget_entity_id=budget_error.entity_id if budget_error else None, error_budget_limit=budget_error.max_budget if budget_error else None, error_budget_spend=budget_error.current_cost if budget_error else None, ) @staticmethod def get_error_information_for_logging_payload( metadata: dict, original_exception: Exception | None, error_str: str | None, ) -> tuple[StandardLoggingPayloadErrorInformation, str | None]: error_information = StandardLoggingPayloadSetup.get_error_information( original_exception=original_exception, ) if not metadata.get("client_disconnected"): return error_information, error_str client_disconnect_error: Final = metadata.get("error_information") if isinstance(client_disconnect_error, dict): error_information = cast( StandardLoggingPayloadErrorInformation, client_disconnect_error, ) else: error_information = cast( StandardLoggingPayloadErrorInformation, { "error_code": "499", "error_message": "Client disconnected the request", "error_class": "ClientDisconnected", }, ) if not error_str: error_str = "Client disconnected the request" return error_information, error_str @staticmethod def get_response_time( start_time_float: float, end_time_float: float, completion_start_time_float: float, stream: bool, ) -> float: """ Get the response time for the LLM response Args: start_time_float: float - start time of the LLM call end_time_float: float - end time of the LLM call completion_start_time_float: float - time to first token of the LLM response (for streaming responses) stream: bool - True when a stream response is returned Returns: float: The response time for the LLM response """ if stream is True: return completion_start_time_float - start_time_float else: return end_time_float - start_time_float @staticmethod def get_standard_logging_payload_trace_id( logging_obj: Logging, litellm_params: Mapping[str, Any], ) -> str: """ Returns the `litellm_trace_id` for this request This helps link sessions when multiple requests are made in a single session Gated behind `litellm.request_correlation_in_logs`: - Off (default): legacy behavior, preserved for backward compatibility - `litellm_session_id` takes priority over `litellm_trace_id` since historically this field doubled as the session-grouping field. - On: `litellm_trace_id` takes priority - trace_id and session_id are independent, see `get_standard_logging_payload_session_id` for session tracking. """ dynamic_litellm_session_id: Final = litellm_params.get("litellm_session_id") dynamic_litellm_trace_id: Final = litellm_params.get("litellm_trace_id") metadata: Final[Mapping[str, object] | None] = litellm_params.get("metadata") metadata_session_id: Final = metadata.get("session_id") if metadata else None metadata_trace_id: Final = metadata.get("trace_id") if metadata else None ordered_candidates: Final[tuple[object, object, object, object]] = ( (dynamic_litellm_trace_id, dynamic_litellm_session_id, metadata_trace_id, metadata_session_id) if litellm.request_correlation_in_logs else (dynamic_litellm_session_id, dynamic_litellm_trace_id, metadata_session_id, metadata_trace_id) ) for candidate in ordered_candidates: if candidate: return str(candidate) return logging_obj.litellm_trace_id @staticmethod def get_standard_logging_payload_session_id( logging_obj: Logging, litellm_params: Mapping[str, Any], ) -> str: """ Returns the end-user/conversation `litellm_session_id` for this request, independent of trace_id. Only populated when `litellm.request_correlation_in_logs` is enabled - off by default to avoid changing existing StandardLoggingPayload shape for callers who haven't opted in. Unlike `get_standard_logging_payload_trace_id`, this never falls back to a generated per-call trace id: it's empty when the caller never supplied a session id. """ if not litellm.request_correlation_in_logs: return "" dynamic_litellm_session_id: Final[object] = litellm_params.get("litellm_session_id") if dynamic_litellm_session_id: return str(dynamic_litellm_session_id) metadata: Final[Mapping[str, object] | None] = litellm_params.get("metadata") metadata_session_id: Final = metadata.get("session_id") if metadata else None if metadata_session_id: return str(metadata_session_id) return logging_obj.litellm_session_id @staticmethod def _get_user_agent_tags(proxy_server_request: dict) -> list[str] | None: """ Return the user agent tags from the proxy server request for spend tracking """ if litellm.disable_add_user_agent_to_request_tags is True: return None user_agent_tags: list[str] | None = None headers: Final = proxy_server_request.get("headers", {}) if headers is not None and isinstance(headers, dict): if "user-agent" in headers: user_agent: Final = headers["user-agent"] if user_agent is not None: if user_agent_tags is None: user_agent_tags = [] user_agent_part: str | None = None if "/" in user_agent: user_agent_part = user_agent.split("/")[0] if user_agent_part is not None: user_agent_tags.append("User-Agent: " + user_agent_part) if user_agent is not None: user_agent_tags.append("User-Agent: " + user_agent) return user_agent_tags @staticmethod def _get_extra_header_tags(proxy_server_request: dict) -> list[str] | None: """ Extract additional header tags for spend tracking based on config. """ extra_headers: Final[list[str]] = getattr(litellm, "extra_spend_tag_headers", None) or [] if not extra_headers: return None headers: Final = proxy_server_request.get("headers", {}) if not isinstance(headers, dict): return None header_tags: Final = [] for header_name in extra_headers: header_value = headers.get(header_name) if header_value: header_tags.append(f"{header_name}: {header_value}") return header_tags if header_tags else None @staticmethod def _get_request_tags(litellm_params: dict, proxy_server_request: dict) -> list[str]: # check for 'tags' in both 'metadata' and 'litellm_metadata' metadata: Final = litellm_params.get("metadata") or {} litellm_metadata: Final = litellm_params.get("litellm_metadata") or {} if metadata.get("tags", []): request_tags = metadata.get("tags", []).copy() elif litellm_metadata.get("tags", []): request_tags = litellm_metadata.get("tags", []).copy() else: request_tags = [] user_agent_tags: Final = StandardLoggingPayloadSetup._get_user_agent_tags(proxy_server_request) additional_header_tags: Final = StandardLoggingPayloadSetup._get_extra_header_tags(proxy_server_request) if user_agent_tags is not None: request_tags.extend(user_agent_tags) if additional_header_tags is not None: request_tags.extend(additional_header_tags) return request_tags def _get_status_fields( status: StandardLoggingPayloadStatus, guardrail_information: list[dict] | None, error_str: str | None, ) -> "StandardLoggingPayloadStatusFields": """ Determine status fields based on request status and guardrail information. Args: status: Overall request status ("success" or "failure") guardrail_information: Guardrail information from metadata error_str: Error string if any Returns: StandardLoggingPayloadStatusFields with llm_api_status and guardrail_status """ # Mapping for legacy guardrail status values to new GuardrailStatus values GUARDRAIL_STATUS_MAP: Final[dict[str, GuardrailStatus]] = { "success": "success", "blocked": "guardrail_intervened", # legacy "guardrail_intervened": "guardrail_intervened", # direct "failure": "guardrail_failed_to_respond", # legacy "guardrail_failed_to_respond": "guardrail_failed_to_respond", # direct "not_run": "not_run", } # Set LLM API status llm_api_status: Final[StandardLoggingPayloadStatus] = status ######################################################### # Map - guardrail_information.guardrail_status to guardrail_status ######################################################### guardrail_status: GuardrailStatus = "not_run" if guardrail_information and isinstance(guardrail_information, list): for information in guardrail_information: if isinstance(information, dict): raw_status = information.get("guardrail_status", "not_run") if raw_status != "not_run": guardrail_status = GUARDRAIL_STATUS_MAP.get(raw_status, "not_run") break return StandardLoggingPayloadStatusFields(llm_api_status=llm_api_status, guardrail_status=guardrail_status) def _extract_response_obj_and_hidden_params( init_response_obj: Any | BaseModel | dict, original_exception: Exception | None, ) -> tuple[dict, dict | None]: """Extract response_obj and hidden_params from init_response_obj.""" hidden_params: dict | None = None if init_response_obj is None: response_obj = {} elif isinstance(init_response_obj, BaseModel): response_obj = init_response_obj.model_dump() hidden_params = getattr(init_response_obj, "_hidden_params", None) elif isinstance(init_response_obj, dict): response_obj = init_response_obj else: response_obj = {} if original_exception is not None and hidden_params is None: response_headers: Final = _get_response_headers(original_exception) if response_headers is not None: hidden_params = dict( StandardLoggingHiddenParams( additional_headers=StandardLoggingPayloadSetup.get_additional_headers(dict(response_headers)), model_id=None, cache_key=None, api_base=None, response_cost=None, litellm_overhead_time_ms=None, batch_models=None, batch_successful_requests=None, batch_failed_requests=None, litellm_model_name=None, usage_object=None, ) ) return response_obj, hidden_params def _autorouter_savings_for_payload( request_metadata: Mapping[str, object], model: str | None, custom_llm_provider: str | None, model_id: str | None, usage_object: Mapping[str, object] | None, cost_breakdown: Mapping[str, object] | None, ) -> float | None: """The auto-router savings figure for the payload, or ``None`` when there is none. Lazy proxy import: the savings module lives with the spend trackers that own the math, and SDK-only installs have no proxy package to import. """ try: from litellm.proxy.spend_tracking.savings import autorouter_savings_for_logging_payload except Exception: # noqa: BLE001 # SDK-only install: no savings driver to run return None try: return autorouter_savings_for_logging_payload( request_metadata=request_metadata, model=model, custom_llm_provider=custom_llm_provider, model_id=model_id, usage_object=usage_object, cost_breakdown=cost_breakdown, ) except Exception as e: # noqa: BLE001 # a savings figure must never fail request logging verbose_logger.debug("autorouter savings skipped on logging payload: %s", e) return None def get_standard_logging_object_payload( kwargs: dict | None, init_response_obj: Any | BaseModel | dict, start_time: dt_object, end_time: dt_object, logging_obj: Logging, status: StandardLoggingPayloadStatus, error_str: str | None = None, original_exception: Exception | None = None, standard_built_in_tools_params: StandardBuiltInToolsParams | None = None, ) -> StandardLoggingPayload | None: try: kwargs = kwargs or {} response_obj, hidden_params = _extract_response_obj_and_hidden_params(init_response_obj, original_exception) # standardize this function to be used across, s3, dynamoDB, langfuse logging litellm_params: Final = kwargs.get("litellm_params", {}) or {} proxy_server_request: Final = litellm_params.get("proxy_server_request") or {} # Merge both litellm_metadata and metadata to get complete metadata metadata: Final[dict] = StandardLoggingPayloadSetup.merge_litellm_metadata(litellm_params) completion_start_time: Final = kwargs.get("completion_start_time", end_time) call_type: Final = kwargs.get("call_type") cache_hit: Final = kwargs.get("cache_hit", False) # Extract usage as a plain dict, avoiding Pydantic round-trip raw_usage_dict: Final = StandardLoggingPayloadSetup.get_usage_as_dict( response_obj=None if is_unbilled_non_inference_call(call_type, metadata, response_obj) else response_obj, combined_usage_object=cast(Usage | None, kwargs.get("combined_usage_object")), ) usage_dict: Final = ( {**raw_usage_dict, "output_image_count": len(init_response_obj.data)} if isinstance(init_response_obj, ImageResponse) and init_response_obj.data else raw_usage_dict ) id = response_obj.get("id", kwargs.get("litellm_call_id")) _model_id: Final = metadata.get("model_info", {}).get("id", "") _model_group: Final = metadata.get("model_group", "") request_tags: Final = StandardLoggingPayloadSetup._get_request_tags( litellm_params=litellm_params, proxy_server_request=proxy_server_request ) request_model_access_groups: Final = request_model_access_groups_from_litellm_params(litellm_params) # cleanup timestamps ( start_time_float, end_time_float, completion_start_time_float, ) = StandardLoggingPayloadSetup.cleanup_timestamps( start_time=start_time, end_time=end_time, completion_start_time=completion_start_time, ) response_time: Final = StandardLoggingPayloadSetup.get_response_time( start_time_float=start_time_float, end_time_float=end_time_float, completion_start_time_float=completion_start_time_float, stream=kwargs.get("stream", False), ) # clean up litellm metadata clean_metadata: Final = StandardLoggingPayloadSetup.get_standard_logging_metadata( metadata=metadata, litellm_params=litellm_params, prompt_integration=kwargs.get("prompt_integration", None), applied_guardrails=kwargs.get("applied_guardrails", None), mcp_tool_call_metadata=kwargs.get("mcp_tool_call_metadata", None), vector_store_request_metadata=kwargs.get("vector_store_request_metadata", None), usage_object=usage_dict, proxy_server_request=proxy_server_request, start_time=start_time, response_id=id, ) _request_body: Final = proxy_server_request.get("body", {}) end_user_id: Final = clean_metadata["user_api_key_end_user_id"] or _request_body.get( "user", None ) # maintain backwards compatibility with old request body check saved_cache_cost: float = 0.0 if cache_hit is True: id = f"{id}_cache_hit{time.time()}" # do not duplicate the request id saved_cache_cost = ( logging_obj._response_cost_calculator( result=init_response_obj, cache_hit=False, ) or 0.0 ) ## Get model cost information ## base_model = _get_base_model_from_metadata(model_call_details=kwargs) # The router overrides completion_response.model to the model-group alias before # this payload is built, so cost-map lookup via that alias always misses. # Fall back to the actual deployment model set by the router in metadata. if base_model is None: base_model = metadata.get("deployment") custom_pricing: Final = use_custom_pricing_for_model(litellm_params=litellm_params) raw_response_cost: Final = kwargs.get("response_cost") llm_response_cost: Final[float] = raw_response_cost or 0.0 guardrail_cost: Final = guardrail_information_cost(metadata.get("standard_logging_guardrail_information")) response_cost: Final[float] = llm_response_cost + guardrail_cost # clean up litellm hidden params clean_hidden_params: Final = StandardLoggingPayloadSetup.get_hidden_params(hidden_params) if clean_hidden_params["response_cost"] is None and raw_response_cost is not None: clean_hidden_params["response_cost"] = llm_response_cost model_cost_information: Final = StandardLoggingPayloadSetup.get_model_cost_information( base_model=base_model, custom_pricing=custom_pricing, custom_llm_provider=kwargs.get("custom_llm_provider"), init_response_obj=init_response_obj, api_base=litellm_params.get("api_base"), ) error_information, error_str = StandardLoggingPayloadSetup.get_error_information_for_logging_payload( metadata=metadata, original_exception=original_exception, error_str=error_str, ) ## get final response object ## final_response_obj: Final = StandardLoggingPayloadSetup.get_final_response_obj( response_obj=response_obj, init_response_obj=init_response_obj, kwargs=kwargs, ) stream: bool | None = None if ( kwargs.get("complete_streaming_response") is not None or kwargs.get("async_complete_streaming_response") is not None ) and kwargs.get("stream") is True: stream = True # Reconstruct full model name with provider prefix for logging # This ensures Bedrock models like "us.anthropic.claude-3-5-sonnet-20240620-v1:0" # are logged as "bedrock/us.anthropic.claude-3-5-sonnet-20240620-v1:0" custom_llm_provider: Final = cast(str | None, kwargs.get("custom_llm_provider")) model_name = reconstruct_model_name(kwargs.get("model", "") or "", custom_llm_provider, metadata) response_model_name: str | None = None if isinstance(final_response_obj, dict): response_model_name = final_response_obj.get("model") # For Azure Model Router, preserve the actual model in the top-level standard # logging payload. from litellm.llms.azure_ai.common_utils import AzureFoundryModelInfo requested_model: Final = kwargs.get("model") stamped_selected_model: Final = AzureFoundryModelInfo.get_model_router_selected_model(hidden_params) if stamped_selected_model is not None: model_name = stamped_selected_model elif ( AzureFoundryModelInfo.is_model_router_call(model=requested_model, hidden_params=hidden_params) and isinstance(response_model_name, str) and response_model_name ): model_name = response_model_name request_cost_breakdown: Final = cost_breakdown_with_guardrail(logging_obj.cost_breakdown, guardrail_cost) autorouter_savings: Final = _autorouter_savings_for_payload( request_metadata=metadata, model=model_name, custom_llm_provider=custom_llm_provider, model_id=_model_id, usage_object=usage_dict, cost_breakdown=request_cost_breakdown, ) payload: Final[StandardLoggingPayload] = StandardLoggingPayload( id=str(id), litellm_call_id=kwargs.get("litellm_call_id") or litellm_params.get("litellm_call_id"), trace_id=StandardLoggingPayloadSetup.get_standard_logging_payload_trace_id( logging_obj=logging_obj, litellm_params=litellm_params, ), session_id=StandardLoggingPayloadSetup.get_standard_logging_payload_session_id( logging_obj=logging_obj, litellm_params=litellm_params, ), call_type=call_type or "", cache_hit=cache_hit, stream=stream, status=status, status_fields=_get_status_fields( status=status, guardrail_information=metadata.get("standard_logging_guardrail_information", None), error_str=error_str, ), custom_llm_provider=custom_llm_provider, saved_cache_cost=saved_cache_cost, startTime=start_time_float, endTime=end_time_float, completionStartTime=completion_start_time_float, response_time=response_time, model=model_name, metadata=clean_metadata, cache_key=clean_hidden_params["cache_key"], response_cost=response_cost, cost_breakdown=request_cost_breakdown, autorouter_savings=autorouter_savings, total_tokens=usage_dict.get("total_tokens", 0), prompt_tokens=usage_dict.get("prompt_tokens", 0), completion_tokens=usage_dict.get("completion_tokens", 0), request_tags=request_tags, request_model_access_groups=request_model_access_groups, end_user=end_user_id, api_base=StandardLoggingPayloadSetup.strip_trailing_slash(litellm_params.get("api_base", "")) or "", model_group=_model_group, model_id=_model_id, requester_ip_address=clean_metadata.get("requester_ip_address", None), user_agent=clean_metadata.get("user_agent", None), messages=truncate_base64_in_messages( StandardLoggingPayloadSetup.append_system_prompt_messages( kwargs=kwargs, messages=kwargs.get("messages") ) ), response=final_response_obj, model_parameters=ModelParamHelper.get_standard_logging_model_parameters( kwargs.get("optional_params", None) or {} ), hidden_params=clean_hidden_params, model_map_information=model_cost_information, error_str=error_str, error_information=error_information, response_cost_failure_debug_info=kwargs.get("response_cost_failure_debug_information"), guardrail_information=metadata.get("standard_logging_guardrail_information", None), standard_built_in_tools_params=standard_built_in_tools_params, ) # emit_standard_logging_payload(payload) - Moved to success_handler to prevent double emitting return payload except Exception as e: verbose_logger.exception("Error creating standard logging object - %s", e) return None def emit_standard_logging_payload(payload: StandardLoggingPayload): if os.getenv("LITELLM_PRINT_STANDARD_LOGGING_PAYLOAD"): print(json.dumps(payload, indent=4), flush=True) # noqa: T201 def get_standard_logging_metadata( metadata: dict[str, Any] | None, ) -> StandardLoggingMetadata: """ Clean and filter the metadata dictionary to include only the specified keys in StandardLoggingMetadata. Args: metadata (Optional[Dict[str, Any]]): The original metadata dictionary. Returns: StandardLoggingMetadata: A StandardLoggingMetadata object containing the cleaned metadata. Note: - If the input metadata is None or not a dictionary, an empty StandardLoggingMetadata object is returned. - If 'user_api_key' is present in metadata and is a valid SHA256 hash, it's stored as 'user_api_key_hash'. """ # Initialize with default values clean_metadata: Final = StandardLoggingMetadata( user_api_key_hash=None, user_api_key_alias=None, user_api_key_spend=None, user_api_key_max_budget=None, user_api_key_budget_reset_at=None, user_api_key_user_spend=None, user_api_key_user_max_budget=None, user_api_key_team_spend=None, user_api_key_team_max_budget=None, user_api_key_team_id=None, user_api_key_org_id=None, user_api_key_org_alias=None, user_api_key_project_id=None, user_api_key_project_alias=None, user_api_key_user_id=None, user_api_key_user_email=None, user_api_key_team_alias=None, spend_logs_metadata=None, requester_ip_address=None, user_agent=None, requester_metadata=None, user_api_key_end_user_id=None, prompt_management_metadata=None, applied_guardrails=None, mcp_tool_call_metadata=None, vector_store_request_metadata=None, routing_decision=None, usage_object=None, requester_custom_headers=None, user_api_key_request_route=None, cold_storage_object_key=None, user_api_key_auth_metadata=None, team_alias=None, team_id=None, ) if isinstance(metadata, dict): # Update the clean_metadata with values from input metadata that match StandardLoggingMetadata fields for key in StandardLoggingMetadata.__annotations__: if key in metadata: clean_metadata[key] = metadata[key] if metadata.get("user_api_key") is not None: if is_valid_sha256_hash(str(metadata.get("user_api_key"))): clean_metadata["user_api_key_hash"] = metadata.get("user_api_key") # this is the hash return clean_metadata def scrub_sensitive_keys_in_metadata(litellm_params: dict | None): if litellm_params is None: litellm_params = {} metadata: Final = litellm_params.get("metadata", {}) or {} ## Extract provider-specific callable values (like langfuse_masking_function) ## Store them separately so only the intended logger can access them ## This prevents callables from leaking to other logging integrations if "langfuse_masking_function" in metadata: masking_fn: Final = metadata.pop("langfuse_masking_function", None) if callable(masking_fn): litellm_params["_langfuse_masking_function"] = masking_fn litellm_params["metadata"] = metadata return litellm_params # integration helper function def modify_integration(integration_name, integration_params): global supabaseClient if integration_name == "supabase": if "table_name" in integration_params: Supabase.supabase_table_name = integration_params["table_name"] @lru_cache(maxsize=16) def _get_traceback_str_for_error(error_str: str) -> str: """ function wrapped with lru_cache to limit the number of times `traceback.format_exc()` is called """ return traceback.format_exc() from decimal import Decimal # used for unit testing from typing import Any, Optional, Union def create_dummy_standard_logging_payload() -> StandardLoggingPayload: # First create the nested objects with proper typing model_info: Final = StandardLoggingModelInformation(model_map_key="gpt-3.5-turbo", model_map_value=None) metadata: Final = StandardLoggingMetadata( user_api_key_hash="test_hash", user_api_key_alias="test_alias", user_api_key_team_id="test_team", user_api_key_user_id="test_user", user_api_key_team_alias="test_team_alias", user_api_key_user_spend=None, user_api_key_user_max_budget=None, user_api_key_team_spend=None, user_api_key_team_max_budget=None, user_api_key_org_id=None, spend_logs_metadata=None, requester_ip_address="127.0.0.1", requester_metadata=None, user_api_key_end_user_id="test_end_user", ) hidden_params: Final = StandardLoggingHiddenParams( model_id=None, cache_key=None, api_base=None, response_cost=None, additional_headers=None, litellm_overhead_time_ms=None, batch_models=None, batch_successful_requests=None, batch_failed_requests=None, litellm_model_name=None, usage_object=None, ) # Convert numeric values to appropriate types response_cost: Final = Decimal("0.1") start_time: Final = Decimal("1234567890.0") end_time: Final = Decimal("1234567891.0") completion_start_time: Final = Decimal("1234567890.5") saved_cache_cost: Final = Decimal("0.0") # Create messages and response with proper typing messages: Final[list[dict[str, str]]] = [{"role": "user", "content": "Hello, world!"}] response: Final[dict[str, list[dict[str, dict[str, str]]]]] = {"choices": [{"message": {"content": "Hi there!"}}]} # Main payload initialization return StandardLoggingPayload( id="test_id", call_type="completion", stream=False, response_cost=response_cost, autorouter_savings=None, response_cost_failure_debug_info=None, status="success", total_tokens=int(DEFAULT_MOCK_RESPONSE_PROMPT_TOKEN_COUNT + DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT), prompt_tokens=int(DEFAULT_MOCK_RESPONSE_PROMPT_TOKEN_COUNT), completion_tokens=int(DEFAULT_MOCK_RESPONSE_COMPLETION_TOKEN_COUNT), startTime=start_time, endTime=end_time, completionStartTime=completion_start_time, model_map_information=model_info, model="gpt-3.5-turbo", model_id="model-123", model_group="openai-gpt", custom_llm_provider="openai", api_base="https://api.openai.com", metadata=metadata, cache_hit=False, cache_key=None, saved_cache_cost=saved_cache_cost, request_tags=[], request_model_access_groups=(), end_user=None, requester_ip_address="127.0.0.1", messages=messages, response=response, error_str=None, model_parameters={"stream": True}, hidden_params=hidden_params, )