([^<]+)")
+_TRANSIENT_ERRORS: Final = (httpx.TransportError, LiteLLMTimeout)
+_UPLOAD_ATTEMPTS: Final = 3
+_UPLOAD_ATTEMPT_TIMEOUT: Final = httpx.Timeout(S3_UPLOAD_ATTEMPT_TIMEOUT_SECONDS)
+
+PutOutcome: TypeAlias = httpx.Response | Exception
@dataclass(frozen=True, slots=True)
@@ -105,6 +113,35 @@ def _s3_key_parent(s3_object_key: str) -> str:
return s3_object_key.rsplit("/", 1)[0] if "/" in s3_object_key else ""
+def _backoff_seconds(attempt: int) -> float:
+ return (1 << attempt) * (0.5 + random.random())
+
+
+def _log_retry(level: int, attempt: int, reason: str, wait: float, s3_object_key: str) -> None:
+ verbose_logger.log(
+ level,
+ "S3 upload attempt %s/%s failed with %s, retrying in %.2fs key=%s",
+ attempt + 1,
+ _UPLOAD_ATTEMPTS,
+ reason,
+ wait,
+ s3_object_key,
+ )
+
+
+def _response_or_raise(outcome: PutOutcome) -> httpx.Response:
+ if isinstance(outcome, Exception):
+ raise outcome
+ outcome.raise_for_status()
+ return outcome
+
+
+def _throttled(outcome: PutOutcome) -> bool:
+ if isinstance(outcome, Exception):
+ return True
+ return outcome.status_code in (429, 503) or _s3_error_code(outcome) == "SlowDown"
+
+
class S3BatchUploadError(Exception):
def __init__(self, failed: int, total: int) -> None:
self.failed = failed
@@ -566,7 +603,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
url: Final = self._build_object_url(batch_logging_element.s3_object_key)
- async def signed_put(prepared: _PreparedPut) -> httpx.Response:
+ async def signed_put(prepared: _PreparedPut) -> PutOutcome:
credentials: Final = await asyncified_get_credentials(
aws_access_key_id=self.s3_aws_access_key_id,
aws_secret_access_key=self.s3_aws_secret_access_key,
@@ -582,34 +619,16 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
self._sign_put, credentials, url, prepared.json_string, prepared.headers
)
try:
- return await self.async_httpx_client.put(url, data=prepared.json_string, headers=signed_headers)
+ return await self.async_httpx_client.put(
+ url, data=prepared.json_string, headers=signed_headers, timeout=_UPLOAD_ATTEMPT_TIMEOUT
+ )
except httpx.HTTPStatusError as error:
return error.response
+ except _TRANSIENT_ERRORS as error:
+ return error
- max_retries: Final = 3
prepared: Final = self._prepare_put(batch_logging_element)
- for attempt in range(max_retries):
- response = await self._recorded_put(partial(signed_put, prepared))
- if (
- response.status_code in _RETRYABLE_STATUSES
- and not (self.s3_drop_on_terminal_error and _is_terminal(response))
- and attempt < max_retries - 1
- ):
- wait_time = 1 << attempt # 1s, 2s
- verbose_logger.log(
- logging.DEBUG if _in_flush.get() else logging.WARNING,
- "S3 upload returned %s, retrying in %ss (attempt %s/%s) key=%s",
- response.status_code,
- wait_time,
- attempt + 1,
- max_retries,
- batch_logging_element.s3_object_key,
- )
- self._flush_retries += 1
- await asyncio.sleep(wait_time)
- continue
- response.raise_for_status()
- break
+ await self._put_with_retries(partial(signed_put, prepared), batch_logging_element.s3_object_key)
except Exception as e:
verbose_logger.exception("Error uploading to s3: %s", e)
self.handle_callback_failure(callback_name="S3Logger")
@@ -678,9 +697,13 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
)
dropped: Final = len(failed) - len(requeued)
if dropped:
+ terminal: Final = sum(1 for _, outcome in failed if outcome == "dropped")
verbose_logger.warning(
- "s3 logging: %s uploads dropped (terminal or retrying longer than s3_max_retry_age_seconds=%s)",
+ "s3 logging: %s uploads dropped this flush: %s rejected as terminal by S3, "
+ "%s retrying longer than s3_max_retry_age_seconds=%s",
dropped,
+ terminal,
+ dropped - terminal,
self.s3_max_retry_age_seconds,
)
if not requeued:
@@ -728,22 +751,46 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
return "dropped"
return "retry"
- async def _recorded_put(self, signed_put: Callable[[], Awaitable[httpx.Response]]) -> httpx.Response:
+ async def _recorded_put(self, signed_put: Callable[[], Awaitable[PutOutcome]]) -> PutOutcome:
+ outcome: Final = await signed_put()
limiter: Final = self._upload_limiter
- adaptive: Final = limiter if isinstance(limiter, AdaptiveConcurrencyLimiter) else None
- try:
- response: Final = await signed_put()
- except Exception:
- if adaptive is not None:
- adaptive.record(PutSample(throttled=True))
- raise
- if adaptive is not None:
- adaptive.record(
- PutSample(
- throttled=response.status_code in (429, 503) or _s3_error_code(response) == "SlowDown",
- )
- )
- return response
+ if isinstance(limiter, AdaptiveConcurrencyLimiter):
+ limiter.record(PutSample(throttled=_throttled(outcome)))
+ return outcome
+
+ def _retry_reason(self, outcome: PutOutcome) -> str | None:
+ if isinstance(outcome, Exception):
+ return type(outcome).__name__
+ if outcome.status_code not in _RETRYABLE_STATUSES:
+ return None
+ if self.s3_drop_on_terminal_error and _is_terminal(outcome):
+ return None
+ return str(outcome.status_code)
+
+ async def _put_with_retries(
+ self, signed_put: Callable[[], Awaitable[PutOutcome]], s3_object_key: str
+ ) -> httpx.Response:
+ for attempt in range(_UPLOAD_ATTEMPTS - 1):
+ outcome = await self._recorded_put(signed_put)
+ reason = self._retry_reason(outcome)
+ if reason is None:
+ return _response_or_raise(outcome)
+ wait = _backoff_seconds(attempt)
+ _log_retry(logging.DEBUG if _in_flush.get() else logging.WARNING, attempt, reason, wait, s3_object_key)
+ self._flush_retries += 1
+ await asyncio.sleep(wait)
+ return _response_or_raise(await self._recorded_put(signed_put))
+
+ def _sync_put_with_retries(self, signed_put: Callable[[], PutOutcome], s3_object_key: str) -> httpx.Response:
+ for attempt in range(_UPLOAD_ATTEMPTS - 1):
+ outcome = signed_put()
+ reason = self._retry_reason(outcome)
+ if reason is None:
+ return _response_or_raise(outcome)
+ wait = _backoff_seconds(attempt)
+ _log_retry(logging.WARNING, attempt, reason, wait, s3_object_key)
+ time.sleep(wait)
+ return _response_or_raise(signed_put())
def _batch_file_elements(self, batch: tuple[s3BatchLoggingElement, ...]) -> tuple[s3BatchLoggingElement, ...]:
now: Final = datetime.now(timezone.utc)
@@ -879,7 +926,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
params=({"ssl_verify": self.s3_verify} if self.s3_verify is not None else None)
)
- def signed_put(prepared_put: _PreparedPut) -> httpx.Response:
+ def signed_put(prepared_put: _PreparedPut) -> PutOutcome:
credentials: Final = self.get_credentials(
aws_access_key_id=self.s3_aws_access_key_id,
aws_secret_access_key=self.s3_aws_secret_access_key,
@@ -887,29 +934,14 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
aws_region_name=self.s3_region_name,
)
signed_headers: Final = self._sign_put(credentials, url, prepared_put.json_string, prepared_put.headers)
- return httpx_client.put(url, data=prepared_put.json_string, headers=signed_headers)
-
- max_retries: Final = 3
- for attempt in range(max_retries):
- response = signed_put(prepared)
- if (
- response.status_code in _RETRYABLE_STATUSES
- and not (self.s3_drop_on_terminal_error and _is_terminal(response))
- and attempt < max_retries - 1
- ):
- wait_time = 1 << attempt # 1s, 2s
- verbose_logger.warning(
- "S3 upload returned %s, retrying in %ss (attempt %s/%s) key=%s",
- response.status_code,
- wait_time,
- attempt + 1,
- max_retries,
- batch_logging_element.s3_object_key,
+ try:
+ return httpx_client.put(
+ url, data=prepared_put.json_string, headers=signed_headers, timeout=_UPLOAD_ATTEMPT_TIMEOUT
)
- time.sleep(wait_time)
- continue
- response.raise_for_status()
- break
+ except _TRANSIENT_ERRORS as error:
+ return error
+
+ self._sync_put_with_retries(partial(signed_put, prepared), batch_logging_element.s3_object_key)
except Exception as e:
verbose_logger.exception("Error uploading to s3: %s", e)
self.handle_callback_failure(callback_name="S3Logger")
diff --git a/litellm/litellm_core_utils/env_utils.py b/litellm/litellm_core_utils/env_utils.py
index d641884b4cd..6883af98bbd 100644
--- a/litellm/litellm_core_utils/env_utils.py
+++ b/litellm/litellm_core_utils/env_utils.py
@@ -4,23 +4,65 @@ Utility helpers for reading and parsing environment variables.
import logging
import os
-from typing import Final
+from typing import Final, TypeVar
+
+_NumberT: Final = TypeVar("_NumberT", int, float)
+
+# Marked before logging: the LiteLLM loggers' own filters read env knobs through these helpers, so
+# warning first would re-enter the helper from inside the log call and recurse until RecursionError.
+_reported_unparsable_values: Final[set[tuple[str, str]]] = set() # mutable-ok: process-wide warn-once registry
+
+
+def _warn_unparsable_once(env_var: str, raw: str, number_type: type[int] | type[float]) -> None:
+ key: Final = (env_var, raw)
+ if key in _reported_unparsable_values:
+ return
+ _reported_unparsable_values.add(key)
+ logging.getLogger("LiteLLM").warning(
+ "%s=%r is not a valid %s. Treating it as unset.", env_var, raw, number_type.__name__
+ )
+
+
+def _get_env_number_or_none(env_var: str, number_type: type[_NumberT]) -> _NumberT | None:
+ raw: Final = os.getenv(env_var)
+ if raw is None or not raw.strip():
+ return None
+ try:
+ return number_type(raw.strip())
+ except ValueError:
+ _warn_unparsable_once(env_var, raw, number_type)
+ return None
+
+
+def get_env_int_or_none(env_var: str) -> int | None:
+ """Parse an environment variable as an integer, returning None when it is unset, blank or unparsable.
+
+ Use this instead of `get_env_int` when callers must distinguish "explicitly configured"
+ from "left at the default", for example when an override should take precedence over a
+ value resolved from somewhere else.
+ """
+ return _get_env_number_or_none(env_var, int)
+
+
+def get_env_float_or_none(env_var: str) -> float | None:
+ """Float counterpart of `get_env_int_or_none`."""
+ return _get_env_number_or_none(env_var, float)
def get_env_int(env_var: str, default: int) -> int:
- """Parse an environment variable as an integer, falling back to default on invalid values.
+ """Parse an environment variable as an integer, falling back to default when it is unset, blank or unparsable.
- Handles empty strings, whitespace, and non-numeric values gracefully
- so that misconfiguration doesn't crash the process at import time.
+ An unparsable value is warned about once, so misconfiguration neither crashes the process at
+ import time nor goes unnoticed.
"""
- raw = os.getenv(env_var)
- if raw is None:
- return default
- raw = raw.strip()
- try:
- return int(raw)
- except (ValueError, TypeError):
- return default
+ value: Final = get_env_int_or_none(env_var)
+ return default if value is None else value
+
+
+def get_env_float(env_var: str, default: float) -> float:
+ """Float counterpart of `get_env_int`."""
+ value: Final = get_env_float_or_none(env_var)
+ return default if value is None else value
def get_env_int_in_range(env_var: str, default: int, minimum: int, maximum: int) -> int:
@@ -41,19 +83,3 @@ def get_env_int_in_range(env_var: str, default: int, minimum: int, maximum: int)
default,
)
return default
-
-
-def get_env_int_or_none(env_var: str) -> int | None:
- """Parse an environment variable as an integer, returning None when it is unset or unusable.
-
- Use this instead of `get_env_int` when callers must distinguish "explicitly configured"
- from "left at the default", for example when an override should take precedence over a
- value resolved from somewhere else.
- """
- raw: Final = os.getenv(env_var)
- if raw is None:
- return None
- try:
- return int(raw.strip())
- except (ValueError, TypeError):
- return None
diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py
index 7b965dea28a..9b10b95956f 100644
--- a/litellm/litellm_core_utils/litellm_logging.py
+++ b/litellm/litellm_core_utils/litellm_logging.py
@@ -22,7 +22,6 @@ from httpx import Response
from pydantic import BaseModel, JsonValue
import litellm
-from litellm import _custom_logger_compatible_callbacks_literal
from litellm._internal_context import post_response_phase
from litellm._logging import (
is_debugging_on,
@@ -652,6 +651,10 @@ class Logging(LiteLLMLoggingBaseClass):
self.stream = stream
self.start_time = start_time # log the call start time
self.call_type = call_type
+ self.relays_llm_passthrough: Final = call_type in (
+ CallTypes.llm_passthrough_route.value,
+ CallTypes.allm_passthrough_route.value,
+ )
self.litellm_call_id = litellm_call_id
self.litellm_trace_id: str = litellm_trace_id if litellm_trace_id else str(uuid.uuid4())
@@ -2389,10 +2392,7 @@ class Logging(LiteLLMLoggingBaseClass):
service_tier=ws_service_tier,
)
- elif (
- self.call_type == CallTypes.llm_passthrough_route.value
- or self.call_type == CallTypes.allm_passthrough_route.value
- ) and isinstance(result, Response):
+ elif self.relays_llm_passthrough and isinstance(result, Response):
from litellm.utils import ProviderConfigManager
provider_config: Final = ProviderConfigManager.get_provider_passthrough_config(
@@ -2589,7 +2589,9 @@ class Logging(LiteLLMLoggingBaseClass):
logging_result: Final = self.normalize_logging_result(result=result)
- if isinstance(result, Response) and isinstance(logging_result, (ModelResponse, EmbeddingResponse)):
+ if isinstance(result, Response) and isinstance(
+ logging_result, (ModelResponse, EmbeddingResponse, TextCompletionResponse)
+ ):
result = logging_result
if standard_logging_object is None and result is not None and self.stream is not True:
@@ -4575,712 +4577,738 @@ def set_callbacks(
raise e
+def build_custom_logger_compatible_class(
+ logging_integration: str,
+ internal_usage_cache: DualCache | None,
+ llm_router: object,
+ custom_logger_init_args: dict[str, object] | None = None,
+) -> CustomLogger | None:
+ custom_logger_init_args_value: Final[dict[str, object]] = 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_value.get("dd_api_key")
+ _dd_site: Final = custom_logger_init_args_value.get("dd_site")
+ _dd_agent_host: Final = custom_logger_init_args_value.get("dd_agent_host")
+ _dd_agent_port: Final = custom_logger_init_args_value.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=cast( # cast-ok: callback options come from dynamic config
+ StandardCallbackDynamicParams, custom_logger_init_args_value
+ ),
+ 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 == "clickhouse":
+ for callback in _in_memory_loggers:
+ if isinstance(callback, ClickHouseSpendLogger):
+ return callback
+
+ _clickhouse_spend_logger: Final = ClickHouseSpendLogger()
+ _in_memory_loggers.append(_clickhouse_spend_logger)
+ return _clickhouse_spend_logger
+ elif logging_integration == "pointfive":
+ for callback in _in_memory_loggers:
+ if isinstance(callback, PointFiveLogger):
+ return callback
+
+ _pointfive_logger: Final = PointFiveLogger()
+ _in_memory_loggers.append(_pointfive_logger)
+ return _pointfive_logger
+ elif logging_integration == "zerobus":
+ for callback in _in_memory_loggers:
+ if isinstance(callback, ZerobusLogger):
+ return callback
+
+ _zerobus_logger: Final = ZerobusLogger()
+ _in_memory_loggers.append(_zerobus_logger)
+ return _zerobus_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, build_otel_v2_logger
+ from litellm.integrations.otel.model.config import OpenTelemetryV2Config
+
+ for callback in _in_memory_loggers:
+ if isinstance(callback, OpenTelemetryV2):
+ return callback
+ otel_settings: Final = _get_custom_logger_settings_from_proxy_server(callback_name=logging_integration)
+ otel_logger_v2: Final = build_otel_v2_logger(
+ config=OpenTelemetryV2Config(**otel_settings), settings=otel_settings
+ )
+ _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 == "ternary":
+ from litellm.integrations.ternary.ternary_logger import TernaryLogger
+
+ for callback in _in_memory_loggers:
+ if isinstance(callback, TernaryLogger):
+ return callback
+ ternary_logger: Final = TernaryLogger()
+ _in_memory_loggers.append(ternary_logger)
+ return ternary_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: Final = 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=langtrace_trace_endpoint(os.getenv("LANGTRACE_API_HOST")),
+ headers=f"x-api-key={os.environ['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 == "signoz":
+ from litellm.integrations.otel.presets.signoz import (
+ SIGNOZ_INGESTION_ENDPOINT_ENV,
+ )
+
+ _signoz_endpoint: Final = os.getenv(SIGNOZ_INGESTION_ENDPOINT_ENV)
+ if not _signoz_endpoint:
+ raise ValueError(f"{SIGNOZ_INGESTION_ENDPOINT_ENV} not found in environment variables")
+
+ _signoz_v2: Final = _maybe_construct_otel_v2("signoz", _in_memory_loggers)
+ if _signoz_v2 is not None:
+ return _signoz_v2
+
+ from litellm.integrations.opentelemetry import (
+ OpenTelemetry,
+ OpenTelemetryConfig,
+ )
+
+ _signoz_base: Final = _signoz_endpoint.rstrip("/")
+ _signoz_key: Final = os.getenv("SIGNOZ_INGESTION_KEY")
+ _signoz_config: Final = OpenTelemetryConfig(
+ exporter="otlp_http",
+ endpoint=(_signoz_base if _signoz_base.endswith("/v1/traces") else f"{_signoz_base}/v1/traces"),
+ headers=(f"signoz-ingestion-key={_signoz_key}" if _signoz_key else None),
+ )
+ for callback in _in_memory_loggers:
+ if isinstance(callback, OpenTelemetry) and callback.callback_name == "signoz":
+ return callback
+ _signoz_logger: Final = OpenTelemetry(config=_signoz_config, callback_name="signoz")
+ _in_memory_loggers.append(_signoz_logger)
+ return _signoz_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":
+ pagerduty_loggers: Final = _enterprise_alerting_loggers()
+ for callback in _in_memory_loggers:
+ if isinstance(callback, pagerduty_loggers.pagerduty):
+ return callback
+ pagerduty_logger: Final = pagerduty_loggers.pagerduty(**custom_logger_init_args_value)
+ _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":
+ resend_email_loggers: Final = _enterprise_alerting_loggers()
+ for callback in _in_memory_loggers:
+ if isinstance(callback, resend_email_loggers.resend_email):
+ return callback
+ resend_email_logger: Final = resend_email_loggers.resend_email()
+ _in_memory_loggers.append(resend_email_logger)
+ return resend_email_logger
+ elif logging_integration == "sendgrid_email":
+ sendgrid_email_loggers: Final = _enterprise_alerting_loggers()
+ for callback in _in_memory_loggers:
+ if isinstance(callback, sendgrid_email_loggers.sendgrid_email):
+ return callback
+ sendgrid_email_logger: Final = sendgrid_email_loggers.sendgrid_email()
+ _in_memory_loggers.append(sendgrid_email_logger)
+ return sendgrid_email_logger
+ elif logging_integration == "smtp_email":
+ smtp_email_loggers: Final = _enterprise_alerting_loggers()
+ for callback in _in_memory_loggers:
+ if isinstance(callback, smtp_email_loggers.smtp_email):
+ return callback
+ smtp_email_logger: Final = smtp_email_loggers.smtp_email()
+ _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_value.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=cast( # cast-ok: callback options come from dynamic config
+ StandardCallbackDynamicParams, custom_logger_init_args_value
+ ),
+ 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
+
+
def _init_custom_logger_compatible_class(
- logging_integration: _custom_logger_compatible_callbacks_literal,
+ logging_integration: str,
internal_usage_cache: DualCache | None,
llm_router: object,
custom_logger_init_args: dict[str, object] | None = {},
+ raise_import_errors: bool = False,
) -> CustomLogger | None:
"""
Initialize a custom logger compatible class
"""
try:
- custom_logger_init_args_value: Final[dict[str, object]] = 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_value.get("dd_api_key")
- _dd_site: Final = custom_logger_init_args_value.get("dd_site")
- _dd_agent_host: Final = custom_logger_init_args_value.get("dd_agent_host")
- _dd_agent_port: Final = custom_logger_init_args_value.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=cast( # cast-ok: callback options come from dynamic config
- StandardCallbackDynamicParams, custom_logger_init_args_value
- ),
- 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 == "clickhouse":
- for callback in _in_memory_loggers:
- if isinstance(callback, ClickHouseSpendLogger):
- return callback
-
- _clickhouse_spend_logger: Final = ClickHouseSpendLogger()
- _in_memory_loggers.append(_clickhouse_spend_logger)
- return _clickhouse_spend_logger
- elif logging_integration == "pointfive":
- for callback in _in_memory_loggers:
- if isinstance(callback, PointFiveLogger):
- return callback
-
- _pointfive_logger: Final = PointFiveLogger()
- _in_memory_loggers.append(_pointfive_logger)
- return _pointfive_logger
- elif logging_integration == "zerobus":
- for callback in _in_memory_loggers:
- if isinstance(callback, ZerobusLogger):
- return callback
-
- _zerobus_logger: Final = ZerobusLogger()
- _in_memory_loggers.append(_zerobus_logger)
- return _zerobus_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, build_otel_v2_logger
- from litellm.integrations.otel.model.config import OpenTelemetryV2Config
-
- for callback in _in_memory_loggers:
- if isinstance(callback, OpenTelemetryV2):
- return callback
- otel_settings: Final = _get_custom_logger_settings_from_proxy_server(callback_name=logging_integration)
- otel_logger_v2: Final = build_otel_v2_logger(
- config=OpenTelemetryV2Config(**otel_settings), settings=otel_settings
- )
- _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 == "ternary":
- from litellm.integrations.ternary.ternary_logger import TernaryLogger
-
- for callback in _in_memory_loggers:
- if isinstance(callback, TernaryLogger):
- return callback
- ternary_logger: Final = TernaryLogger()
- _in_memory_loggers.append(ternary_logger)
- return ternary_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: Final = 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=langtrace_trace_endpoint(os.getenv("LANGTRACE_API_HOST")),
- headers=f"x-api-key={os.environ['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 == "signoz":
- from litellm.integrations.otel.presets.signoz import (
- SIGNOZ_INGESTION_ENDPOINT_ENV,
- )
-
- _signoz_endpoint: Final = os.getenv(SIGNOZ_INGESTION_ENDPOINT_ENV)
- if not _signoz_endpoint:
- raise ValueError(f"{SIGNOZ_INGESTION_ENDPOINT_ENV} not found in environment variables")
-
- _signoz_v2: Final = _maybe_construct_otel_v2("signoz", _in_memory_loggers)
- if _signoz_v2 is not None:
- return _signoz_v2
-
- from litellm.integrations.opentelemetry import (
- OpenTelemetry,
- OpenTelemetryConfig,
- )
-
- _signoz_base: Final = _signoz_endpoint.rstrip("/")
- _signoz_key: Final = os.getenv("SIGNOZ_INGESTION_KEY")
- _signoz_config: Final = OpenTelemetryConfig(
- exporter="otlp_http",
- endpoint=(_signoz_base if _signoz_base.endswith("/v1/traces") else f"{_signoz_base}/v1/traces"),
- headers=(f"signoz-ingestion-key={_signoz_key}" if _signoz_key else None),
- )
- for callback in _in_memory_loggers:
- if isinstance(callback, OpenTelemetry) and callback.callback_name == "signoz":
- return callback
- _signoz_logger: Final = OpenTelemetry(config=_signoz_config, callback_name="signoz")
- _in_memory_loggers.append(_signoz_logger)
- return _signoz_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":
- pagerduty_loggers: Final = _enterprise_alerting_loggers()
- for callback in _in_memory_loggers:
- if isinstance(callback, pagerduty_loggers.pagerduty):
- return callback
- pagerduty_logger: Final = pagerduty_loggers.pagerduty(**custom_logger_init_args_value)
- _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":
- resend_email_loggers: Final = _enterprise_alerting_loggers()
- for callback in _in_memory_loggers:
- if isinstance(callback, resend_email_loggers.resend_email):
- return callback
- resend_email_logger: Final = resend_email_loggers.resend_email()
- _in_memory_loggers.append(resend_email_logger)
- return resend_email_logger
- elif logging_integration == "sendgrid_email":
- sendgrid_email_loggers: Final = _enterprise_alerting_loggers()
- for callback in _in_memory_loggers:
- if isinstance(callback, sendgrid_email_loggers.sendgrid_email):
- return callback
- sendgrid_email_logger: Final = sendgrid_email_loggers.sendgrid_email()
- _in_memory_loggers.append(sendgrid_email_logger)
- return sendgrid_email_logger
- elif logging_integration == "smtp_email":
- smtp_email_loggers: Final = _enterprise_alerting_loggers()
- for callback in _in_memory_loggers:
- if isinstance(callback, smtp_email_loggers.smtp_email):
- return callback
- smtp_email_logger: Final = smtp_email_loggers.smtp_email()
- _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_value.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=cast( # cast-ok: callback options come from dynamic config
- StandardCallbackDynamicParams, custom_logger_init_args_value
- ),
- 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
+ return build_custom_logger_compatible_class(
+ logging_integration,
+ internal_usage_cache=internal_usage_cache,
+ llm_router=llm_router,
+ custom_logger_init_args=custom_logger_init_args,
+ )
except Exception as e:
+ if raise_import_errors and isinstance(e, ImportError):
+ raise ImportError(
+ f"Could not build the '{logging_integration}' callback configured in litellm_settings: {e}"
+ ) from e
verbose_logger.exception("[Non-Blocking Error] Error initializing custom logger: %s", e)
return None
- return None
+
+
+def init_custom_logger_compatible_class_at_startup(
+ callback: str,
+ internal_usage_cache: DualCache | None,
+ llm_router: object,
+) -> CustomLogger | None:
+ return _init_custom_logger_compatible_class(
+ callback, internal_usage_cache=internal_usage_cache, llm_router=llm_router, raise_import_errors=True
+ )
def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[CustomLogger]) -> "OpenTelemetryV2 | None":
diff --git a/litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py b/litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py
index 6f16e1e3fb4..ad33c47fc51 100644
--- a/litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py
+++ b/litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py
@@ -6,7 +6,7 @@ import traceback
from collections.abc import Mapping, Sequence
from typing import Final, Literal, cast
-from pydantic import ConfigDict, TypeAdapter, with_config
+from pydantic import ConfigDict, TypeAdapter, ValidationError, with_config
from typing_extensions import ReadOnly, TypedDict
import litellm
@@ -222,6 +222,9 @@ async def convert_to_streaming_response_async(
role=choice["message"]["role"],
function_call=choice["message"].get("function_call", None),
tool_calls=choice["message"].get("tool_calls", None),
+ provider_specific_fields=_cached_provider_specific_fields(
+ _CACHED_MESSAGE_PROVIDER_FIELDS.validate_python(choice["message"])
+ ),
)
finish_reason = choice.get("finish_reason", None)
@@ -243,6 +246,9 @@ async def convert_to_streaming_response_async(
completion_tokens=response_object["usage"].get("completion_tokens", 0),
prompt_tokens=response_object["usage"].get("prompt_tokens", 0),
total_tokens=response_object["usage"].get("total_tokens", 0),
+ completion_tokens_details=_cached_completion_tokens_details(
+ _CACHED_USAGE_DETAILS.validate_python(response_object["usage"])
+ ),
),
)
@@ -334,6 +340,9 @@ def convert_to_streaming_response(
model_response_object.usage.completion_tokens = response_object["usage"].get("completion_tokens", 0)
model_response_object.usage.prompt_tokens = response_object["usage"].get("prompt_tokens", 0)
model_response_object.usage.total_tokens = response_object["usage"].get("total_tokens", 0)
+ model_response_object.usage.completion_tokens_details = _cached_completion_tokens_details(
+ _CACHED_USAGE_DETAILS.validate_python(response_object["usage"])
+ )
if "id" in response_object:
model_response_object.id = response_object["id"]
@@ -390,6 +399,35 @@ class _ParallelToolUseArgs(TypedDict):
_PARALLEL_TOOL_USE_ARGS: Final = TypeAdapter(_ParallelToolUseArgs)
+class _CachedMessageProviderFields(TypedDict, total=False):
+ provider_specific_fields: ReadOnly[object]
+
+
+class _CachedUsageDetails(TypedDict, total=False):
+ completion_tokens_details: ReadOnly[object]
+
+
+_CACHED_MESSAGE_PROVIDER_FIELDS: Final = TypeAdapter(_CachedMessageProviderFields)
+_CACHED_USAGE_DETAILS: Final = TypeAdapter(_CachedUsageDetails)
+_FIELD_MAP: Final = TypeAdapter(dict[str, object])
+_COMPLETION_TOKENS_DETAILS: Final = TypeAdapter(CompletionTokensDetailsWrapper)
+
+
+def _cached_provider_specific_fields(message: _CachedMessageProviderFields) -> Mapping[str, object] | None:
+ fields: Final = message.get("provider_specific_fields")
+ return _FIELD_MAP.validate_python(fields) if isinstance(fields, dict) else None
+
+
+def _cached_completion_tokens_details(usage: _CachedUsageDetails) -> CompletionTokensDetailsWrapper | None:
+ details: Final = usage.get("completion_tokens_details")
+ if not isinstance(details, dict):
+ return None
+ try:
+ return _COMPLETION_TOKENS_DETAILS.validate_python(details)
+ except ValidationError:
+ return None
+
+
def handle_invalid_parallel_tool_calls(
tool_calls: list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall],
) -> list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall] | None:
diff --git a/litellm/litellm_core_utils/prompt_templates/image_handling.py b/litellm/litellm_core_utils/prompt_templates/image_handling.py
index 7924bb9bf4c..4cf76e5bd5e 100644
--- a/litellm/litellm_core_utils/prompt_templates/image_handling.py
+++ b/litellm/litellm_core_utils/prompt_templates/image_handling.py
@@ -4,12 +4,14 @@ Helper functions to handle images passed in messages
import asyncio
import base64
+import time
from collections.abc import Callable, Iterable, Mapping
from dataclasses import dataclass
from itertools import chain
from types import MappingProxyType
from typing import Final
+import httpx
from httpx import Response
import litellm
@@ -22,6 +24,7 @@ from litellm.types.llms.openai import AllMessageValues
MAX_IMGS_IN_MEMORY: Final = 10
MAX_CONCURRENT_REMOTE_MEDIA_FETCHES: Final = 20
+_FETCH_ATTEMPTS: Final = 3
in_memory_cache: Final = InMemoryCache(max_size_in_memory=MAX_IMGS_IN_MEMORY)
@@ -73,6 +76,13 @@ def _process_image_response(response: Response, url: str) -> str:
return result
+def _attempt_timeouts(timeout: float | httpx.Timeout | None) -> Iterable[float | httpx.Timeout | None]:
+ if timeout is None or isinstance(timeout, httpx.Timeout):
+ return (timeout,) * _FETCH_ATTEMPTS
+ deadline: Final = time.monotonic() + timeout
+ return (remaining for _ in range(_FETCH_ATTEMPTS) if (remaining := deadline - time.monotonic()) > 0)
+
+
def _rejected_image_fetch(url: str, verdict: SSRFError) -> "litellm.ImageFetchError":
verbose_logger.warning("Image fetch of %s rejected before any request went out: %s", url, verdict)
return litellm.ImageFetchError(
@@ -81,7 +91,7 @@ def _rejected_image_fetch(url: str, verdict: SSRFError) -> "litellm.ImageFetchEr
)
-async def async_convert_url_to_base64(url: str) -> str:
+async def async_convert_url_to_base64(url: str, timeout: float | httpx.Timeout | None = None) -> str:
if url.startswith("data:") and ";base64," in url:
return url
@@ -96,9 +106,9 @@ async def async_convert_url_to_base64(url: str) -> str:
return cached_result
client: Final = litellm.module_level_aclient
- for _ in range(3):
+ for attempt_timeout in _attempt_timeouts(timeout):
try:
- response = await async_safe_get(client, url)
+ response = await async_safe_get(client, url, timeout=attempt_timeout)
return _process_image_response(response, url)
except litellm.ImageFetchError:
raise
@@ -109,7 +119,7 @@ async def async_convert_url_to_base64(url: str) -> str:
raise litellm.ImageFetchError(f"Error: Unable to fetch image from URL after 3 attempts. url={url}")
-def convert_url_to_base64(url: str) -> str:
+def convert_url_to_base64(url: str, timeout: float | httpx.Timeout | None = None) -> str:
if url.startswith("data:") and ";base64," in url:
return url
@@ -124,9 +134,9 @@ def convert_url_to_base64(url: str) -> str:
return cached_result
client: Final = litellm.module_level_client
- for _ in range(3):
+ for attempt_timeout in _attempt_timeouts(timeout):
try:
- response = safe_get(client, url)
+ response = safe_get(client, url, timeout=attempt_timeout)
return _process_image_response(response, url)
except litellm.ImageFetchError:
raise
@@ -286,14 +296,16 @@ def _inline_message(
return inlined_message # pyright: ignore[reportReturnType] # the same message with its remote parts inlined
-async def _fetch_data_url(url: str, in_flight: asyncio.Semaphore) -> str:
+async def _fetch_data_url(url: str, in_flight: asyncio.Semaphore, timeout: float | httpx.Timeout | None) -> str:
async with in_flight:
- return await async_convert_url_to_base64(url)
+ return await async_convert_url_to_base64(url, timeout=timeout)
-async def _fetch_data_urls(remote_urls: tuple[str, ...]) -> tuple[str, ...]:
+async def async_convert_urls_to_base64(
+ remote_urls: tuple[str, ...], timeout: float | httpx.Timeout | None = None
+) -> tuple[str, ...]:
in_flight: Final = asyncio.Semaphore(MAX_CONCURRENT_REMOTE_MEDIA_FETCHES)
- fetches: Final = tuple(asyncio.create_task(_fetch_data_url(url, in_flight)) for url in remote_urls)
+ fetches: Final = tuple(asyncio.create_task(_fetch_data_url(url, in_flight, timeout)) for url in remote_urls)
try:
return tuple(await asyncio.gather(*fetches))
except BaseException:
@@ -329,6 +341,6 @@ async def async_inline_remote_media(
remote_urls: Final = _remote_urls_to_inline(messages, should_inline)
if not remote_urls:
return messages
- data_urls: Final = await _fetch_data_urls(remote_urls)
+ data_urls: Final = await async_convert_urls_to_base64(remote_urls)
inlined: Final = MappingProxyType(dict(zip(remote_urls, data_urls, strict=True)))
return [_inline_message(message, inlined, should_inline) for message in messages]
diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py
index a78e6f17cc9..484978bae04 100644
--- a/litellm/litellm_core_utils/streaming_handler.py
+++ b/litellm/litellm_core_utils/streaming_handler.py
@@ -1,6 +1,7 @@
import asyncio
import collections.abc
import datetime
+import inspect
import json
import logging
import threading
@@ -234,6 +235,7 @@ class CustomStreamWrapper:
self.completion_stream = completion_stream
self.sent_first_chunk = False
self.sent_last_chunk = False
+ self._closed = False
self._stream_created_time: float = time.time()
litellm_params: Final[GenericLiteLLMParams] = GenericLiteLLMParams.model_validate(
@@ -278,6 +280,7 @@ class CustomStreamWrapper:
self._response_headers = _response_headers
self.response_id: str | None = None
+ self.received_model_id: str | None = None
self.logging_loop = None
self.rules = Rules()
self.stream_options = stream_options or getattr(logging_obj, "stream_options", None)
@@ -328,6 +331,18 @@ class CustomStreamWrapper:
def __aiter__(self) -> AsyncIterator["ModelResponseStream"]:
return self
+ def __enter__(self) -> "CustomStreamWrapper":
+ return self
+
+ def __exit__(self, *exc_info: object) -> None:
+ self.close()
+
+ async def __aenter__(self) -> "CustomStreamWrapper":
+ return self
+
+ async def __aexit__(self, *exc_info: object) -> None:
+ await self.aclose()
+
def _restore_consumer_correlation_context(self, *, guarded: bool = False) -> None:
"""Restore trace_id/session_id in the *consuming* thread/task/context.
@@ -392,11 +407,26 @@ class CustomStreamWrapper:
"""
self._restore_consumer_correlation_context(guarded=True)
+ def close(self) -> None:
+ stream_to_close: Final = cast(object, self.completion_stream) # cast-ok: completion_stream is untyped
+ self.completion_stream = None
+ self._closed = True
+ close_stream: Final = getattr(stream_to_close, "close", None)
+ if callable(close_stream):
+ try:
+ result: Final = close_stream()
+ if inspect.iscoroutine(result):
+ result.close()
+ except Exception as e: # noqa: BLE001 # best-effort cleanup; must not raise into the caller
+ verbose_logger.debug("CustomStreamWrapper.close: error closing completion_stream: %s", e)
+ self._restore_consumer_correlation_context()
+
async def aclose(self):
# Restore the consumer's outer context only after the underlying
# provider stream's own close (and its diagnostic logging below, if
# closing fails) completes - not before - so those log lines still
# carry this closing stream's own trace_id/session_id.
+ self._closed = True
if self.completion_stream is not None:
stream_to_close: Final = self.completion_stream
self.completion_stream = None
@@ -809,6 +839,7 @@ class CustomStreamWrapper:
self.response_id = id
if id and isinstance(id, str) and id.strip():
+ self.received_model_id = id
model_response.hidden_params["received_model_id"] = id
if self.response_id is not None and isinstance(self.response_id, str):
@@ -1727,7 +1758,9 @@ class CustomStreamWrapper:
self.logging_obj.success_handler(processed_chunk, None, None, cache_hit)
def finish_reason_handler(self):
- model_response: Final = self.model_response_creator()
+ model_response: Final = self.model_response_creator(
+ hidden_params=None if self.received_model_id is None else {"received_model_id": self.received_model_id}
+ )
_finish_reason: Final = self.received_finish_reason or self.intermittent_finish_reason
if _finish_reason is not None:
model_response.choices[0].finish_reason = _finish_reason
@@ -1780,6 +1813,8 @@ class CustomStreamWrapper:
response.hidden_params["additional_headers"]["llm_provider-x-litellm-response-cost"] = _cost
def __next__(self) -> "ModelResponseStream":
+ if self._closed:
+ raise StopIteration
cache_hit = False
if self.custom_llm_provider is not None and self.custom_llm_provider == "cached_response":
cache_hit = True
@@ -1950,7 +1985,7 @@ class CustomStreamWrapper:
if self.sent_stream_usage is False and self.send_stream_usage is True:
self.sent_stream_usage = True
return response
- self._restore_consumer_correlation_context()
+ self.close()
raise # Re-raise StopIteration
else:
self.sent_last_chunk = True
@@ -1986,6 +2021,7 @@ class CustomStreamWrapper:
traceback_exception: Final = traceback.format_exc()
# LOG FAILURE - handle streaming failure logging in the _next_ object, remove `handle_failure` once it's deprecated
threading.Thread(target=self.logging_obj.failure_handler, args=(e, traceback_exception)).start()
+ self.close()
self._handle_stream_fallback_error(e)
def fetch_sync_stream(self):
@@ -2005,6 +2041,8 @@ class CustomStreamWrapper:
return self.completion_stream
async def __anext__(self) -> "ModelResponseStream":
+ if self._closed:
+ raise StopAsyncIteration
cache_hit = False
if self.custom_llm_provider is not None and self.custom_llm_provider == "cached_response":
cache_hit = True
@@ -2125,12 +2163,15 @@ class CustomStreamWrapper:
asyncio.create_task(
self.logging_obj.dispatch_failure_handlers(e, traceback_exception, prefer_async_handlers=True)
)
+ await self.aclose()
self._handle_stream_fallback_error(e)
except (httpx.ReadError, httpx.RemoteProtocolError) as e:
if self.received_finish_reason is None:
+ await self.aclose()
self._log_stream_failure_and_raise(e)
return await self._finalize_completed_stream(cache_hit=cache_hit)
except Exception as e:
+ await self.aclose()
self._log_stream_failure_and_raise(e)
async def _finalize_completed_stream(self, cache_hit: bool) -> "ModelResponseStream":
@@ -2224,7 +2265,7 @@ class CustomStreamWrapper:
)
)
- self._restore_consumer_correlation_context()
+ await self.aclose()
raise StopAsyncIteration # Re-raise StopIteration
else:
self.sent_last_chunk = True
diff --git a/litellm/llms/anthropic/chat/handler.py b/litellm/llms/anthropic/chat/handler.py
index eb0de9d9b60..4e5b761b9a3 100644
--- a/litellm/llms/anthropic/chat/handler.py
+++ b/litellm/llms/anthropic/chat/handler.py
@@ -25,7 +25,10 @@ from litellm.litellm_core_utils.json_fragment_accumulator import JSONFragmentAcc
from litellm.llms.base_llm.base_utils import (
RESPONSE_FORMAT_ENVELOPE_NAME_KEYS,
ResponseFormatSchema,
+ names_a_response_format_property,
+ partial_response_format_answer,
response_format_envelope_contents,
+ split_response_format_answer,
)
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
@@ -63,7 +66,7 @@ from litellm.types.utils import (
)
from ...base import BaseLLM
-from ..common_utils import AnthropicError, process_anthropic_headers
+from ..common_utils import AnthropicError, process_anthropic_headers, refusal_provider_specific_fields
from .transformation import ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY, AnthropicConfig, JsonModeRequest
if TYPE_CHECKING:
@@ -85,6 +88,9 @@ _NAME_VALUE_PREFIX: Final = re.compile(
r'^\s*\{\s*"(?:' + _ENVELOPE_NAME_KEY_PATTERN + r')"\s*:\s*(?:"(?:[^"\\]|\\.)*\\?)?$'
)
_STRING_VALUE_THEN_KEY: Final = re.compile(r'^"(?:[^"\\]|\\.)*"\s*,\s*"((?:[^"\\]|\\.)*)"\s*:')
+_LABEL_THEN_KEY: Final = re.compile(r'"((?:[^"\\]|\\.)*)"\s*,\s*"((?:[^"\\]|\\.)*)"\s*:\s*')
+_LEADING_LABELS: Final = re.compile(r'(?:"(?:[^"\\]|\\.)*"\s*,\s*"(?:[^"\\]|\\.)*"\s*:\s*)*')
+_CLOSED_STRING: Final = re.compile(r'"((?:[^"\\]|\\.)*)"')
_UNDECIDED_PREFIX_LIMIT: Final = 4096
@@ -95,14 +101,45 @@ def _key_after_a_string_verdict(string_value_start: str, schema: ResponseFormatS
return "pass_through" if next_key.group(1) in schema.property_names else "hold"
-def _first_value_verdict(after_first_key: str, schema: ResponseFormatSchema) -> _EnvelopeVerdict:
+def _is_a_label(text: str, schema: ResponseFormatSchema, tool_name: str) -> bool:
+ return text in schema.property_names or text == tool_name
+
+
+def _label_verdict(label: str, next_key: str, schema: ResponseFormatSchema, tool_name: str) -> _EnvelopeVerdict:
+ if _is_a_label(label, schema, tool_name):
+ return "hold"
+ return "pass_through" if next_key in schema.property_names else "undecided"
+
+
+def _leading_labels_verdict(string_value_start: str, schema: ResponseFormatSchema, tool_name: str) -> _EnvelopeVerdict:
+ last_value_start: Final = _LEADING_LABELS.sub("", string_value_start, count=1)
+ labelled: Final = string_value_start[: len(string_value_start) - len(last_value_start)]
+ verdicts: Final[tuple[_EnvelopeVerdict, ...]] = tuple(
+ _label_verdict(pair.group(1), pair.group(2), schema, tool_name) for pair in _LABEL_THEN_KEY.finditer(labelled)
+ )
+ decided: Final[_EnvelopeVerdict | None] = next((verdict for verdict in verdicts if verdict != "undecided"), None)
+ if decided is not None:
+ return decided
+ last_label: Final = _CLOSED_STRING.match(last_value_start)
+ if last_label is not None and _is_a_label(last_label.group(1), schema, tool_name):
+ return "hold"
+ return "undecided" if not last_value_start or last_value_start.startswith('"') else "pass_through"
+
+
+def _string_value_verdict(string_value_start: str, schema: ResponseFormatSchema, tool_name: str) -> _EnvelopeVerdict:
+ if schema.required_names:
+ return _key_after_a_string_verdict(string_value_start, schema)
+ return _leading_labels_verdict(string_value_start, schema, tool_name)
+
+
+def _first_value_verdict(after_first_key: str, schema: ResponseFormatSchema, tool_name: str) -> _EnvelopeVerdict:
value_start: Final = after_first_key.lstrip()
if not value_start:
return "undecided"
- if value_start.startswith("{"):
+ if value_start.startswith('"'):
+ return _string_value_verdict(value_start, schema, tool_name)
+ if value_start.startswith("{") and not schema.accepts_a_lone_undeclared_key:
return "hold"
- if value_start.startswith('"') and schema.required_names:
- return _key_after_a_string_verdict(value_start, schema)
return "pass_through"
@@ -116,9 +153,11 @@ def _prefix_verdict(buffered_json: str, schema: ResponseFormatSchema, tool_name:
return "hold"
if first_value is None and _NAME_VALUE_PREFIX.match(buffered_json):
return "undecided"
- if first_key.group(1) in schema.property_names or not schema.property_names or schema.accepts_a_lone_undeclared_key:
+ if first_key.group(1) == "values" and "values" not in schema.property_names:
+ return "hold"
+ if first_key.group(1) in schema.property_names or not schema.property_names:
return "pass_through"
- return _first_value_verdict(buffered_json[first_key.end() :], schema)
+ return _first_value_verdict(buffered_json[first_key.end() :], schema, tool_name)
def _held_answer(held: str, schema: ResponseFormatSchema) -> str:
@@ -130,6 +169,13 @@ def _held_answer(held: str, schema: ResponseFormatSchema) -> str:
return held if contents is None else json.dumps(contents)
+def _split_answer_text(blocks: tuple[str, ...], schema: ResponseFormatSchema) -> str:
+ answer: Final = split_response_format_answer(blocks, schema)
+ if answer is not None:
+ return json.dumps(answer)
+ return "".join(_held_answer(block, schema) for block in blocks)
+
+
def _envelope_verdict(buffered_json: str, schema: ResponseFormatSchema, tool_name: str) -> _EnvelopeVerdict:
verdict: Final = _prefix_verdict(buffered_json, schema, tool_name)
if verdict == "undecided" and len(buffered_json) > _UNDECIDED_PREFIX_LIMIT:
@@ -186,6 +232,7 @@ async def make_call(
speed=speed,
tool_name_reverse_map=tool_name_reverse_map,
response_format_schema=response_format_schema,
+ response=response,
)
# LOGGING
@@ -256,6 +303,7 @@ def make_sync_call(
speed=speed,
tool_name_reverse_map=tool_name_reverse_map,
response_format_schema=response_format_schema,
+ response=response,
)
# LOGGING
@@ -629,9 +677,12 @@ class ModelResponseIterator:
speed: str | None = None,
tool_name_reverse_map: dict[str, str] | None = None,
response_format_schema: ResponseFormatSchema | None = None,
+ response: httpx.Response | None = None,
):
self.streaming_response = streaming_response
self.response_iterator = self.streaming_response
+ self.sync_stream = sync_stream
+ self._response = response
self.content_blocks: list[ContentBlockDelta] = []
self.tool_index = -1
self.json_mode = json_mode
@@ -655,6 +706,9 @@ class ModelResponseIterator:
self.response_format_schema = response_format_schema
self._response_format_buffer: io.StringIO = io.StringIO()
self._response_format_verdict: _EnvelopeVerdict = "undecided"
+ self._split_answer_blocks: tuple[str, ...] = ()
+ self._withheld_answer_close: str = ""
+ self._streamed_answer_keys: frozenset[str] | None = None
# For handling partial JSON chunks from fragmentation
# See: https://github.com/BerriAI/litellm/issues/17473
@@ -917,7 +971,7 @@ class ModelResponseIterator:
# Track current content block type for filtering deltas
self.current_content_block_type = content_block_start["content_block"]["type"]
if content_block_start["content_block"]["type"] == "text":
- text = content_block_start["content_block"]["text"]
+ text = self._release_answer_close() + content_block_start["content_block"]["text"]
elif (
content_block_start["content_block"]["type"] == "tool_use"
or content_block_start["content_block"]["type"] == "server_tool_use"
@@ -1048,7 +1102,9 @@ class ModelResponseIterator:
# These are automatically handled by Anthropic API, we just pass them through
pass
elif type_chunk == "message_delta":
- finish_reason, usage, container = self._handle_message_delta(chunk)
+ finish_reason, usage, container, refusal_fields = self._handle_message_delta(chunk)
+ provider_specific_fields.update(refusal_fields)
+ text = self._release_answer_close() + self._drain_split_answer_blocks()
if container:
provider_specific_fields["container"] = container
# Store container_id and re-emit code_interpreter_results
@@ -1155,6 +1211,8 @@ class ModelResponseIterator:
# (per Anthropic's fine-grained streaming pattern)
tool_name: Final = tool_use.get("function", {}).get("name", "")
self.is_response_format_tool = tool_name == RESPONSE_FORMAT_TOOL_NAME
+ if self.is_response_format_tool:
+ self.tool_index -= 1
if self.is_response_format_tool and self.response_format_schema is not None:
self.converted_response_format_tool = True
@@ -1176,27 +1234,73 @@ class ModelResponseIterator:
) -> str:
if tool_use.get("id") is not None:
self._response_format_buffer = io.StringIO()
- self._response_format_verdict = "undecided"
+ held_back: Final = self._split_answer_blocks or self._streamed_answer_keys is not None
+ self._response_format_verdict = "hold" if held_back else "undecided"
return ""
fragment: Final = tool_use.get("function", {}).get("arguments") or ""
- if self._response_format_verdict == "pass_through":
- return fragment
self._response_format_buffer.write(fragment)
+ if self._response_format_verdict == "pass_through":
+ return self._withhold_answer_close(fragment)
if self._response_format_verdict == "hold":
return ""
buffered: Final = self._response_format_buffer.getvalue()
self._response_format_verdict = _envelope_verdict(buffered, schema, RESPONSE_FORMAT_TOOL_NAME)
- return buffered if self._response_format_verdict == "pass_through" else ""
+ return self._withhold_answer_close(buffered) if self._response_format_verdict == "pass_through" else ""
+
+ def _withhold_answer_close(self, text: str) -> str:
+ pending: Final = self._withheld_answer_close + text
+ body: Final = pending.rstrip()
+ cut: Final = len(body) - 1 if body.endswith("}") else len(pending)
+ self._withheld_answer_close = pending[cut:]
+ return pending[:cut]
+
+ def _release_answer_close(self) -> str:
+ close: Final = self._withheld_answer_close
+ self._withheld_answer_close = ""
+ self._streamed_answer_keys = None
+ return close
+
+ def _join_streamed_answer(self, held: str, schema: ResponseFormatSchema) -> str | None:
+ streamed: Final = self._streamed_answer_keys
+ fields: Final = partial_response_format_answer(held, schema)
+ if streamed is None or fields is None or not streamed.isdisjoint(fields):
+ return None
+ self._streamed_answer_keys = streamed | frozenset(fields)
+ separator: Final = ", " if streamed else ""
+ return f"{separator}{json.dumps(fields)[1:-1]}" if fields else ""
def _flush_response_format_buffer(self) -> str:
- if self.response_format_schema is None or self._response_format_verdict == "pass_through":
+ schema: Final = self.response_format_schema
+ if schema is None:
return ""
held: Final = self._response_format_buffer.getvalue()
+ verdict: Final = self._response_format_verdict
self._response_format_buffer = io.StringIO()
self._response_format_verdict = "undecided"
- return _held_answer(held, self.response_format_schema) if held else ""
+ if verdict == "pass_through":
+ answer: Final = partial_response_format_answer(held, schema)
+ close: Final = "" if answer is not None else self._release_answer_close()
+ self._streamed_answer_keys = None if answer is None else frozenset(answer)
+ return close
+ joined: Final = self._join_streamed_answer(held, schema)
+ if joined is not None:
+ return joined
+ released: Final = self._release_answer_close()
+ if self._split_answer_blocks or names_a_response_format_property(held, schema):
+ self._split_answer_blocks = (*self._split_answer_blocks, held)
+ return released
+ return released + (_held_answer(held, schema) if held else "")
- def _handle_message_delta(self, chunk: dict) -> tuple[str, Usage | None, dict[str, Any] | None]:
+ def _drain_split_answer_blocks(self) -> str:
+ blocks: Final = self._split_answer_blocks
+ self._split_answer_blocks = ()
+ if self.response_format_schema is None or not blocks:
+ return ""
+ return _split_answer_text(blocks, self.response_format_schema)
+
+ def _handle_message_delta(
+ self, chunk: dict
+ ) -> tuple[str, Usage | None, dict[str, Any] | None, Mapping[str, object]]:
"""
Handle message_delta event for finish_reason, usage, and container.
@@ -1207,16 +1311,18 @@ class ModelResponseIterator:
Tuple of (finish_reason, usage, container)
"""
message_delta: Final = MessageBlockDelta(**chunk)
- finish_reason = map_finish_reason(finish_reason=message_delta["delta"].get("stop_reason", "stop") or "stop")
- # Override finish_reason to "stop" if we converted response_format tools
- # (matches OpenAI behavior and non-streaming Anthropic implementation)
- if self.converted_response_format_tool:
- finish_reason = "stop"
+ provider_finish_reason: Final = map_finish_reason(
+ finish_reason=message_delta["delta"].get("stop_reason", "stop") or "stop"
+ )
+ called_only_the_response_format_tool: Final = (
+ provider_finish_reason == "tool_calls" and self.converted_response_format_tool and self.tool_index < 0
+ )
+ finish_reason: Final = "stop" if called_only_the_response_format_tool else provider_finish_reason
usage: Final = (
self._handle_usage(anthropic_usage_chunk=message_delta["usage"]) if "usage" in message_delta else None
)
container: Final = message_delta["delta"].get("container")
- return finish_reason, usage, container
+ return finish_reason, usage, container, refusal_provider_specific_fields(message_delta["delta"])
def _handle_accumulated_json_chunk(self, data_str: str, is_final: bool = False) -> ModelResponseStream | None:
"""
@@ -1277,6 +1383,22 @@ class ModelResponseIterator:
return self._handle_accumulated_json_chunk(data_str)
# Sync iterator
+ def close(self) -> None:
+ response: Final = self._response
+ self._response = None
+ if response is not None:
+ response.close()
+
+ async def aclose(self) -> None:
+ response: Final = self._response
+ self._response = None
+ if response is None:
+ return
+ if self.sync_stream:
+ response.close()
+ return
+ await response.aclose()
+
def __iter__(self):
return self
diff --git a/litellm/llms/anthropic/chat/transformation.py b/litellm/llms/anthropic/chat/transformation.py
index c43e5f9b05b..487f231e1db 100644
--- a/litellm/llms/anthropic/chat/transformation.py
+++ b/litellm/llms/anthropic/chat/transformation.py
@@ -39,6 +39,7 @@ from litellm.llms.base_llm.base_utils import (
ResponseFormatSchema,
as_json_object,
response_format_schema,
+ split_response_format_answer,
type_to_response_format_param,
unwrap_response_format_envelope,
)
@@ -103,10 +104,12 @@ from litellm.utils import (
)
from ..common_utils import (
+ REFUSAL_STOP_FIELDS,
AnthropicError,
AnthropicModelInfo,
eager_input_streaming_flag,
process_anthropic_headers,
+ refusal_provider_specific_fields,
requires_native_compaction_beta,
strip_advisor_blocks_from_messages,
)
@@ -293,6 +296,22 @@ DROP_UNSUPPORTED_ADAPTIVE_THINKING_WARNING: Final = (
DROP_UNSUPPORTED_SPEED_WARNING: Final = (
"Dropping unsupported `speed` for model=%s (drop_params=True). Fast mode is only supported on select Opus models."
)
+SERVICE_TIER_PROVIDERS: Final = frozenset({"anthropic", "azure_ai"})
+
+OPENAI_TO_ANTHROPIC_SERVICE_TIER: Final = MappingProxyType(
+ {"auto": "auto", "default": "standard_only", "standard_only": "standard_only"}
+)
+
+DROP_UNSUPPORTED_SERVICE_TIER_WARNING: Final = (
+ "Dropping unsupported `service_tier` for model=%s (drop_params=True). "
+ "Anthropic only accepts 'auto' and 'standard_only'."
+)
+
+MESSAGES_SERVICE_TIER_PROVIDERS: Final = SERVICE_TIER_PROVIDERS | {"bedrock_mantle"}
+
+DROP_SERVICE_TIER_PROVIDER_WARNING: Final = (
+ "Dropping `service_tier` for model=%s on provider=%s. Only the Anthropic API routes accept it."
+)
class AnthropicConfig(AnthropicModelInfo, BaseConfig):
@@ -335,6 +354,11 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
def _resolved_provider(self) -> str:
return self.custom_llm_provider or "anthropic"
+ def uses_native_structured_output(self, model: str) -> bool:
+ return AnthropicConfig._supports_model_capability(
+ model, "supports_native_structured_output", self._resolved_provider
+ )
+
@classmethod
def get_config(cls, *, model: str | None = None):
config: Final = super().get_config()
@@ -495,6 +519,23 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
)
optional_params.pop("speed", None)
+ @staticmethod
+ def _map_service_tier(value: str, model: str, drop_params: bool) -> str | None:
+ anthropic_tier: Final = OPENAI_TO_ANTHROPIC_SERVICE_TIER.get(value)
+ if anthropic_tier is not None:
+ return anthropic_tier
+ if not (litellm.drop_params or drop_params):
+ raise litellm.utils.UnsupportedParamsError(
+ message=(
+ f"{model} does not support service_tier={value!r}. Anthropic accepts 'auto', "
+ "'standard_only', or OpenAI's 'default' (sent as 'standard_only'). "
+ "To drop unsupported params, set `litellm.drop_params = True`."
+ ),
+ status_code=400,
+ )
+ litellm.verbose_logger.warning(DROP_UNSUPPORTED_SERVICE_TIER_WARNING, model)
+ return None
+
@classmethod
def maybe_drop_speed_param(
cls,
@@ -505,6 +546,24 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
) -> None:
return cls._maybe_drop_speed_param(model, optional_params, drop_params, custom_llm_provider)
+ @staticmethod
+ def provider_supports_service_tier(custom_llm_provider: str | None) -> bool:
+ """Anthropic's ``service_tier`` is accepted by the direct API and the Foundry and Mantle gateways in
+ front of it. Vertex rejects the field and Bedrock Invoke only takes its own tier names."""
+ return custom_llm_provider is None or custom_llm_provider in MESSAGES_SERVICE_TIER_PROVIDERS
+
+ @classmethod
+ def maybe_drop_service_tier_param(
+ cls,
+ model: str,
+ optional_params: dict[str, object], # mutable-ok: mirrors override contract
+ custom_llm_provider: str | None = None,
+ ) -> None:
+ if "service_tier" not in optional_params or cls.provider_supports_service_tier(custom_llm_provider):
+ return
+ litellm.verbose_logger.warning(DROP_SERVICE_TIER_PROVIDER_WARNING, model, custom_llm_provider)
+ optional_params.pop("service_tier", None)
+
@staticmethod
def _raise_invalid_reasoning_effort(model: str, value: object, llm_provider: str) -> NoReturn:
"""Raise a ``BadRequestError`` for an unrecognised ``reasoning_effort``.
@@ -568,6 +627,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
params.append("thinking")
params.append("reasoning_effort")
+ if self._resolved_provider in SERVICE_TIER_PROVIDERS:
+ params.append("service_tier")
+
return params
@staticmethod
@@ -680,12 +742,12 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
# Already handled above
continue
- if key == "properties" and isinstance(value, dict):
- result[key] = {k: AnthropicConfig.filter_anthropic_output_schema(v) for k, v in value.items()}
+ if key in ("properties", "$defs") and isinstance(value, dict):
+ nested_schemas: dict[str, Any] = value # mutable-ok: filter_anthropic_output_schema takes a dict
+ result[key] = {k: AnthropicConfig.filter_anthropic_output_schema(v) for k, v in nested_schemas.items()}
elif key == "items" and isinstance(value, dict):
- result[key] = AnthropicConfig.filter_anthropic_output_schema(value)
- elif key == "$defs" and isinstance(value, dict):
- result[key] = {k: AnthropicConfig.filter_anthropic_output_schema(v) for k, v in value.items()}
+ nested_schema: dict[str, Any] = value # mutable-ok: filter_anthropic_output_schema takes a dict
+ result[key] = AnthropicConfig.filter_anthropic_output_schema(nested_schema)
elif key == "anyOf" and isinstance(value, list):
result["anyOf"] = result.get("anyOf", []) + [
AnthropicConfig.filter_anthropic_output_schema(item) for item in value
@@ -1600,11 +1662,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
output_key=param,
)
elif param == "response_format" and isinstance(value, dict):
- if AnthropicConfig._supports_model_capability(
- model,
- "supports_native_structured_output",
- self._resolved_provider,
- ):
+ if self.uses_native_structured_output(model):
_output_format = self.map_response_format_to_anthropic_output_format(value)
if _output_format is not None:
optional_params["output_format"] = _output_format
@@ -1719,6 +1777,13 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
elif param == "cache_control" and isinstance(value, dict):
# Pass through top-level cache_control for automatic prompt caching
optional_params["cache_control"] = value
+ elif (
+ param == "service_tier"
+ and isinstance(value, str)
+ and self._resolved_provider in SERVICE_TIER_PROVIDERS
+ and (anthropic_tier := AnthropicConfig._map_service_tier(value, model, drop_params)) is not None
+ ):
+ optional_params["service_tier"] = anthropic_tier # rebind-ok: in-place fill is this method's contract
## handle thinking tokens
self.update_optional_params_with_thinking_tokens(
@@ -2229,13 +2294,12 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if json_tool.get("function", {}).get("arguments") is None:
return None, tool_calls, None
_message: Final = AnthropicConfig.convert_tool_response_to_message(
- tool_calls=[json_tool], response_format_schema=response_format_schema
+ tool_calls=tool_calls, response_format_schema=response_format_schema
)
return _message, [], None
- first_json: Final = tool_calls[json_indices[0]]
json_msg: Final = AnthropicConfig.convert_tool_response_to_message(
- [first_json], response_format_schema=response_format_schema
+ [tool_calls[i] for i in json_indices], response_format_schema=response_format_schema
)
extra_content: Final[str | None] = json_msg.content if json_msg is not None else None
filtered_tools: Final = [t for i, t in enumerate(tool_calls) if i not in json_indices]
@@ -2661,7 +2725,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if compaction_blocks is not None:
provider_specific_fields["compaction_blocks"] = compaction_blocks
- return provider_specific_fields
+ refusal_fields: Final = refusal_provider_specific_fields(
+ REFUSAL_STOP_FIELDS.validate_python(completion_response)
+ )
+ return {**provider_specific_fields, **refusal_fields}
def transform_parsed_response(
self,
@@ -2746,7 +2813,12 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
_message.provider_specific_fields = provider_specific_fields
if json_mode_message is not None:
+ refusal_fields: Final = refusal_provider_specific_fields(
+ REFUSAL_STOP_FIELDS.validate_python(completion_response)
+ )
completion_response["stop_reason"] = "stop"
+ if refusal_fields:
+ json_mode_message.provider_specific_fields = dict(refusal_fields)
_message = json_mode_message
model_response.choices[0].message = _message
@@ -2856,6 +2928,16 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
json_mode_content_str: Final[str | None] = tool_calls[0]["function"].get("arguments")
if json_mode_content_str is None:
return None
+ split_answer: Final = (
+ None
+ if response_format_schema is None
+ else split_response_format_answer(
+ tuple(tool_call["function"].get("arguments") or "" for tool_call in tool_calls),
+ response_format_schema,
+ )
+ )
+ if split_answer is not None:
+ return litellm.Message(content=json.dumps(split_answer))
try:
payload: Final[object] = json.loads(json_mode_content_str)
except (json.JSONDecodeError, RecursionError):
@@ -2906,11 +2988,13 @@ class JsonModeRequest(BaseModel):
return None if json_tool is None else response_format_schema(json_tool.get("input_schema"))
-def _valid_user_id(user_id: str) -> bool:
+def _valid_user_id(user_id: object) -> bool:
"""
- Validate that user_id is not an email or phone number.
- Returns: bool: True if valid (not email or phone), False otherwise
+ Validate that user_id is a string and not an email or phone number.
+ Returns: bool: True if valid (string, not email or phone), False otherwise
"""
+ if not isinstance(user_id, str):
+ return False
email_pattern: Final = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
phone_pattern: Final = r"^\+?[\d\s\(\)-]{7,}$"
diff --git a/litellm/llms/anthropic/common_utils.py b/litellm/llms/anthropic/common_utils.py
index 47a82aa097c..dbcb7ea4962 100644
--- a/litellm/llms/anthropic/common_utils.py
+++ b/litellm/llms/anthropic/common_utils.py
@@ -8,11 +8,12 @@ import re
from collections.abc import Mapping, MutableMapping, Sequence
from datetime import datetime, timezone
from types import MappingProxyType
-from typing import Any, ClassVar, Final, Literal, TypeVar
+from typing import Annotated, Any, ClassVar, Final, Literal, TypeVar
from urllib.parse import quote, urlparse
import httpx
-from pydantic import ConfigDict, Field, StrictBool, TypeAdapter, ValidationError
+from pydantic import BeforeValidator, ConfigDict, Field, StrictBool, TypeAdapter, ValidationError
+from typing_extensions import ReadOnly, TypedDict
import litellm
from litellm.constants import (
@@ -2150,3 +2151,35 @@ def create_anthropic_model_list_response(
"first_id": data[0]["id"] if data else None,
"last_id": data[-1]["id"] if data else None,
}
+
+
+class RefusalStopFields(TypedDict, total=False):
+ stop_reason: ReadOnly[object]
+ stop_details: ReadOnly[object]
+
+
+def _is_dict(value: object) -> bool:
+ return isinstance(value, dict)
+
+
+def _no_stop_fields() -> Mapping[str, object]:
+ return {}
+
+
+def _dict_or_empty(value: object) -> object:
+ return value if _is_dict(value) else _no_stop_fields()
+
+
+REFUSAL_STOP_FIELDS: Final[TypeAdapter[RefusalStopFields]] = TypeAdapter(
+ Annotated[RefusalStopFields, BeforeValidator(_dict_or_empty)]
+)
+
+
+def refusal_provider_specific_fields(message: Mapping[str, object]) -> Mapping[str, object]:
+ stop_reason: Final = message.get("stop_reason")
+ stop_details: Final = message.get("stop_details")
+ if stop_details is not None:
+ return {"stop_reason": stop_reason, "stop_details": stop_details}
+ if stop_reason == "refusal":
+ return {"stop_reason": stop_reason}
+ return {}
diff --git a/litellm/llms/anthropic/pass_through/adapters/handler.py b/litellm/llms/anthropic/pass_through/adapters/handler.py
index 1bdb2b2db2c..9c8ab02d552 100644
--- a/litellm/llms/anthropic/pass_through/adapters/handler.py
+++ b/litellm/llms/anthropic/pass_through/adapters/handler.py
@@ -1,4 +1,4 @@
-from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping
+from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
from typing import (
TYPE_CHECKING,
Final,
@@ -6,6 +6,7 @@ from typing import (
cast,
)
+from pydantic import TypeAdapter
from typing_extensions import TypedDict
import litellm
@@ -19,10 +20,20 @@ from litellm.llms.anthropic.pass_through.context_management import (
PolyfillResult,
apply_context_management,
)
+from litellm.llms.anthropic.pass_through.safeguards import (
+ SafeguardsEvaluator,
+ build_safeguards_evaluator,
+ read_classifier_model_setting,
+ requested_dangerous_tool_use,
+ with_safeguard_results,
+)
from litellm.llms.anthropic.pass_through.utils import (
+ ProxySubCallLoggingKwargs,
is_reasoning_auto_summary_enabled,
litellm_logging_obj_from_kwargs,
local_model_name,
+ proxy_sub_call_logging_kwargs,
+ proxy_sub_call_trace_id,
)
from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
from litellm.types.llms.anthropic_messages.anthropic_response import (
@@ -38,6 +49,7 @@ if TYPE_CHECKING:
# Anthropic-only keys already mapped by the translator; strip on extra_kwargs re-merge.
ANTHROPIC_ONLY_REQUEST_KEYS: Final[frozenset[str]] = frozenset({"output_config", "safeguards"})
+_REQUEST_KWARGS: Final = TypeAdapter(Mapping[str, object])
_AnthropicMessages: TypeAlias = "list[dict[str, object]]"
_AnthropicSystem: TypeAlias = "str | list[dict[str, object]] | None"
@@ -69,6 +81,10 @@ def _proxy_router_fallback() -> "Router | None":
return _proxy_router
+def _router_or_proxy_fallback(requested: "Router | None") -> "Router | None":
+ return requested if requested is not None else _proxy_router_fallback()
+
+
def _extract_proxy_litellm_metadata(
kwargs: Mapping[str, object],
) -> "tuple[dict[str, object], UserAPIKeyAuth | None] | tuple[None, None]":
@@ -90,6 +106,31 @@ def _extract_proxy_litellm_metadata(
return litellm_metadata, user_api_key_auth
+def _requested_router(kwargs: Mapping[str, object]) -> "Router | None":
+ from litellm.router import Router
+
+ requested: Final = kwargs.get("litellm_router")
+ return requested if isinstance(requested, Router) else None
+
+
+def request_safeguards_evaluator(
+ kwargs: Mapping[str, object], messages: Sequence[Mapping[str, object]]
+) -> SafeguardsEvaluator | None:
+ safeguards: Final = kwargs.get("safeguards")
+ if requested_dangerous_tool_use(safeguards) is None or read_classifier_model_setting() is None:
+ return None
+ proxy_litellm_metadata, user_api_key_auth = _extract_proxy_litellm_metadata(kwargs)
+ return build_safeguards_evaluator(
+ safeguards=safeguards,
+ messages=messages,
+ litellm_metadata=proxy_litellm_metadata,
+ user_api_key_auth=user_api_key_auth,
+ llm_router=_router_or_proxy_fallback(_requested_router(kwargs)),
+ logging_kwargs=proxy_sub_call_logging_kwargs(kwargs),
+ parent_trace_id=proxy_sub_call_trace_id(kwargs),
+ )
+
+
async def _prepare_context_managed_request(
*,
model: str,
@@ -101,6 +142,8 @@ async def _prepare_context_managed_request(
additional_drop_params: list[str] | None,
llm_router: "Router | None",
user_api_key_auth: "UserAPIKeyAuth | None" = None,
+ sub_call_logging_kwargs: ProxySubCallLoggingKwargs | None = None,
+ parent_trace_id: str | None = None,
) -> PolyfillResult | None:
"""Apply client compaction history, then optional context_management polyfill."""
from litellm.llms.anthropic.pass_through.context_management.editors.compact import (
@@ -140,6 +183,8 @@ async def _prepare_context_managed_request(
additional_drop_params=additional_drop_params,
llm_router=llm_router,
user_api_key_auth=user_api_key_auth,
+ sub_call_logging_kwargs=sub_call_logging_kwargs,
+ parent_trace_id=parent_trace_id,
)
if polyfill_result is not None:
@@ -262,6 +307,8 @@ async def _run_polyfill_if_enabled(
additional_drop_params: list[str] | None,
llm_router: "Router | None",
user_api_key_auth: "UserAPIKeyAuth | None" = None,
+ sub_call_logging_kwargs: ProxySubCallLoggingKwargs | None = None,
+ parent_trace_id: str | None = None,
) -> PolyfillResult | None:
"""Run the async context_management polyfill if a spec is present.
@@ -288,6 +335,8 @@ async def _run_polyfill_if_enabled(
litellm_metadata=litellm_metadata,
llm_router=llm_router,
user_api_key_auth=user_api_key_auth,
+ sub_call_logging_kwargs=sub_call_logging_kwargs,
+ parent_trace_id=parent_trace_id,
)
except AnthropicContextManagementError:
# Surface validation errors so the endpoint can emit an Anthropic-format
@@ -622,12 +671,12 @@ class LiteLLMMessagesToCompletionTransformationHandler:
"""Handle non-Anthropic models asynchronously using the adapter"""
context_management: Final = kwargs.pop("context_management", None)
additional_drop_params: Final[list[str] | None] = kwargs.get("additional_drop_params", None)
- requested_router: Final[Router | None] = kwargs.pop("litellm_router", None)
- litellm_router: Final[Router | None] = (
- requested_router if requested_router is not None else _proxy_router_fallback()
- )
+ litellm_router: Final[Router | None] = _router_or_proxy_fallback(kwargs.pop("litellm_router", None))
proxy_litellm_metadata, user_api_key_auth = _extract_proxy_litellm_metadata(kwargs)
+ request_kwargs: Final = _REQUEST_KWARGS.validate_python(kwargs)
+ sub_call_logging_kwargs: Final = proxy_sub_call_logging_kwargs(request_kwargs)
+ sub_call_trace_id: Final = proxy_sub_call_trace_id(request_kwargs)
polyfill_result: Final = await _prepare_context_managed_request(
model=model,
@@ -639,10 +688,21 @@ class LiteLLMMessagesToCompletionTransformationHandler:
additional_drop_params=additional_drop_params,
llm_router=litellm_router,
user_api_key_auth=user_api_key_auth,
+ sub_call_logging_kwargs=sub_call_logging_kwargs,
+ parent_trace_id=sub_call_trace_id,
)
effective_messages: Final = polyfill_result.messages if polyfill_result is not None else messages
effective_system: Final = polyfill_result.system if polyfill_result is not None else system
+ safeguards_evaluator: Final = build_safeguards_evaluator(
+ safeguards=kwargs.get("safeguards"),
+ messages=messages,
+ litellm_metadata=proxy_litellm_metadata,
+ user_api_key_auth=user_api_key_auth,
+ llm_router=litellm_router,
+ logging_kwargs=sub_call_logging_kwargs,
+ parent_trace_id=sub_call_trace_id,
+ )
(
completion_kwargs,
@@ -675,6 +735,7 @@ class LiteLLMMessagesToCompletionTransformationHandler:
polyfill_result=polyfill_result,
is_async=True,
litellm_logging_obj=litellm_logging_obj_from_kwargs(kwargs),
+ safeguards_evaluator=safeguards_evaluator,
)
if transformed_stream is not None:
return transformed_stream
@@ -686,7 +747,7 @@ class LiteLLMMessagesToCompletionTransformationHandler:
polyfill_result=polyfill_result,
)
if anthropic_response is not None:
- return anthropic_response
+ return await with_safeguard_results(anthropic_response, safeguards_evaluator)
raise ValueError("Failed to transform response to Anthropic format")
@staticmethod
@@ -763,6 +824,7 @@ class LiteLLMMessagesToCompletionTransformationHandler:
polyfill_result: PolyfillResult | None = None
else:
proxy_litellm_metadata, user_api_key_auth = _extract_proxy_litellm_metadata(kwargs)
+ request_kwargs: Final = _REQUEST_KWARGS.validate_python(kwargs)
polyfill_result = run_async_function(
_prepare_context_managed_request,
model=model,
@@ -774,6 +836,8 @@ class LiteLLMMessagesToCompletionTransformationHandler:
additional_drop_params=additional_drop_params,
llm_router=litellm_router,
user_api_key_auth=user_api_key_auth,
+ sub_call_logging_kwargs=proxy_sub_call_logging_kwargs(request_kwargs),
+ parent_trace_id=proxy_sub_call_trace_id(request_kwargs),
)
effective_messages: Final = polyfill_result.messages if polyfill_result is not None else messages
diff --git a/litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py
index 636051d34ce..8d26b1e6387 100644
--- a/litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py
+++ b/litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py
@@ -15,12 +15,15 @@ from typing import (
get_args,
)
+from pydantic import TypeAdapter
from typing_extensions import assert_never
from litellm._logging import verbose_logger
from litellm._uuid import uuid
from litellm.exceptions import ContextWindowExceededError, MidStreamFallbackError
from litellm.litellm_core_utils.hidden_params import set_hidden_params
+from litellm.llms.anthropic.pass_through.safeguards import StreamedSafeguardResults
+from litellm.llms.anthropic.pass_through.utils import anthropic_sse_frame
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.types.llms.anthropic import (
AppliedEdit,
@@ -37,11 +40,13 @@ from litellm.types.utils import AdapterCompletionStreamWrapper, Delta
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
+ from litellm.llms.anthropic.pass_through.safeguards import SafeguardsEvaluator
from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import ModelResponseStream
_STREAMING_DELTA_TYPES: Final = frozenset(get_args(StreamingContentBlockDeltaType))
+_SSE_EVENT: Final = TypeAdapter(Mapping[str, object])
class _UsageDeltaWithIterations(UsageDelta, total=False):
@@ -342,6 +347,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
compaction_block: CompactionBlock | None = None,
iterations_usage: list[UsageIteration] | None = None,
litellm_logging_obj: "LiteLLMLoggingObject | None" = None,
+ safeguards_evaluator: "SafeguardsEvaluator | None" = None,
):
# Wrap the upstream stream so chunks that carry both content and a
# finish_reason (fake-streamed providers) are split into two — see
@@ -352,6 +358,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
if litellm_logging_obj is not None:
litellm_logging_obj.record_streamed_anthropic_message_id(self._message_id)
self.litellm_logging_obj = litellm_logging_obj
+ self.safeguards_evaluator: Final = safeguards_evaluator
# Mapping of truncated tool names to original names (for OpenAI's 64-char limit)
self.tool_name_mapping = tool_name_mapping or {}
# Polyfill applied_edits on final message_delta.
@@ -1048,14 +1055,17 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
Async version of anthropic_sse_wrapper.
Convert AnthropicStreamWrapper dict chunks to Server-Sent Events format.
"""
+ safeguard_results: Final = (
+ StreamedSafeguardResults(self.safeguards_evaluator) if self.safeguards_evaluator is not None else None
+ )
try:
async for chunk in self:
- if isinstance(chunk, dict):
- event_type: str = str(chunk.get("type", "message"))
- payload = f"event: {event_type}\ndata: {json.dumps(chunk)}\n\n"
- yield payload.encode()
- else:
+ if not isinstance(chunk, dict):
yield chunk
+ elif safeguard_results is None:
+ yield anthropic_sse_frame(chunk)
+ else:
+ yield anthropic_sse_frame(await safeguard_results.observe(_SSE_EVENT.validate_python(chunk)))
except Exception as e: # noqa: BLE001 # boundary before the socket: any upstream failure becomes an Anthropic error event
verbose_logger.exception("Anthropic Adapter - mid-stream error: %s", e)
logging_obj: Final = self.litellm_logging_obj
diff --git a/litellm/llms/anthropic/pass_through/adapters/transformation.py b/litellm/llms/anthropic/pass_through/adapters/transformation.py
index d2a0849ffef..1e276b7c9c7 100644
--- a/litellm/llms/anthropic/pass_through/adapters/transformation.py
+++ b/litellm/llms/anthropic/pass_through/adapters/transformation.py
@@ -206,6 +206,7 @@ from .streaming_iterator import AnthropicSSEStream, AnthropicStreamWrapper
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
+ from litellm.llms.anthropic.pass_through.safeguards import SafeguardsEvaluator
from litellm.types.llms.anthropic import ContentBlockContentBlockDict
ToolResultContent: TypeAlias = str | list[ToolMessageContentPart]
@@ -312,6 +313,7 @@ class AnthropicAdapter:
polyfill_result: PolyfillResult | None = None,
is_async: bool = True,
litellm_logging_obj: "LiteLLMLoggingObject | None" = None,
+ safeguards_evaluator: "SafeguardsEvaluator | None" = None,
) -> AsyncIterator[bytes] | Iterator[bytes] | None:
"""
Translate OpenAI streaming response to Anthropic format.
@@ -339,6 +341,7 @@ class AnthropicAdapter:
compaction_block=compaction_block,
iterations_usage=iterations_usage,
litellm_logging_obj=litellm_logging_obj,
+ safeguards_evaluator=safeguards_evaluator,
)
# Return the SSE-wrapped version for proper event formatting.
if is_async:
diff --git a/litellm/llms/anthropic/pass_through/caller_model_checks.py b/litellm/llms/anthropic/pass_through/caller_model_checks.py
new file mode 100644
index 00000000000..7eb6a8f9266
--- /dev/null
+++ b/litellm/llms/anthropic/pass_through/caller_model_checks.py
@@ -0,0 +1,498 @@
+"""Checks a proxy-made sub-call runs against the caller of the parent request.
+
+The compaction summary call and the safeguards classifier call go to a model the
+admin configured, not the one the client asked for, and they never pass back
+through the proxy's auth, budget, or rate-limit hooks. These checks apply the
+caller's model allowlists, per-model budgets, and RPM/TPM limits to that model
+before the sub-call runs.
+"""
+
+from collections.abc import Awaitable, Mapping, Sequence
+from typing import TYPE_CHECKING, Final, Literal, Optional, Protocol, TypeAlias
+
+from pydantic import TypeAdapter, ValidationError
+
+import litellm
+from litellm._logging import verbose_logger
+
+if TYPE_CHECKING:
+ from starlette.requests import Request
+
+ from litellm.proxy.hooks.parallel_request_limiter_v3 import (
+ RateLimitDescriptor,
+ RateLimitDescriptorRateLimitObject,
+ RateLimitResponse,
+ )
+ from litellm.router import Router
+ from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth
+
+CallerModelVerdict: TypeAlias = Literal["allowed", "denied", "denied_for_now"]
+_DEFAULT_SUB_CALL_ROUTE: Final = "/v1/messages"
+_MODEL_SCOPED_DESCRIPTOR_PREFIX: Final = "model_per_"
+_HEADERS: Final = TypeAdapter(Mapping[str, str])
+_LASTING_REFUSAL_STATUSES: Final = frozenset({401, 403})
+_LASTING_REFUSAL_CODES: Final = frozenset(str(status) for status in _LASTING_REFUSAL_STATUSES)
+
+
+class _TeamMemberModelBudgetLimiter(Protocol):
+ async def is_team_member_within_model_budget(
+ self,
+ user_id: str,
+ team_id: str,
+ team_member_model_max_budget: Mapping[str, object],
+ model: str,
+ ) -> bool: ...
+
+
+class _CreateRateLimitDescriptors(Protocol):
+ def __call__(
+ self,
+ *,
+ user_api_key_dict: "UserAPIKeyAuth",
+ data: Mapping[str, str],
+ rpm_limit_type: object,
+ tpm_limit_type: object,
+ model_has_failures: bool,
+ ) -> "Sequence[RateLimitDescriptor]": ...
+
+
+class _AddModelRateLimitDescriptor(Protocol):
+ def __call__(
+ self,
+ *,
+ user_api_key_dict: "UserAPIKeyAuth",
+ requested_model: str,
+ descriptors: "Sequence[RateLimitDescriptor]",
+ ) -> None: ...
+
+
+class _CreateOrgRateLimitDescriptors(Protocol):
+ def __call__(
+ self, user_api_key_dict: "UserAPIKeyAuth", requested_model: str | None = None
+ ) -> "Sequence[RateLimitDescriptor]": ...
+
+
+class _GetProxyHook(Protocol):
+ def __call__(self, hook: str) -> object: ...
+
+
+class _ShouldRateLimit(Protocol):
+ def __call__(
+ self,
+ *,
+ descriptors: "Sequence[RateLimitDescriptor]",
+ parent_otel_span: object,
+ read_only: bool,
+ ) -> "Awaitable[RateLimitResponse]": ...
+
+
+async def caller_model_verdict(
+ user_api_key_auth: Optional["UserAPIKeyAuth"],
+ model: str,
+ llm_router: Optional["Router"],
+ parent_litellm_metadata: Mapping[str, object] | None = None,
+ parent_trace_id: str | None = None,
+) -> CallerModelVerdict:
+ """``denied`` when an allowlist or a per-model budget refuses ``model`` for the caller, which only an
+ admin or the next budget window changes; ``denied_for_now`` when a rate-limit window is full or a check
+ could not run, which the caller's next request may clear."""
+ access: Final = await caller_can_call_model(
+ user_api_key_auth=user_api_key_auth,
+ model=model,
+ llm_router=llm_router,
+ parent_litellm_metadata=parent_litellm_metadata,
+ parent_trace_id=parent_trace_id,
+ )
+ if access != "allowed":
+ return access
+ budget: Final = await caller_within_model_budget(user_api_key_auth=user_api_key_auth, model=model)
+ if budget != "allowed":
+ return budget
+ return await caller_within_model_rate_limit(user_api_key_auth=user_api_key_auth, model=model)
+
+
+async def caller_can_call_model(
+ user_api_key_auth: Optional["UserAPIKeyAuth"],
+ model: str,
+ llm_router: Optional["Router"],
+ parent_litellm_metadata: Mapping[str, object] | None = None,
+ parent_trace_id: str | None = None,
+) -> CallerModelVerdict:
+ """Run the proxy's own model-access checks for ``model`` against the caller of the parent request.
+
+ These are the checks the parent request passed for its own model, so every scope they cover (key, team,
+ team member, user, project, end user, managed agent grants) applies to the sub-call's model too, and a
+ deployment that skips them for the parent (no-auth dev mode, custom auth without common checks) skips
+ them here. Budgets are checked too, so a spent budget the parent skipped for a free model still refuses
+ a paid one. ``denied`` answers a budget refusal or a 401 or 403, which only an admin or the next budget
+ window changes. ``denied_for_now`` answers any other failure, such as a team read that 404s, since the
+ parent passed the same checks this turn and the next turn may clear it. ``allowed`` answers a caller
+ outside the proxy.
+ """
+ if user_api_key_auth is None:
+ return "allowed"
+ try:
+ from fastapi import HTTPException
+
+ from litellm.proxy._types import ProxyException
+ from litellm.proxy.auth.user_api_key_auth import (
+ enforce_key_and_fallback_model_access, # pyright: ignore[reportUnknownVariableType] # takes the bare dict and list the proxy passes
+ run_centralized_common_checks,
+ )
+ from litellm.proxy.proxy_server import (
+ llm_model_list, # pyright: ignore[reportUnknownVariableType] # proxy global has untyped deployment entries
+ )
+ except ImportError:
+ return "allowed"
+
+ route: Final = user_api_key_auth.request_route or _DEFAULT_SUB_CALL_ROUTE
+ try:
+ request: Final = _sub_call_request(route, parent_litellm_metadata)
+ request_data: Final = _sub_call_request_data(model, parent_litellm_metadata, parent_trace_id)
+ caller: Final = user_api_key_auth.model_copy(update={"budget_reservation": None})
+ await enforce_key_and_fallback_model_access(
+ valid_token=caller,
+ request_data=request_data,
+ route=route,
+ request=request,
+ llm_model_list=llm_model_list, # pyright: ignore[reportUnknownArgumentType] # proxy global has untyped deployment entries
+ llm_router=llm_router,
+ )
+ await run_centralized_common_checks(caller, request, request_data, route, sub_call=True)
+ except litellm.BudgetExceededError:
+ return "denied"
+ except HTTPException as e:
+ return "denied" if e.status_code in _LASTING_REFUSAL_STATUSES else _check_failed_this_turn(model, e)
+ except ProxyException as e:
+ return "denied" if e.code in _LASTING_REFUSAL_CODES else _check_failed_this_turn(model, e)
+ except Exception as e: # noqa: BLE001 # a check that cannot run must not invoke the model with the caller's limits dropped
+ return _check_failed_this_turn(model, e)
+ return "allowed"
+
+
+def _check_failed_this_turn(model: str, error: Exception) -> CallerModelVerdict:
+ verbose_logger.warning(
+ "proxy sub-call: model access check for model=%s failed this turn; denying for now: %s", model, error
+ )
+ return "denied_for_now"
+
+
+def _sub_call_request(route: str, parent_litellm_metadata: Mapping[str, object] | None) -> "Request":
+ from starlette.requests import Request
+
+ headers: Final = _parent_headers(parent_litellm_metadata)
+ return Request(
+ scope={
+ "type": "http",
+ "method": "POST",
+ "path": route,
+ "scheme": "http",
+ "query_string": b"",
+ "root_path": "",
+ "headers": tuple(
+ (name.lower().encode("latin-1"), value.encode("latin-1")) for name, value in headers.items()
+ ),
+ }
+ )
+
+
+def _parent_headers(parent_litellm_metadata: Mapping[str, object] | None) -> Mapping[str, str]:
+ if not parent_litellm_metadata:
+ return {}
+ try:
+ return _HEADERS.validate_python(parent_litellm_metadata.get("headers") or {})
+ except ValidationError:
+ return {}
+
+
+def _sub_call_request_data(
+ model: str, parent_litellm_metadata: Mapping[str, object] | None, parent_trace_id: str | None
+) -> dict[str, object]: # mutable-ok: the proxy's checks merge metadata and tags into the body they are handed
+ parent: Final[Mapping[str, object]] = parent_litellm_metadata or {}
+ end_user_id: Final = parent.get("user_api_key_end_user_id")
+ metadata_trace_id: Final = parent.get("trace_id")
+ trace_id: Final = metadata_trace_id if isinstance(metadata_trace_id, str) and metadata_trace_id else parent_trace_id
+ return {
+ "model": model,
+ **({"user": end_user_id} if isinstance(end_user_id, str) and end_user_id else {}),
+ **({"litellm_trace_id": trace_id} if isinstance(trace_id, str) and trace_id else {}),
+ }
+
+
+async def caller_within_model_budget(
+ user_api_key_auth: Optional["UserAPIKeyAuth"],
+ model: str,
+) -> CallerModelVerdict:
+ """Return ``allowed`` when the caller is within their per-model budget for ``model``.
+
+ Mirrors the per-model budget enforcement that ``user_api_key_auth`` runs for
+ the client-requested model. Returns ``allowed`` outside the proxy or when no
+ per-model budget is configured, ``denied`` over a budget, and ``denied_for_now``
+ when a budget read fails.
+
+ Every scope is checked because the sub-call's spend is charged to every
+ scope: the key, team, user, team-member and end-user budgets ride into the
+ sub-call's metadata, so skipping one of them would let the sub-call
+ increment a counter it can never be refused by.
+ """
+ if user_api_key_auth is None:
+ return "allowed"
+ try:
+ from litellm.proxy.proxy_server import model_max_budget_limiter
+ except Exception:
+ return "allowed"
+
+ model_max_budget: Final = getattr(user_api_key_auth, "model_max_budget", None)
+ token: Final = getattr(user_api_key_auth, "token", None)
+ if isinstance(model_max_budget, dict) and model_max_budget and token is not None:
+ try:
+ await model_max_budget_limiter.is_key_within_model_budget(
+ user_api_key_dict=user_api_key_auth,
+ model=model,
+ )
+ except litellm.BudgetExceededError:
+ return "denied"
+ except Exception as e:
+ verbose_logger.warning(
+ "proxy sub-call: unexpected error during key model-budget check for model=%s; denying: %s",
+ model,
+ e,
+ )
+ return "denied_for_now"
+
+ user_model_max_budget: Final = user_api_key_auth.user_model_max_budget
+ user_id: Final = user_api_key_auth.user_id
+ if isinstance(user_model_max_budget, dict) and user_model_max_budget and user_id is not None:
+ try:
+ await model_max_budget_limiter.is_user_within_model_budget(
+ user_id=user_id,
+ user_model_max_budget=user_model_max_budget,
+ model=model,
+ )
+ except litellm.BudgetExceededError:
+ return "denied"
+ except Exception as e: # noqa: BLE001 # a budget gate denies on any failure, as the key and end-user scopes do
+ verbose_logger.warning(
+ "proxy sub-call: unexpected error during user model-budget check for model=%s; denying: %s",
+ model,
+ e,
+ )
+ return "denied_for_now"
+
+ team_model_max_budget: Final = user_api_key_auth.team_model_max_budget
+ team_id: Final = user_api_key_auth.team_id
+ if isinstance(team_model_max_budget, dict) and team_model_max_budget and team_id is not None:
+ try:
+ await model_max_budget_limiter.is_team_within_model_budget(
+ team_id=team_id,
+ team_model_max_budget=team_model_max_budget,
+ key_model_max_budget=model_max_budget if isinstance(model_max_budget, dict) else None,
+ model=model,
+ )
+ except litellm.BudgetExceededError:
+ return "denied"
+ except Exception as e: # noqa: BLE001 # a budget gate denies on any failure, as the other scopes do
+ verbose_logger.warning(
+ "proxy sub-call: unexpected error during team model-budget check for model=%s; denying: %s",
+ model,
+ e,
+ )
+ return "denied_for_now"
+
+ end_user_model_max_budget: Final[dict[str, object] | None] = getattr(
+ user_api_key_auth, "end_user_model_max_budget", None
+ )
+ end_user_id: Final[str | None] = getattr(user_api_key_auth, "end_user_id", None)
+ if isinstance(end_user_model_max_budget, dict) and end_user_model_max_budget and end_user_id is not None:
+ try:
+ await model_max_budget_limiter.is_end_user_within_model_budget(
+ end_user_id=end_user_id,
+ end_user_model_max_budget=end_user_model_max_budget,
+ model=model,
+ )
+ except litellm.BudgetExceededError:
+ return "denied"
+ except Exception as e:
+ verbose_logger.warning(
+ "proxy sub-call: unexpected error during end-user model-budget check for model=%s; denying: %s",
+ model,
+ e,
+ )
+ return "denied_for_now"
+
+ return await _caller_within_team_member_model_budget(
+ user_api_key_auth=user_api_key_auth,
+ model_max_budget_limiter=model_max_budget_limiter,
+ model=model,
+ )
+
+
+async def _caller_within_team_member_model_budget(
+ user_api_key_auth: "UserAPIKeyAuth",
+ model_max_budget_limiter: _TeamMemberModelBudgetLimiter,
+ model: str,
+) -> CallerModelVerdict:
+ team_member_model_max_budget: Final = user_api_key_auth.team_member_model_max_budget
+ user_id: Final = user_api_key_auth.user_id
+ team_id: Final = user_api_key_auth.team_id
+ if (
+ not isinstance(team_member_model_max_budget, Mapping)
+ or not team_member_model_max_budget
+ or not isinstance(user_id, str)
+ or not isinstance(team_id, str)
+ ):
+ return "allowed"
+ try:
+ await model_max_budget_limiter.is_team_member_within_model_budget(
+ user_id=user_id,
+ team_id=team_id,
+ team_member_model_max_budget=team_member_model_max_budget,
+ model=model,
+ )
+ except litellm.BudgetExceededError:
+ return "denied"
+ except Exception as e: # noqa: BLE001 # a budget gate denies on any failure, as the other scopes do
+ verbose_logger.warning(
+ "proxy sub-call: unexpected error during team member model-budget check for model=%s; denying: %s",
+ model,
+ e,
+ )
+ return "denied_for_now"
+ return "allowed"
+
+
+def _sub_call_descriptors(
+ descriptors: "Sequence[RateLimitDescriptor]",
+) -> "tuple[RateLimitDescriptor, ...]":
+ return tuple(_sub_call_descriptor(descriptor) for descriptor in descriptors)
+
+
+def _sub_call_descriptor(descriptor: "RateLimitDescriptor") -> "RateLimitDescriptor":
+ rate_limit: Final = descriptor.get("rate_limit")
+ if rate_limit is None:
+ return descriptor
+ caller_wide: Final = not descriptor["key"].startswith(_MODEL_SCOPED_DESCRIPTOR_PREFIX)
+ windowed_limits: Final[RateLimitDescriptorRateLimitObject] = {
+ "requests_per_unit": None if caller_wide else rate_limit.get("requests_per_unit"),
+ "tokens_per_unit": rate_limit.get("tokens_per_unit"),
+ "window_size": rate_limit.get("window_size"),
+ }
+ return {**descriptor, "rate_limit": windowed_limits}
+
+
+async def caller_within_model_rate_limit(
+ user_api_key_auth: Optional["UserAPIKeyAuth"],
+ model: str,
+) -> CallerModelVerdict:
+ """Return ``allowed`` when the caller is within their configured RPM/TPM limits for ``model``.
+
+ This mirrors the read side of
+ ``_PROXY_MaxParallelRequestsHandler_v3.async_pre_call_hook`` for ``model``:
+ it builds the same descriptor set and runs the check in ``read_only`` mode
+ so no counter is reserved or incremented. The sub-call's tokens are still
+ charged once by the limiter's post-call success hook (via the propagated
+ ``litellm_metadata``), but it takes no request slot, so it does not count
+ toward the caller's RPM. ``max_parallel_requests`` gauges and the RPM of
+ every caller-wide scope (key, user, team, org, ...) are left out of the
+ check: the sub-call runs inside the caller's already admitted request,
+ whose own slot would otherwise count against it. The RPM of a limit on
+ ``model`` itself still gates the sub-call, as do token limits.
+
+ Returns ``allowed`` outside the proxy, when the active limiter does not
+ expose the read-only descriptor check (legacy limiter), or when the
+ descriptor set cannot be built. ``denied_for_now`` comes from a definitive
+ ``OVER_LIMIT`` response, which the next window clears, and from the limiter's own fail-closed rejection
+ (``RateLimitUnverifiableError``, raised when ``fail_closed_rate_limit_enforcement``
+ is on and the counters could not be verified), so any other internal error
+ here lets the sub-call run rather than blocking every one.
+ """
+ if user_api_key_auth is None:
+ return "allowed"
+ try:
+ from litellm.proxy.hooks.parallel_request_limiter_v3 import RateLimitUnverifiableError
+ from litellm.proxy.proxy_server import proxy_logging_obj
+ except Exception:
+ return "allowed"
+
+ get_proxy_hook: Final[_GetProxyHook | None] = getattr(proxy_logging_obj, "get_proxy_hook", None)
+ limiter: Final[object] = get_proxy_hook("parallel_request_limiter") if get_proxy_hook is not None else None
+ return await rate_limit_verdict(
+ user_api_key_auth=user_api_key_auth, model=model, limiter=limiter, unverifiable=RateLimitUnverifiableError
+ )
+
+
+async def rate_limit_verdict(
+ user_api_key_auth: "UserAPIKeyAuth",
+ model: str,
+ limiter: object,
+ unverifiable: type[Exception],
+) -> CallerModelVerdict:
+ should_rate_limit_check: Final[_ShouldRateLimit | None] = getattr(limiter, "should_rate_limit", None)
+ create_descriptors: Final[_CreateRateLimitDescriptors | None] = getattr(
+ limiter, "create_rate_limit_descriptors", None
+ )
+ add_team_descriptor: Final[_AddModelRateLimitDescriptor | None] = getattr(
+ limiter, "_add_team_model_rate_limit_descriptor_from_metadata", None
+ )
+ add_project_descriptor: Final[_AddModelRateLimitDescriptor | None] = getattr(
+ limiter, "_add_project_model_rate_limit_descriptor_from_metadata", None
+ )
+ create_org_descriptors: Final[_CreateOrgRateLimitDescriptors | None] = getattr(
+ limiter, "create_organization_rate_limit_descriptor", None
+ )
+ if (
+ limiter is None
+ or should_rate_limit_check is None
+ or create_descriptors is None
+ or add_team_descriptor is None
+ or add_project_descriptor is None
+ or create_org_descriptors is None
+ ):
+ return "allowed"
+
+ try:
+ metadata: Final[Mapping[str, object]] = getattr(user_api_key_auth, "metadata", None) or {}
+ data: Final = {"model": model}
+ base_descriptors: Final = create_descriptors(
+ user_api_key_dict=user_api_key_auth,
+ data=data,
+ rpm_limit_type=metadata.get("rpm_limit_type"),
+ tpm_limit_type=metadata.get("tpm_limit_type"),
+ model_has_failures=False,
+ )
+ add_team_descriptor(
+ user_api_key_dict=user_api_key_auth,
+ requested_model=model,
+ descriptors=base_descriptors,
+ )
+ add_project_descriptor(
+ user_api_key_dict=user_api_key_auth,
+ requested_model=model,
+ descriptors=base_descriptors,
+ )
+ descriptors: Final = _sub_call_descriptors(
+ (*base_descriptors, *create_org_descriptors(user_api_key_auth, model))
+ )
+ if not descriptors:
+ return "allowed"
+ parent_otel_span: Final[object] = getattr(user_api_key_auth, "parent_otel_span", None)
+ response: Final[RateLimitResponse] = await should_rate_limit_check(
+ descriptors=descriptors,
+ parent_otel_span=parent_otel_span,
+ read_only=True,
+ )
+ except unverifiable as e:
+ verbose_logger.warning(
+ "proxy sub-call: rate-limit counters for model=%s could not be verified; denying for now: %s",
+ model,
+ e,
+ )
+ return "denied_for_now"
+ except Exception as e:
+ verbose_logger.warning(
+ "proxy sub-call: unexpected error during rate-limit check for model=%s; allowing: %s",
+ model,
+ e,
+ )
+ return "allowed"
+ return "allowed" if response.get("overall_code") != "OVER_LIMIT" else "denied_for_now"
diff --git a/litellm/llms/anthropic/pass_through/context_management/dispatcher.py b/litellm/llms/anthropic/pass_through/context_management/dispatcher.py
index 9b4e12ef2f9..27b7412292e 100644
--- a/litellm/llms/anthropic/pass_through/context_management/dispatcher.py
+++ b/litellm/llms/anthropic/pass_through/context_management/dispatcher.py
@@ -13,6 +13,7 @@ from .editors import apply_clear_tool_uses_20250919, apply_compact_20260112
from .result import PolyfillResult
if TYPE_CHECKING:
+ from litellm.llms.anthropic.pass_through.utils import ProxySubCallLoggingKwargs
from litellm.router import Router
from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth
@@ -82,6 +83,8 @@ async def apply_context_management(
litellm_metadata: dict[str, object] | None = None,
llm_router: "Router | None" = None,
user_api_key_auth: "UserAPIKeyAuth | None" = None,
+ sub_call_logging_kwargs: "ProxySubCallLoggingKwargs | None" = None,
+ parent_trace_id: str | None = None,
) -> PolyfillResult:
"""Run edits in order; return a single ``PolyfillResult``.
@@ -123,6 +126,8 @@ async def apply_context_management(
litellm_metadata=litellm_metadata,
llm_router=llm_router,
user_api_key_auth=user_api_key_auth,
+ sub_call_logging_kwargs=sub_call_logging_kwargs,
+ parent_trace_id=parent_trace_id,
)
if editor_is_async
else await asyncify(editor)(
diff --git a/litellm/llms/anthropic/pass_through/context_management/editors/compact.py b/litellm/llms/anthropic/pass_through/context_management/editors/compact.py
index deb3d849ae9..09f49e9b663 100644
--- a/litellm/llms/anthropic/pass_through/context_management/editors/compact.py
+++ b/litellm/llms/anthropic/pass_through/context_management/editors/compact.py
@@ -14,13 +14,23 @@ Mirrors Anthropic's native ``compact_20260112`` for non-Anthropic providers:
import re
from collections.abc import Awaitable, Mapping, Sequence
-from typing import TYPE_CHECKING, Final, Literal, Optional, Protocol, TypeVar, Union, cast
+from typing import TYPE_CHECKING, Final, Optional, Protocol, TypeVar, Union, cast
from typing_extensions import NotRequired, ReadOnly, TypedDict, Unpack
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.asyncify import asyncify
+from litellm.llms.anthropic.pass_through.caller_model_checks import (
+ caller_can_call_model,
+ caller_within_model_budget,
+ caller_within_model_rate_limit,
+)
+from litellm.llms.anthropic.pass_through.utils import (
+ ProxySubCallLoggingKwargs,
+ proxy_general_settings,
+ proxy_sub_call_metadata,
+)
from litellm.types.llms.anthropic import (
AppliedEdit,
CompactionBlock,
@@ -29,11 +39,6 @@ from litellm.types.llms.anthropic import (
if TYPE_CHECKING:
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
- from litellm.proxy.hooks.parallel_request_limiter_v3 import (
- RateLimitDescriptor,
- RateLimitDescriptorRateLimitObject,
- RateLimitResponse,
- )
from litellm.router import Router
from litellm.types.llms.anthropic import (
AllAnthropicPassThroughMessageValues,
@@ -58,38 +63,6 @@ from ..constants import (
from ..errors import AnthropicContextManagementError
from ..result import PolyfillResult
-# Auth metadata fields propagated from the parent request to the summary call
-# so the summary's spend is attributed to the same scopes. The list mirrors the
-# fields populated by
-# ``LiteLLMProxyRequestSetup.add_user_api_key_auth_to_request_metadata``.
-# The three ``*_model_max_budget`` fields
-# are what ``_PROXY_VirtualKeyModelMaxBudgetLimiter`` reads post-call to update
-# the per-model spend caches, so without them the summary spend would never
-# count against the caller's model budget. ``user_api_key_end_user_id`` /
-# ``user_api_key_project_id`` are the scope identifiers the post-call spend hook
-# and rate limiter key their counters on, and ``user_api_end_user_max_budget``
-# is the end-user budget the cost callback enforces — without these the summary
-# tokens escape the caller's end-user/project budgets and counters.
-_PROPAGATED_METADATA_KEYS: Final = (
- "user_api_key",
- "user_api_key_alias",
- "user_api_key_team_id",
- "user_api_key_team_alias",
- "user_api_key_user_id",
- "user_api_key_user_email",
- "user_api_key_org_id",
- "user_api_key_project_id",
- "user_api_key_end_user_id",
- "user_api_end_user_max_budget",
- "user_api_key_model_max_budget",
- "user_api_key_team_model_max_budget",
- "user_api_key_user_model_max_budget",
- "user_api_key_team_member_model_max_budget",
- "user_api_key_end_user_model_max_budget",
- "litellm_call_id",
- "litellm_parent_otel_span",
-)
-
_SUMMARY_TAG_RE: Final = re.compile(r"AccessDeniedSignatureDoesNotMatchSignatureDoesNotMatch{emptyMessage}
; @@ -237,6 +268,7 @@ export function AccessGroupDetail({ accessGroupId, onBack }: AccessGroupDetailPr