refactor(otel/v2): satisfy the type-discipline and strict-lint ceilings

Never-rebound names carry Final, deliberate rebinds carry rebind-ok with the reason, and the endpoint suffix check collapses into one endswith call. Two hook parameters and the constructor kwargs move from Any to object, which is both stricter and what the bodies actually rely on.

The remaining constructions keep a mutable-ok naming what needs a concrete dict: pydantic model_copy payloads, the OTel SDK Resource call, and the LRU caches that are mutable by design.

litellm/integrations/otel/destination_logger.py

@@ -13,7 +13,7 @@ owning logger, and including it here would export the same call twice.
 from collections.abc import Mapping
 from datetime import datetime
 from functools import lru_cache
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Final
 
 from opentelemetry.trace import Span
 
@@ -34,14 +34,14 @@ if TYPE_CHECKING:
 
 def _vocabulary_config(backend: str) -> OpenTelemetryV2Config:
     """The backend's span vocabulary with no exporter of its own."""
-    preset_fn = PRESET_BY_CALLBACK.get(backend)
+    preset_fn: Final = PRESET_BY_CALLBACK.get(backend)
     if preset_fn is None:
         return OpenTelemetryV2Config()
     try:
-        config = preset_fn(allow_missing_credentials=True)
+        config: Final = preset_fn(allow_missing_credentials=True)
     except Exception:  # noqa: BLE001  # an unbuildable preset still has a usable default vocabulary
         return OpenTelemetryV2Config()
-    return config.model_copy(update={"exporters": ()})
+    return config.model_copy(update={"exporters": ()})  # mutable-ok: pydantic model_copy takes a plain update mapping
 
 
 class _DestinationOnlyOtel(OpenTelemetryV2):
@@ -89,7 +89,7 @@ class _DestinationOnlyOtel(OpenTelemetryV2):
         Emitting a second one here would deliver the same discovery call twice; returning
         ``True`` still stops the caller from closing it as an LLM call.
         """
-        raw_payload = kwargs.get("standard_logging_object")
+        raw_payload: Final = kwargs.get("standard_logging_object")
         return isinstance(raw_payload, Mapping) and is_mcp_list_tools(raw_payload)
 
     def export_to_destinations(
@@ -114,15 +114,15 @@ class _DestinationOnlyOtel(OpenTelemetryV2):
 class AdminDestinationLogger(CustomLogger):
     """Delivers each request's gen-AI span to the destinations its identity resolved."""
 
-    def __init__(self, **kwargs: Any) -> None:
+    def __init__(self, **kwargs: object) -> None:  # kwargs-ok: CustomLogger's constructor signature
         super().__init__(**kwargs)
         self._emitters: dict[str, _DestinationOnlyOtel] = {}  # mutable-ok: bounded per-backend emitter cache
 
     def _emitter_for(self, backend: str) -> _DestinationOnlyOtel:
-        existing = self._emitters.get(backend)
+        existing: Final = self._emitters.get(backend)
         if existing is not None:
             return existing
-        emitter = _DestinationOnlyOtel(config=_vocabulary_config(backend), callback_name=backend)
+        emitter: Final = _DestinationOnlyOtel(config=_vocabulary_config(backend), callback_name=backend)
         self._emitters[backend] = emitter
         return emitter
 
@@ -132,8 +132,11 @@ class AdminDestinationLogger(CustomLogger):
         start_time: "datetime | float | None",
         end_time: "datetime | float | None",
     ) -> None:
-        owned = otel_v2_owned_backends()
-        for backend in sorted({d.callback_name for d in request_destinations() if d.callback_name} - owned):
+        owned: Final = otel_v2_owned_backends()
+        for backend in sorted(
+            {d.callback_name for d in request_destinations() if d.callback_name}
+            - owned  # mutable-ok: handed to a model or SDK that needs a concrete dict
+        ):  # mutable-ok: construction is handed to an API that needs a concrete dict
             try:
                 self._emitter_for(backend).export_to_destinations(kwargs, start_time, end_time)
             except Exception as exc:  # noqa: BLE001  # one destination's failure must not break the request or the others
@@ -142,7 +145,7 @@ class AdminDestinationLogger(CustomLogger):
     async def async_log_success_event(
         self,
         kwargs: "Mapping[str, Any]",
-        response_obj: Any,
+        response_obj: object,
         start_time: "datetime | float | None",
         end_time: "datetime | float | None",
     ) -> None:
@@ -151,7 +154,7 @@ class AdminDestinationLogger(CustomLogger):
     async def async_log_failure_event(
         self,
         kwargs: "Mapping[str, Any]",
-        response_obj: Any,
+        response_obj: object,
         start_time: "datetime | float | None",
         end_time: "datetime | float | None",
     ) -> None:
@@ -165,7 +168,7 @@ def admin_destination_logger() -> AdminDestinationLogger:
 
 def register_admin_destination_logger() -> None:
     """Put the destination sink on the proxy's async callback lists, once."""
-    sink = admin_destination_logger()
-    for bucket in (litellm._async_success_callback, litellm._async_failure_callback):
+    sink: Final = admin_destination_logger()
+    for bucket in (litellm._async_success_callback, litellm._async_failure_callback):  # pyright: ignore[reportPrivateUsage]  # the proxy's own callback buckets; the sink must register on them like any built-in logger
         if not any(callback is sink for callback in bucket):
             bucket.append(sink)

litellm/integrations/otel/emitter.py

@@ -238,16 +238,16 @@ class SpanEmitter:
         # alone coalesced the successful attempt into the failed one it replaced and the
         # retried call reached its destinations only as the failure. The response id splits
         # the attempts apart while a sync+async double-firing of one attempt still shares it.
