diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py
index e8f3b305139..301cf970caf 100644
--- a/litellm/integrations/opentelemetry.py
+++ b/litellm/integrations/opentelemetry.py
@@ -363,6 +363,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
logger_provider: object | None = None,
meter_provider: object | None = None,
max_dynamic_tracer_providers: int = _MAX_DYNAMIC_TRACER_PROVIDERS,
+ register_on_proxy: bool = True,
**kwargs,
):
team_metadata_keys_override: Final = kwargs.pop("baggage_team_metadata_keys", None)
@@ -414,7 +415,8 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
# Sample env-var / config / message_logging at init so subsequent
# _capture_in_span / _capture_in_event calls are deterministic.
self._capture_mode_cached = self._compute_capture_mode_from_init_state()
- self._init_otel_logger_on_litellm_proxy()
+ if register_on_proxy:
+ self._init_otel_logger_on_litellm_proxy()
@staticmethod
def _get_litellm_resource(config: OpenTelemetryConfig) -> "_Resource":
diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py
index d2a32ef73b6..66be53c4dc6 100644
--- a/litellm/integrations/otel/logger.py
+++ b/litellm/integrations/otel/logger.py
@@ -172,6 +172,7 @@ class OpenTelemetryV2(CustomLogger):
tracer_provider: TracerProvider | None = None,
logger_provider: LoggerProvider | None = None,
meter_provider: "MeterProvider | None" = None,
+ register_on_proxy: bool = True,
**kwargs: Any,
) -> None:
super().__init__(**kwargs)
@@ -191,7 +192,8 @@ class OpenTelemetryV2(CustomLogger):
)
self._tenant_tracers = TenantTracerCache(self.config, callback_name, LITELLM_TRACER_NAME)
self._open_llm_calls: OrderedDict[str, _LLMCallSpan] = OrderedDict()
- self._init_otel_logger_on_litellm_proxy()
+ if register_on_proxy:
+ self._init_otel_logger_on_litellm_proxy()
def _init_metrics(self, meter_provider: "MeterProvider | None") -> "GenAIMetricRecorder | None":
"""Create the six GenAI histograms when metrics are enabled, else ``None``.
diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py
index a8672b5c112..558254950ba 100644
--- a/litellm/litellm_core_utils/litellm_logging.py
+++ b/litellm/litellm_core_utils/litellm_logging.py
@@ -246,6 +246,16 @@ else:
_PAGERDUTY_ALERTING_FACTORY: Final = PagerDutyAlerting
_in_memory_loggers: Final[list[CustomLogger]] = []
+
+def _register_dashboard_callback(callback: CustomLogger, callback_name: str) -> None:
+ """Record the dashboard identity of a factory-created callback."""
+ from litellm.litellm_core_utils.logging_callback_manager import (
+ register_dashboard_callback,
+ )
+
+ register_dashboard_callback(callback, callback_name)
+
+
_STANDARD_LOGGING_METADATA_KEYS: Final[frozenset[str]] = frozenset(StandardLoggingMetadata.__annotations__.keys())
### GLOBAL VARIABLES ###
@@ -651,6 +661,7 @@ class Logging(LiteLLMLoggingBaseClass):
internal_usage_cache=None,
llm_router=None,
custom_logger_init_args=_custom_logger_init_args,
+ **_get_request_local_otel_init_options(callback),
)
if callback_class is None:
return ()
@@ -659,7 +670,13 @@ class Logging(LiteLLMLoggingBaseClass):
# resolve the name again without creds so the trace logger (OTel v2 /
# legacy agent) keeps receiving this request.
_newrelic_trace_class: Final = (
- _init_custom_logger_compatible_class(callback, internal_usage_cache=None, llm_router=None)
+ _init_custom_logger_compatible_class(
+ callback,
+ internal_usage_cache=None,
+ llm_router=None,
+ custom_logger_init_args=None,
+ **_get_request_local_otel_init_options(callback),
+ )
if callback == "newrelic" and _custom_logger_init_args and _custom_logger_init_args.get("newrelic_api_key")
else None
)
@@ -1062,6 +1079,8 @@ class Logging(LiteLLMLoggingBaseClass):
logging_integration=callback_name,
internal_usage_cache=None,
llm_router=None,
+ custom_logger_init_args=None,
+ **_get_request_local_otel_init_options(callback_name),
)
if custom_logger is not None:
self.model_call_details["prompt_integration"] = model.split("/")[0]
@@ -4081,11 +4100,25 @@ def set_callbacks(callback_list, function_id=None):
raise e
+def _get_request_local_otel_init_options(callback_name: str) -> dict[str, bool]:
+ """Prevent a request-local generic OTel logger from becoming proxy-global."""
+ if callback_name != "otel":
+ return {} # mutable-ok: factory accepts a mutable keyword mapping
+ return { # mutable-ok: factory accepts a mutable keyword mapping
+ "register_dashboard_provenance": False,
+ "register_on_proxy": False,
+ "cache_in_memory": False,
+ }
+
+
def _init_custom_logger_compatible_class(
logging_integration: _custom_logger_compatible_callbacks_literal,
internal_usage_cache: DualCache | None,
llm_router: Any | None, # expect litellm.Router, but typing errors due to circular import
custom_logger_init_args: dict | None = {},
+ register_dashboard_provenance: bool = True,
+ register_on_proxy: bool = True,
+ cache_in_memory: bool = True,
) -> CustomLogger | None:
"""
Initialize a custom logger compatible class
@@ -4093,7 +4126,13 @@ def _init_custom_logger_compatible_class(
try:
custom_logger_init_args = custom_logger_init_args or {}
if logging_integration == "agentops": # Add AgentOps initialization
- _v2 = _maybe_construct_otel_v2("agentops", _in_memory_loggers)
+ _v2 = _maybe_construct_otel_v2(
+ "agentops",
+ _in_memory_loggers,
+ register_dashboard_provenance=register_dashboard_provenance,
+ register_on_proxy=register_on_proxy,
+ cache_in_memory=cache_in_memory,
+ )
if _v2 is not None:
return _v2
for callback in _in_memory_loggers:
@@ -4266,7 +4305,13 @@ def _init_custom_logger_compatible_class(
_in_memory_loggers.append(_opik_logger)
return _opik_logger
elif logging_integration == "arize":
- _v2 = _maybe_construct_otel_v2("arize", _in_memory_loggers)
+ _v2 = _maybe_construct_otel_v2(
+ "arize",
+ _in_memory_loggers,
+ register_dashboard_provenance=register_dashboard_provenance,
+ register_on_proxy=register_on_proxy,
+ cache_in_memory=cache_in_memory,
+ )
if _v2 is not None:
return _v2
from litellm.integrations.opentelemetry import (
@@ -4291,11 +4336,22 @@ def _init_custom_logger_compatible_class(
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)
+ _arize_otel_logger: Final = ArizeLogger(
+ config=otel_config,
+ callback_name="arize",
+ register_on_proxy=register_on_proxy,
+ )
+ if cache_in_memory:
+ _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)
+ _v2 = _maybe_construct_otel_v2(
+ "arize_phoenix",
+ _in_memory_loggers,
+ register_dashboard_provenance=register_dashboard_provenance,
+ register_on_proxy=register_on_proxy,
+ cache_in_memory=cache_in_memory,
+ )
if _v2 is not None:
return _v2
from litellm.integrations.opentelemetry import (
@@ -4317,11 +4373,22 @@ def _init_custom_logger_compatible_class(
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)
+ _arize_phoenix_otel_logger: Final = ArizePhoenixLogger(
+ config=otel_config,
+ callback_name="arize_phoenix",
+ register_on_proxy=register_on_proxy,
+ )
+ if cache_in_memory:
+ _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)
+ _v2 = _maybe_construct_otel_v2(
+ "levo",
+ _in_memory_loggers,
+ register_dashboard_provenance=register_dashboard_provenance,
+ register_on_proxy=register_on_proxy,
+ cache_in_memory=cache_in_memory,
+ )
if _v2 is not None:
return _v2
from litellm.integrations.levo.levo import LevoLogger
@@ -4342,8 +4409,13 @@ def _init_custom_logger_compatible_class(
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)
+ _levo_otel_logger: Final = LevoLogger(
+ config=otel_config,
+ callback_name="levo",
+ register_on_proxy=register_on_proxy,
+ )
+ if cache_in_memory:
+ _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,
@@ -4357,29 +4429,51 @@ def _init_custom_logger_compatible_class(
from litellm.integrations.otel.logger import OpenTelemetryV2
for callback in _in_memory_loggers:
- if type(callback) is OpenTelemetryV2:
+ if type(callback) is OpenTelemetryV2 and getattr(callback, "callback_name", None) == "otel":
+ if register_dashboard_provenance:
+ _register_dashboard_callback(callback, logging_integration)
return callback
+ otel_settings: Final = _get_custom_logger_settings_from_proxy_server(callback_name=logging_integration)
+ otel_settings.setdefault("callback_name", "otel")
+ otel_settings.pop("register_on_proxy", None)
otel_logger_v2: Final = OpenTelemetryV2(
- **_get_custom_logger_settings_from_proxy_server(callback_name=logging_integration)
+ register_on_proxy=register_on_proxy,
+ **otel_settings,
)
- _in_memory_loggers.append(otel_logger_v2)
- _maybe_auto_initialize_arize_phoenix(_in_memory_loggers)
+ if register_dashboard_provenance:
+ _register_dashboard_callback(otel_logger_v2, logging_integration)
+ if cache_in_memory:
+ _in_memory_loggers.append(otel_logger_v2)
+ if register_on_proxy:
+ _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:
+ if type(callback) is OpenTelemetry and getattr(callback, "callback_name", None) == "otel":
+ if register_dashboard_provenance:
+ _register_dashboard_callback(callback, logging_integration)
return callback
- otel_logger: Final = OpenTelemetry(
- **_get_custom_logger_settings_from_proxy_server(callback_name=logging_integration)
+ legacy_otel_settings: Final = _get_custom_logger_settings_from_proxy_server(
+ callback_name=logging_integration
)
- _in_memory_loggers.append(otel_logger)
+ legacy_otel_settings.setdefault("callback_name", "otel")
+ legacy_otel_settings.pop("register_on_proxy", None)
+ otel_logger: Final = OpenTelemetry(
+ register_on_proxy=register_on_proxy,
+ **legacy_otel_settings,
+ )
+ if register_dashboard_provenance:
+ _register_dashboard_callback(otel_logger, logging_integration)
+ if cache_in_memory:
+ _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)
+ if register_on_proxy:
+ _maybe_auto_initialize_arize_phoenix(_in_memory_loggers)
return otel_logger
@@ -4455,8 +4549,12 @@ def _init_custom_logger_compatible_class(
# 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)
+ _otel_logger = OpenTelemetry(
+ config=otel_config,
+ register_on_proxy=register_on_proxy,
+ )
+ if cache_in_memory:
+ _in_memory_loggers.append(_otel_logger)
return _otel_logger
elif logging_integration == "dynamic_rate_limiter":
from litellm.proxy.hooks.dynamic_rate_limiter import (
@@ -4497,7 +4595,13 @@ def _init_custom_logger_compatible_class(
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)
+ _v2 = _maybe_construct_otel_v2(
+ "langtrace",
+ _in_memory_loggers,
+ register_dashboard_provenance=register_dashboard_provenance,
+ register_on_proxy=register_on_proxy,
+ cache_in_memory=cache_in_memory,
+ )
if _v2 is not None:
return _v2
@@ -4514,8 +4618,13 @@ def _init_custom_logger_compatible_class(
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)
+ _otel_logger = OpenTelemetry(
+ config=otel_config,
+ callback_name="langtrace",
+ register_on_proxy=register_on_proxy,
+ )
+ if cache_in_memory:
+ _in_memory_loggers.append(_otel_logger)
return _otel_logger
elif logging_integration == "mlflow":
@@ -4535,7 +4644,13 @@ def _init_custom_logger_compatible_class(
_in_memory_loggers.append(langfuse_logger)
return langfuse_logger
elif logging_integration == "langfuse_otel":
- _v2 = _maybe_construct_otel_v2("langfuse_otel", _in_memory_loggers)
+ _v2 = _maybe_construct_otel_v2(
+ "langfuse_otel",
+ _in_memory_loggers,
+ register_dashboard_provenance=register_dashboard_provenance,
+ register_on_proxy=register_on_proxy,
+ cache_in_memory=cache_in_memory,
+ )
if _v2 is not None:
return _v2
from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
@@ -4545,11 +4660,22 @@ def _init_custom_logger_compatible_class(
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)
+ _otel_logger = LangfuseOtelLogger(
+ config=None,
+ callback_name="langfuse_otel",
+ register_on_proxy=register_on_proxy,
+ )
+ if cache_in_memory:
+ _in_memory_loggers.append(_otel_logger)
return _otel_logger
elif logging_integration == "weave_otel":
- _v2 = _maybe_construct_otel_v2("weave_otel", _in_memory_loggers)
+ _v2 = _maybe_construct_otel_v2(
+ "weave_otel",
+ _in_memory_loggers,
+ register_dashboard_provenance=register_dashboard_provenance,
+ register_on_proxy=register_on_proxy,
+ cache_in_memory=cache_in_memory,
+ )
if _v2 is not None:
return _v2
from litellm.integrations.opentelemetry import OpenTelemetryConfig
@@ -4569,8 +4695,13 @@ def _init_custom_logger_compatible_class(
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)
+ _otel_logger = WeaveOtelLogger(
+ config=otel_config,
+ callback_name="weave_otel",
+ register_on_proxy=register_on_proxy,
+ )
+ if cache_in_memory:
+ _in_memory_loggers.append(_otel_logger)
return _otel_logger
elif logging_integration == "pagerduty":
for callback in _in_memory_loggers:
@@ -4696,7 +4827,13 @@ def _init_custom_logger_compatible_class(
in_memory_dynamic_logger_cache=in_memory_dynamic_logger_cache,
)
- _v2 = _maybe_construct_otel_v2("newrelic", _in_memory_loggers)
+ _v2 = _maybe_construct_otel_v2(
+ "newrelic",
+ _in_memory_loggers,
+ register_dashboard_provenance=register_dashboard_provenance,
+ register_on_proxy=register_on_proxy,
+ cache_in_memory=cache_in_memory,
+ )
if _v2 is not None:
return _v2
for callback in _in_memory_loggers:
@@ -4712,7 +4849,28 @@ def _init_custom_logger_compatible_class(
return None
-def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[CustomLogger]) -> "OpenTelemetryV2 | None":
+def promote_otel_callback_to_global(callback: CustomLogger) -> None:
+ """Activate and retain a generic OTel logger after global callback promotion."""
+ from litellm.integrations.opentelemetry import OpenTelemetry
+ from litellm.integrations.otel.logger import OpenTelemetryV2
+
+ if type(callback) not in (OpenTelemetry, OpenTelemetryV2):
+ return
+ if getattr(callback, "callback_name", None) != "otel":
+ return
+ callback._init_otel_logger_on_litellm_proxy()
+ if not any(cached_callback is callback for cached_callback in _in_memory_loggers):
+ _in_memory_loggers.append(callback)
+ _maybe_auto_initialize_arize_phoenix(_in_memory_loggers)
+
+
+def _maybe_construct_otel_v2(
+ callback_name: str,
+ _in_memory_loggers: list[CustomLogger],
+ register_dashboard_provenance: bool = True,
+ register_on_proxy: bool = True,
+ cache_in_memory: bool = True,
+) -> "OpenTelemetryV2 | None":
"""If ``LITELLM_OTEL_V2`` is on, build (or reuse) a single ``OpenTelemetryV2``
instance configured via the preset for ``callback_name``.
@@ -4731,6 +4889,8 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom
return None
for callback in _in_memory_loggers:
if isinstance(callback, OpenTelemetryV2) and getattr(callback, "callback_name", None) == callback_name:
+ if register_dashboard_provenance:
+ _register_dashboard_callback(callback, callback_name)
return callback
try:
config: Final = preset_fn()
@@ -4738,8 +4898,15 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom
# If env vars are missing or the preset raises, defer to the legacy path
# so customers get the same error story they had before V2 landed.
return None
- v2_logger: Final = OpenTelemetryV2(config=config, callback_name=callback_name)
- _in_memory_loggers.append(v2_logger)
+ v2_logger: Final = OpenTelemetryV2(
+ config=config,
+ callback_name=callback_name,
+ register_on_proxy=register_on_proxy,
+ )
+ if register_dashboard_provenance:
+ _register_dashboard_callback(v2_logger, callback_name)
+ if cache_in_memory:
+ _in_memory_loggers.append(v2_logger)
return v2_logger
diff --git a/litellm/litellm_core_utils/logging_callback_manager.py b/litellm/litellm_core_utils/logging_callback_manager.py
index 9b612993a69..859261101e3 100644
--- a/litellm/litellm_core_utils/logging_callback_manager.py
+++ b/litellm/litellm_core_utils/logging_callback_manager.py
@@ -1,5 +1,8 @@
-from collections.abc import Callable
-from typing import TYPE_CHECKING, Final
+import sys
+import weakref
+from collections.abc import Callable, Mapping
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final, NamedTuple
import litellm
from litellm._logging import verbose_logger
@@ -17,6 +20,172 @@ else:
_generic_api_logger_cache: Final[dict[str, GenericAPILogger]] = {}
+class _DashboardCallbackRegistration(NamedTuple):
+ callback_name: str
+ callback_type: str
+
+
+_DASHBOARD_CALLBACK_ALIASES: Final[Mapping[str, str]] = MappingProxyType(
+ {
+ "arize": "arize",
+ "aws_sqs": "sqs",
+ "azure_sentinel": "azure_sentinel",
+ "braintrust": "braintrust",
+ "custom_callback_api": "generic_api",
+ "datadog": "datadog",
+ "datadog_cost_management": "datadog_cost_management",
+ "datadog_metrics": "datadog_metrics",
+ "galileo": "galileo",
+ "generic_api": "generic_api",
+ "lago": "lago",
+ "langfuse": "langfuse",
+ "langfuse_otel": "langfuse_otel",
+ "langsmith": "langsmith",
+ "newrelic": "newrelic",
+ "openmeter": "openmeter",
+ "opentelemetry": "otel",
+ "otel": "otel",
+ "s3": "s3",
+ "s3_v2": "s3",
+ "sqs": "sqs",
+ "traceloop": "traceloop",
+ }
+)
+
+
+def get_dashboard_callback_name(callback_name: str) -> str | None:
+ """Return the dashboard identity for a supported callback name."""
+ return _DASHBOARD_CALLBACK_ALIASES.get(callback_name.lower())
+
+
+def is_generic_api_callback(
+ callback_name: str,
+ callback_settings: Mapping[str, object] | None = None,
+) -> bool:
+ """Return whether a configured callback resolves to Generic API."""
+ if get_dashboard_callback_name(callback_name) == "generic_api":
+ return True
+ callback_settings_to_use: Final = callback_settings if callback_settings is not None else litellm.callback_settings
+ callback_config: Final = callback_settings_to_use.get(callback_name)
+ if isinstance(callback_config, dict) and callback_config.get("callback_type") == "generic_api":
+ return True
+ from litellm.integrations.generic_api.generic_api_callback import (
+ is_callback_compatible,
+ )
+
+ return is_callback_compatible(callback_name)
+
+
+class _DashboardCallbackProvenance(NamedTuple):
+ callback_ref: weakref.ReferenceType[CustomLogger]
+ callback_names: tuple[str, ...]
+
+
+_DASHBOARD_CALLBACK_PROVENANCE: Final[
+ dict[int, _DashboardCallbackProvenance]
+] = {} # mutable-ok: weak-reference callbacks remove their collected logger entries
+
+_AMBIGUOUS_DASHBOARD_CALLBACK_TYPES: Final[Mapping[tuple[str, str], str]] = MappingProxyType(
+ {
+ ("litellm.integrations.opentelemetry", "OpenTelemetry"): "otel",
+ ("litellm.integrations.otel.logger", "OpenTelemetryV2"): "otel",
+ }
+)
+
+_SUPPORTED_DASHBOARD_CALLBACK_TYPES: Final[Mapping[tuple[str, str], str]] = MappingProxyType(
+ {
+ ("litellm.integrations.arize.arize", "ArizeLogger"): "arize",
+ ("litellm.integrations.azure_sentinel.azure_sentinel", "AzureSentinelLogger"): "azure_sentinel",
+ ("litellm.integrations.braintrust_logging", "BraintrustLogger"): "braintrust",
+ ("litellm.integrations.generic_api.generic_api_callback", "GenericAPILogger"): "generic_api",
+ ("litellm.integrations.datadog.datadog", "DataDogLogger"): "datadog",
+ (
+ "litellm.integrations.datadog.datadog_cost_management",
+ "DatadogCostManagementLogger",
+ ): "datadog_cost_management",
+ ("litellm.integrations.datadog.datadog_metrics", "DatadogMetricsLogger"): "datadog_metrics",
+ ("litellm.integrations.galileo", "GalileoObserve"): "galileo",
+ ("litellm.integrations.lago", "LagoLogger"): "lago",
+ ("litellm.integrations.langfuse.langfuse_otel", "LangfuseOtelLogger"): "langfuse_otel",
+ ("litellm.integrations.langfuse.langfuse_prompt_management", "LangfusePromptManagement"): "langfuse",
+ ("litellm.integrations.langsmith", "LangsmithLogger"): "langsmith",
+ ("litellm.integrations.newrelic.newrelic", "NewRelicLogger"): "newrelic",
+ ("litellm.integrations.openmeter", "OpenMeterLogger"): "openmeter",
+ ("litellm.integrations.s3_v2", "S3Logger"): "s3",
+ ("litellm.integrations.sqs", "SQSLogger"): "sqs",
+ }
+)
+
+
+def _get_dashboard_callback_type_name(
+ callback: CustomLogger,
+ callback_types: Mapping[tuple[str, str], str],
+) -> str | None:
+ callback_type: Final = type(callback)
+ callback_key: Final = (callback_type.__module__, callback_type.__name__)
+ dashboard_callback_name: Final = callback_types.get(callback_key)
+ if dashboard_callback_name is None:
+ return None
+ callback_module: Final = sys.modules.get(callback_type.__module__)
+ if callback_module is None or getattr(callback_module, callback_type.__name__, None) is not callback_type:
+ return None
+ return dashboard_callback_name
+
+
+def register_dashboard_callback(callback: CustomLogger, callback_name: str) -> None:
+ """Record the dashboard identity of a factory-created OTel logger."""
+ dashboard_callback_name: Final = get_dashboard_callback_name(callback_name)
+ if (
+ dashboard_callback_name is None
+ or _get_dashboard_callback_type_name(callback, _AMBIGUOUS_DASHBOARD_CALLBACK_TYPES) is None
+ ):
+ return
+
+ callback_id: Final = id(callback)
+ registered_callback: Final = _DASHBOARD_CALLBACK_PROVENANCE.get(callback_id)
+ if registered_callback is not None and registered_callback.callback_ref() is callback:
+ if dashboard_callback_name not in registered_callback.callback_names:
+ _DASHBOARD_CALLBACK_PROVENANCE[callback_id] = _DashboardCallbackProvenance(
+ registered_callback.callback_ref,
+ registered_callback.callback_names + (dashboard_callback_name,),
+ )
+ return
+
+ def _remove_callback(reference: weakref.ReferenceType[CustomLogger]) -> None:
+ registered_callback: Final = _DASHBOARD_CALLBACK_PROVENANCE.get(callback_id)
+ if registered_callback is not None and registered_callback.callback_ref is reference:
+ _DASHBOARD_CALLBACK_PROVENANCE.pop(callback_id, None)
+
+ try:
+ _DASHBOARD_CALLBACK_PROVENANCE[callback_id] = _DashboardCallbackProvenance(
+ weakref.ref(callback, _remove_callback),
+ (dashboard_callback_name,),
+ )
+ except TypeError:
+ return
+
+
+def _get_registered_dashboard_callback_names(callback: CustomLogger) -> tuple[str, ...]:
+ registered_callback: Final = _DASHBOARD_CALLBACK_PROVENANCE.get(id(callback))
+ if registered_callback is None or registered_callback.callback_ref() is not callback:
+ return ()
+ return registered_callback.callback_names
+
+
+def _get_dashboard_callback_registry_callbacks() -> tuple[CustomLogger | Callable | str, ...]:
+ return tuple(
+ callback
+ for callback_registry in (
+ litellm.success_callback,
+ getattr(litellm, "_async_success_callback", ()),
+ litellm.failure_callback,
+ getattr(litellm, "_async_failure_callback", ()),
+ litellm.callbacks,
+ )
+ for callback in callback_registry
+ )
+
+
class LoggingCallbackManager:
"""
A centralized class that allows easy add / remove callbacks for litellm.
@@ -244,6 +413,18 @@ class LoggingCallbackManager:
return callback
+ def reset_dashboard_callback_registrations(self) -> None:
+ """Drop provenance for callbacks no longer active in dashboard registries."""
+ active_callbacks: Final = _get_dashboard_callback_registry_callbacks()
+ stale_callback_ids: Final = tuple(
+ callback_id
+ for callback_id, registration in _DASHBOARD_CALLBACK_PROVENANCE.items()
+ if registration.callback_ref() is None
+ or not any(registration.callback_ref() is callback for callback in active_callbacks)
+ )
+ for callback_id in stale_callback_ids:
+ _DASHBOARD_CALLBACK_PROVENANCE.pop(callback_id, None)
+
def _safe_add_callback_to_list(
self,
callback: CustomLogger | Callable | str,
@@ -340,6 +521,7 @@ class LoggingCallbackManager:
litellm._async_success_callback = []
litellm._async_failure_callback = []
litellm.callbacks = []
+ self.reset_dashboard_callback_registrations()
def _get_all_callbacks(self) -> list[CustomLogger | Callable | str]:
"""
@@ -441,6 +623,80 @@ class LoggingCallbackManager:
return result
+ def get_dashboard_callback_registrations(self) -> tuple[_DashboardCallbackRegistration, ...]:
+ """Return active callbacks recognized by the Logging & Alerts dashboard."""
+ self.reset_dashboard_callback_registrations()
+ success_callbacks: Final = tuple(litellm.success_callback) + tuple(
+ getattr(litellm, "_async_success_callback", ())
+ )
+ failure_callbacks: Final = tuple(litellm.failure_callback) + tuple(
+ getattr(litellm, "_async_failure_callback", ())
+ )
+ registrations: Final = tuple(
+ registration
+ for callback_type, callbacks in (
+ ("success", success_callbacks),
+ ("failure", failure_callbacks),
+ ("success_and_failure", tuple(litellm.callbacks)),
+ )
+ for callback in callbacks
+ for registration in self._get_dashboard_callback_registration(callback, callback_type)
+ )
+ callback_names: Final = tuple(
+ registration.callback_name
+ for registration_index, registration in enumerate(registrations)
+ if registration.callback_name
+ not in tuple(prior.callback_name for prior in registrations[:registration_index])
+ )
+ return tuple(
+ _DashboardCallbackRegistration(callback_name, callback_type)
+ for callback_name in callback_names
+ for callback_type in self._get_dashboard_callback_types(callback_name, registrations)
+ )
+
+ @staticmethod
+ def _get_dashboard_callback_types(
+ callback_name: str,
+ registrations: tuple[_DashboardCallbackRegistration, ...],
+ ) -> tuple[str, ...]:
+ callback_types: Final = tuple(
+ registration.callback_type for registration in registrations if registration.callback_name == callback_name
+ )
+ if "success_and_failure" in callback_types or ("success" in callback_types and "failure" in callback_types):
+ return ("success_and_failure",)
+ if "success" in callback_types:
+ return ("success",)
+ if "failure" in callback_types:
+ return ("failure",)
+ return ()
+
+ @staticmethod
+ def _get_dashboard_callback_registration(
+ callback: CustomLogger | Callable | str,
+ callback_type: str,
+ ) -> tuple[_DashboardCallbackRegistration, ...]:
+ return tuple(
+ _DashboardCallbackRegistration(callback_name, callback_type)
+ for callback_name in LoggingCallbackManager._get_dashboard_callback_names(callback)
+ )
+
+ @staticmethod
+ def _get_dashboard_callback_names(callback: CustomLogger | Callable | str) -> tuple[str, ...]:
+ if isinstance(callback, str):
+ callback_name: Final = get_dashboard_callback_name(callback)
+ return (callback_name,) if callback_name is not None else ()
+ if not isinstance(callback, CustomLogger):
+ return ()
+
+ dashboard_callback_name: Final = _get_dashboard_callback_type_name(
+ callback, _SUPPORTED_DASHBOARD_CALLBACK_TYPES
+ )
+ if dashboard_callback_name is not None:
+ return (dashboard_callback_name,)
+ if _get_dashboard_callback_type_name(callback, _AMBIGUOUS_DASHBOARD_CALLBACK_TYPES) is not None:
+ return _get_registered_dashboard_callback_names(callback)
+ return ()
+
def _get_callback_string(self, callback: CustomLogger | Callable | str) -> str:
from litellm.litellm_core_utils.custom_logger_registry import (
CustomLoggerRegistry,
diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py
index f40b0632398..5ce648ec153 100644
--- a/litellm/proxy/_types.py
+++ b/litellm/proxy/_types.py
@@ -3462,6 +3462,45 @@ class AllCallbacks(LiteLLMPydanticObjectBase):
],
)
+ arize: CallbackOnUI = CallbackOnUI(
+ litellm_callback_name="arize",
+ ui_callback_name="Arize",
+ litellm_callback_params=[ # mutable-ok: UI callback parameter order is user-facing
+ "ARIZE_SPACE_ID",
+ "ARIZE_SPACE_KEY",
+ "ARIZE_API_KEY",
+ "ARIZE_PROJECT_NAME",
+ "ARIZE_ENDPOINT",
+ "ARIZE_HTTP_ENDPOINT",
+ ],
+ )
+
+ datadog_metrics: CallbackOnUI = CallbackOnUI(
+ litellm_callback_name="datadog_metrics",
+ ui_callback_name="Datadog Metrics",
+ litellm_callback_params=[ # mutable-ok: UI callback parameter order is user-facing
+ "DD_API_KEY",
+ "DD_APP_KEY",
+ "DD_SITE",
+ ],
+ )
+
+ datadog_cost_management: CallbackOnUI = CallbackOnUI(
+ litellm_callback_name="datadog_cost_management",
+ ui_callback_name="Datadog Cost Management",
+ litellm_callback_params=[ # mutable-ok: UI callback parameter order is user-facing
+ "DD_API_KEY",
+ "DD_APP_KEY",
+ "DD_SITE",
+ ],
+ )
+
+ sqs: CallbackOnUI = CallbackOnUI(
+ litellm_callback_name="sqs",
+ ui_callback_name="SQS",
+ litellm_callback_params=[], # mutable-ok: CallbackOnUI requires a mutable list
+ )
+
openmeter: CallbackOnUI = CallbackOnUI(
litellm_callback_name="openmeter",
ui_callback_name="OpenMeter",
diff --git a/litellm/proxy/management_helpers/audit_logs.py b/litellm/proxy/management_helpers/audit_logs.py
index ecd6abea3c3..a705a9c2658 100644
--- a/litellm/proxy/management_helpers/audit_logs.py
+++ b/litellm/proxy/management_helpers/audit_logs.py
@@ -85,6 +85,10 @@ def _resolve_audit_log_callback(name: str) -> CustomLogger | None:
logging_integration=name,
internal_usage_cache=None,
llm_router=None,
+ custom_logger_init_args=None,
+ register_dashboard_provenance=False,
+ register_on_proxy=False,
+ cache_in_memory=False,
)
if instance is not None:
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index 597acdf661f..cde1b055ade 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -64,10 +64,15 @@ from litellm.constants import (
from litellm.litellm_core_utils.litellm_logging import (
_init_custom_logger_compatible_class,
)
+from litellm.litellm_core_utils.logging_callback_manager import (
+ get_dashboard_callback_name,
+ is_generic_api_callback,
+)
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
from litellm.proxy._types import (
UI_TEAM_ID,
+ AllCallbacks,
CallbackDelete,
CallInfo,
CommonProxyErrors,
@@ -16356,6 +16361,55 @@ def _apply_callback_role_gate(entries: list, is_full_admin: bool) -> list:
return [{**entry, "variables": _redact_callback_env_vars(entry.get("variables") or {})} for entry in entries]
+def _get_configured_dashboard_callback_name(
+ callback_name: str,
+ callback_settings: object,
+) -> str | None:
+ dashboard_callback_name: Final = get_dashboard_callback_name(callback_name)
+ if dashboard_callback_name is not None:
+ return dashboard_callback_name
+ if is_generic_api_callback(
+ callback_name,
+ callback_settings
+ if isinstance(callback_settings, Mapping)
+ else {}, # mutable-ok: utility expects mapping-compatible settings
+ ):
+ return "generic_api"
+ return None
+
+
+def _get_read_only_runtime_callback_row(
+ callback_name: str,
+ callback_type: str,
+ environment_variables: dict,
+) -> dict:
+ callback_row: Final = process_callback(
+ callback_name,
+ callback_type,
+ environment_variables,
+ )
+ return callback_row | {"read_only": True} # mutable-ok: HTTP response rows are mutable dictionaries
+
+
+def _get_runtime_callback_rows(
+ configured_callback_registrations: frozenset[tuple[str, str]],
+ environment_variables: dict,
+) -> tuple[dict, ...]:
+ return tuple(
+ _get_read_only_runtime_callback_row(callback_name, callback_type, environment_variables)
+ for callback_name, callback_type in litellm.logging_callback_manager.get_dashboard_callback_registrations()
+ if not (
+ (callback_name, "success_and_failure") in configured_callback_registrations
+ or (callback_name, callback_type) in configured_callback_registrations
+ or (
+ callback_type == "success_and_failure"
+ and (callback_name, "success") in configured_callback_registrations
+ and (callback_name, "failure") in configured_callback_registrations
+ )
+ )
+ )
+
+
class _AlertingDestinationEntry(TypedDict):
name: ReadOnly[str]
variables: ReadOnly[Mapping[str, str | None]]
@@ -16975,6 +17029,21 @@ async def get_config(
for _callback in _success_and_failure_callbacks:
_data_to_return.append(process_callback(_callback, "success_and_failure", environment_variables))
+ callback_settings: Final = config_data.get("callback_settings")
+ configured_callback_registrations: Final = frozenset(
+ (canonical_callback_name, callback_type)
+ for callback_type, callback_group in (
+ ("success", _success_callbacks),
+ ("failure", _failure_callbacks),
+ ("success_and_failure", _success_and_failure_callbacks),
+ )
+ for callback in callback_group
+ if isinstance(callback, str)
+ for canonical_callback_name in (_get_configured_dashboard_callback_name(callback, callback_settings),)
+ if canonical_callback_name is not None
+ )
+ _data_to_return.extend(_get_runtime_callback_rows(configured_callback_registrations, environment_variables))
+
_data_to_return = _apply_callback_role_gate(_data_to_return, is_full_admin)
# Check if slack alerting is on
diff --git a/litellm/utils.py b/litellm/utils.py
index fa2226dbf2c..7481c687bcc 100644
--- a/litellm/utils.py
+++ b/litellm/utils.py
@@ -859,18 +859,39 @@ def function_setup(
all_callbacks: Final = get_dynamic_callbacks(dynamic_callbacks=dynamic_callbacks)
if len(all_callbacks) > 0:
+ from litellm.litellm_core_utils.litellm_logging import (
+ promote_otel_callback_to_global,
+ )
+ from litellm.litellm_core_utils.logging_callback_manager import (
+ register_dashboard_callback,
+ )
+
for callback in all_callbacks:
# check if callback is a string - e.g. "lago", "openmeter"
if isinstance(callback, str):
- callback = litellm.litellm_core_utils.litellm_logging._init_custom_logger_compatible_class(
- callback,
- internal_usage_cache=None,
- llm_router=None,
- )
+ dashboard_callback_name: str | None = callback if callback == "otel" else None
+ if dashboard_callback_name is not None:
+ callback = litellm.litellm_core_utils.litellm_logging._init_custom_logger_compatible_class(
+ callback,
+ internal_usage_cache=None,
+ llm_router=None,
+ custom_logger_init_args=None,
+ register_dashboard_provenance=False,
+ register_on_proxy=False,
+ cache_in_memory=False,
+ )
+ else:
+ callback = litellm.litellm_core_utils.litellm_logging._init_custom_logger_compatible_class(
+ callback,
+ internal_usage_cache=None,
+ llm_router=None,
+ )
if callback is None or any(
type(cb) is type(callback) for cb in litellm._async_success_callback
): # don't double add a callback
continue
+ else:
+ dashboard_callback_name = None
if callback not in litellm.input_callback:
litellm.input_callback.append(callback)
if callback not in litellm.success_callback:
@@ -881,6 +902,9 @@ def function_setup(
litellm.logging_callback_manager.add_litellm_async_success_callback(callback)
if callback not in litellm._async_failure_callback:
litellm.logging_callback_manager.add_litellm_async_failure_callback(callback)
+ if dashboard_callback_name is not None:
+ promote_otel_callback_to_global(callback)
+ register_dashboard_callback(callback, dashboard_callback_name)
print_verbose(f"Initialized litellm callbacks, Async Success Callbacks: {litellm._async_success_callback}")
if (
diff --git a/tests/litellm_utils_tests/test_logging_callback_manager.py b/tests/litellm_utils_tests/test_logging_callback_manager.py
index ebd5b473ebb..14204d890d7 100644
--- a/tests/litellm_utils_tests/test_logging_callback_manager.py
+++ b/tests/litellm_utils_tests/test_logging_callback_manager.py
@@ -1,17 +1,17 @@
import json
import os
-import time
from datetime import datetime
-from unittest.mock import AsyncMock, patch, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
+
import pytest
import litellm
from litellm.integrations.custom_logger import CustomLogger
-from litellm.litellm_core_utils.logging_callback_manager import LoggingCallbackManager
from litellm.integrations.langfuse.langfuse_prompt_management import (
LangfusePromptManagement,
)
from litellm.integrations.opentelemetry import OpenTelemetry
+from litellm.litellm_core_utils.logging_callback_manager import LoggingCallbackManager
# Test fixtures
@@ -33,6 +33,493 @@ def mock_custom_logger():
# Test cases
+def test_dashboard_callback_names_cover_callback_config_catalogue():
+ from pathlib import Path
+
+ from litellm.litellm_core_utils.logging_callback_manager import (
+ get_dashboard_callback_name,
+ )
+
+ callback_config_path = Path(__file__).parents[2] / "litellm/integrations/callback_configs.json"
+ with callback_config_path.open() as config_file:
+ callback_configs = json.load(config_file)
+
+ assert all(get_dashboard_callback_name(callback_config["id"]) is not None for callback_config in callback_configs)
+
+
+def test_dashboard_callback_names_cover_dashboard_callback_catalogue():
+ from litellm.litellm_core_utils.logging_callback_manager import (
+ get_dashboard_callback_name,
+ )
+ from litellm.proxy._types import AllCallbacks
+
+ callback_names = {callback["litellm_callback_name"] for callback in AllCallbacks().model_dump().values()}
+
+ assert all(get_dashboard_callback_name(callback_name) is not None for callback_name in callback_names)
+
+
+def test_dashboard_callback_inventory_includes_only_public_callbacks(callback_manager, monkeypatch):
+ class UnhashableCustomLogger(CustomLogger):
+ __hash__ = None
+
+ def __eq__(self, other):
+ return self is other
+
+ def custom_callback(*args, **kwargs):
+ pass
+
+ unhashable_callback = UnhashableCustomLogger()
+ monkeypatch.setattr(
+ litellm,
+ "success_callback",
+ ["opentelemetry", "s3_v2", "custom_callback_api", unhashable_callback, custom_callback],
+ )
+ monkeypatch.setattr(litellm, "_async_success_callback", ["langsmith"])
+ monkeypatch.setattr(litellm, "failure_callback", ["aws_sqs", "langsmith"])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", ["langfuse_otel", "_PROXY_VirtualKeyModelMaxBudgetLimiter"])
+
+ assert callback_manager.get_dashboard_callback_registrations() == (
+ ("otel", "success"),
+ ("s3", "success"),
+ ("generic_api", "success"),
+ ("langsmith", "success_and_failure"),
+ ("sqs", "failure"),
+ ("langfuse_otel", "success_and_failure"),
+ )
+
+
+def test_dashboard_callback_inventory_recognizes_azure_sentinel_and_traceloop(
+ callback_manager,
+ monkeypatch,
+):
+ from litellm.integrations.azure_sentinel.azure_sentinel import AzureSentinelLogger
+
+ # azure_sentinel is promoted to a logger object by the factory; traceloop is a
+ # legacy string callback the success handler dispatches by name.
+ monkeypatch.setattr(litellm, "success_callback", [object.__new__(AzureSentinelLogger), "traceloop"])
+
+ assert callback_manager.get_dashboard_callback_registrations() == (
+ ("azure_sentinel", "success"),
+ ("traceloop", "success"),
+ )
+
+
+def test_dashboard_callback_inventory_rejects_callback_name_spoof(callback_manager, monkeypatch):
+ class SpoofedCallback(CustomLogger):
+ callback_name = "s3_v2"
+
+ monkeypatch.setattr(litellm, "success_callback", [SpoofedCallback()])
+
+ assert callback_manager.get_dashboard_callback_registrations() == ()
+
+
+def test_dashboard_callback_inventory_rejects_spoofed_trusted_subclasses(
+ callback_manager,
+ monkeypatch,
+):
+ from litellm.integrations.arize.arize import ArizeLogger
+ from litellm.integrations.generic_api.generic_api_callback import (
+ GenericAPILogger,
+ )
+ from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
+ from litellm.integrations.otel.logger import OpenTelemetryV2
+
+ spoofed_callback_types = tuple(
+ type(
+ callback_name,
+ (callback_type,),
+ {},
+ )
+ for callback_name, callback_type in (
+ ("SpoofedArizeLogger", ArizeLogger),
+ ("SpoofedGenericAPILogger", GenericAPILogger),
+ ("SpoofedLangfuseOtelLogger", LangfuseOtelLogger),
+ ("SpoofedOpenTelemetry", OpenTelemetry),
+ ("SpoofedOpenTelemetryV2", OpenTelemetryV2),
+ )
+ )
+ monkeypatch.setattr(
+ litellm, "success_callback", [object.__new__(callback_type) for callback_type in spoofed_callback_types]
+ )
+ for callback in litellm.success_callback:
+ callback.callback_name = "generic_api"
+
+ assert callback_manager.get_dashboard_callback_registrations() == ()
+
+
+def test_dashboard_callback_inventory_rejects_unregistered_otel_callback(
+ callback_manager,
+ monkeypatch,
+):
+ otel_callback = object.__new__(OpenTelemetry)
+ otel_callback.callback_name = "langfuse"
+ monkeypatch.setattr(litellm, "success_callback", [otel_callback])
+
+ assert callback_manager.get_dashboard_callback_registrations() == ()
+
+
+@pytest.mark.asyncio
+async def test_dashboard_callback_inventory_recognizes_supported_logger_types(
+ callback_manager,
+ monkeypatch,
+):
+ from litellm.integrations.datadog.datadog_cost_management import (
+ DatadogCostManagementLogger,
+ )
+ from litellm.integrations.datadog.datadog_metrics import DatadogMetricsLogger
+ from litellm.integrations.lago import LagoLogger
+ from litellm.integrations.langfuse.langfuse_prompt_management import (
+ LangfusePromptManagement,
+ )
+ from litellm.integrations.langsmith import LangsmithLogger
+ from litellm.integrations.openmeter import OpenMeterLogger
+ from litellm.integrations.s3_v2 import S3Logger
+ from litellm.integrations.sqs import SQSLogger
+
+ monkeypatch.setenv("LAGO_API_KEY", "test-key")
+ monkeypatch.setenv("LAGO_API_BASE", "https://example.com")
+ monkeypatch.setenv("LAGO_API_EVENT_CODE", "test-event")
+ monkeypatch.setenv("OPENMETER_API_KEY", "test-key")
+ monkeypatch.setattr(
+ litellm,
+ "success_callback",
+ [
+ LagoLogger(),
+ OpenMeterLogger(),
+ LangfusePromptManagement(),
+ LangsmithLogger(),
+ DatadogMetricsLogger(),
+ DatadogCostManagementLogger(),
+ S3Logger(),
+ SQSLogger(),
+ ],
+ )
+
+ assert callback_manager.get_dashboard_callback_registrations() == (
+ ("lago", "success"),
+ ("openmeter", "success"),
+ ("langfuse", "success"),
+ ("langsmith", "success"),
+ ("datadog_metrics", "success"),
+ ("datadog_cost_management", "success"),
+ ("s3", "success"),
+ ("sqs", "success"),
+ )
+
+
+@pytest.mark.asyncio
+async def test_dashboard_callback_inventory_includes_builtin_generic_api_callback(
+ callback_manager,
+ monkeypatch,
+):
+ from litellm.litellm_core_utils import litellm_logging as logging_module
+
+ monkeypatch.setenv("GENERIC_LOGGER_ENDPOINT", "https://example.com/logs")
+ logging_module._in_memory_loggers.clear()
+ try:
+ generic_api_callback = logging_module._init_custom_logger_compatible_class(
+ logging_integration="generic_api",
+ internal_usage_cache=None,
+ llm_router=None,
+ custom_logger_init_args={},
+ )
+
+ assert generic_api_callback is not None
+ assert generic_api_callback.callback_name is None
+
+ monkeypatch.setattr(litellm, "success_callback", [generic_api_callback])
+
+ assert callback_manager.get_dashboard_callback_registrations() == (("generic_api", "success"),)
+ finally:
+ logging_module._in_memory_loggers.clear()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("callback_name", [None, "logfire"])
+async def test_dashboard_callback_inventory_includes_factory_created_otel_callback(
+ callback_manager,
+ monkeypatch,
+ callback_name,
+):
+ from litellm.litellm_core_utils import litellm_logging as logging_module
+
+ monkeypatch.setattr(
+ litellm,
+ "callback_settings",
+ {"otel": {"callback_name": callback_name}},
+ )
+ logging_module._in_memory_loggers.clear()
+ try:
+ otel_callback = logging_module._init_custom_logger_compatible_class(
+ logging_integration="otel",
+ internal_usage_cache=None,
+ llm_router=None,
+ custom_logger_init_args={},
+ )
+ assert otel_callback is not None
+ assert otel_callback.callback_name == callback_name
+ monkeypatch.setattr(litellm, "success_callback", [otel_callback])
+
+ assert callback_manager.get_dashboard_callback_registrations() == (("otel", "success"),)
+ finally:
+ logging_module._in_memory_loggers.clear()
+
+
+@pytest.mark.asyncio
+async def test_dashboard_otel_provenance_keeps_legacy_otel_separate_from_preset(
+ callback_manager,
+ monkeypatch,
+):
+ from litellm.integrations.otel.model.config import is_otel_v2_enabled
+ from litellm.litellm_core_utils import litellm_logging as logging_module
+
+ monkeypatch.setenv("LANGTRACE_API_KEY", "test-key")
+ monkeypatch.setenv("LITELLM_OTEL_V2", "false")
+ is_otel_v2_enabled.cache_clear()
+ logging_module._in_memory_loggers.clear()
+ try:
+ langtrace = logging_module._init_custom_logger_compatible_class("langtrace", None, None, {})
+ otel = logging_module._init_custom_logger_compatible_class("otel", None, None, {})
+
+ assert otel is not langtrace
+ monkeypatch.setattr(litellm, "success_callback", [otel])
+ assert callback_manager.get_dashboard_callback_registrations() == (("otel", "success"),)
+ finally:
+ logging_module._in_memory_loggers.clear()
+ is_otel_v2_enabled.cache_clear()
+
+
+@pytest.mark.asyncio
+async def test_dashboard_otel_provenance_registered_on_v2_preset_cache_hit_after_reset(
+ callback_manager,
+ monkeypatch,
+):
+ from litellm.integrations.otel.model.config import is_otel_v2_enabled
+ from litellm.litellm_core_utils import litellm_logging as logging_module
+
+ monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+ is_otel_v2_enabled.cache_clear()
+ logging_module._in_memory_loggers.clear()
+ try:
+ arize = logging_module._init_custom_logger_compatible_class("arize", None, None, {})
+ callback_manager._reset_all_callbacks()
+ cached_arize = logging_module._init_custom_logger_compatible_class("arize", None, None, {})
+ assert cached_arize is arize
+ monkeypatch.setattr(litellm, "success_callback", [cached_arize])
+
+ assert callback_manager.get_dashboard_callback_registrations() == (("arize", "success"),)
+ finally:
+ logging_module._in_memory_loggers.clear()
+ is_otel_v2_enabled.cache_clear()
+
+
+@pytest.mark.asyncio
+async def test_dashboard_otel_provenance_registered_on_v2_otel_cache_hit(
+ callback_manager,
+ monkeypatch,
+):
+ from litellm.integrations.otel.model.config import is_otel_v2_enabled
+ from litellm.litellm_core_utils import litellm_logging as logging_module
+
+ monkeypatch.setenv("LITELLM_OTEL_V2", "true")
+ is_otel_v2_enabled.cache_clear()
+ logging_module._in_memory_loggers.clear()
+ try:
+ arize = logging_module._init_custom_logger_compatible_class("arize", None, None, {})
+ otel = logging_module._init_custom_logger_compatible_class("otel", None, None, {})
+ assert otel is not arize
+ monkeypatch.setattr(litellm, "success_callback", [otel])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+
+ assert callback_manager.get_dashboard_callback_registrations() == (("otel", "success"),)
+ finally:
+ logging_module._in_memory_loggers.clear()
+ is_otel_v2_enabled.cache_clear()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+ ("otel_v2_enabled", "expected_type"),
+ ((True, "OpenTelemetryV2"), (False, "OpenTelemetry")),
+)
+async def test_dynamic_otel_callback_resolution_is_request_local(
+ callback_manager,
+ monkeypatch,
+ otel_v2_enabled,
+ expected_type,
+):
+ from litellm.integrations.opentelemetry import OpenTelemetry
+ from litellm.integrations.otel.logger import OpenTelemetryV2
+ from litellm.integrations.otel.model.config import is_otel_v2_enabled
+ from litellm.litellm_core_utils import litellm_logging as logging_module
+
+ monkeypatch.setenv("LITELLM_OTEL_V2", str(otel_v2_enabled).lower())
+ is_otel_v2_enabled.cache_clear()
+ monkeypatch.setattr(logging_module, "_in_memory_loggers", [])
+ monkeypatch.setattr(litellm, "input_callback", [])
+ monkeypatch.setattr(litellm, "service_callback", [])
+ monkeypatch.setattr(litellm, "success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ request_logging = logging_module.Logging(
+ model="test-model",
+ messages=[],
+ stream=False,
+ call_type="completion",
+ start_time=None,
+ litellm_call_id="request-id",
+ function_id="function-id",
+ dynamic_success_callbacks=["otel"],
+ )
+
+ dynamic_callback = request_logging.dynamic_success_callbacks[0]
+ expected_callback_type = OpenTelemetryV2 if expected_type == "OpenTelemetryV2" else OpenTelemetry
+ assert type(dynamic_callback) is expected_callback_type
+ assert dynamic_callback.callback_name == "otel"
+ assert logging_module._in_memory_loggers == []
+ assert all(
+ dynamic_callback is not registered_callback
+ for callback_registry in (
+ litellm.input_callback,
+ litellm.service_callback,
+ litellm.success_callback,
+ litellm.failure_callback,
+ litellm._async_success_callback,
+ litellm._async_failure_callback,
+ litellm.callbacks,
+ )
+ for registered_callback in callback_registry
+ )
+ assert callback_manager.get_dashboard_callback_registrations() == ()
+ is_otel_v2_enabled.cache_clear()
+
+
+@pytest.mark.parametrize(
+ ("otel_v2_enabled", "expected_type"),
+ ((True, "OpenTelemetryV2"), (False, "OpenTelemetry")),
+)
+def test_prompt_management_otel_resolution_is_request_local(
+ callback_manager,
+ monkeypatch,
+ otel_v2_enabled,
+ expected_type,
+):
+ from litellm.integrations.opentelemetry import OpenTelemetry
+ from litellm.integrations.otel.logger import OpenTelemetryV2
+ from litellm.integrations.otel.model.config import is_otel_v2_enabled
+ from litellm.litellm_core_utils import litellm_logging as logging_module
+
+ monkeypatch.setenv("LITELLM_OTEL_V2", str(otel_v2_enabled).lower())
+ is_otel_v2_enabled.cache_clear()
+ monkeypatch.setattr(logging_module, "_in_memory_loggers", [])
+ monkeypatch.setattr(litellm, "input_callback", [])
+ monkeypatch.setattr(litellm, "service_callback", [])
+ monkeypatch.setattr(litellm, "success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ request_logging = logging_module.Logging(
+ model="otel/prompt",
+ messages=[],
+ stream=False,
+ call_type="completion",
+ start_time=None,
+ litellm_call_id="request-id",
+ function_id="function-id",
+ )
+ prompt_logger = request_logging.get_custom_logger_for_prompt_management(
+ model="otel/prompt",
+ non_default_params={},
+ )
+
+ expected_callback_type = OpenTelemetryV2 if expected_type == "OpenTelemetryV2" else OpenTelemetry
+ assert type(prompt_logger) is expected_callback_type
+ assert prompt_logger.callback_name == "otel"
+ assert logging_module._in_memory_loggers == []
+ assert all(
+ prompt_logger is not registered_callback
+ for callback_registry in (
+ litellm.input_callback,
+ litellm.service_callback,
+ litellm.success_callback,
+ litellm.failure_callback,
+ litellm._async_success_callback,
+ litellm._async_failure_callback,
+ litellm.callbacks,
+ )
+ for registered_callback in callback_registry
+ )
+ assert callback_manager.get_dashboard_callback_registrations() == ()
+ is_otel_v2_enabled.cache_clear()
+
+
+@pytest.mark.parametrize(
+ ("otel_v2_enabled", "expected_type"),
+ ((True, "OpenTelemetryV2"), (False, "OpenTelemetry")),
+)
+def test_function_setup_promotes_cold_otel_only_after_duplicate_check(
+ callback_manager,
+ monkeypatch,
+ otel_v2_enabled,
+ expected_type,
+):
+ from litellm.integrations.opentelemetry import OpenTelemetry
+ from litellm.integrations.otel.logger import OpenTelemetryV2
+ from litellm.integrations.otel.model.config import is_otel_v2_enabled
+ from litellm.litellm_core_utils import litellm_logging as logging_module
+ from litellm.proxy import proxy_server
+
+ checker = MagicMock()
+ checker.is_async_callable.return_value = False
+ logging_object = MagicMock()
+ monkeypatch.setenv("LITELLM_OTEL_V2", str(otel_v2_enabled).lower())
+ monkeypatch.setattr(litellm.utils.litellm_utils, "get_coroutine_checker", lambda: checker)
+ monkeypatch.setattr(litellm.utils, "get_litellm_logging_class", lambda: logging_object)
+ monkeypatch.setattr(litellm.utils, "callback_list", ["already-initialized"])
+ monkeypatch.setattr(logging_module, "_in_memory_loggers", [])
+ monkeypatch.setattr(litellm, "input_callback", [])
+ monkeypatch.setattr(litellm, "service_callback", [])
+ monkeypatch.setattr(litellm, "success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+ monkeypatch.setattr(proxy_server, "open_telemetry_logger", None)
+ is_otel_v2_enabled.cache_clear()
+ try:
+ litellm.utils.function_setup(
+ original_function="acompletion",
+ rules_obj=litellm.utils.Rules(),
+ start_time=datetime.now(),
+ model="gpt-4",
+ messages=[],
+ litellm_call_id="cold-global-promotion",
+ callbacks=["otel"],
+ )
+
+ otel_callback = litellm._async_success_callback[0]
+ expected_callback_type = OpenTelemetryV2 if expected_type == "OpenTelemetryV2" else OpenTelemetry
+ assert type(otel_callback) is expected_callback_type
+ assert otel_callback.callback_name == "otel"
+ assert otel_callback in logging_module._in_memory_loggers
+ assert otel_callback in litellm.input_callback
+ assert otel_callback in litellm.service_callback
+ assert otel_callback in litellm._async_failure_callback
+ assert proxy_server.open_telemetry_logger is otel_callback
+ assert callback_manager.get_dashboard_callback_registrations() == (("otel", "success_and_failure"),)
+ finally:
+ is_otel_v2_enabled.cache_clear()
+
+
def test_add_string_callback():
"""
Test adding a string callback to litellm.callbacks - only 1 instance of the string callback should be added
@@ -54,7 +541,6 @@ def test_duplicate_langfuse_logger_test():
for _ in range(10):
langfuse_logger = LangfusePromptManagement()
manager.add_litellm_success_callback(langfuse_logger)
- print("litellm.success_callback: ", litellm.success_callback)
assert len(litellm.success_callback) == 1
@@ -65,24 +551,13 @@ def test_duplicate_multiple_loggers_test():
otel_logger = OpenTelemetry()
manager.add_litellm_success_callback(langfuse_logger)
manager.add_litellm_success_callback(otel_logger)
- print("litellm.success_callback: ", litellm.success_callback)
assert len(litellm.success_callback) == 2
# Check exactly one instance of each logger type
- langfuse_count = sum(
- 1
- for callback in litellm.success_callback
- if isinstance(callback, LangfusePromptManagement)
- )
- otel_count = sum(
- 1
- for callback in litellm.success_callback
- if isinstance(callback, OpenTelemetry)
- )
+ langfuse_count = sum(1 for callback in litellm.success_callback if isinstance(callback, LangfusePromptManagement))
+ otel_count = sum(1 for callback in litellm.success_callback if isinstance(callback, OpenTelemetry))
- assert (
- langfuse_count == 1
- ), "Should have exactly one LangfusePromptManagement instance"
+ assert langfuse_count == 1, "Should have exactly one LangfusePromptManagement instance"
assert otel_count == 1, "Should have exactly one OpenTelemetry instance"
@@ -237,23 +712,18 @@ async def test_slack_alerting_callback_registration(callback_manager):
when outage_alerts or region_outage_alerts are enabled
"""
from litellm.caching.caching import DualCache
- from litellm.proxy.utils import ProxyLogging
from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting
- from unittest.mock import patch
+ from litellm.proxy.utils import ProxyLogging
# Mock the async HTTP handler
- with patch(
- "litellm.integrations.SlackAlerting.slack_alerting.get_async_httpx_client"
- ) as mock_http:
+ with patch("litellm.integrations.SlackAlerting.slack_alerting.get_async_httpx_client") as mock_http:
mock_http.return_value = AsyncMock()
# Create a fresh ProxyLogging instance
proxy_logging = ProxyLogging(user_api_key_cache=DualCache())
# Test 1: No callbacks should be added when alerting is None
- proxy_logging.update_values(
- alerting=None, alert_types=["outage_alerts", "region_outage_alerts"]
- )
+ proxy_logging.update_values(alerting=None, alert_types=["outage_alerts", "region_outage_alerts"])
assert len(litellm.callbacks) == 0
# Test 2: Callbacks should be added when slack alerting is enabled with outage alerts
@@ -263,16 +733,15 @@ async def test_slack_alerting_callback_registration(callback_manager):
# Test 3: Callbacks should be added when slack alerting is enabled with region outage alerts
callback_manager._reset_all_callbacks() # Reset callbacks
- proxy_logging.update_values(
- alerting=["slack"], alert_types=["region_outage_alerts"]
- )
+ proxy_logging.update_values(alerting=["slack"], alert_types=["region_outage_alerts"])
assert len(litellm.callbacks) == 1
assert isinstance(litellm.callbacks[0], SlackAlerting)
# Test 4: No callbacks should be added for other alert types
callback_manager._reset_all_callbacks() # Reset callbacks
proxy_logging.update_values(
- alerting=["slack"], alert_types=["budget_alerts"] # Some other alert type
+ alerting=["slack"],
+ alert_types=["budget_alerts"], # Some other alert type
)
assert len(litellm.callbacks) == 0
@@ -282,9 +751,7 @@ async def test_slack_alerting_callback_registration(callback_manager):
assert len(litellm.callbacks) == 1 # Regular callback for outage alerts
assert isinstance(litellm.callbacks[0], SlackAlerting)
# response_taking_too_long_callback is async, so it should be in the async success callback list
- response_taking_too_long_callback = (
- proxy_logging.slack_alerting_instance.response_taking_too_long_callback
- )
+ response_taking_too_long_callback = proxy_logging.slack_alerting_instance.response_taking_too_long_callback
assert len(litellm._async_success_callback) == 1
assert litellm._async_success_callback[0] == response_taking_too_long_callback
@@ -305,28 +772,18 @@ async def test_generic_api_compatible_callbacks_json():
with patch.dict(os.environ, {"SUMOLOGIC_WEBHOOK_URL": test_sumologic_url}):
# Test that sumologic callback is recognized from JSON file
- result = LoggingCallbackManager._add_custom_callback_generic_api_str(
- "sumologic"
- )
+ result = LoggingCallbackManager._add_custom_callback_generic_api_str("sumologic")
# Verify a GenericAPILogger instance is returned
- assert isinstance(
- result, GenericAPILogger
- ), "Should return GenericAPILogger instance for sumologic callback"
+ assert isinstance(result, GenericAPILogger), "Should return GenericAPILogger instance for sumologic callback"
# Verify the endpoint is correctly loaded from environment variable
- assert (
- result.endpoint == test_sumologic_url
- ), f"Endpoint should be {test_sumologic_url}"
+ assert result.endpoint == test_sumologic_url, f"Endpoint should be {test_sumologic_url}"
# Verify headers only contain Content-Type (no Authorization for SumoLogic)
assert "Content-Type" in result.headers, "Should have Content-Type header"
- assert (
- result.headers["Content-Type"] == "application/json"
- ), "Content-Type should be application/json"
- assert (
- "Authorization" not in result.headers
- ), "Should not have Authorization header for SumoLogic"
+ assert result.headers["Content-Type"] == "application/json", "Content-Type should be application/json"
+ assert "Authorization" not in result.headers, "Should not have Authorization header for SumoLogic"
@pytest.mark.asyncio
@@ -349,28 +806,20 @@ async def test_generic_api_compatible_callbacks_json_rubrik():
result = LoggingCallbackManager._add_custom_callback_generic_api_str("rubrik")
# Verify a GenericAPILogger instance is returned
- assert isinstance(
- result, GenericAPILogger
- ), "Should return GenericAPILogger instance for rubrik callback"
+ assert isinstance(result, GenericAPILogger), "Should return GenericAPILogger instance for rubrik callback"
# Verify the endpoint is correctly loaded
- assert (
- result.endpoint == test_rubrik_url
- ), f"Endpoint should be {test_rubrik_url}"
+ assert result.endpoint == test_rubrik_url, f"Endpoint should be {test_rubrik_url}"
# Verify headers include Authorization with Bearer token
assert "Content-Type" in result.headers, "Should have Content-Type header"
- assert (
- "Authorization" in result.headers
- ), "Should have Authorization header for Rubrik"
- assert (
- result.headers["Authorization"] == f"Bearer {test_rubrik_api_key}"
- ), "Authorization should have correct API key"
+ assert "Authorization" in result.headers, "Should have Authorization header for Rubrik"
+ assert result.headers["Authorization"] == f"Bearer {test_rubrik_api_key}", (
+ "Authorization should have correct API key"
+ )
# Verify event_types filter (rubrik only logs success events)
- assert result.event_types == [
- "llm_api_success"
- ], "Rubrik should only log success events"
+ assert result.event_types == ["llm_api_success"], "Rubrik should only log success events"
def test_generic_api_compatible_callbacks_json_unknown_callback():
@@ -378,9 +827,7 @@ def test_generic_api_compatible_callbacks_json_unknown_callback():
Test that unknown callbacks (not in JSON or callback_settings) are returned unchanged
"""
# Test with a callback that doesn't exist in the JSON file
- result = LoggingCallbackManager._add_custom_callback_generic_api_str(
- "unknown_callback"
- )
+ result = LoggingCallbackManager._add_custom_callback_generic_api_str("unknown_callback")
# Should return the string unchanged
assert result == "unknown_callback", "Unknown callback should be returned as-is"
@@ -409,9 +856,7 @@ async def test_generic_api_callback_settings_retry_config():
}
try:
- result = LoggingCallbackManager._add_custom_callback_generic_api_str(
- callback_name
- )
+ result = LoggingCallbackManager._add_custom_callback_generic_api_str(callback_name)
assert isinstance(result, GenericAPILogger)
assert result.endpoint == "https://example.com/api/logs"
diff --git a/tests/test_litellm/proxy/management_helpers/test_audit_log_callbacks.py b/tests/test_litellm/proxy/management_helpers/test_audit_log_callbacks.py
index b1d111bf1f9..404db4cf717 100644
--- a/tests/test_litellm/proxy/management_helpers/test_audit_log_callbacks.py
+++ b/tests/test_litellm/proxy/management_helpers/test_audit_log_callbacks.py
@@ -18,6 +18,7 @@ from litellm.proxy.management_helpers.audit_logs import (
_audit_log_task_done_callback,
_build_audit_log_payload,
_dispatch_audit_log_to_callbacks,
+ _resolve_audit_log_callback,
create_audit_log_for_update,
is_audit_logging_enabled,
)
@@ -26,8 +27,13 @@ from litellm.types.utils import StandardAuditLogPayload
@pytest.fixture(autouse=True)
def reset_audit_log_callbacks(monkeypatch: pytest.MonkeyPatch) -> None:
- """Every test starts with no audit log callbacks registered."""
+ """Every test starts with no audit log callbacks or cached resolvers."""
+ from litellm.proxy.management_helpers.audit_logs import reset_audit_log_callback_cache
+
monkeypatch.setattr(litellm, "audit_log_callbacks", [])
+ reset_audit_log_callback_cache()
+ yield
+ reset_audit_log_callback_cache()
def _make_audit_log(
@@ -47,6 +53,58 @@ def _make_audit_log(
)
+@pytest.mark.parametrize(
+ ("otel_v2_enabled", "expected_type"),
+ ((True, "OpenTelemetryV2"), (False, "OpenTelemetry")),
+)
+def test_resolving_an_audit_callback_is_local_and_not_dashboard_visible(
+ monkeypatch: pytest.MonkeyPatch,
+ otel_v2_enabled: bool,
+ expected_type: str,
+):
+ from litellm.integrations.opentelemetry import OpenTelemetry
+ from litellm.integrations.otel.logger import OpenTelemetryV2
+ from litellm.integrations.otel.model.config import is_otel_v2_enabled
+ from litellm.litellm_core_utils import litellm_logging
+ from litellm.litellm_core_utils.logging_callback_manager import LoggingCallbackManager
+
+ manager = LoggingCallbackManager()
+ monkeypatch.setenv("LITELLM_OTEL_V2", str(otel_v2_enabled).lower())
+ is_otel_v2_enabled.cache_clear()
+ monkeypatch.setattr(litellm_logging, "_in_memory_loggers", [])
+ monkeypatch.setattr(litellm, "input_callback", [])
+ monkeypatch.setattr(litellm, "service_callback", [])
+ monkeypatch.setattr(litellm, "success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+ try:
+ audit_logger = _resolve_audit_log_callback("otel")
+
+ expected_callback_type = OpenTelemetryV2 if expected_type == "OpenTelemetryV2" else OpenTelemetry
+ assert type(audit_logger) is expected_callback_type
+ assert audit_logger.callback_name == "otel"
+ assert _resolve_audit_log_callback("otel") is audit_logger
+ assert litellm_logging._in_memory_loggers == []
+ assert all(
+ audit_logger is not registered_callback
+ for callback_registry in (
+ litellm.input_callback,
+ litellm.service_callback,
+ litellm.success_callback,
+ litellm.failure_callback,
+ litellm._async_success_callback,
+ litellm._async_failure_callback,
+ litellm.callbacks,
+ )
+ for registered_callback in callback_registry
+ )
+ assert manager.get_dashboard_callback_registrations() == ()
+ finally:
+ is_otel_v2_enabled.cache_clear()
+
+
@pytest.mark.parametrize(
("premium_user", "configured_value", "environment_value", "expected"),
(
@@ -156,9 +214,7 @@ class TestDispatchAuditLogToCallbacks:
async def test_nonblocking_on_callback_failure(self, monkeypatch: pytest.MonkeyPatch):
"""Callback errors should not propagate."""
mock_logger = MagicMock(spec=CustomLogger)
- mock_logger.async_log_audit_log_event = AsyncMock(
- side_effect=RuntimeError("boom")
- )
+ mock_logger.async_log_audit_log_event = AsyncMock(side_effect=RuntimeError("boom"))
monkeypatch.setattr(litellm, "audit_log_callbacks", [mock_logger])
audit_log = _make_audit_log()
@@ -264,9 +320,7 @@ class TestCreateAuditLogForUpdateWithCallbacks:
patch("litellm.store_audit_logs", True),
patch("litellm.proxy.proxy_server.prisma_client") as mock_prisma,
):
- mock_prisma.db.litellm_auditlog.create = AsyncMock(
- side_effect=RuntimeError("DB connection lost")
- )
+ mock_prisma.db.litellm_auditlog.create = AsyncMock(side_effect=RuntimeError("DB connection lost"))
audit_log = _make_audit_log()
await create_audit_log_for_update(audit_log)
@@ -282,9 +336,7 @@ class TestAuditLogTaskDoneCallback:
mock_task = MagicMock(spec=asyncio.Task)
mock_task.exception.return_value = RuntimeError("callback failed")
- with patch(
- "litellm.proxy.management_helpers.audit_logs.verbose_proxy_logger"
- ) as mock_logger:
+ with patch("litellm.proxy.management_helpers.audit_logs.verbose_proxy_logger") as mock_logger:
_audit_log_task_done_callback(mock_task)
mock_logger.error.assert_called_once()
assert "callback failed" in str(mock_logger.error.call_args)
@@ -294,9 +346,7 @@ class TestAuditLogTaskDoneCallback:
mock_task = MagicMock(spec=asyncio.Task)
mock_task.exception.return_value = None
- with patch(
- "litellm.proxy.management_helpers.audit_logs.verbose_proxy_logger"
- ) as mock_logger:
+ with patch("litellm.proxy.management_helpers.audit_logs.verbose_proxy_logger") as mock_logger:
_audit_log_task_done_callback(mock_task)
mock_logger.error.assert_not_called()
@@ -305,9 +355,7 @@ class TestAuditLogTaskDoneCallback:
mock_task = MagicMock(spec=asyncio.Task)
mock_task.exception.side_effect = asyncio.CancelledError()
- with patch(
- "litellm.proxy.management_helpers.audit_logs.verbose_proxy_logger"
- ) as mock_logger:
+ with patch("litellm.proxy.management_helpers.audit_logs.verbose_proxy_logger") as mock_logger:
_audit_log_task_done_callback(mock_task)
mock_logger.error.assert_not_called()
@@ -386,9 +434,7 @@ class TestS3AuditCallbackParamsDecoupling:
from litellm.proxy.management_helpers import audit_logs as ll_audit_logs
monkeypatch.setattr(litellm, "s3_callback_params", litellm.s3_callback_params)
- monkeypatch.setattr(
- litellm, "s3_audit_callback_params", getattr(litellm, "s3_audit_callback_params", None)
- )
+ monkeypatch.setattr(litellm, "s3_audit_callback_params", getattr(litellm, "s3_audit_callback_params", None))
ll_audit_logs._audit_log_callback_cache.clear()
ll_logging._in_memory_loggers.clear()
yield
@@ -452,7 +498,6 @@ class TestS3AuditCallbackParamsDecoupling:
def test_empty_dict_opts_in(self, monkeypatch: pytest.MonkeyPatch):
"""`s3_audit_callback_params = {}` is opt-in (truthy-by-presence) and
produces a separate instance with no bucket configured (env/IAM-only)."""
- from litellm.integrations.s3_v2 import S3Logger
from litellm.litellm_core_utils.litellm_logging import (
_init_custom_logger_compatible_class,
)
diff --git a/tests/test_litellm/proxy/proxy_server/test_routes_config.py b/tests/test_litellm/proxy/proxy_server/test_routes_config.py
index ad3c470acf3..8e896588073 100644
--- a/tests/test_litellm/proxy/proxy_server/test_routes_config.py
+++ b/tests/test_litellm/proxy/proxy_server/test_routes_config.py
@@ -94,10 +94,7 @@ def test_config_update_no_db_error(client, auth_as, monkeypatch):
json={"general_settings": {"alerting": ["slack"]}},
)
assert response.status_code != 200
- assert (
- "db" in str(response.json()).lower()
- or "connect" in str(response.json()).lower()
- )
+ assert "db" in str(response.json()).lower() or "connect" in str(response.json()).lower()
# ---------------------------------------------------------------------------
@@ -138,9 +135,7 @@ def test_config_field_update_happy_admin(client, auth_as, mock_prisma, monkeypat
}
-def test_config_field_update_non_admin_rejected(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_config_field_update_non_admin_rejected(client, auth_as, mock_prisma, monkeypatch):
"""Non-admin cannot update config fields — returns 400 with not-allowed
detail (handler uses 400 for the auth gate, not 403)."""
from litellm.proxy import proxy_server as ps
@@ -200,9 +195,7 @@ def test_config_field_info_happy_admin(client, auth_as, mock_prisma, monkeypatch
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN):
- response = client.get(
- "/config/field/info", params={"field_name": "max_parallel_requests"}
- )
+ response = client.get("/config/field/info", params={"field_name": "max_parallel_requests"})
assert response.status_code == 200
assert normalize(response.json()) == {
"field_name": "max_parallel_requests",
@@ -210,9 +203,7 @@ def test_config_field_info_happy_admin(client, auth_as, mock_prisma, monkeypatch
}
-def test_config_field_info_non_admin_rejected(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_config_field_info_non_admin_rejected(client, auth_as, mock_prisma, monkeypatch):
"""Non-admin (INTERNAL_USER) is denied — admin-view gate fires."""
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
@@ -221,9 +212,7 @@ def test_config_field_info_non_admin_rejected(
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.INTERNAL_USER):
- response = client.get(
- "/config/field/info", params={"field_name": "max_parallel_requests"}
- )
+ response = client.get("/config/field/info", params={"field_name": "max_parallel_requests"})
assert response.status_code == 400
assert "error" in response.json().get("detail", {})
@@ -240,16 +229,12 @@ def test_config_field_info_field_not_in_db(client, auth_as, mock_prisma, monkeyp
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN):
- response = client.get(
- "/config/field/info", params={"field_name": "max_parallel_requests"}
- )
+ response = client.get("/config/field/info", params={"field_name": "max_parallel_requests"})
assert response.status_code == 400
assert "not in DB" in response.json().get("detail", {}).get("error", "")
-def test_config_field_info_redacts_nested_secret_for_view_only_admin(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_config_field_info_redacts_nested_secret_for_view_only_admin(client, auth_as, mock_prisma, monkeypatch):
"""A view-only admin reading a structured field must not receive nested
credentials. database_args carries aws_web_identity_token (a DynamoDB
role-assumption credential); it must come back redacted while non-secret
@@ -270,9 +255,7 @@ def test_config_field_info_redacts_nested_secret_for_view_only_admin(
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
- response = client.get(
- "/config/field/info", params={"field_name": "database_args"}
- )
+ response = client.get("/config/field/info", params={"field_name": "database_args"})
assert response.status_code == 200
value = response.json()["field_value"]
assert value["aws_web_identity_token"] == "REDACTED"
@@ -280,9 +263,7 @@ def test_config_field_info_redacts_nested_secret_for_view_only_admin(
assert value["user_table_name"] == "LiteLLM_UserTable"
-def test_config_field_info_full_admin_sees_nested_secret(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_config_field_info_full_admin_sees_nested_secret(client, auth_as, mock_prisma, monkeypatch):
"""The redaction must not over-redact for a full PROXY_ADMIN, who needs
the real nested value to populate the edit form."""
from litellm.proxy import proxy_server as ps
@@ -300,18 +281,14 @@ def test_config_field_info_full_admin_sees_nested_secret(
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN):
- response = client.get(
- "/config/field/info", params={"field_name": "database_args"}
- )
+ response = client.get("/config/field/info", params={"field_name": "database_args"})
assert response.status_code == 200
value = response.json()["field_value"]
assert value["aws_web_identity_token"] == "sk-super-secret-token"
assert value["region_name"] == "us-east-1"
-def test_config_field_info_redacts_top_level_scalar_for_view_only(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_config_field_info_redacts_top_level_scalar_for_view_only(client, auth_as, mock_prisma, monkeypatch):
"""The top-level scalar branch must also redact for a view-only admin.
database_url carries DB credentials and is not caught by the name masker,
so it is in the explicit secret set."""
@@ -325,9 +302,7 @@ def test_config_field_info_redacts_top_level_scalar_for_view_only(
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
- response = client.get(
- "/config/field/info", params={"field_name": "database_url"}
- )
+ response = client.get("/config/field/info", params={"field_name": "database_url"})
assert response.status_code == 200
assert response.json()["field_value"] == "REDACTED"
@@ -341,17 +316,12 @@ def test_redact_general_setting_value_recurses_list_of_dicts():
{"path": "/foo", "headers": {"Authorization": "Bearer sk-x"}},
{"path": "/bar", "client_secret": "sk-y"},
]
- redacted = ps._redact_general_setting_value(
- "some_list_field", value, is_full_admin=False
- )
+ redacted = ps._redact_general_setting_value("some_list_field", value, is_full_admin=False)
assert redacted[0]["headers"]["Authorization"] == "REDACTED"
assert redacted[0]["path"] == "/foo"
assert redacted[1]["client_secret"] == "REDACTED"
assert redacted[1]["path"] == "/bar"
- assert (
- ps._redact_general_setting_value("some_list_field", value, is_full_admin=True)
- == value
- )
+ assert ps._redact_general_setting_value("some_list_field", value, is_full_admin=True) == value
def test_redact_secret_values_in_obj_fails_closed_at_max_depth():
@@ -369,22 +339,16 @@ def test_redact_secret_values_in_obj_fails_closed_at_max_depth():
for _ in range(ps._REDACT_SECRET_MAX_DEPTH + 2):
nested = {"wrap": nested}
- out = ps._redact_general_setting_value(
- "some_struct_field", nested, is_full_admin=False
- )
+ out = ps._redact_general_setting_value("some_struct_field", nested, is_full_admin=False)
# the secret must not survive anywhere in the returned tree
assert "sk-leak-bottom" not in repr(out)
# full admin is unaffected by the cap — the value comes back untouched
- admin_out = ps._redact_general_setting_value(
- "some_struct_field", nested, is_full_admin=True
- )
+ admin_out = ps._redact_general_setting_value("some_struct_field", nested, is_full_admin=True)
assert admin_out is nested
-def test_config_list_redacts_pass_through_secret_for_view_only(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_config_list_redacts_pass_through_secret_for_view_only(client, auth_as, mock_prisma, monkeypatch):
"""/config/list must not leak pass_through_endpoints upstream credentials
to a view-only admin. pass_through_endpoints is a known secret-bearing
field, so a non-admin gets it redacted; a full admin still sees it."""
@@ -411,24 +375,16 @@ def test_config_list_redacts_pass_through_secret_for_view_only(
)
def _pass_through_value(body):
- return next(
- entry["field_value"]
- for entry in body
- if entry["field_name"] == "pass_through_endpoints"
- )
+ return next(entry["field_value"] for entry in body if entry["field_name"] == "pass_through_endpoints")
with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
- view_resp = client.get(
- "/config/list", params={"config_type": "general_settings"}
- )
+ view_resp = client.get("/config/list", params={"config_type": "general_settings"})
assert view_resp.status_code == 200
assert "sk-UPSTREAM-SECRET" not in view_resp.text
assert _pass_through_value(view_resp.json()) == "REDACTED"
with auth_as(LitellmUserRoles.PROXY_ADMIN):
- admin_resp = client.get(
- "/config/list", params={"config_type": "general_settings"}
- )
+ admin_resp = client.get("/config/list", params={"config_type": "general_settings"})
assert admin_resp.status_code == 200
admin_value = _pass_through_value(admin_resp.json())
assert admin_value[0]["headers"]["Authorization"] == "Bearer sk-UPSTREAM-SECRET"
@@ -452,9 +408,7 @@ def test_config_list_happy_admin(client, auth_as, mock_prisma, monkeypatch):
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.PROXY_ADMIN):
- response = client.get(
- "/config/list", params={"config_type": "general_settings"}
- )
+ response = client.get("/config/list", params={"config_type": "general_settings"})
assert response.status_code == 200
body = response.json()
assert isinstance(body, list)
@@ -560,9 +514,7 @@ def test_config_list_non_admin_rejected(client, auth_as, mock_prisma, monkeypatc
monkeypatch.setattr(ps, "prisma_client", mock_prisma)
with auth_as(LitellmUserRoles.INTERNAL_USER):
- response = client.get(
- "/config/list", params={"config_type": "general_settings"}
- )
+ response = client.get("/config/list", params={"config_type": "general_settings"})
assert response.status_code == 400
assert "role" in response.json().get("detail", {}).get("error", "").lower()
@@ -575,9 +527,7 @@ def test_config_list_no_db_error(client, auth_as, monkeypatch):
monkeypatch.setattr(ps, "prisma_client", None)
with auth_as(LitellmUserRoles.PROXY_ADMIN):
- response = client.get(
- "/config/list", params={"config_type": "general_settings"}
- )
+ response = client.get("/config/list", params={"config_type": "general_settings"})
assert response.status_code == 400
assert "error" in response.json().get("detail", {})
@@ -621,9 +571,7 @@ def test_config_field_delete_happy_admin(client, auth_as, mock_prisma, monkeypat
}
-def test_config_field_delete_non_admin_rejected(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_config_field_delete_non_admin_rejected(client, auth_as, mock_prisma, monkeypatch):
"""Non-admin caller hits the 400 not-allowed branch with role in detail."""
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
@@ -643,9 +591,7 @@ def test_config_field_delete_non_admin_rejected(
assert "role" in response.json().get("detail", {}).get("error", "").lower()
-def test_config_field_delete_field_not_in_config(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_config_field_delete_field_not_in_config(client, auth_as, mock_prisma, monkeypatch):
"""If there is no general_settings row at all, returns 400 'not in config'."""
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
@@ -690,9 +636,7 @@ def test_config_callback_delete_happy_admin(client, auth_as, mock_prisma, monkey
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
with auth_as(LitellmUserRoles.PROXY_ADMIN):
- response = client.post(
- "/config/callback/delete", json={"callback_name": "langfuse"}
- )
+ response = client.post("/config/callback/delete", json={"callback_name": "langfuse"})
assert response.status_code == 200
# `deleted_at` is an ISO timestamp generated at request time — extend
# the volatile set just for this assertion so dict-equality still works.
@@ -705,9 +649,7 @@ def test_config_callback_delete_happy_admin(client, auth_as, mock_prisma, monkey
}
-def test_config_callback_delete_non_admin_rejected(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_config_callback_delete_non_admin_rejected(client, auth_as, mock_prisma, monkeypatch):
"""Non-admin caller is rejected with 400 not-allowed."""
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
@@ -717,9 +659,7 @@ def test_config_callback_delete_non_admin_rejected(
monkeypatch.setattr(ps, "store_model_in_db", True)
with auth_as(LitellmUserRoles.INTERNAL_USER):
- response = client.post(
- "/config/callback/delete", json={"callback_name": "langfuse"}
- )
+ response = client.post("/config/callback/delete", json={"callback_name": "langfuse"})
assert response.status_code == 400
assert "role" in response.json().get("detail", {}).get("error", "").lower()
@@ -734,22 +674,15 @@ def test_config_callback_delete_not_found(client, auth_as, mock_prisma, monkeypa
monkeypatch.setattr(ps, "store_model_in_db", True)
fake_proxy_config = MagicMock()
- fake_proxy_config.get_config = AsyncMock(
- return_value={"litellm_settings": {"success_callback": ["slack"]}}
- )
+ fake_proxy_config.get_config = AsyncMock(return_value={"litellm_settings": {"success_callback": ["slack"]}})
monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
with auth_as(LitellmUserRoles.PROXY_ADMIN):
- response = client.post(
- "/config/callback/delete", json={"callback_name": "langfuse"}
- )
+ response = client.post("/config/callback/delete", json={"callback_name": "langfuse"})
# The handler re-raises HTTPException(404) verbatim (only generic
# `Exception` becomes a 500 ProxyException), so pin 404 strictly.
assert response.status_code == 404
- assert (
- "langfuse" in str(response.json()).lower()
- or "not found" in str(response.json()).lower()
- )
+ assert "langfuse" in str(response.json()).lower() or "not found" in str(response.json()).lower()
# ---------------------------------------------------------------------------
@@ -813,10 +746,367 @@ def test_get_config_callbacks_internal_error(client, auth_as, mock_prisma, monke
with auth_as(LitellmUserRoles.PROXY_ADMIN):
response = client.get("/get/config/callbacks")
assert response.status_code >= 400
- assert (
- "boom" in str(response.json()).lower()
- or "error" in str(response.json()).lower()
+ assert "boom" in str(response.json()).lower() or "error" in str(response.json()).lower()
+
+
+def test_get_config_callbacks_redacts_runtime_only_callback_for_view_only_admin(
+ client, auth_as, mock_prisma, monkeypatch
+):
+ import litellm
+ from litellm.proxy import proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ _install_litellm_config(mock_prisma)
+ monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+ monkeypatch.setattr(ps, "llm_router", None)
+ monkeypatch.setattr(litellm, "success_callback", ["langsmith"])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ fake_proxy_config = MagicMock()
+ fake_proxy_config.get_config = AsyncMock(
+ return_value={
+ "litellm_settings": {},
+ "general_settings": {},
+ "environment_variables": {"LANGSMITH_API_KEY": "secret"},
+ }
)
+ monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
+
+ with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
+ response = client.get("/get/config/callbacks")
+
+ assert response.status_code == 200
+ assert "secret" not in response.text
+ assert (
+ next(callback for callback in response.json()["callbacks"] if callback["name"] == "langsmith")["variables"][
+ "LANGSMITH_API_KEY"
+ ]
+ == "REDACTED"
+ )
+
+
+def test_get_config_callbacks_includes_runtime_callback_with_different_type(client, auth_as, mock_prisma, monkeypatch):
+ import litellm
+ from litellm.proxy import proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ _install_litellm_config(mock_prisma)
+ monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+ monkeypatch.setattr(ps, "llm_router", None)
+ monkeypatch.setattr(litellm, "success_callback", ["langsmith"])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", ["langsmith"])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ fake_proxy_config = MagicMock()
+ fake_proxy_config.get_config = AsyncMock(
+ return_value={
+ "litellm_settings": {"success_callback": ["langsmith"]},
+ "general_settings": {},
+ "environment_variables": {},
+ }
+ )
+ monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
+
+ with auth_as(LitellmUserRoles.PROXY_ADMIN):
+ response = client.get("/get/config/callbacks")
+
+ assert response.status_code == 200
+ callbacks = response.json()["callbacks"]
+ assert [
+ (callback["type"], callback.get("read_only", False))
+ for callback in callbacks
+ if callback["name"] == "langsmith"
+ ] == [("success", False), ("success_and_failure", True)]
+
+
+def test_get_config_callbacks_canonicalizes_runtime_callback_aliases(client, auth_as, mock_prisma, monkeypatch):
+ import litellm
+ from litellm.proxy import proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ _install_litellm_config(mock_prisma)
+ monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+ monkeypatch.setattr(ps, "llm_router", None)
+ monkeypatch.setattr(
+ litellm,
+ "success_callback",
+ ["s3_v2", "opentelemetry", "aws_sqs", "langfuse_otel"],
+ )
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ fake_proxy_config = MagicMock()
+ fake_proxy_config.get_config = AsyncMock(
+ return_value={
+ "litellm_settings": {"success_callback": ["s3", "otel"]},
+ "general_settings": {},
+ "environment_variables": {},
+ }
+ )
+ monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
+
+ with auth_as(LitellmUserRoles.PROXY_ADMIN):
+ response = client.get("/get/config/callbacks")
+
+ assert response.status_code == 200
+ callbacks = response.json()["callbacks"]
+ assert [(callback["name"], callback["type"], callback.get("read_only", False)) for callback in callbacks] == [
+ ("s3", "success", False),
+ ("otel", "success", False),
+ ("sqs", "success", True),
+ ("langfuse_otel", "success", True),
+ ]
+
+
+def test_get_config_callbacks_does_not_duplicate_configured_combined_runtime_callback(
+ client, auth_as, mock_prisma, monkeypatch
+):
+ import litellm
+ from litellm.proxy import proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ _install_litellm_config(mock_prisma)
+ monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+ monkeypatch.setattr(ps, "llm_router", None)
+ monkeypatch.setattr(litellm, "success_callback", ["langsmith"])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", ["langsmith"])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ fake_proxy_config = MagicMock()
+ fake_proxy_config.get_config = AsyncMock(
+ return_value={
+ "litellm_settings": {"callbacks": ["langsmith"]},
+ "general_settings": {},
+ "environment_variables": {},
+ }
+ )
+ monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
+
+ with auth_as(LitellmUserRoles.PROXY_ADMIN):
+ response = client.get("/get/config/callbacks")
+
+ assert response.status_code == 200
+ assert [
+ (callback["name"], callback["type"], callback.get("read_only", False))
+ for callback in response.json()["callbacks"]
+ ] == [("langsmith", "success_and_failure", False)]
+
+
+def test_get_config_callbacks_does_not_duplicate_named_generic_api_callback(client, auth_as, mock_prisma, monkeypatch):
+ import litellm
+ from litellm.integrations.generic_api.generic_api_callback import GenericAPILogger
+ from litellm.proxy import proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ _install_litellm_config(mock_prisma)
+ monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+ monkeypatch.setattr(ps, "llm_router", None)
+ monkeypatch.setattr(litellm, "success_callback", [object.__new__(GenericAPILogger)])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ fake_proxy_config = MagicMock()
+ fake_proxy_config.get_config = AsyncMock(
+ return_value={
+ "litellm_settings": {"success_callback": ["my_webhook"]},
+ "callback_settings": {
+ "my_webhook": {
+ "callback_type": "generic_api",
+ "endpoint": "https://example.com/events",
+ "headers": {"Content-Type": "application/json"},
+ }
+ },
+ "general_settings": {},
+ "environment_variables": {},
+ }
+ )
+ monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
+
+ with auth_as(LitellmUserRoles.PROXY_ADMIN):
+ response = client.get("/get/config/callbacks")
+
+ assert response.status_code == 200
+ assert [
+ (callback["name"], callback["type"], callback.get("read_only", False))
+ for callback in response.json()["callbacks"]
+ ] == [("my_webhook", "success", False)]
+
+
+def test_get_config_callbacks_does_not_duplicate_compatible_generic_api_callback(
+ client, auth_as, mock_prisma, monkeypatch
+):
+ import litellm
+ from litellm.integrations.generic_api.generic_api_callback import GenericAPILogger
+ from litellm.proxy import proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ _install_litellm_config(mock_prisma)
+ monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+ monkeypatch.setattr(ps, "llm_router", None)
+ monkeypatch.setattr(litellm, "success_callback", [object.__new__(GenericAPILogger)])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ fake_proxy_config = MagicMock()
+ fake_proxy_config.get_config = AsyncMock(
+ return_value={
+ "litellm_settings": {"success_callback": ["sumologic"]},
+ "general_settings": {},
+ "environment_variables": {},
+ }
+ )
+ monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
+
+ with auth_as(LitellmUserRoles.PROXY_ADMIN):
+ response = client.get("/get/config/callbacks")
+
+ assert response.status_code == 200
+ assert [
+ (callback["name"], callback["type"], callback.get("read_only", False))
+ for callback in response.json()["callbacks"]
+ ] == [("sumologic", "success", False)]
+
+
+def test_get_config_callbacks_does_not_duplicate_compatible_generic_api_callback_with_malformed_settings(
+ client, auth_as, mock_prisma, monkeypatch
+):
+ import litellm
+ from litellm.integrations.generic_api.generic_api_callback import GenericAPILogger
+ from litellm.proxy import proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ _install_litellm_config(mock_prisma)
+ monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+ monkeypatch.setattr(ps, "llm_router", None)
+ monkeypatch.setattr(litellm, "callback_settings", {"unrelated": {"callback_type": "generic_api"}})
+ monkeypatch.setattr(litellm, "success_callback", [object.__new__(GenericAPILogger)])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ fake_proxy_config = MagicMock()
+ fake_proxy_config.get_config = AsyncMock(
+ return_value={
+ "litellm_settings": {"success_callback": ["sumologic"]},
+ "callback_settings": "malformed",
+ "general_settings": {},
+ "environment_variables": {},
+ }
+ )
+ monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
+
+ with auth_as(LitellmUserRoles.PROXY_ADMIN):
+ response = client.get("/get/config/callbacks")
+
+ assert response.status_code == 200
+ assert [
+ (callback["name"], callback["type"], callback.get("read_only", False))
+ for callback in response.json()["callbacks"]
+ ] == [("sumologic", "success", False)]
+
+
+def test_get_config_callbacks_includes_runtime_only_callback(client, auth_as, mock_prisma, monkeypatch):
+ import litellm
+ from litellm.proxy import proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ _install_litellm_config(mock_prisma)
+ monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+ monkeypatch.setattr(ps, "llm_router", None)
+ monkeypatch.setattr(litellm, "success_callback", ["langfuse", "langsmith", "internal_callback"])
+ monkeypatch.setattr(litellm, "_async_success_callback", ["langsmith"])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ fake_proxy_config = MagicMock()
+ fake_proxy_config.get_config = AsyncMock(
+ return_value={
+ "litellm_settings": {"success_callback": ["langfuse"]},
+ "general_settings": {},
+ "environment_variables": {},
+ }
+ )
+ monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
+
+ with auth_as(LitellmUserRoles.PROXY_ADMIN):
+ response = client.get("/get/config/callbacks")
+
+ assert response.status_code == 200
+ callbacks = response.json()["callbacks"]
+ callback_names = [callback["name"] for callback in callbacks]
+ assert callback_names.count("langfuse") == 1
+ assert callback_names.count("langsmith") == 1
+ assert "internal_callback" not in callback_names
+ assert next(callback for callback in callbacks if callback["name"] == "langsmith") == {
+ "name": "langsmith",
+ "variables": {
+ "LANGSMITH_API_KEY": None,
+ "LANGSMITH_PROJECT": None,
+ "LANGSMITH_DEFAULT_RUN_NAME": None,
+ },
+ "type": "success",
+ "read_only": True,
+ }
+
+
+def test_get_config_callbacks_includes_and_redacts_runtime_only_azure_sentinel(
+ client, auth_as, mock_prisma, monkeypatch
+):
+ import litellm
+ from litellm.integrations.azure_sentinel.azure_sentinel import AzureSentinelLogger
+ from litellm.proxy import proxy_server as ps
+ from litellm.proxy._types import LitellmUserRoles
+
+ _install_litellm_config(mock_prisma)
+ monkeypatch.setattr(ps, "prisma_client", mock_prisma)
+ monkeypatch.setattr(ps, "llm_router", None)
+ monkeypatch.setattr(litellm, "success_callback", ["langfuse", object.__new__(AzureSentinelLogger)])
+ monkeypatch.setattr(litellm, "_async_success_callback", [])
+ monkeypatch.setattr(litellm, "failure_callback", [])
+ monkeypatch.setattr(litellm, "_async_failure_callback", [])
+ monkeypatch.setattr(litellm, "callbacks", [])
+
+ fake_proxy_config = MagicMock()
+ fake_proxy_config.get_config = AsyncMock(
+ return_value={
+ "litellm_settings": {"success_callback": ["langfuse"]},
+ "general_settings": {},
+ "environment_variables": {
+ "AZURE_SENTINEL_CLIENT_SECRET": "azure-sentinel-secret",
+ "AZURE_SENTINEL_ENDPOINT": "https://example.com",
+ },
+ }
+ )
+ monkeypatch.setattr(ps, "proxy_config", fake_proxy_config)
+
+ with auth_as(LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
+ response = client.get("/get/config/callbacks")
+
+ assert response.status_code == 200
+ assert "azure-sentinel-secret" not in response.text
+ callbacks = response.json()["callbacks"]
+ assert [(callback["name"], callback["type"], callback.get("read_only", False)) for callback in callbacks] == [
+ ("langfuse", "success", False),
+ ("azure_sentinel", "success", True),
+ ]
+ azure_sentinel = callbacks[-1]
+ assert azure_sentinel["variables"]["AZURE_SENTINEL_CLIENT_SECRET"] == "REDACTED"
+ assert azure_sentinel["variables"]["AZURE_SENTINEL_ENDPOINT"] == "https://example.com"
_CALLBACK_ENV_FIXTURE = {
@@ -850,14 +1140,10 @@ def _install_callbacks_config(monkeypatch, mock_prisma):
def _callback_variables(body: dict, name: str) -> dict:
- return next(
- cb["variables"] for cb in body["callbacks"] if cb["name"] == name
- )
+ return next(cb["variables"] for cb in body["callbacks"] if cb["name"] == name)
-def test_get_config_callbacks_redacts_secret_env_vars_for_view_only_admin(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_get_config_callbacks_redacts_secret_env_vars_for_view_only_admin(client, auth_as, mock_prisma, monkeypatch):
from litellm.proxy._types import LitellmUserRoles
_install_callbacks_config(monkeypatch, mock_prisma)
@@ -889,9 +1175,7 @@ def test_get_config_callbacks_redacts_secret_env_vars_for_view_only_admin(
assert otel_vars["OTEL_ENDPOINT"] == _CALLBACK_ENV_FIXTURE["OTEL_ENDPOINT"]
-def test_get_config_callbacks_full_admin_still_sees_secret_env_vars(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_get_config_callbacks_full_admin_still_sees_secret_env_vars(client, auth_as, mock_prisma, monkeypatch):
from litellm.proxy._types import LitellmUserRoles
_install_callbacks_config(monkeypatch, mock_prisma)
@@ -912,9 +1196,7 @@ def test_get_config_callbacks_full_admin_still_sees_secret_env_vars(
assert otel_vars["OTEL_HEADERS"] == _CALLBACK_ENV_FIXTURE["OTEL_HEADERS"]
-def test_get_config_callbacks_redacts_slack_webhook_urls_for_view_only_admin(
- client, auth_as, mock_prisma, monkeypatch
-):
+def test_get_config_callbacks_redacts_slack_webhook_urls_for_view_only_admin(client, auth_as, mock_prisma, monkeypatch):
from litellm.proxy import proxy_server as ps
from litellm.proxy._types import LitellmUserRoles
@@ -1048,9 +1330,7 @@ def test_config_yaml_returns_demo_payload(client, auth_as):
response = client.request("GET", "/config/yaml", json={})
shape = {
"status": response.status_code,
- "media_type_yaml": response.headers.get("content-type", "").startswith(
- "application/json"
- ),
+ "media_type_yaml": response.headers.get("content-type", "").startswith("application/json"),
"has_body": len(response.content) > 0,
}
assert shape == {
diff --git a/ui/litellm-dashboard/src/components/Settings/LoggingAndAlerts/LoggingCallbacks/LoggingCallbacksTable.test.tsx b/ui/litellm-dashboard/src/components/Settings/LoggingAndAlerts/LoggingCallbacks/LoggingCallbacksTable.test.tsx
index 3ef7dee01b0..f895184a9d5 100644
--- a/ui/litellm-dashboard/src/components/Settings/LoggingAndAlerts/LoggingCallbacks/LoggingCallbacksTable.test.tsx
+++ b/ui/litellm-dashboard/src/components/Settings/LoggingAndAlerts/LoggingCallbacks/LoggingCallbacksTable.test.tsx
@@ -97,6 +97,18 @@ describe("LoggingCallbacksTable", () => {
expect(onDelete).toHaveBeenCalledWith(callback);
});
+ it("should not render actions for a read-only callback", () => {
+ const callback = {
+ name: "langsmith",
+ type: "success" as const,
+ read_only: true,
+ variables: baseVars,
+ };
+ render();
+
+ expect(screen.queryByTestId("callback-actions-langsmith-success")).not.toBeInTheDocument();
+ });
+
// Regression: `/get_callbacks` returns the same `name` twice when a
// callback is registered for both success and failure (e.g. `generic_api`
// → POST to spend-log on both 200 and 4xx/5xx). The UI used to ignore
diff --git a/ui/litellm-dashboard/src/components/Settings/LoggingAndAlerts/LoggingCallbacks/LoggingCallbacksTableColumns.tsx b/ui/litellm-dashboard/src/components/Settings/LoggingAndAlerts/LoggingCallbacks/LoggingCallbacksTableColumns.tsx
index 2263fe03b3d..c0f5748670e 100644
--- a/ui/litellm-dashboard/src/components/Settings/LoggingAndAlerts/LoggingCallbacks/LoggingCallbacksTableColumns.tsx
+++ b/ui/litellm-dashboard/src/components/Settings/LoggingAndAlerts/LoggingCallbacks/LoggingCallbacksTableColumns.tsx
@@ -50,6 +50,10 @@ interface CallbackRowActionsProps {
}
function CallbackRowActions({ callback, onTest, onEdit, onDelete }: CallbackRowActionsProps) {
+ if (callback.read_only) {
+ return null;
+ }
+
return (