-        attempt_id = data.response_id if isinstance(data, LLMCallSpanData) else None
-        dedup_key = (
+        attempt_id: Final = data.response_id if isinstance(data, LLMCallSpanData) else None
+        dedup_key: Final = (
             (f"{data.identity.call_id}:{attempt_id}" if attempt_id else data.identity.call_id)
             if isinstance(data, (LLMCallSpanData, MCPToolCallSpanData))
             else None
         )
         if self._seen(dedup_key, role):
             return None
-        name = _NAME_BUILDERS[role](data)
-        first: Span | None = None
+        name: Final = _NAME_BUILDERS[role](data)
+        first: Span | None = None  # rebind-ok: set to the first span produced below
         for tracer in tracers:
             span = self.start_span(
                 role,

litellm/integrations/otel/logger.py

@@ -212,7 +212,7 @@ class OpenTelemetryV2(CustomLogger):
         other backend, and each one's own exporter went dark. Matching the name as well
         lets each backend register itself while still de-duplicating a repeat of itself.
         """
-        already_otel = any(
+        already_otel: Final = any(
             cb.__class__.__module__.startswith(_OTEL_MODULES)
             and (not per_backend or getattr(cb, "callback_name", None) == self.callback_name)
             for cb in callbacks
@@ -290,15 +290,15 @@ class OpenTelemetryV2(CustomLogger):
         # by the previous attempt is cleared here; leaving it made the close callback
         # short-circuit and the successful attempt after a failure went untraced.
         self._closed_call_ids.pop(call_id, None)
-        start_time_ns = to_ns(datetime.now())
-        spans: tuple[Span, ...] = ()
+        start_time_ns: Final = to_ns(datetime.now())
+        spans: tuple[Span, ...] = ()  # rebind-ok: reassigned on the branch below
         # Parent to the request's anchored root span (stable across the request),
         # falling back to ambient on the SDK path. Open the span live only when
         # that resolves to a recordable parent; otherwise defer to the close
         # callback (the thread-pool case, where the anchor isn't visible here).
         parent_context: Final = resolve_request_span_context()
         if is_recordable_span(get_current_span(parent_context)):
-            spans = tuple(
+            spans = tuple(  # rebind-ok: replaces the empty default declared above
                 self._emitter.start_span(
                     SpanRole.LLM_CALL,
                     call.provisional_span_name,
@@ -415,7 +415,7 @@ class OpenTelemetryV2(CustomLogger):
         # The tool-call span carries ``gen_ai.operation.name``, so the fan-out processor
         # treats it as a gen-AI span and skips it; route it to the request's admin
         # destinations like the LLM-call span, or it reaches only the global exporter.
-        call = LLMCallEvent.from_dict(kwargs)
+        call: Final = LLMCallEvent.from_dict(kwargs)
         self._emitter.emit_fanout(
             SpanRole.MCP_TOOL_CALL,
             data,
@@ -487,11 +487,11 @@ class OpenTelemetryV2(CustomLogger):
         """
         from litellm.integrations.otel.presets import dynamic_otlp_headers
 
-        call = LLMCallEvent.from_dict(kwargs)
-        call_id = call.call_id
+        call: Final = LLMCallEvent.from_dict(kwargs)
+        call_id: Final = call.call_id
 
-        carrier = self._open_llm_calls.pop(call_id, None) if call_id else None
-        payload = call.payload
+        carrier: Final = self._open_llm_calls.pop(call_id, None) if call_id else None
+        payload: Final = call.payload
 
         # The closed marker guards the carrier-less path only, where it stops a repeat
         # success/failure callback re-emitting a span that already shipped. An open
@@ -503,13 +503,13 @@ class OpenTelemetryV2(CustomLogger):
         # its own payload id (the provider's response id, falling back to the call id).
         # Keying on the call id made the successful attempt after a failure look like a
         # duplicate, so a destination saw only the failure.
-        emit_key = (payload.get("id") if payload else None) or call_id
+        emit_key: Final = (payload.get("id") if payload else None) or call_id
         if carrier is None and emit_key and emit_key in self._closed_call_ids:
             return None
 
         if carrier is None:
-            destinations = self._destinations_for_backend(call)
-            own_credentials = bool(dynamic_otlp_headers(self.callback_name, call.dynamic_params))
+            destinations: Final = self._destinations_for_backend(call)
+            own_credentials: Final = bool(dynamic_otlp_headers(self.callback_name, call.dynamic_params))
             if call.is_no_upstream_call or payload is None or not (destinations or own_credentials):
                 return None
             self._mark_closed(emit_key)
@@ -522,14 +522,14 @@ class OpenTelemetryV2(CustomLogger):
                 call.dynamic_params,
             )
 
-        end_time_ns = to_ns(end_time)
+        end_time_ns: Final = to_ns(end_time)
         self._mark_closed(emit_key)
         if payload is None:
             for span in carrier.spans:
                 span.end(end_time=end_time_ns)
             return None
 
-        data = LLMCallSpanData.from_standard_logging_payload(
+        data: Final = LLMCallSpanData.from_standard_logging_payload(
             payload,
             capture_content=self.config.capture_span_content,
             time_to_first_chunk_seconds=call.time_to_first_chunk_seconds,
@@ -569,7 +569,7 @@ class OpenTelemetryV2(CustomLogger):
         Two callers: the SDK thread-pool path and the destination-resolver path. Both anchor to
         the request's root span via the worker-copied context and seed identity Baggage.
         """
-        data = LLMCallSpanData.from_standard_logging_payload(
+        data: Final = LLMCallSpanData.from_standard_logging_payload(
             payload,
             capture_content=self.config.capture_span_content,
             time_to_first_chunk_seconds=time_to_first_chunk_seconds,
@@ -878,7 +878,7 @@ def publish_global_otel_v2_provider(
         register_admin_destination_logger,
     )
 
-    logger = select_global_otel_v2_logger(in_memory_loggers, registered=registered)
+    logger: Final = select_global_otel_v2_logger(in_memory_loggers, registered=registered)
     set_global_provider(logger._tracer_provider)
     register_admin_destination_logger()
     return logger

litellm/integrations/otel/model/config.py

@@ -71,7 +71,7 @@ class _OTelV2Flag(BaseSettings):
         """
         if not isinstance(value, str):
             return value
-        stripped = value.strip()
+        stripped: Final = value.strip()
         if not stripped:
             return False
         try:
@@ -289,7 +289,7 @@ class OpenTelemetryV2Config(BaseSettings):
         # span (including prompt and completion content) to stdout synchronously on the
         # request path. An explicitly chosen exporter (even ``console``), a non-console
         # kind, or an endpoint all still fold.
-        console_by_default = self.exporter == "console" and "exporter" not in self.model_fields_set
+        console_by_default: Final = self.exporter == "console" and "exporter" not in self.model_fields_set
         if not self.exporters and (self.endpoint or not console_by_default):
             self.exporters = [
                 ExporterSpec(

litellm/integrations/otel/plumbing/providers.py

@@ -1,6 +1,7 @@
 """Provider / exporter factory + the Baggage span processor."""
 
 from collections.abc import Callable, Iterable, Mapping
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final
 
 from opentelemetry import _logs, baggage, metrics
@@ -112,7 +113,7 @@ def _otlp_traces_endpoint(endpoint: str | None) -> str | None:
         return endpoint
     endpoint = endpoint.rstrip("/")
     # Splunk Observability uses ``/v2/trace/otlp``; never rewrite it.
-    if endpoint.endswith("/v1/traces") or "/v2/trace/otlp" in endpoint or endpoint.endswith("/api/trace"):
+    if endpoint.endswith(("/v1/traces", "/api/trace")) or "/v2/trace/otlp" in endpoint:
         return endpoint
     for other_signal in ("/v1/logs", "/v1/metrics"):
         if endpoint.endswith(other_signal):
@@ -120,7 +121,7 @@ def _otlp_traces_endpoint(endpoint: str | None) -> str | None:
     return endpoint + "/v1/traces"
 
 
-_GRPC_BACKENDS = frozenset({"arize"})
+_GRPC_BACKENDS: Final = frozenset({"arize"})
 
 
 def default_otlp_kind_for_backend(callback_name: "str | None") -> str:
@@ -133,7 +134,7 @@ def destination_resource_attrs(destination: "OtelDestination") -> Mapping[str, s
     ``model_id`` / ``arize.project.name``; empty for header-routed backends), read
     by both export paths so the gen-AI span and its parents share one Resource.
     """
-    return dict(destination.resource_attributes)
+    return MappingProxyType(dict(destination.resource_attributes))
 
 
 def parse_headers(raw: str | None) -> dict[str, str]:

litellm/integrations/otel/plumbing/routing.py

@@ -67,7 +67,7 @@ class TenantTracerCache:
         self._callback_name = callback_name
         self._tracer_name = tracer_name
         self._providers: OrderedDict[tuple[object, ...], TracerProvider] = (
-            OrderedDict()
+            OrderedDict()  # mutable-ok: LRU cache, mutated by design
         )  # mutable-ok: bounded LRU tracer-provider cache
 
     def _evict_if_full(self) -> None:
@@ -96,7 +96,7 @@ class TenantTracerCache:
         """
         if not destinations:
             return (default,)
-        groups = tuple(
+        groups: Final = tuple(
             self._tracer_for_group(resource_key, group, include_base=False)
             for resource_key, group in self._group_by_resource(destinations)
         )
@@ -115,7 +115,7 @@ class TenantTracerCache:
         headers: Final = dynamic_otlp_headers(self._callback_name, dynamic_params)
         if not headers:
             return self.tracers_for(default, destinations)
-        dynamic = self._credential_scoped_tracer(headers, dynamic_params)
+        dynamic: Final = self._credential_scoped_tracer(headers, dynamic_params)
         if not destinations:
             return (dynamic,)
         return (dynamic, *self.tracers_for(default, destinations, include_base_on_first=False))
@@ -135,18 +135,20 @@ class TenantTracerCache:
         """
         from litellm.integrations.otel.presets import dynamic_otlp_destination
 
-        destination = dynamic_otlp_destination(self._callback_name, dynamic_params)
-        cache_key: tuple[object, ...] = (
+        destination: Final = dynamic_otlp_destination(self._callback_name, dynamic_params)
+        cache_key: Final[tuple[object, ...]] = (
             "dynamic",
             tuple(sorted(headers.items())),
             destination.endpoint if destination is not None else None,
             destination.protocol if destination is not None else None,
         )
-        provider = self._providers.get(cache_key)
+        provider = self._providers.get(cache_key)  # rebind-ok: replaced below on a cache miss
         if provider is not None:
             self._providers.move_to_end(cache_key)
         else:
-            provider = build_tracer_provider(self._config_with_headers(headers, dynamic_params))
+            provider = build_tracer_provider(  # rebind-ok: reassigned on the branch below
+                self._config_with_headers(headers, dynamic_params)
+            )  # rebind-ok: cache miss
             self._providers[cache_key] = provider
             self._evict_if_full()
         return get_tracer(provider, self._tracer_name)
@@ -171,18 +173,25 @@ class TenantTracerCache:
         same builder an equivalent admin destination would use so the team reaches the
         account its ``callback_vars`` name.
         """
-        header_str = ",".join(f"{key}={value}" for key, value in headers.items())
-        owns_exporter = any(
+        header_str: Final = ",".join(f"{key}={value}" for key, value in headers.items())
+        owns_exporter: Final = any(
             spec.owner == self._callback_name and spec.kind.lower() not in _NON_OTLP_KINDS
             for spec in self._config.exporters
         )
         if not owns_exporter:
             return self._config.model_copy(
-                update={"exporters": [*self._config.exporters, *self._synthesized_exporter(header_str, dynamic_params)]}
+                update={  # mutable-ok: handed to a model or SDK that needs a concrete dict
+                    "exporters": [
+                        *self._config.exporters,
+                        *self._synthesized_exporter(header_str, dynamic_params),
+                    ]  # mutable-ok: handed to a model or SDK that needs a concrete dict
+                }  # mutable-ok: pydantic model_copy takes a plain update mapping
             )
-        exporters = [
+        exporters: Final = [
             (
-                spec.model_copy(update={"headers": header_str})
+                spec.model_copy(
+                    update={"headers": header_str}  # mutable-ok: handed to a model or SDK that needs a concrete dict
+                )  # mutable-ok: pydantic model_copy takes a plain update mapping
                 if spec.owner == self._callback_name and spec.kind.lower() not in _NON_OTLP_KINDS
                 else spec
             )
@@ -204,7 +213,7 @@ class TenantTracerCache:
         """
         from litellm.integrations.otel.presets import dynamic_otlp_destination
 
-        destination = dynamic_otlp_destination(self._callback_name, dynamic_params)
+        destination: Final = dynamic_otlp_destination(self._callback_name, dynamic_params)
         if destination is None or not destination.endpoint:
             return ()
         return (
@@ -228,12 +237,12 @@ class TenantTracerCache:
             destination_resource_attrs,
         )
 
-        groups: OrderedDict[tuple[tuple[str, str], ...], list[OtelDestination]] = (
-            OrderedDict()
-        )  # mutable-ok: insertion-order grouping accumulator, frozen before return
+        groups: Final[  # mutable-ok: insertion-order grouping accumulator, frozen before return
+            OrderedDict[tuple[tuple[str, str], ...], list[OtelDestination]]
+        ] = OrderedDict()
         for destination in destinations:
             key = tuple(sorted(destination_resource_attrs(destination).items()))
-            groups.setdefault(key, []).append(destination)
+            groups.setdefault(key, []).append(destination)  # mutable-ok: grouping accumulator, frozen before return
         return tuple((key, tuple(group)) for key, group in sorted(groups.items()))
 
     def _tracer_for_group(
@@ -243,16 +252,16 @@ class TenantTracerCache:
         *,
         include_base: bool,
     ) -> Tracer:
-        cache_key: tuple[object, ...] = (
+        cache_key: Final[tuple[object, ...]] = (
             resource_key,
             tuple(sorted((d.endpoint, tuple(sorted(d.headers.items())), d.protocol or "") for d in group)),
             include_base,
         )
-        provider = self._providers.get(cache_key)
+        provider = self._providers.get(cache_key)  # rebind-ok: replaced below on a cache miss
         if provider is not None:
             self._providers.move_to_end(cache_key)
         else:
-            provider = build_tracer_provider(
+            provider = build_tracer_provider(  # rebind-ok: cache miss
                 self._config_with_destinations(tuple(group), include_base_exporters=include_base)
             )
             self._providers[cache_key] = provider
@@ -289,8 +298,8 @@ class TenantTracerCache:
             destination_resource_attrs,
         )
 
-        kind = self._owned_otlp_kind()
-        appended = tuple(
+        kind: Final = self._owned_otlp_kind()
+        appended: Final = tuple(
             ExporterSpec(
                 kind=d.protocol or kind,
                 endpoint=d.endpoint,
@@ -299,22 +308,27 @@ class TenantTracerCache:
             )
             for d in destinations
         )
-        base_exporters = (*self._config.exporters,) if include_base_exporters else ()
-        merged_resource_attrs = {
+        base_exporters: Final = (*self._config.exporters,) if include_base_exporters else ()
+        merged_resource_attrs: Final = {  # mutable-ok: handed to a model or SDK that needs a concrete dict
             **self._config.resource_attributes,
-            **{key: value for d in destinations for key, value in destination_resource_attrs(d).items()},
+            **{  # mutable-ok: handed to a model or SDK that needs a concrete dict
+                key: value for d in destinations for key, value in destination_resource_attrs(d).items()
+            },  # mutable-ok: construction is handed to an API that needs a concrete dict
         }
         return self._config.model_copy(
-            update={
-                "exporters": [*base_exporters, *appended],
+            update={  # mutable-ok: pydantic model_copy takes a plain update mapping
+                "exporters": [  # mutable-ok: handed to a model or SDK that needs a concrete dict
+                    *base_exporters,
+                    *appended,
+                ],  # mutable-ok: pydantic model_copy takes a plain update mapping
                 "resource_attributes": merged_resource_attrs,
             }
         )
 
 
-_MAX_CACHED_PROCESSORS = 256
+_MAX_CACHED_PROCESSORS: Final = 256
 
-_GENAI_SPAN_ATTR = "gen_ai.operation.name"
+_GENAI_SPAN_ATTR: Final = "gen_ai.operation.name"
 
 
 def _processor_key(destination: OtelDestination) -> "tuple[str, tuple[tuple[str, str], ...], str | None]":
@@ -322,7 +336,7 @@ def _processor_key(destination: OtelDestination) -> "tuple[str, tuple[tuple[str,
 
 
 def _is_genai_span(span: ReadableSpan) -> bool:
-    attributes = span.attributes or {}
+    attributes: Final = span.attributes or {}  # mutable-ok: read-only fallback for a span with no attributes
     return _GENAI_SPAN_ATTR in attributes
 
 
@@ -333,10 +347,12 @@ def _with_destination_resource(span: ReadableSpan, destination: OtelDestination)
         destination_resource_attrs,
     )
 
-    extra = destination_resource_attrs(destination)
+    extra: Final = destination_resource_attrs(destination)
     if not extra:
         return span
-    merged = Resource.create({**dict(span.resource.attributes), **extra})
+    merged: Final = Resource.create(
+        {**dict(span.resource.attributes), **extra}  # mutable-ok: handed to a model or SDK that needs a concrete dict
+    )  # mutable-ok: the OTel SDK takes a concrete attribute mapping
     return _ResourceWrappedReadableSpan(span, merged)
 
 
@@ -371,14 +387,14 @@ class TenantFanOutSpanProcessor(SpanProcessor):
     def __init__(self, owner_callback_name: str | None) -> None:
         self._owner = owner_callback_name
         self._processors: OrderedDict[tuple, SpanProcessor] = (
-            OrderedDict()
+            OrderedDict()  # mutable-ok: LRU cache, mutated by design
         )  # mutable-ok: bounded LRU span-processor cache
 
     def on_start(self, span: Span, parent_context: Context | None = None) -> None:
         return None
 
     def on_end(self, span: ReadableSpan) -> None:
-        destinations = request_destinations()
+        destinations: Final = request_destinations()
         if not destinations:
             return
         if _is_genai_span(span):
@@ -409,7 +425,7 @@ class TenantFanOutSpanProcessor(SpanProcessor):
         self._processors.clear()
 
     def force_flush(self, timeout_millis: int = 30000) -> bool:
-        all_ok = True
+        all_ok = True  # rebind-ok: cleared by any failing exporter below
         # Snapshot before iterating (see ``shutdown``): a concurrent ``on_end`` mutating
         # the processor cache must not abort the flush and drop the remaining destinations'
         # buffered spans.
@@ -422,28 +438,28 @@ class TenantFanOutSpanProcessor(SpanProcessor):
         return all_ok
 
     def _processor_for(self, destination: OtelDestination) -> SpanProcessor | None:
-        key = _processor_key(destination)
-        cached = self._processors.get(key)
+        key: Final = _processor_key(destination)
+        cached: Final = self._processors.get(key)
         if cached is not None:
             self._processors.move_to_end(key)
             return cached
         from litellm.integrations.otel.plumbing.providers import (
-            _exporter_from_spec,
+            _exporter_from_spec,  # pyright: ignore[reportPrivateUsage]  # shared with providers.py inside the otel plumbing package
             default_otlp_kind_for_backend,
         )
         from litellm.integrations.otel.plumbing.providers import (
-            _processor_for as _build_processor,
+            _processor_for as _build_processor,  # pyright: ignore[reportPrivateUsage]  # shared with providers.py inside the otel plumbing package
         )
 
         try:
-            spec = ExporterSpec(
+            spec: Final = ExporterSpec(
                 kind=destination.protocol or default_otlp_kind_for_backend(destination.callback_name),
                 endpoint=destination.endpoint,
                 headers=destination.header_string(),
                 owner=None,
             )
-            exporter = _exporter_from_spec(spec)
-            processor = _build_processor(exporter, use_simple=False)
+            exporter: Final = _exporter_from_spec(spec)
+            processor: Final = _build_processor(exporter, use_simple=False)
         except Exception as exc:  # noqa: BLE001  # a malformed destination spec must not break fan-out; skip this destination
             verbose_logger.debug(
                 "OTel V2 fan-out: failed to build processor for %s: %s",

litellm/integrations/otel/presets/__init__.py

@@ -16,7 +16,7 @@ those per-request headers.
 """
 
 from collections.abc import Callable
-from typing import Final, TYPE_CHECKING
+from typing import TYPE_CHECKING, Final
 
 from litellm.integrations.otel.presets.agentops import agentops_preset
 from litellm.integrations.otel.presets.arize import arize_dynamic_headers, arize_preset
@@ -39,7 +39,7 @@ if TYPE_CHECKING:
 #: routing). Only integrations that support dynamic credentials appear here —
 #: Arize-Phoenix/Langtrace/Levo/AgentOps/generic don't, so they use the logger's
 #: default tracer.
-DYNAMIC_HEADERS_BY_CALLBACK: dict[str, Callable[[StandardCallbackDynamicParams], dict[str, str]]] = {
+DYNAMIC_HEADERS_BY_CALLBACK: Final[dict[str, Callable[[StandardCallbackDynamicParams], dict[str, str]]]] = {
     "arize": arize_dynamic_headers,
     "langfuse_otel": langfuse_dynamic_headers,
     "weave_otel": weave_dynamic_headers,
@@ -76,13 +76,15 @@ def dynamic_otlp_destination(
 
     if callback_name not in DYNAMIC_HEADERS_BY_CALLBACK or not dynamic_params:
         return None
-    values = {str(key): str(value) for key, value in dynamic_params.items() if isinstance(value, str)}
+    values: Final = {  # mutable-ok: handed to a model or SDK that needs a concrete dict
+        str(key): str(value) for key, value in dynamic_params.items() if isinstance(value, str)
+    }  # mutable-ok: handed to a model or SDK that needs a concrete dict
     return build_destination(callback_name or "", values)
 
 
 #: Callback name → preset. The ``Preset`` annotation makes mypy verify every
 #: registered value matches the preset interface.
-PRESET_BY_CALLBACK: dict[str, Preset] = {
+PRESET_BY_CALLBACK: Final[dict[str, Preset]] = {  # mutable-ok: module registry, read-only after import
     "agentops": agentops_preset,
     "arize": arize_preset,
     "arize_phoenix": phoenix_preset,

litellm/integrations/otel/presets/agentops.py

@@ -50,21 +50,28 @@ def agentops_preset(
     """
     settings: Final = _AgentOpsSettings()
     base: Final = config_overrides or OpenTelemetryV2Config()
-    global_exporter = (
+    global_exporter: Final = (
         ()
         if allow_missing_credentials and not settings.api_key
         else (
             ExporterSpec(
                 kind=_AGENTOPS_EXPORTER_KIND,
                 endpoint=_AGENTOPS_ENDPOINT,
-                options=({"api_key": settings.api_key} if settings.api_key else None),
+                options=(
+                    {"api_key": settings.api_key}
+                    if settings.api_key
+                    else None  # mutable-ok: handed to a model or SDK that needs a concrete dict
+                ),  # mutable-ok: the exporter spec carries a concrete options mapping
                 owner=ExporterOwner.AGENTOPS,
             ),
         )
     )
     return base.model_copy(
         update={
-            "exporters": [*base.exporters, *global_exporter],
+            "exporters": [
+                *base.exporters,
+                *global_exporter,
+            ],  # mutable-ok: pydantic model_copy takes a plain update mapping
             "resource_attributes": {
                 **base.resource_attributes,
                 "service.name": settings.service_name,

litellm/integrations/otel/presets/arize.py

@@ -14,7 +14,7 @@ from litellm.integrations.otel.model.config import (
 from litellm.integrations.otel.presets.utils import ensure_mappers
 from litellm.types.utils import StandardCallbackDynamicParams
 
-ARIZE_PUBLIC_OTLP_ENDPOINT = "https://otlp.arize.com/v1"
+ARIZE_PUBLIC_OTLP_ENDPOINT: Final = "https://otlp.arize.com/v1"
 
 
 class _ArizeSettings(BaseSettings):
@@ -33,11 +33,11 @@ def arize_preset(
     allow_missing_credentials: bool = False,
 ) -> OpenTelemetryV2Config:
     arize_cfg: Final = _V1ArizeLogger.get_arize_config()
-    settings = _ArizeSettings()
-    has_own_endpoint = bool(settings.grpc_endpoint or settings.http_endpoint)
-    has_credentials = bool(arize_cfg.space_id or arize_cfg.space_key or arize_cfg.api_key)
+    settings: Final = _ArizeSettings()
+    has_own_endpoint: Final = bool(settings.grpc_endpoint or settings.http_endpoint)
+    has_credentials: Final = bool(arize_cfg.space_id or arize_cfg.space_key or arize_cfg.api_key)
     base: Final = config_overrides or OpenTelemetryV2Config()
-    global_exporter = (
+    global_exporter: Final = (
         ()
         if allow_missing_credentials and not has_credentials and not has_own_endpoint
         else (
@@ -51,7 +51,10 @@ def arize_preset(
     )
     return base.model_copy(
         update={
-            "exporters": [*base.exporters, *global_exporter],
+            "exporters": [
+                *base.exporters,
+                *global_exporter,
+            ],  # mutable-ok: pydantic model_copy takes a plain update mapping
             "mapper_names": ensure_mappers(base.mapper_names, "openinference"),
             "resource_attributes": {
                 **base.resource_attributes,
@@ -77,9 +80,11 @@ def _arize_headers(arize_cfg, settings: "_ArizeSettings", has_own_endpoint: bool
 
 def arize_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]:
     """Per-request Arize OTLP headers from team/key dynamic params."""
-    headers: dict[str, str] = {}
+    headers: dict[  # rebind-ok: reassigned on the branch below
+        str, str
+    ] = {}  # rebind-ok: populated by the optional-credential branches below  # mutable-ok: construction is handed to an API that needs a concrete dict
     # ``arize_space_key`` is the suggested param and wins over ``arize_space_id``.
-    space = params.get("arize_space_key") or params.get("arize_space_id")
+    space: Final = params.get("arize_space_key") or params.get("arize_space_id")
     if space:
         headers["arize-space-id"] = space
     api_key: Final = params.get("arize_api_key")

litellm/integrations/otel/presets/destinations.py

@@ -20,7 +20,9 @@ from litellm.integrations.langfuse.langfuse_otel import (
     LangfuseOtelLogger,
 )
 from litellm.integrations.otel.model.destination import OtelDestination
-from litellm.integrations.weave.weave_otel import _get_weave_authorization_header
+from litellm.integrations.weave.weave_otel import (
+    _get_weave_authorization_header,  # pyright: ignore[reportPrivateUsage]  # reuse the backend's own header builder rather than duplicating its auth scheme
+)
 
 
 def _parse_header_string(raw: str) -> Mapping[str, str]:
@@ -40,7 +42,7 @@ def _langfuse_destination(values: Mapping[str, str]) -> OtelDestination | None:
         return None
     host: Final = values.get("langfuse_host")
     endpoint: Final = _langfuse_endpoint(host) if host else LANGFUSE_CLOUD_US_ENDPOINT
-    auth: Final = LangfuseOtelLogger._get_langfuse_authorization_header(public_key=public_key, secret_key=secret_key)
+    auth: Final = LangfuseOtelLogger._get_langfuse_authorization_header(public_key=public_key, secret_key=secret_key)  # pyright: ignore[reportPrivateUsage]  # reuse the backend's own header builder rather than duplicating its auth scheme
     return OtelDestination(endpoint=endpoint, headers=MappingProxyType({"Authorization": auth}))
 
 

litellm/integrations/otel/presets/langfuse.py

@@ -19,12 +19,12 @@ def langfuse_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str,
     public_key: Final = params.get("langfuse_public_key")
     secret_key: Final = params.get("langfuse_secret_key")
     if public_key and secret_key:
-        return {
+        return {  # mutable-ok: handed to a model or SDK that needs a concrete dict
             "Authorization": _V1Langfuse._get_langfuse_authorization_header(
                 public_key=public_key, secret_key=secret_key
             )
         }
-    return {}
+    return {}  # mutable-ok: handed to a model or SDK that needs a concrete dict
 
 
 def langfuse_preset(
@@ -33,13 +33,15 @@ def langfuse_preset(
     allow_missing_credentials: bool = False,
 ) -> OpenTelemetryV2Config:
     base: Final = config_overrides or OpenTelemetryV2Config()
-    mappers = ensure_mappers(base.mapper_names, "langfuse")
+    mappers: Final = ensure_mappers(base.mapper_names, "langfuse")
     try:
-        cfg = _V1Langfuse.get_langfuse_otel_config()
+        cfg: Final = _V1Langfuse.get_langfuse_otel_config()
     except Exception:
         if not allow_missing_credentials:
             raise
-        return base.model_copy(update={"mapper_names": mappers})
+        return base.model_copy(
+            update={"mapper_names": mappers}  # mutable-ok: handed to a model or SDK that needs a concrete dict
+        )  # mutable-ok: pydantic model_copy takes a plain update mapping
     kind: Final = cfg.exporter if isinstance(cfg.exporter, str) else "otlp_http"
     return base.model_copy(
         update={

litellm/integrations/otel/presets/levo.py

@@ -17,7 +17,7 @@ def levo_preset(
 ) -> OpenTelemetryV2Config:
     base: Final = config_overrides or OpenTelemetryV2Config()
     try:
-        cfg = _V1Levo.get_levo_config()
+        cfg: Final = _V1Levo.get_levo_config()
     except Exception:
         if not allow_missing_credentials:
             raise

litellm/integrations/otel/presets/phoenix.py

@@ -34,7 +34,7 @@ def phoenix_preset(
     headers: Final = cfg.otlp_auth_headers if hasattr(cfg, "otlp_auth_headers") else None
     project_name: Final = _PhoenixSettings().project_name
     base: Final = config_overrides or OpenTelemetryV2Config()
-    global_exporter = (
+    global_exporter: Final = (
         ExporterSpec(
             kind=cfg.protocol if hasattr(cfg, "protocol") else "otlp_http",
             endpoint=cfg.endpoint,
@@ -44,7 +44,10 @@ def phoenix_preset(
     )
     return base.model_copy(
         update={
-            "exporters": [*base.exporters, *global_exporter],
+            "exporters": [
+                *base.exporters,
+                *global_exporter,
+            ],  # mutable-ok: pydantic model_copy takes a plain update mapping
             "mapper_names": ensure_mappers(base.mapper_names, "openinference"),
             "resource_attributes": {
                 **base.resource_attributes,

litellm/integrations/otel/presets/weave.py

@@ -17,7 +17,9 @@ from litellm.types.utils import StandardCallbackDynamicParams
 
 def weave_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str, str]:
     """Per-request Weave OTLP headers from team/key dynamic params."""
-    headers: dict[str, str] = {}
+    headers: dict[  # rebind-ok: reassigned on the branch below
+        str, str
+    ] = {}  # rebind-ok: populated by the optional-credential branches below  # mutable-ok: construction is handed to an API that needs a concrete dict
     api_key: Final = params.get("wandb_api_key")
     if api_key:
         headers["Authorization"] = _get_weave_authorization_header(api_key=api_key)
@@ -33,13 +35,15 @@ def weave_preset(
     allow_missing_credentials: bool = False,
 ) -> OpenTelemetryV2Config:
     base: Final = config_overrides or OpenTelemetryV2Config()
-    mappers = ensure_mappers(base.mapper_names, "openinference", "weave")
+    mappers: Final = ensure_mappers(base.mapper_names, "openinference", "weave")
     try:
-        weave_cfg = get_weave_otel_config()
+        weave_cfg: Final = get_weave_otel_config()
     except Exception:
         if not allow_missing_credentials:
             raise
-        return base.model_copy(update={"mapper_names": mappers})
+        return base.model_copy(
+            update={"mapper_names": mappers}  # mutable-ok: handed to a model or SDK that needs a concrete dict
+        )  # mutable-ok: pydantic model_copy takes a plain update mapping
     return base.model_copy(
         update={
             "exporters": [

litellm/litellm_core_utils/litellm_logging.py

@@ -4025,11 +4025,16 @@ def _init_custom_logger_compatible_class(
                 for callback in _in_memory_loggers:
                     if type(callback) is OpenTelemetryV2:
                         return callback
-                settings = _get_custom_logger_settings_from_proxy_server(callback_name=logging_integration)
-                config = OpenTelemetryV2Config(**settings)
+                settings: Final = _get_custom_logger_settings_from_proxy_server(callback_name=logging_integration)
+                config = OpenTelemetryV2Config(**settings)  # rebind-ok: refined below when no exporter resolved
                 if not config.exporters:
-                    config = OpenTelemetryV2Config(**{**settings, "exporter": config.exporter})
-                otel_logger_v2 = OpenTelemetryV2(config=config)
+                    config = OpenTelemetryV2Config(  # rebind-ok: reassigned on the branch below
+                        **{
+                            **settings,
+                            "exporter": config.exporter,
+                        }  # mutable-ok: handed to a model or SDK that needs a concrete dict
+                    )  # mutable-ok: handed to a model or SDK that needs a concrete dict
+                otel_logger_v2: Final = OpenTelemetryV2(config=config)
                 _in_memory_loggers.append(otel_logger_v2)
                 _maybe_auto_initialize_arize_phoenix(_in_memory_loggers)
                 return otel_logger_v2
@@ -4393,7 +4398,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom
         if isinstance(callback, OpenTelemetryV2) and getattr(callback, "callback_name", None) == callback_name:
             return callback
     try:
-        config = preset_fn(allow_missing_credentials=False)
+        config: Final = preset_fn(allow_missing_credentials=False)
     except Exception:
         return None
     if not config.exporters:
@@ -4407,7 +4412,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom
             "register a logging destination for it.",
             callback_name,
         )
-    v2_logger = OpenTelemetryV2(config=config, callback_name=callback_name)
+    v2_logger: Final = OpenTelemetryV2(config=config, callback_name=callback_name)
     _in_memory_loggers.append(v2_logger)
     return v2_logger
 
@@ -4430,7 +4435,7 @@ def otel_v2_owned_backends() -> frozenset[str]:
     """
     from litellm.integrations.otel.logger import OpenTelemetryV2
 
-    dispatched = litellm._async_success_callback
+    dispatched: Final = litellm._async_success_callback
     return frozenset(
         name
         for logger in _in_memory_loggers

litellm/proxy/_experimental/mcp_server/server.py

@@ -4790,7 +4790,9 @@ if MCP_AVAILABLE:
 
         return stored
 
-    async def _refresh_request_otel_destinations(user_api_key_auth: Any) -> None:
+    async def _refresh_request_otel_destinations(
+        user_api_key_auth: object,
+    ) -> None:
         """Re-resolve the caller's admin destinations for THIS JSON-RPC message.
 
         A stateful MCP session dispatches every later message on descendants of the task
@@ -4803,7 +4805,9 @@ if MCP_AVAILABLE:
         if user_api_key_auth is None:
             return
         try:
-            from litellm.proxy.litellm_pre_call_utils import _apply_admin_logging_exporters
+            from litellm.proxy.litellm_pre_call_utils import (
+                _apply_admin_logging_exporters,  # pyright: ignore[reportPrivateUsage]  # the MCP path applies the same pre-call resolution as the HTTP path
+            )
 
             await _apply_admin_logging_exporters(user_api_key_auth)
         except Exception as exc:  # noqa: BLE001  # a resolver failure must not break the MCP call

litellm/proxy/auth/user_api_key_auth.py

@@ -1035,7 +1035,7 @@ async def _hoist_request_destinations(request: Request, user_api_key_dict: UserA
             set_request_destinations,
         )
         from litellm.proxy.litellm_pre_call_utils import (
-            _resolve_logging_exporters,
+            _resolve_logging_exporters,  # pyright: ignore[reportPrivateUsage]  # resolver lives in the pre-call module it is hoisted from
         )
 
         destinations_raw, _backends = await _resolve_logging_exporters(user_api_key_dict)

tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py

@@ -329,3 +329,35 @@ async def test_disclosure_agrees_with_the_resolver_on_a_shared_target(monkeypatc
 
     assert len(destinations) == 1
     assert resolved_logging_exporter_names(None, None) == ("dup-one", "dup-two")
+
+
+
+def test_backend_without_a_preset_routes_under_generic():
+    """Regression: a backend outside ``PRESET_BY_CALLBACK`` is routed as ``generic``.
+
+    Only a registered name gets an ``OpenTelemetryV2`` logger, and that logger is what
+    emits the gen-AI span to its destinations. Routing an unregistered name under itself
+    delivered the proxy-internal spans but never the LLM call. Registered names keep
+    their own routing so each backend's attribute vocabulary is preserved."""
+    from litellm.integrations.otel.presets import PRESET_BY_CALLBACK
+
+    unknown = CredentialItem(
+        credential_name="self-hosted",
+        credential_values={"otel_endpoint": "http://collector.example/v1/traces"},
+        credential_info={"credential_type": "logging", "description": "honeycomb", "access": {"global": True}},
+    )
+    resolved = destination_for_credential(unknown)
+    assert resolved is not None
+    assert resolved[0] == "generic"
+    assert resolved[1].endpoint == "http://collector.example/v1/traces"
+
+    for registered in ("arize", "langfuse_otel", "generic"):
+        assert registered in PRESET_BY_CALLBACK
+    known = CredentialItem(
+        credential_name="lf",
+        credential_values={"langfuse_public_key": "pk", "langfuse_secret_key": "sk"},
+        credential_info={"credential_type": "logging", "description": "langfuse_otel", "access": {"global": True}},
+    )
+    known_resolved = destination_for_credential(known)
+    assert known_resolved is not None
+    assert known_resolved[0] == "langfuse_otel"

tests/test_litellm/proxy/test_litellm_pre_call_utils.py

@@ -6917,3 +6917,25 @@ async def test_apply_admin_logging_exporters_degrades_when_flag_on_without_opent
         env={**os.environ, "PYTHONPATH": repo_root, "LITELLM_OTEL_V2": "true"},
     )
     assert "REQUEST_PATH_OK" in result.stdout, f"flag-on request path raised:\n{result.stderr[-3000:]}"
+
+
+
+@pytest.mark.asyncio
+async def test_apply_admin_logging_exporters_registers_on_failure(_seeded_logging_credentials, monkeypatch):
+    """An admin-owned destination must capture a FAILED upstream call, not only a
+    successful one.
+
+    The destination sink is one process-wide logger, so it has to sit on the failure
+    list as well as the success list; registering it on success alone means a
+    401/timeout never reaches the destination and the trace lands with no error
+    gen-AI span. Registration is idempotent.
+    """
+    import litellm
+    from litellm.integrations.otel.destination_logger import admin_destination_logger
+    from litellm.integrations.otel.logger import publish_global_otel_v2_provider
+
+    sink = admin_destination_logger()
+    publish_global_otel_v2_provider([], lambda provider: None)
+    publish_global_otel_v2_provider([], lambda provider: None)
+    for bucket in (litellm._async_success_callback, litellm._async_failure_callback):
+        assert sum(1 for callback in bucket if callback is sink) == 1