scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-29 01:42:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 79

Merge origin/litellm_internal_staging into litellm_feat/v1.84.0-mcp-gateway-jwt-auth

Browse source
This commit is contained in:
mateo-berri 2026-05-26 21:10:42 +00:00
commit adc4a838bd
No known key found for this signature in database
33 changed files with 2510 additions and 344 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
deploy/charts/litellm-helm/templates/deployment.yaml
View file

@ -30,7 +30,7 @@ spec:
checksum/config: {{ include (print $.Template.BasePath "/configmap-litellm.yaml") . | sha256sum }}
{{- end }}
{{- with .Values.podAnnotations }}
{{- toYaml . | nindent 8 }}
{{- tpl (toYaml .) $ | nindent 8 }}
{{- end }}
labels:
{{- include "litellm.labels" . | nindent 8 }}

25
deploy/charts/litellm-helm/tests/deployment_tests.yaml
View file

@ -377,3 +377,28 @@ tests:
content:
name: sidecar-tpl
image: "ghcr.io/berriai/litellm-database:test"
- it: should support tpl in podAnnotations
template: deployment.yaml
set:
image:
repository: ghcr.io/berriai/litellm-database
tag: test
# Mirrors the real-world scenario this feature unblocks:
# user disables the built-in ConfigMap (and its built-in checksum/config
# annotation) and re-implements checksum/config themselves via tpl.
proxyConfigMap:
create: false
podAnnotations:
checksum/config: "{{ .Values.image.tag }}"
example.com/some-key: "{{ .Values.image.repository }}"
example.com/literal: "plain-string-value"
asserts:
- equal:
path: spec.template.metadata.annotations["checksum/config"]
value: "test"
- equal:
path: spec.template.metadata.annotations["example.com/some-key"]
value: "ghcr.io/berriai/litellm-database"
- equal:
path: spec.template.metadata.annotations["example.com/literal"]
value: "plain-string-value"

34
litellm/cost_calculator.py
View file

@ -417,9 +417,36 @@ def cost_per_token( # noqa: PLR0915
prompt_tokens_cost_usd_dollar: float = 0
completion_tokens_cost_usd_dollar: float = 0
model_cost_ref = litellm.model_cost
# Only callers that explicitly pass `custom_llm_provider` get the
# dedup/prefix-join treatment. When provider is omitted, preserve legacy
# behavior: `model_with_provider` stays equal to the raw `model` string
# (provider is detected below for downstream use only).
caller_supplied_provider = custom_llm_provider is not None
# `model` is normally a string, but callers that mock the transport can pass
# non-string objects. Only run the string-based dedup/prefix-join when it is
# actually a string — e.g. a MagicMock's `.startswith()` is always truthy and
# its slices return new mocks, which would spin the dedup loop forever.
model_is_str = isinstance(model, str)
# Router/proxy deployments may repeat the provider segment (e.g. model_name
# "openai/openai/gpt-5.5"). Strip duplicated `{provider}/` chains before joining.
if caller_supplied_provider and model_is_str:
_dup_prefix = f"{custom_llm_provider}/"
while model.startswith(_dup_prefix):
_remainder = model[len(_dup_prefix) :]
if _remainder.startswith(_dup_prefix):
model = _remainder
else:
break
model_with_provider = model
if custom_llm_provider is not None:
model_with_provider = custom_llm_provider + "/" + model
if caller_supplied_provider:
_prov_prefix = f"{custom_llm_provider}/"
if model_is_str and model.startswith(_prov_prefix):
model_with_provider = model
else:
model_with_provider = f"{custom_llm_provider}/{model}"
if region_name is not None:
model_with_provider_and_region = (
f"{custom_llm_provider}/{region_name}/{model}"
@ -430,6 +457,9 @@ def cost_per_token( # noqa: PLR0915
model_with_provider = model_with_provider_and_region
else:
_, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model)
assert custom_llm_provider is not None # caller-supplied or get_llm_provider
model_without_prefix = model
model_parts = model.split("/", 1)
if len(model_parts) > 1:

397
litellm/integrations/arize/arize_phoenix.py
View file

@ -1,5 +1,7 @@
import os
from typing import TYPE_CHECKING, Any, Optional, Union
import threading
from collections import OrderedDict
from typing import TYPE_CHECKING, Any, Optional, Tuple, Union
from litellm._logging import verbose_logger
from litellm.integrations.arize import _utils
@ -8,8 +10,10 @@ from litellm.types.integrations.arize_phoenix import ArizePhoenixConfig
if TYPE_CHECKING:
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SpanProcessor
from opentelemetry.trace import Span as _Span
from opentelemetry.trace import SpanKind
from opentelemetry.trace import Tracer
from litellm.integrations.opentelemetry import OpenTelemetry as _OpenTelemetry
from litellm.integrations.opentelemetry import (
@ -21,20 +25,27 @@ if TYPE_CHECKING:
OpenTelemetryConfig = _OpenTelemetryConfig
Span = Union[_Span, Any]
OpenTelemetry = _OpenTelemetry
LITELLM_TRACER_NAME: str
else:
Protocol = Any
OpenTelemetryConfig = Any
Span = Any
Tracer = Any
TracerProvider = Any
SpanKind = Any
# Import OpenTelemetry at runtime
SpanProcessor = Any
try:
from litellm.integrations.opentelemetry import OpenTelemetry
from litellm.integrations.opentelemetry import (
LITELLM_TRACER_NAME,
OpenTelemetry,
)
except ImportError:
LITELLM_TRACER_NAME = "litellm"
OpenTelemetry = None # type: ignore
ARIZE_HOSTED_PHOENIX_ENDPOINT = "https://otlp.arize.com/v1/traces"
_MAX_PROJECT_PROVIDERS = 64
class ArizePhoenixLogger(OpenTelemetry): # type: ignore
@ -48,37 +59,142 @@ class ArizePhoenixLogger(OpenTelemetry): # type: ignore
def _init_tracing(self, tracer_provider):
"""
Override to always create a *private* TracerProvider for Arize Phoenix.
Override to create per-project TracerProviders (LRU-cached) for Arize Phoenix.
The base ``OpenTelemetry._init_tracing`` falls back to the global
TracerProvider when one already exists. That causes whichever
integration initialises second to silently reuse the first one's
exporter, so spans only reach one destination.
By creating our own provider we guarantee Arize Phoenix always gets
its own exporter pipeline, regardless of initialisation order.
"""
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.trace import SpanKind
if tracer_provider is not None:
# Explicitly supplied (e.g. in tests) — honour it.
self.tracer = tracer_provider.get_tracer("litellm")
self._use_injected_tracer_provider = True
self._shared_span_processor = None
self.tracer = tracer_provider.get_tracer(LITELLM_TRACER_NAME)
self.span_kind = SpanKind
return
# Always create a dedicated provider — never touch the global one.
provider = TracerProvider(resource=self._get_litellm_resource(self.config))
provider.add_span_processor(self._get_span_processor())
self.tracer = provider.get_tracer("litellm")
self._use_injected_tracer_provider = False
self._project_providers: OrderedDict[str, TracerProvider] = OrderedDict()
self._project_providers_lock = threading.Lock()
self._shared_span_processor = self._get_span_processor()
self.span_kind = SpanKind
default_project = self._resolve_project_name({})
self.tracer = self._get_tracer_for(default_project)
verbose_logger.debug(
"ArizePhoenixLogger: Created dedicated TracerProvider "
"(endpoint=%s, exporter=%s)",
"ArizePhoenixLogger: Initialized per-project TracerProvider cache "
"(default_project=%s, endpoint=%s, exporter=%s)",
default_project,
self.config.endpoint,
self.config.exporter,
)
def flush_tracer_providers(self) -> None:
"""
Flush all cached per-project providers and the shared span processor.
Call on graceful proxy shutdown. Do not call on LRU eviction — in-flight
spans may still reference evicted providers.
"""
if getattr(self, "_use_injected_tracer_provider", False):
return
shared_processor = getattr(self, "_shared_span_processor", None)
if shared_processor is not None:
try:
shared_processor.force_flush()
except Exception as e:
verbose_logger.debug(
"ArizePhoenixLogger: shared span processor force_flush failed: %s",
e,
)
with getattr(self, "_project_providers_lock", threading.Lock()):
providers = list(getattr(self, "_project_providers", {}).values())
for provider in providers:
try:
provider.force_flush()
except Exception as e:
verbose_logger.debug(
"ArizePhoenixLogger: TracerProvider force_flush failed: %s", e
)
def _get_litellm_resource_for_project(self, project_name: str):
"""
Build an OTEL Resource with project routing attrs that win over env detector.
Phoenix uses ``openinference.project.name``; Arize AX uses ``model_id`` and
``service.name``. Project attrs are merged last so OTEL_RESOURCE_ATTRIBUTES
from init does not pin every provider to one project.
"""
from opentelemetry.sdk.resources import OTELResourceDetector, Resource
project_attributes: dict[str, str] = {
"openinference.project.name": project_name,
"model_id": project_name,
"service.name": project_name,
}
deployment_environment = getattr(self.config, "deployment_environment", None)
if deployment_environment is not None:
project_attributes["deployment.environment"] = deployment_environment
env_resource = OTELResourceDetector().detect()
project_resource = Resource.create(project_attributes) # type: ignore[arg-type]
return env_resource.merge(project_resource)
def _build_tracer_provider_for_project(self, project_name: str) -> TracerProvider:
"""Create a TracerProvider for *project_name* (caller holds no cache lock)."""
from opentelemetry.sdk.trace import TracerProvider
provider = TracerProvider(
resource=self._get_litellm_resource_for_project(project_name)
)
provider.add_span_processor(self._shared_span_processor)
return provider
def _get_tracer_for(self, project_name: str) -> Tracer:
"""Return a tracer for *project_name*, creating/caching a provider on miss."""
if getattr(self, "_use_injected_tracer_provider", False):
return self.tracer
with self._project_providers_lock:
if project_name in self._project_providers:
self._project_providers.move_to_end(project_name)
return self._project_providers[project_name].get_tracer(
LITELLM_TRACER_NAME
)
# OTELResourceDetector().detect() is synchronous; build outside the lock so
# concurrent requests for other projects are not blocked on cache misses.
new_provider = self._build_tracer_provider_for_project(project_name)
with self._project_providers_lock:
if project_name in self._project_providers:
self._project_providers.move_to_end(project_name)
return self._project_providers[project_name].get_tracer(
LITELLM_TRACER_NAME
)
if len(self._project_providers) >= _MAX_PROJECT_PROVIDERS:
self._project_providers.popitem(last=False)
self._project_providers[project_name] = new_provider
return new_provider.get_tracer(LITELLM_TRACER_NAME)
def _resolve_tracer_for_kwargs(self, kwargs: dict) -> Tuple[str, Tracer]:
"""Resolve project name once and return the matching tracer."""
project_name = self._resolve_project_name(kwargs)
return project_name, self._get_tracer_for(project_name)
def get_tracer_to_use_for_request(self, kwargs: dict) -> Tracer:
"""Route guardrail/raw-request spans to the same per-project tracer as the request."""
if getattr(self, "_use_injected_tracer_provider", False):
return self.tracer
return self._resolve_tracer_for_kwargs(kwargs)[1]
def _init_otel_logger_on_litellm_proxy(self):
"""
Override: Arize Phoenix should NOT overwrite the proxy's
@ -93,56 +209,109 @@ class ArizePhoenixLogger(OpenTelemetry): # type: ignore
@staticmethod
def set_arize_phoenix_attributes(span: Span, kwargs, response_obj):
from litellm.integrations.opentelemetry_utils.base_otel_llm_obs_attributes import (
safe_set_attribute,
)
_utils.set_attributes(span, kwargs, response_obj, ArizeOTELAttributes)
# Dynamic project name: check metadata first, then fall back to env var config
dynamic_project_name = ArizePhoenixLogger._get_dynamic_project_name(kwargs)
if dynamic_project_name:
safe_set_attribute(span, "openinference.project.name", dynamic_project_name)
else:
# Fall back to static config from env var
config = ArizePhoenixLogger.get_arize_phoenix_config()
if config.project_name:
safe_set_attribute(
span, "openinference.project.name", config.project_name
)
return
@staticmethod
def _get_dynamic_project_name(kwargs) -> Optional[str]:
"""
Retrieve dynamic Phoenix project name from request metadata.
def _normalize_project_name(name: Optional[str]) -> Optional[str]:
if name is None:
return None
normalized = str(name).strip()
return normalized if normalized else None
Users can set `metadata.phoenix_project_name` in their request to route
traces to different Phoenix projects dynamically.
"""
standard_logging_payload = kwargs.get("standard_logging_object")
if isinstance(standard_logging_payload, dict):
metadata = standard_logging_payload.get("metadata")
@staticmethod
def _iter_metadata_dicts_from_kwargs(kwargs: dict):
"""Yield request metadata dicts; standard_logging_object before litellm_params."""
for key in ("standard_logging_object", "litellm_params"):
found_key = kwargs.get(key)
if not isinstance(found_key, dict):
continue
metadata = found_key.get("metadata")
if isinstance(metadata, dict):
project_name = metadata.get("phoenix_project_name")
if project_name:
return str(project_name)
yield metadata
# Also check litellm_params.metadata for SDK usage
@staticmethod
def _is_proxy_request(kwargs: dict) -> bool:
"""True when the call is routed through the LiteLLM proxy.
Proxy mode is determined solely by the server-set ``proxy_server_request``
field in ``litellm_params``. Checking request metadata for
``user_api_key_auth_metadata`` is intentionally avoided: that field is
user-supplied and would let an authenticated caller fake proxy-mode
detection to route their telemetry into arbitrary Arize/Phoenix projects.
"""
litellm_params = kwargs.get("litellm_params")
if isinstance(litellm_params, dict):
metadata = litellm_params.get("metadata") or {}
else:
metadata = {}
if isinstance(metadata, dict):
project_name = metadata.get("phoenix_project_name")
if project_name:
return str(project_name)
return isinstance(litellm_params, dict) and bool(
litellm_params.get("proxy_server_request")
)
@staticmethod
def _project_from_metadata_dict(
metadata: dict, metadata_key: str, *, proxy_mode: bool
) -> Optional[str]:
"""
Read a Phoenix project field from proxy/SDK metadata.
On the proxy, only ``user_api_key_auth_metadata`` (team/key config) may
select the project. SDK callers may still set project fields directly on
``metadata``.
"""
auth_metadata = metadata.get("user_api_key_auth_metadata")
if isinstance(auth_metadata, dict):
project = ArizePhoenixLogger._normalize_project_name(
auth_metadata.get(metadata_key)
)
if project:
return project
if not proxy_mode:
return ArizePhoenixLogger._normalize_project_name(
metadata.get(metadata_key)
)
return None
def _get_phoenix_context(self, kwargs):
@staticmethod
def _metadata_project_from_kwargs(kwargs: dict, metadata_key: str) -> Optional[str]:
proxy_mode = ArizePhoenixLogger._is_proxy_request(kwargs)
for metadata in ArizePhoenixLogger._iter_metadata_dicts_from_kwargs(kwargs):
project = ArizePhoenixLogger._project_from_metadata_dict(
metadata, metadata_key, proxy_mode=proxy_mode
)
if project:
return project
return None
@staticmethod
def _resolve_project_name(kwargs: dict) -> str:
"""
Resolve the target Phoenix/Arize project for this request.
Proxy priority: ``user_api_key_auth_metadata.phoenix_project_name_override``,
``user_api_key_auth_metadata.phoenix_project_name``, env, then ``default``.
SDK priority: request metadata fields, then env, then ``default``.
"""
override = ArizePhoenixLogger._metadata_project_from_kwargs(
kwargs, "phoenix_project_name_override"
)
if override:
return override
phoenix_name = ArizePhoenixLogger._metadata_project_from_kwargs(
kwargs, "phoenix_project_name"
)
if phoenix_name:
return phoenix_name
env_name = ArizePhoenixLogger._normalize_project_name(
os.environ.get("PHOENIX_PROJECT_NAME")
or os.environ.get("ARIZE_PROJECT_NAME")
)
if env_name:
return env_name
return "default"
def _get_phoenix_context(self, kwargs, tracer: Optional[Tracer] = None):
"""
Build a trace context for Phoenix's dedicated TracerProvider.
@ -159,11 +328,13 @@ class ArizePhoenixLogger(OpenTelemetry): # type: ignore
"""
from opentelemetry import trace
if tracer is None:
tracer = self._resolve_tracer_for_kwargs(kwargs)[1]
litellm_params = kwargs.get("litellm_params", {}) or {}
proxy_server_request = litellm_params.get("proxy_server_request", {}) or {}
headers = proxy_server_request.get("headers", {}) or {}
# Propagate distributed trace context if the caller sent a traceparent
traceparent_ctx = (
self.get_traceparent_from_header(headers=headers)
if headers.get("traceparent")
@ -173,10 +344,8 @@ class ArizePhoenixLogger(OpenTelemetry): # type: ignore
is_proxy_mode = bool(proxy_server_request)
if is_proxy_mode:
# Create a parent span on Phoenix's own tracer so both parent
# and child are exported to Phoenix.
start_time_val = kwargs.get("start_time", kwargs.get("api_call_start_time"))
parent_span = self.tracer.start_span(
parent_span = tracer.start_span(
name="litellm_proxy_request",
start_time=(
self._to_ns(start_time_val) if start_time_val is not None else None
@ -187,100 +356,77 @@ class ArizePhoenixLogger(OpenTelemetry): # type: ignore
ctx = trace.set_span_in_context(parent_span)
return ctx, parent_span
# SDK mode — no parent span needed
return traceparent_ctx, None
def _handle_success(self, kwargs, response_obj, start_time, end_time):
"""
Override to always create spans on ArizePhoenixLogger's dedicated TracerProvider.
The base class's ``_get_span_context`` would find the parent span created by
the ``otel`` callback on the *global* TracerProvider. That span is invisible
in Phoenix (different exporter pipeline), so we ignore it and build our own
hierarchy via ``_get_phoenix_context``.
"""
from opentelemetry.trace import Status, StatusCode
verbose_logger.debug(
"ArizePhoenixLogger: Logging kwargs: %s, OTEL config settings=%s",
kwargs,
self.config,
self._handle_phoenix_trace(
kwargs, response_obj, start_time, end_time, success=True
)
ctx, parent_span = self._get_phoenix_context(kwargs)
# Create litellm_request span (child of our parent when in proxy mode)
span = self.tracer.start_span(
name=self._get_span_name(kwargs),
start_time=self._to_ns(start_time),
context=ctx,
)
span.set_status(Status(StatusCode.OK))
self.set_attributes(span, kwargs, response_obj)
# Raw-request sub-span (if enabled) — must be created before
# ending the parent span so the hierarchy is valid.
self._maybe_log_raw_request(kwargs, response_obj, start_time, end_time, span)
span.end(end_time=self._to_ns(end_time))
# Guardrail span
self._create_guardrail_span(kwargs=kwargs, context=ctx)
# Annotate and close our proxy parent span
if parent_span is not None:
parent_span.set_status(Status(StatusCode.OK))
self.set_attributes(parent_span, kwargs, response_obj)
parent_span.end(end_time=self._to_ns(end_time))
# Metrics & cost recording
self._record_metrics(kwargs, response_obj, start_time, end_time)
# Semantic logs
if self.config.enable_events:
self._emit_semantic_logs(kwargs, response_obj, span)
def _handle_failure(self, kwargs, response_obj, start_time, end_time):
"""
Override to always create failure spans on ArizePhoenixLogger's dedicated
TracerProvider. Mirrors ``_handle_success`` but sets ERROR status.
"""
self._handle_phoenix_trace(
kwargs, response_obj, start_time, end_time, success=False
)
def _handle_phoenix_trace(
self,
kwargs,
response_obj,
start_time,
end_time,
*,
success: bool,
):
from opentelemetry.trace import Status, StatusCode
verbose_logger.debug(
"ArizePhoenixLogger: Failure - Logging kwargs: %s, OTEL config settings=%s",
"ArizePhoenixLogger: %s - kwargs: %s, OTEL config settings=%s",
"success" if success else "failure",
kwargs,
self.config,
)
ctx, parent_span = self._get_phoenix_context(kwargs)
_project_name, tracer = self._resolve_tracer_for_kwargs(kwargs)
ctx, parent_span = self._get_phoenix_context(kwargs, tracer=tracer)
# Create litellm_request span (child of our parent when in proxy mode)
span = self.tracer.start_span(
status = Status(StatusCode.OK if success else StatusCode.ERROR)
span = tracer.start_span(
name=self._get_span_name(kwargs),
start_time=self._to_ns(start_time),
context=ctx,
)
span.set_status(Status(StatusCode.ERROR))
span.set_status(status)
self.set_attributes(span, kwargs, response_obj)
self._record_exception_on_span(span=span, kwargs=kwargs)
if not success:
self._record_exception_on_span(span=span, kwargs=kwargs)
if success:
self._maybe_log_raw_request(
kwargs, response_obj, start_time, end_time, span
)
span.end(end_time=self._to_ns(end_time))
# Guardrail span
self._create_guardrail_span(kwargs=kwargs, context=ctx)
# Annotate and close our proxy parent span
if parent_span is not None:
parent_span.set_status(Status(StatusCode.ERROR))
parent_span.set_status(status)
self.set_attributes(parent_span, kwargs, response_obj)
self._record_exception_on_span(span=parent_span, kwargs=kwargs)
if not success:
self._record_exception_on_span(span=parent_span, kwargs=kwargs)
parent_span.end(end_time=self._to_ns(end_time))
if success:
self._record_metrics(kwargs, response_obj, start_time, end_time)
if self.config.enable_events:
self._emit_semantic_logs(kwargs, response_obj, span)
@staticmethod
def get_arize_phoenix_config() -> ArizePhoenixConfig:
"""
Retrieves the Arize Phoenix configuration based on environment variables.
Returns:
ArizePhoenixConfig: A Pydantic model containing Arize Phoenix configuration.
"""
api_key = os.environ.get("PHOENIX_API_KEY", None)
@ -295,18 +441,15 @@ class ArizePhoenixLogger(OpenTelemetry): # type: ignore
protocol: Protocol = "otlp_http"
if collector_endpoint:
# Parse the endpoint to determine protocol
if collector_endpoint.startswith("grpc://") or (
":4317" in collector_endpoint and "/v1/traces" not in collector_endpoint
):
endpoint = collector_endpoint
protocol = "otlp_grpc"
else:
# Phoenix Cloud endpoints (app.phoenix.arize.com) include the space in the URL
if "app.phoenix.arize.com" in collector_endpoint:
endpoint = collector_endpoint
protocol = "otlp_http"
# For other HTTP endpoints, ensure they have the correct path
elif "/v1/traces" not in collector_endpoint:
if collector_endpoint.endswith("/v1"):
endpoint = collector_endpoint + "/traces"
@ -318,7 +461,6 @@ class ArizePhoenixLogger(OpenTelemetry): # type: ignore
endpoint = collector_endpoint
protocol = "otlp_http"
else:
# If no endpoint specified, self hosted phoenix
endpoint = "http://localhost:6006/v1/traces"
protocol = "otlp_http"
verbose_logger.debug(
@ -329,12 +471,11 @@ class ArizePhoenixLogger(OpenTelemetry): # type: ignore
if api_key is not None:
otlp_auth_headers = f"Authorization=Bearer {api_key}"
elif "app.phoenix.arize.com" in endpoint:
# Phoenix Cloud requires an API key
raise ValueError(
"PHOENIX_API_KEY must be set when using Phoenix Cloud (app.phoenix.arize.com)."
)
project_name = os.environ.get("PHOENIX_PROJECT_NAME", "default")
project_name = os.environ.get("PHOENIX_PROJECT_NAME") or "default"
return ArizePhoenixConfig(
otlp_auth_headers=otlp_auth_headers,
@ -343,8 +484,6 @@ class ArizePhoenixLogger(OpenTelemetry): # type: ignore
project_name=project_name,
)
## cannot suppress additional proxy server spans, removed previous methods.
async def async_health_check(self):
config = self.get_arize_phoenix_config()

21
litellm/integrations/datadog/datadog_metrics.py
View file

@ -144,7 +144,26 @@ class DatadogMetricsLogger(CustomBatchLogger):
}
self.log_queue.append(series_llm_latency)
# 3. Request Count / Status Code
# 3. LiteLLM Overhead Latency Metric (total - llm_api time)
hidden_params = log.get("hidden_params", {}) or {}
litellm_overhead_time_ms = hidden_params.get("litellm_overhead_time_ms")
if litellm_overhead_time_ms is not None:
overhead_tags = self._extract_tags(log) # no status_code on latency metric
series_overhead: DatadogMetricSeries = {
"metric": "litellm.overhead.latency",
"type": 3, # gauge
"points": [
{
"timestamp": timestamp,
"value": litellm_overhead_time_ms
/ 1000, # convert ms → seconds
}
],
"tags": overhead_tags,
}
self.log_queue.append(series_overhead)
# 4. Request Count / Status Code
series_count: DatadogMetricSeries = {
"metric": "litellm.llm_api.request_count",
"type": 1, # count

25
litellm/litellm_core_utils/litellm_logging.py
View file

@ -3910,31 +3910,6 @@ def _init_custom_logger_compatible_class( # noqa: PLR0915
endpoint=arize_phoenix_config.endpoint,
headers=arize_phoenix_config.otlp_auth_headers,
)
if arize_phoenix_config.project_name:
existing_attrs = os.environ.get("OTEL_RESOURCE_ATTRIBUTES", "")
# Add openinference.project.name attribute
if existing_attrs:
os.environ["OTEL_RESOURCE_ATTRIBUTES"] = (
f"{existing_attrs},openinference.project.name={arize_phoenix_config.project_name}"
)
else:
os.environ["OTEL_RESOURCE_ATTRIBUTES"] = (
f"openinference.project.name={arize_phoenix_config.project_name}"
)
# Set Phoenix project name from environment variable
phoenix_project_name = os.environ.get("PHOENIX_PROJECT_NAME", None)
if phoenix_project_name:
existing_attrs = os.environ.get("OTEL_RESOURCE_ATTRIBUTES", "")
# Add openinference.project.name attribute
if existing_attrs:
os.environ["OTEL_RESOURCE_ATTRIBUTES"] = (
f"{existing_attrs},openinference.project.name={phoenix_project_name}"
)
else:
os.environ["OTEL_RESOURCE_ATTRIBUTES"] = (
f"openinference.project.name={phoenix_project_name}"
)
# auth can be disabled on local deployments of arize phoenix
if arize_phoenix_config.otlp_auth_headers is not None:

5
litellm/litellm_core_utils/llm_cost_calc/utils.py
View file

@ -30,6 +30,9 @@ _IMAGE_RESPONSE_CALL_TYPES = frozenset(
}
)
# Pre-resolved DataResidency enum values for fast membership checks
_VALID_DATA_RESIDENCIES = frozenset(r.value for r in DataResidency)
def _is_above_128k(tokens: float) -> bool:
if tokens > 128000:
@ -636,7 +639,7 @@ def _get_regional_uplift_multiplier(
if data_residency is None:
return 1.0
residency = data_residency.lower()
if residency not in {r.value for r in DataResidency}:
if residency not in _VALID_DATA_RESIDENCIES:
return 1.0
multiplier = model_info.get(f"regional_processing_uplift_multiplier_{residency}")
if multiplier is None:

14
litellm/litellm_core_utils/prompt_templates/factory.py
View file

@ -3997,7 +3997,7 @@ def _convert_to_bedrock_tool_call_invoke(
for tool in tool_calls:
if "function" in tool:
tool_id = tool["id"]
name = tool["function"].get("name", "")
name = make_valid_bedrock_tool_name(tool["function"].get("name", ""))
arguments = tool["function"].get("arguments", "")
if not arguments or not arguments.strip():
@ -5323,16 +5323,10 @@ def _bedrock_converse_messages_pt( # noqa: PLR0915
def make_valid_bedrock_tool_name(input_tool_name: str) -> str:
"""
Replaces any invalid characters in the input tool name with underscores
and ensures the resulting string is a valid identifier for Bedrock tools
"""
"""Normalize tool names to Bedrock pattern [a-zA-Z][a-zA-Z0-9_-]*."""
def replace_invalid(char):
"""
Bedrock tool names only supports alpha-numeric characters and underscores
"""
if char.isalnum() or char == "_":
if char.isalnum() or char in ("_", "-"):
return char
return "_"
@ -5492,7 +5486,7 @@ def _bedrock_tools_pt(
raw_name = f"litellm_unnamed_tool_{tool_idx}"
# related issue: https://github.com/BerriAI/litellm/issues/5007
# Bedrock tool names must satisfy regular expression pattern: [a-zA-Z][a-zA-Z0-9_]* ensure this is true
# Bedrock tool names must satisfy pattern: [a-zA-Z][a-zA-Z0-9_-]*
name = make_valid_bedrock_tool_name(input_tool_name=raw_name)
if _tool_description: # bedrock doesn't accept empty "" or None descriptions
description = _tool_description

25
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -267,18 +267,16 @@ class RealTimeStreaming:
def _make_disable_auto_response_message(self) -> str:
"""Return a session.update that disables VAD auto-response."""
turn_detection: Dict[str, Any] = {
"type": "server_vad",
"create_response": False,
}
if self._backend_uses_beta_protocol:
session: Dict[str, Any] = {
"turn_detection": {"create_response": False},
}
session: Dict[str, Any] = {"turn_detection": turn_detection}
else:
session = {
"type": "realtime",
"audio": {
"input": {
"turn_detection": {"create_response": False},
}
},
"audio": {"input": {"turn_detection": turn_detection}},
}
return json.dumps({"type": "session.update", "session": session})
@ -564,10 +562,19 @@ class RealTimeStreaming:
try:
raw_response = await self.backend_ws.recv( # type: ignore[union-attr]
decode=False
) # improves performance
)
except TypeError:
raw_response = await self.backend_ws.recv() # type: ignore[union-attr, assignment]
if isinstance(raw_response, bytes):
try:
raw_response = raw_response.decode("utf-8")
except UnicodeDecodeError:
verbose_logger.warning(
"Received non-UTF-8 binary frame from backend, skipping."
)
continue
if self.provider_config:
try:
await self._handle_provider_config_message(raw_response)

7
litellm/llms/base_llm/chat/transformation.py
View file

@ -108,10 +108,9 @@ class BaseConfig(ABC):
return type_to_response_format_param(response_format=response_format)
def is_thinking_enabled(self, non_default_params: dict) -> bool:
return (
non_default_params.get("thinking", {}).get("type") == "enabled"
or non_default_params.get("reasoning_effort") is not None
)
return (non_default_params.get("thinking") or {}).get(
"type"
) == "enabled" or non_default_params.get("reasoning_effort") is not None
def is_max_tokens_in_request(self, non_default_params: dict) -> bool:
"""

5
litellm/llms/bedrock/chat/converse_transformation.py
View file

@ -30,6 +30,7 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
BedrockConverseMessagesProcessor,
_bedrock_converse_messages_pt,
_bedrock_tools_pt,
make_valid_bedrock_tool_name,
)
from litellm.llms.anthropic.chat.transformation import (
DROP_UNSUPPORTED_OUTPUT_CONFIG_WARNING,
@ -595,7 +596,9 @@ class AmazonConverseConfig(BaseConfig):
elif isinstance(tool_choice, dict):
# only supported for anthropic + mistral models - https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ToolChoice.html
specific_tool = SpecificToolChoiceBlock(
name=tool_choice.get("function", {}).get("name", "")
name=make_valid_bedrock_tool_name(
tool_choice.get("function", {}).get("name", "")
)
)
return ToolChoiceValuesBlock(tool=specific_tool)
else:

328
litellm/model_prices_and_context_window_backup.json
View file

@ -17919,22 +17919,9 @@
},
"github_copilot/claude-haiku-4.5": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_output_tokens": 16000,
"max_tokens": 16000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions"
],
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_vision": true
},
"github_copilot/claude-opus-4.5": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_output_tokens": 16000,
"max_tokens": 16000,
"max_input_tokens": 200000,
"max_output_tokens": 32000,
"max_tokens": 32000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions"
@ -17942,7 +17929,22 @@
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_vision": true,
"supports_minimal_reasoning_effort": true
"supports_reasoning": true
},
"github_copilot/claude-opus-4.5": {
"litellm_provider": "github_copilot",
"max_input_tokens": 200000,
"max_output_tokens": 32000,
"max_tokens": 32000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions"
],
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_vision": true,
"supports_minimal_reasoning_effort": true,
"supports_reasoning": true
},
"github_copilot/claude-opus-4.6-fast": {
"litellm_provider": "github_copilot",
@ -17957,6 +17959,22 @@
"supports_parallel_function_calling": true,
"supports_vision": true
},
"github_copilot/claude-opus-4.7": {
"litellm_provider": "github_copilot",
"max_input_tokens": 200000,
"max_output_tokens": 64000,
"max_tokens": 64000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions",
"/v1/messages"
],
"supports_vision": true,
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_reasoning": true
},
"github_copilot/claude-opus-41": {
"litellm_provider": "github_copilot",
"max_input_tokens": 80000,
@ -17983,16 +18001,33 @@
},
"github_copilot/claude-sonnet-4.5": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_output_tokens": 16000,
"max_tokens": 16000,
"max_input_tokens": 200000,
"max_output_tokens": 32000,
"max_tokens": 32000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions"
],
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_vision": true
"supports_vision": true,
"supports_reasoning": true
},
"github_copilot/claude-sonnet-4.6": {
"litellm_provider": "github_copilot",
"max_input_tokens": 200000,
"max_output_tokens": 32000,
"max_tokens": 32000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions",
"/v1/messages"
],
"supports_vision": true,
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_reasoning": true
},
"github_copilot/gemini-2.5-pro": {
"litellm_provider": "github_copilot",
@ -18002,7 +18037,25 @@
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_vision": true
"supports_vision": true,
"supported_endpoints": [
"/v1/chat/completions"
],
"supports_reasoning": true
},
"github_copilot/gemini-3-flash-preview": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_output_tokens": 64000,
"max_tokens": 64000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions"
],
"supports_vision": true,
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_reasoning": true
},
"github_copilot/gemini-3-pro-preview": {
"litellm_provider": "github_copilot",
@ -18014,13 +18067,30 @@
"supports_parallel_function_calling": true,
"supports_vision": true
},
"github_copilot/gemini-3.1-pro-preview": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_output_tokens": 64000,
"max_tokens": 64000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions"
],
"supports_vision": true,
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_reasoning": true
},
"github_copilot/gpt-3.5-turbo": {
"litellm_provider": "github_copilot",
"max_input_tokens": 16384,
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supports_function_calling": true
"supports_function_calling": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-3.5-turbo-0613": {
"litellm_provider": "github_copilot",
@ -18028,7 +18098,10 @@
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supports_function_calling": true
"supports_function_calling": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4": {
"litellm_provider": "github_copilot",
@ -18036,7 +18109,22 @@
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supports_function_calling": true
"supports_function_calling": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4-0125-preview": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions"
],
"supports_function_calling": true,
"supports_parallel_function_calling": true
},
"github_copilot/gpt-4-0613": {
"litellm_provider": "github_copilot",
@ -18044,16 +18132,22 @@
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supports_function_calling": true
"supports_function_calling": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4-o-preview": {
"litellm_provider": "github_copilot",
"max_input_tokens": 64000,
"max_input_tokens": 128000,
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true
"supports_parallel_function_calling": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4.1": {
"litellm_provider": "github_copilot",
@ -18064,7 +18158,10 @@
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_vision": true
"supports_vision": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4.1-2025-04-14": {
"litellm_provider": "github_copilot",
@ -18075,68 +18172,89 @@
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_vision": true
"supports_vision": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-41-copilot": {
"litellm_provider": "github_copilot",
"mode": "completion"
"mode": "chat"
},
"github_copilot/gpt-4o": {
"litellm_provider": "github_copilot",
"max_input_tokens": 64000,
"max_input_tokens": 128000,
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_vision": true
"supports_vision": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4o-2024-05-13": {
"litellm_provider": "github_copilot",
"max_input_tokens": 64000,
"max_input_tokens": 128000,
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_vision": true
"supports_vision": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4o-2024-08-06": {
"litellm_provider": "github_copilot",
"max_input_tokens": 64000,
"max_output_tokens": 16384,
"max_tokens": 16384,
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true
},
"github_copilot/gpt-4o-2024-11-20": {
"litellm_provider": "github_copilot",
"max_input_tokens": 64000,
"max_input_tokens": 128000,
"max_output_tokens": 16384,
"max_tokens": 16384,
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_vision": true
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4o-2024-11-20": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_output_tokens": 16384,
"max_tokens": 16384,
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_vision": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4o-mini": {
"litellm_provider": "github_copilot",
"max_input_tokens": 64000,
"max_input_tokens": 128000,
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true
"supports_parallel_function_calling": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-4o-mini-2024-07-18": {
"litellm_provider": "github_copilot",
"max_input_tokens": 64000,
"max_input_tokens": 128000,
"max_output_tokens": 4096,
"max_tokens": 4096,
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true
"supports_parallel_function_calling": true,
"supported_endpoints": [
"/v1/chat/completions"
]
},
"github_copilot/gpt-5": {
"litellm_provider": "github_copilot",
@ -18155,14 +18273,19 @@
},
"github_copilot/gpt-5-mini": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_input_tokens": 264000,
"max_output_tokens": 64000,
"max_tokens": 64000,
"mode": "chat",
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_vision": true
"supports_vision": true,
"supported_endpoints": [
"/v1/chat/completions",
"/v1/responses"
],
"supports_reasoning": true
},
"github_copilot/gpt-5.1": {
"litellm_provider": "github_copilot",
@ -18195,7 +18318,7 @@
},
"github_copilot/gpt-5.2": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_input_tokens": 264000,
"max_output_tokens": 64000,
"max_tokens": 64000,
"mode": "chat",
@ -18206,11 +18329,27 @@
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_vision": true
"supports_vision": true,
"supports_reasoning": true
},
"github_copilot/gpt-5.2-codex": {
"litellm_provider": "github_copilot",
"max_input_tokens": 400000,
"max_output_tokens": 128000,
"max_tokens": 128000,
"mode": "responses",
"supported_endpoints": [
"/v1/responses"
],
"supports_vision": true,
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_reasoning": true
},
"github_copilot/gpt-5.3-codex": {
"litellm_provider": "github_copilot",
"max_input_tokens": 128000,
"max_input_tokens": 400000,
"max_output_tokens": 128000,
"max_tokens": 128000,
"mode": "responses",
@ -18220,25 +18359,96 @@
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_vision": true
"supports_vision": true,
"supports_reasoning": true
},
"github_copilot/gpt-5.4": {
"litellm_provider": "github_copilot",
"max_input_tokens": 400000,
"max_output_tokens": 128000,
"max_tokens": 128000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions",
"/v1/responses"
],
"supports_vision": true,
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_reasoning": true
},
"github_copilot/gpt-5.4-mini": {
"litellm_provider": "github_copilot",
"max_input_tokens": 400000,
"max_output_tokens": 128000,
"max_tokens": 128000,
"mode": "responses",
"supported_endpoints": [
"/v1/responses"
],
"supports_vision": true,
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_reasoning": true
},
"github_copilot/gpt-5.5": {
"litellm_provider": "github_copilot",
"max_input_tokens": 400000,
"max_output_tokens": 128000,
"max_tokens": 128000,
"mode": "responses",
"supported_endpoints": [
"/v1/responses"
],
"supports_vision": true,
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_reasoning": true
},
"github_copilot/oswe-vscode-prime": {
"litellm_provider": "github_copilot",
"max_input_tokens": 264000,
"max_output_tokens": 64000,
"max_tokens": 64000,
"mode": "chat",
"supported_endpoints": [
"/v1/chat/completions",
"/v1/responses"
],
"supports_vision": true,
"supports_function_calling": true,
"supports_parallel_function_calling": true,
"supports_response_schema": true
},
"github_copilot/text-embedding-3-small": {
"litellm_provider": "github_copilot",
"max_input_tokens": 8191,
"max_tokens": 8191,
"mode": "embedding"
"mode": "embedding",
"supported_endpoints": [
"/v1/embeddings"
]
},
"github_copilot/text-embedding-3-small-inference": {
"litellm_provider": "github_copilot",
"max_input_tokens": 8191,
"max_tokens": 8191,
"mode": "embedding"
"mode": "embedding",
"supported_endpoints": [
"/v1/embeddings"
]
},
"github_copilot/text-embedding-ada-002": {
"litellm_provider": "github_copilot",
"max_input_tokens": 8191,
"max_tokens": 8191,
"mode": "embedding"
"mode": "embedding",
"supported_endpoints": [
"/v1/embeddings"
]
},
"chatgpt/gpt-5.4": {
"litellm_provider": "chatgpt",

74
litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
View file

@ -719,25 +719,32 @@ class MCPRequestHandler:
)
)
key_access_group_extras = (
await MCPRequestHandler._get_key_access_group_mcp_server_extras(
user_api_key_auth
)
)
#########################################################
# Calculate key/team allowed servers using inheritance and intersection logic
#########################################################
allowed_mcp_servers: List[str] = []
has_lower_level_mcp_restrictions = (
len(allowed_mcp_servers_for_key) > 0
or len(allowed_mcp_servers_for_team) > 0
)
if len(allowed_mcp_servers_for_team) > 0:
if len(allowed_mcp_servers_for_key) > 0:
# Key has its own MCP permissions - use intersection with team permissions
for _mcp_server in allowed_mcp_servers_for_key:
if _mcp_server in allowed_mcp_servers_for_team:
allowed_mcp_servers.append(_mcp_server)
else:
# Key has no MCP permissions - inherit from team
allowed_mcp_servers = allowed_mcp_servers_for_team
key_set = set(allowed_mcp_servers_for_key)
team_set = set(allowed_mcp_servers_for_team)
extras_set = set(key_access_group_extras)
has_lower_level_mcp_restrictions = bool(key_set or team_set or extras_set)
# 1. Team-gated base scope.
if not team_set:
base = key_set # no team restriction
elif not key_set:
base = team_set # key has no own perms → inherits team
else:
allowed_mcp_servers = allowed_mcp_servers_for_key
base = key_set & team_set # both restrict → intersect
# 2. Extend with access-group extras (LIT-3189 — bypasses team
# ceiling, gated by group's assigned_team_ids / assigned_key_ids).
allowed_mcp_servers: List[str] = list(base | extras_set)
#########################################################
# Check end_user permissions if end_user_id is set
@ -1030,6 +1037,43 @@ class MCPRequestHandler:
return True
return False
@staticmethod
async def _get_key_access_group_mcp_server_extras(
user_api_key_auth: Optional[UserAPIKeyAuth] = None,
) -> List[str]:
"""
Resolve the key's unified `access_group_ids` (LiteLLM_AccessGroupTable) to
MCP server IDs, gated by the access group's `assigned_team_ids` /
`assigned_key_ids`. These servers extend the team's MCP scope rather
than being capped by it. Tag-style `mcp_access_groups` (per-server tags)
are intentionally not handled here — they have no assignment fields and
remain subject to the team ceiling.
"""
if user_api_key_auth is None:
return []
try:
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
from litellm.proxy.auth.auth_checks import (
get_authorized_resources_from_key_access_groups,
)
raw_server_ids = await get_authorized_resources_from_key_access_groups(
valid_token=user_api_key_auth,
team_object=None,
resource_field="access_mcp_server_ids",
)
if not raw_server_ids:
return []
# Permission entries may be server_ids OR names/aliases — expand to ids.
return global_mcp_server_manager.expand_permission_list(raw_server_ids)
except Exception as e:
verbose_logger.warning(
f"Failed to get key access group MCP server extras: {str(e)}"
)
return []
@staticmethod
async def _get_allowed_mcp_servers_for_key(
user_api_key_auth: Optional[UserAPIKeyAuth] = None,

100
litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py
View file

@ -1,4 +1,5 @@
import asyncio
import html as _html
import json
import time
from typing import Any, Dict, Optional, Tuple
@ -689,8 +690,105 @@ async def token_endpoint(
)
# Per RFC 6749 §4.1.2.1, an IdP that rejects an OAuth authorization request
# redirects back to the configured redirect URI with ``error`` /
# ``error_description`` / ``error_uri`` query params and no ``code``. The MCP
# loopback flow funnels that response through this /callback endpoint, so
# the endpoint must accept either a successful (``code``+``state``) or an
# error response. Declaring ``code``/``state`` as required would cause
# FastAPI to reject the error response with a 422 before the handler runs,
# which strands the MCP client waiting on the loopback (see LIT-2750).
def _render_oauth_error_html(error: str, description: Optional[str]) -> HTMLResponse:
"""Render an actionable HTML page for an IdP-reported OAuth error.
Used when we cannot propagate the error back to the registered
``redirect_uri`` (state missing or undecryptable). Returned with a 400
status so the failure is observable to operators while still being a
human-readable page for the end user.
"""
safe_error = _html.escape(error or "unknown_error")
safe_description = _html.escape(description) if description else ""
description_html = f"<p>{safe_description}</p>" if safe_description else ""
body = (
"<html><body>"
"<h2>Authentication failed</h2>"
f"<p><strong>Error:</strong> {safe_error}</p>"
f"{description_html}"
"<p>You can close this window and try again.</p>"
"</body></html>"
)
return HTMLResponse(body, status_code=400)
@router.get("/callback")
async def callback(request: Request, code: str, state: str):
async def callback(
request: Request,
code: Optional[str] = None,
state: Optional[str] = None,
error: Optional[str] = None,
error_description: Optional[str] = None,
error_uri: Optional[str] = None,
):
"""OAuth 2.0 authorization response handler for MCP loopback clients.
Accepts either:
- A successful authorization response (``code`` + ``state``), which is
forwarded back to the validated client ``redirect_uri`` with the
original (un-wrapped) ``state``.
- An error response (``error``[+``error_description``/``error_uri``]), per
RFC 6749 §4.1.2.1. When ``state`` is present and decodes to a trusted
``redirect_uri``, the error params are propagated back to the client so
its OAuth library can surface them. Otherwise we render an HTML error
page so the user is not left on an opaque 422 / blank screen.
"""
# 1. IdP-reported error path (e.g. ``?error=access_denied``).
if error:
verbose_logger.info(
"MCP /callback received IdP error: error=%s, error_description=%s",
error,
error_description,
)
if state:
try:
state_data = decode_state_hash(state)
original_state = state_data.get("original_state")
redirect_uri = _get_validated_client_redirect_uri(request, state_data)
except HTTPException:
# Untrusted/invalid client redirect_uri — surface inline rather
# than blindly forwarding the error to an attacker-controlled URL.
return _render_oauth_error_html(error, error_description)
except Exception:
# State could not be decrypted (expired key, tampered, etc.).
return _render_oauth_error_html(error, error_description)
params: Dict[str, str] = {"error": error}
if error_description:
params["error_description"] = error_description
if error_uri:
params["error_uri"] = error_uri
if original_state is not None:
params["state"] = original_state
complete_returned_url = _append_query_params(redirect_uri, params)
return RedirectResponse(url=complete_returned_url, status_code=302)
# No state — nothing to round-trip to. Show the user the error.
return _render_oauth_error_html(error, error_description)
# 2. Neither success nor error parameters present — most likely a stray
# GET / dropped SSO redirect chain. Surface a 400 instead of 422.
if not code or not state:
missing = [
name for name, value in (("code", code), ("state", state)) if not value
]
return _render_oauth_error_html(
"invalid_request",
f"Missing authorization {' and '.join(repr(m) for m in missing)} parameter(s).",
)
# 3. Successful authorization response.
try:
state_data = decode_state_hash(state)
original_state = state_data["original_state"]

61
litellm/proxy/auth/auth_checks.py
View file

@ -3163,44 +3163,40 @@ async def can_team_access_model(
raise
async def _key_access_group_grants_model(
model: Union[str, List[str]],
async def get_authorized_resources_from_key_access_groups(
valid_token: Optional[UserAPIKeyAuth],
team_object: Optional[LiteLLM_TeamTable],
llm_router: Optional[Router],
) -> bool:
resource_field: Literal[
"access_model_names", "access_mcp_server_ids", "access_agent_ids"
],
) -> List[str]:
"""
Returns True if the key's `access_group_ids` expand to models that grant
access to `model`. Used to let a key's access group override a team's
model restriction in `common_checks`.
A key's access group only counts if the access group itself authorizes the
caller as an owner — that is, the group's `assigned_team_ids` includes the
key's `team_id`, or the group's `assigned_key_ids` includes the key's
token. This preserves the team-as-owner boundary (a team member cannot
escalate by naming a group assigned to a different team) while still
letting a group reach the key without first being added to the team's
`access_group_ids` list.
For each access_group_id on the key, fetch the LiteLLM_AccessGroupTable row
and contribute its `resource_field` only if the group authorizes the caller
as an owner — that is, the group's `assigned_team_ids` includes the key's
`team_id`, or the group's `assigned_key_ids` includes the key's token. This
preserves the team-as-owner boundary while still letting a group reach the
key without first being added to the team's `access_group_ids` list.
"""
if valid_token is None:
return False
return []
key_access_group_ids = list(valid_token.access_group_ids or [])
if not key_access_group_ids:
return False
return []
from litellm.proxy.proxy_server import prisma_client as _prisma_client
from litellm.proxy.proxy_server import proxy_logging_obj as _proxy_logging_obj
from litellm.proxy.proxy_server import user_api_key_cache as _user_api_key_cache
if _prisma_client is None or _user_api_key_cache is None:
return False
return []
key_team_id = valid_token.team_id or (
team_object.team_id if team_object is not None else None
)
key_token = valid_token.token
authorized_models: List[str] = []
authorized_resources: List[str] = []
for ag_id in key_access_group_ids:
try:
ag = await get_access_object(
@ -3216,17 +3212,36 @@ async def _key_access_group_grants_model(
)
key_authorized = bool(key_token and key_token in (ag.assigned_key_ids or []))
if team_authorized or key_authorized:
authorized_models.extend(ag.access_model_names or [])
authorized_resources.extend(getattr(ag, resource_field, []) or [])
return list(set(authorized_resources))
async def _key_access_group_grants_model(
model: Union[str, List[str]],
valid_token: Optional[UserAPIKeyAuth],
team_object: Optional[LiteLLM_TeamTable],
llm_router: Optional[Router],
) -> bool:
"""
Returns True if the key's `access_group_ids` expand to models that grant
access to `model`. Used to let a key's access group override a team's
model restriction in `common_checks`.
"""
authorized_models = await get_authorized_resources_from_key_access_groups(
valid_token=valid_token,
team_object=team_object,
resource_field="access_model_names",
)
if not authorized_models:
return False
try:
_can_object_call_model(
model=model,
llm_router=llm_router,
models=list(set(authorized_models)),
team_model_aliases=valid_token.team_model_aliases,
team_id=valid_token.team_id,
models=authorized_models,
team_model_aliases=valid_token.team_model_aliases if valid_token else None,
team_id=valid_token.team_id if valid_token else None,
object_type="key",
)
return True

6
litellm/proxy/auth/auth_utils.py
View file

@ -213,6 +213,12 @@ _EXTRA_BANNED_OBSERVABILITY_PARAMS: FrozenSet[str] = frozenset(
{
"posthog_api_url",
"phoenix_project_name",
"phoenix_project_name_override",
# Server-reserved: written exclusively by add_user_api_key_auth_to_request_metadata
# from the authenticated key's database record. A caller-supplied value
# would survive the server merge and let an authenticated user redirect
# their Arize/Phoenix telemetry into arbitrary projects.
"user_api_key_auth_metadata",
"wandb_api_key",
"weave_project_id",
}

6
litellm/proxy/batches_endpoints/endpoints.py
View file

@ -15,6 +15,9 @@ from litellm.batches.main import CancelBatchRequest, RetrieveBatchRequest
from litellm.proxy._types import *
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
from litellm.proxy.common_utils.callback_utils import (
sanitize_openai_provider_metadata,
)
from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
from litellm.proxy.common_utils.openai_endpoint_utils import (
get_custom_llm_provider_from_request_headers,
@ -120,6 +123,9 @@ async def create_batch( # noqa: PLR0915
or get_custom_llm_provider_from_request_headers(request=request)
or "openai"
)
if isinstance(data.get("metadata"), dict):
data["metadata"] = sanitize_openai_provider_metadata(data["metadata"])
_create_batch_data = LiteLLMBatchCreateRequest(**data)
# Apply team-level batch output expiry enforcement

128
litellm/proxy/common_utils/callback_utils.py
View file

@ -409,11 +409,15 @@ def get_remaining_tokens_and_requests_from_request_data(data: Dict) -> Dict[str,
def get_logging_caching_headers(request_data: Dict) -> Optional[Dict]:
_metadata = request_data.get("metadata", None)
if not _metadata:
_metadata = request_data.get("litellm_metadata", None)
if not isinstance(_metadata, dict):
_metadata = {}
_metadata: Dict = {}
metadata_bucket = request_data.get("metadata")
litellm_metadata_bucket = request_data.get("litellm_metadata")
if isinstance(metadata_bucket, dict):
_metadata.update(metadata_bucket)
if isinstance(litellm_metadata_bucket, dict):
# Batch/file routes store proxy tracking in litellm_metadata while
# user-facing metadata stays in metadata; merge both for headers.
_metadata.update(litellm_metadata_bucket)
headers = {}
if "applied_guardrails" in _metadata:
headers["x-litellm-applied-guardrails"] = ",".join(
@ -452,19 +456,103 @@ def get_logging_caching_headers(request_data: Dict) -> Optional[Dict]:
return headers
def get_metadata_variable_name_from_kwargs(
kwargs: dict,
) -> Literal["metadata", "litellm_metadata"]:
"""
Helper to return what the "metadata" field should be called in the request data
- New endpoints return `litellm_metadata`
- Old endpoints return `metadata`
Context:
- LiteLLM used `metadata` as an internal field for storing metadata
- OpenAI then started using this field for their metadata
- LiteLLM is now moving to using `litellm_metadata` for our metadata
"""
return "litellm_metadata" if "litellm_metadata" in kwargs else "metadata"
LITELLM_PROXY_INTERNAL_METADATA_KEYS = frozenset(
{
"applied_policies",
"applied_guardrails",
"policy_sources",
"guardrails",
"guardrail_config",
"_guardrail_pipelines",
"_pipeline_managed_guardrails",
"disable_global_guardrails",
"disable_global_guardrail",
"opted_out_global_guardrails",
"pillar_response_headers",
"_pillar_response_headers_trusted",
"pillar_flagged",
"pillar_scanners",
"pillar_evidence",
"pillar_evidence_truncated",
"pillar_session_id_response",
"standard_logging_object",
"proxy_server_request",
"secret_fields",
}
)
def _get_or_create_proxy_metadata_bucket(
request_data: Dict,
) -> tuple[Literal["metadata", "litellm_metadata"], dict]:
"""
Return the proxy-internal metadata bucket for this request.
Batch/file routes store proxy state in ``litellm_metadata`` so the OpenAI
``metadata`` field can remain provider-safe (string values only).
"""
metadata_key = get_metadata_variable_name_from_kwargs(request_data)
metadata_bucket = request_data.get(metadata_key)
if not isinstance(metadata_bucket, dict):
metadata_bucket = {}
request_data[metadata_key] = metadata_bucket
return metadata_key, metadata_bucket
def sanitize_openai_provider_metadata(
metadata: Optional[Dict[str, Any]],
) -> Optional[Dict[str, str]]:
"""
Keep only provider-safe OpenAI metadata entries (string keys -> string values).
Strips LiteLLM proxy-internal tracking fields that must not be forwarded to
OpenAI batch/file APIs.
"""
if not metadata:
return metadata
sanitized: Dict[str, str] = {}
for key, value in metadata.items():
if key in LITELLM_PROXY_INTERNAL_METADATA_KEYS:
continue
if isinstance(value, str):
sanitized[key] = value
else:
verbose_proxy_logger.debug(
"sanitize_openai_provider_metadata: dropping key %r with non-string value of type %s",
key,
type(value).__name__,
)
return sanitized or None
def add_guardrail_to_applied_guardrails_header(
request_data: Dict, guardrail_name: Optional[str]
):
if guardrail_name is None:
return
_metadata = request_data.get("metadata", None) or {}
_, _metadata = _get_or_create_proxy_metadata_bucket(request_data)
if "applied_guardrails" in _metadata:
if guardrail_name not in _metadata["applied_guardrails"]:
_metadata["applied_guardrails"].append(guardrail_name)
else:
_metadata["applied_guardrails"] = [guardrail_name]
# Ensure metadata is set back to request_data (important when metadata didn't exist)
request_data["metadata"] = _metadata
def add_policy_to_applied_policies_header(
@ -478,14 +566,12 @@ def add_policy_to_applied_policies_header(
"""
if policy_name is None:
return
_metadata = request_data.get("metadata", None) or {}
_, _metadata = _get_or_create_proxy_metadata_bucket(request_data)
if "applied_policies" in _metadata:
if policy_name not in _metadata["applied_policies"]:
_metadata["applied_policies"].append(policy_name)
else:
_metadata["applied_policies"] = [policy_name]
# Ensure metadata is set back to request_data (important when metadata didn't exist)
request_data["metadata"] = _metadata
def add_policy_sources_to_metadata(request_data: Dict, policy_sources: Dict[str, str]):
@ -498,13 +584,12 @@ def add_policy_sources_to_metadata(request_data: Dict, policy_sources: Dict[str,
"""
if not policy_sources:
return
_metadata = request_data.get("metadata", None) or {}
_, _metadata = _get_or_create_proxy_metadata_bucket(request_data)
existing = _metadata.get("policy_sources", {})
if not isinstance(existing, dict):
existing = {}
existing.update(policy_sources)
_metadata["policy_sources"] = existing
request_data["metadata"] = _metadata
def add_guardrail_response_to_standard_logging_object(
@ -527,23 +612,6 @@ def add_guardrail_response_to_standard_logging_object(
return standard_logging_object
def get_metadata_variable_name_from_kwargs(
kwargs: dict,
) -> Literal["metadata", "litellm_metadata"]:
"""
Helper to return what the "metadata" field should be called in the request data
- New endpoints return `litellm_metadata`
- Old endpoints return `metadata`
Context:
- LiteLLM used `metadata` as an internal field for storing metadata
- OpenAI then started using this field for their metadata
- LiteLLM is now moving to using `litellm_metadata` for our metadata
"""
return "litellm_metadata" if "litellm_metadata" in kwargs else "metadata"
def process_callback(
_callback: str, callback_type: str, environment_variables: dict
) -> dict:

4
litellm/proxy/example_config_yaml/oai_misc_config.yaml
View file

@ -23,11 +23,11 @@ model_list:
model: bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0
#########################################################
########## batch specific params ########################
s3_bucket_name: litellm-proxy-941277531214
s3_bucket_name: litellm-proxy-123456789012
s3_region_name: us-west-2
s3_access_key_id: os.environ/AWS_ACCESS_KEY_ID
s3_secret_access_key: os.environ/AWS_SECRET_ACCESS_KEY
aws_batch_role_arn: arn:aws:iam::941277531214:role/service-role/AmazonBedrockExecutionRoleForAgents_BB9HNW6V4CV
aws_batch_role_arn: arn:aws:iam::123456789012:role/service-role/AmazonBedrockExecutionRoleForAgents_EXAMPLE
model_info:
mode: batch

24
litellm/proxy/management_helpers/utils.py
View file

@ -471,10 +471,32 @@ async def _emit_management_endpoint_otel_span(
route = func.__name__
request_body = {}
_CREDENTIAL_FIELDS = frozenset(
{
"key",
"token",
"api_key",
"secret",
"password",
"access_token",
"refresh_token",
"private_key",
"service_account_key",
}
)
_response: Optional[dict] = None
if exception is None and result is not None:
try:
raw = dict(result)
_response = {k: v for k, v in raw.items() if k not in _CREDENTIAL_FIELDS}
except Exception:
_response = None
logging_payload = ManagementEndpointLoggingPayload(
route=route,
request_data=request_body,
response=None,
response=_response,
start_time=start_time,
end_time=end_time,
exception=exception,

2
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View file

@ -908,7 +908,7 @@ async def pass_through_request( # noqa: PLR0915
else {"json": _parsed_body}
)
req = async_client.build_request(
"POST",
request.method,
url,
params=requested_query_params,
headers=headers,

4
tests/llm_translation/test_bedrock_completion.py
View file

@ -1062,7 +1062,7 @@ def test_bedrock_tools_pt_invalid_names():
print("bedrock tools after prompt formatting=", result)
assert len(result) == 2
assert result[0]["toolSpec"]["name"] == "a123_invalid_name"
assert result[0]["toolSpec"]["name"] == "a123-invalid_name"
assert result[1]["toolSpec"]["name"] == "another_invalid_name"
@ -1171,7 +1171,7 @@ def test_bedrock_tools_transformation_valid_params():
assert isinstance(result, list)
assert len(result) == 1
assert "toolSpec" in result[0]
assert result[0]["toolSpec"]["name"] == "a123_invalid_name"
assert result[0]["toolSpec"]["name"] == "a123-invalid_name"
assert result[0]["toolSpec"]["description"] == "Invalid name test"
assert "inputSchema" in result[0]["toolSpec"]
assert "json" in result[0]["toolSpec"]["inputSchema"]

600
tests/test_litellm/integrations/arize/test_arize_phoenix.py
View file

@ -7,7 +7,6 @@ from litellm.integrations.arize.arize_phoenix import (
ArizePhoenixConfig,
ArizePhoenixLogger,
)
from litellm.integrations.arize._utils import ArizeOTELAttributes
class TestArizePhoenixConfig(unittest.TestCase):
@ -217,44 +216,147 @@ def test_get_arize_phoenix_config_expection_on_missing_api_key(monkeypatch, env_
# ---------------------------------------------------------------------------
# Dynamic project naming from metadata
# Per-project routing via Resource (not span attributes)
# ---------------------------------------------------------------------------
class TestGetDynamicProjectName:
"""Tests for _get_dynamic_project_name extraction logic."""
class TestResolveProjectName:
"""Tests for _resolve_project_name priority chain."""
def test_extracts_from_standard_logging_object_metadata(self):
def test_extracts_phoenix_name_from_standard_logging_object_metadata(self):
kwargs = {
"standard_logging_object": {
"metadata": {"phoenix_project_name": "my-project"},
}
}
assert ArizePhoenixLogger._get_dynamic_project_name(kwargs) == "my-project"
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "my-project"
def test_extracts_from_litellm_params_metadata(self):
def test_extracts_phoenix_name_from_litellm_params_metadata(self):
kwargs = {
"litellm_params": {
"metadata": {"phoenix_project_name": "sdk-project"},
}
}
assert ArizePhoenixLogger._get_dynamic_project_name(kwargs) == "sdk-project"
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "sdk-project"
def test_returns_none_when_no_metadata(self):
assert ArizePhoenixLogger._get_dynamic_project_name({}) is None
@patch.dict("os.environ", {"PHOENIX_PROJECT_NAME": "env-project"}, clear=False)
def test_falls_back_to_phoenix_env_when_no_metadata(self):
assert ArizePhoenixLogger._resolve_project_name({}) == "env-project"
@patch.dict(
"os.environ",
{"ARIZE_PROJECT_NAME": "arize-env", "PHOENIX_PROJECT_NAME": ""},
clear=False,
)
def test_falls_back_to_arize_env_when_phoenix_unset(self):
assert ArizePhoenixLogger._resolve_project_name({}) == "arize-env"
@patch.dict("os.environ", {}, clear=True)
def test_falls_back_to_default_when_no_metadata_or_env(self):
assert ArizePhoenixLogger._resolve_project_name({}) == "default"
def test_phoenix_override_beats_phoenix_metadata(self):
kwargs = {
"standard_logging_object": {
"metadata": {
"phoenix_project_name_override": "override-proj",
"phoenix_project_name": "phoenix-proj",
},
}
}
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "override-proj"
def test_whitespace_only_metadata_falls_through_to_default(self):
kwargs = {
"standard_logging_object": {
"metadata": {"phoenix_project_name_override": " "},
}
}
with patch.dict("os.environ", {}, clear=True):
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "default"
def test_strips_whitespace_from_project_name(self):
kwargs = {
"standard_logging_object": {
"metadata": {"phoenix_project_name": " trimmed "},
}
}
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "trimmed"
def test_non_dict_standard_logging_object_does_not_raise(self):
"""isinstance(dict) guard prevents AttributeError on non-dict payloads."""
kwargs = {"standard_logging_object": "not-a-dict"}
assert ArizePhoenixLogger._get_dynamic_project_name(kwargs) is None
with patch.dict("os.environ", {}, clear=True):
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "default"
def test_resolves_override_from_user_api_key_auth_metadata(self):
kwargs = {
"litellm_params": {
"metadata": {
"user_api_key_auth_metadata": {
"phoenix_project_name_override": "claude-code",
},
},
},
}
with patch.dict("os.environ", {}, clear=True):
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "claude-code"
def test_resolves_phoenix_name_from_user_api_key_auth_metadata(self):
kwargs = {
"standard_logging_object": {
"metadata": {
"user_api_key_auth_metadata": {
"phoenix_project_name": "team-project",
},
},
},
}
with patch.dict("os.environ", {}, clear=True):
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "team-project"
def test_proxy_ignores_client_metadata_when_auth_metadata_set(self):
kwargs = {
"litellm_params": {
"proxy_server_request": {
"url": "/v1/chat/completions",
"method": "POST",
"headers": {},
},
"metadata": {
"phoenix_project_name_override": "attacker-project",
"user_api_key_auth_metadata": {
"phoenix_project_name_override": "team-project",
},
},
},
}
with patch.dict("os.environ", {}, clear=True):
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "team-project"
def test_proxy_without_auth_metadata_falls_back_to_env(self):
kwargs = {
"litellm_params": {
"proxy_server_request": {
"url": "/v1/chat/completions",
"method": "POST",
"headers": {},
},
"metadata": {"phoenix_project_name": "attacker-project"},
},
}
with patch.dict(
"os.environ", {"PHOENIX_PROJECT_NAME": "env-project"}, clear=True
):
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "env-project"
class TestDynamicProjectNameOnSpan:
"""set_arize_phoenix_attributes sets openinference.project.name on the span."""
class TestProjectNameNotOnSpan:
"""Project routing uses Resource on TracerProvider, not span attributes."""
@patch.dict("os.environ", {"PHOENIX_PROJECT_NAME": "env-fallback"}, clear=False)
@patch("litellm.integrations.arize._utils.set_attributes")
def test_dynamic_name_sets_span_attribute(self, _mock_set_attrs):
def test_set_arize_phoenix_attributes_does_not_set_project_on_span(
self, _mock_set_attrs
):
span = MagicMock()
kwargs = {
"standard_logging_object": {
@ -263,20 +365,468 @@ class TestDynamicProjectNameOnSpan:
}
ArizePhoenixLogger.set_arize_phoenix_attributes(span, kwargs, response_obj=None)
span.set_attribute.assert_called_once_with(
"openinference.project.name", "dynamic-proj"
for call in span.set_attribute.call_args_list:
assert call[0][0] != "openinference.project.name"
class TestPerProjectTracerProviderCache:
"""Spans for different projects use different Resources on export."""
def test_different_metadata_routes_to_different_resource(self):
from datetime import datetime
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
@patch.dict("os.environ", {"PHOENIX_PROJECT_NAME": "env-project"}, clear=False)
@patch("litellm.integrations.arize._utils.set_attributes")
def test_falls_back_to_env_var_when_no_dynamic_name(self, _mock_set_attrs):
span = MagicMock()
ArizePhoenixLogger.set_arize_phoenix_attributes(span, {}, response_obj=None)
from litellm.integrations.opentelemetry import OpenTelemetryConfig
span.set_attribute.assert_called_once_with(
"openinference.project.name", "env-project"
exporter = InMemorySpanExporter()
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=exporter),
callback_name="arize_phoenix",
)
start = datetime(2024, 1, 1, 12, 0, 0)
end = datetime(2024, 1, 1, 12, 0, 1)
logger._handle_success(
{
"standard_logging_object": {
"metadata": {"phoenix_project_name": "project-a"},
},
},
response_obj={},
start_time=start,
end_time=end,
)
logger._handle_success(
{
"standard_logging_object": {
"metadata": {"phoenix_project_name": "project-b"},
},
},
response_obj={},
start_time=start,
end_time=end,
)
spans = exporter.get_finished_spans()
project_names = {
s.resource.attributes.get("openinference.project.name") for s in spans
}
assert "project-a" in project_names
assert "project-b" in project_names
def test_shared_span_processor_created_once_at_init(self):
from litellm.integrations.opentelemetry import (
OpenTelemetry,
OpenTelemetryConfig,
)
mock_processor = MagicMock()
with patch.object(
OpenTelemetry, "_get_span_processor", return_value=mock_processor
) as mock_get_processor:
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=MagicMock()),
callback_name="arize_phoenix",
)
assert mock_get_processor.call_count == 1
assert logger._shared_span_processor is mock_processor
logger._project_providers.clear()
logger._get_tracer_for("project-a")
logger._get_tracer_for("project-b")
assert mock_get_processor.call_count == 1
def test_lru_eviction_does_not_shutdown_provider(self):
from litellm.integrations.opentelemetry import OpenTelemetryConfig
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=MagicMock()),
callback_name="arize_phoenix",
)
logger._project_providers.clear()
logger._get_tracer_for("project-0")
evicted_provider = logger._project_providers["project-0"]
shutdown_mock = MagicMock()
evicted_provider.shutdown = shutdown_mock # type: ignore[method-assign]
for i in range(1, 65):
logger._get_tracer_for(f"project-{i}")
assert len(logger._project_providers) == 64
assert "project-0" not in logger._project_providers
assert "project-64" in logger._project_providers
shutdown_mock.assert_not_called()
def test_flush_tracer_providers_force_flushes_shared_processor(self):
from litellm.integrations.opentelemetry import OpenTelemetryConfig
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=MagicMock()),
callback_name="arize_phoenix",
)
mock_processor = MagicMock()
logger._shared_span_processor = mock_processor
mock_provider = MagicMock()
logger._project_providers["proj"] = mock_provider
logger.flush_tracer_providers()
mock_processor.force_flush.assert_called_once()
mock_provider.force_flush.assert_called_once()
class TestGetLitellmResourceForProject:
"""Resource attrs used by Phoenix OSS and Arize AX for project routing."""
def test_project_attrs_win_over_otel_resource_attributes_env(self):
from litellm.integrations.opentelemetry import OpenTelemetryConfig
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=MagicMock()),
callback_name="arize_phoenix",
)
with patch.dict(
"os.environ",
{
"OTEL_RESOURCE_ATTRIBUTES": "openinference.project.name=env-pinned,model_id=env-model"
},
clear=False,
):
resource = logger._get_litellm_resource_for_project("dynamic-proj")
assert resource.attributes["openinference.project.name"] == "dynamic-proj"
assert resource.attributes["model_id"] == "dynamic-proj"
assert resource.attributes["service.name"] == "dynamic-proj"
@patch.dict("os.environ", {"OTEL_DEPLOYMENT_ENVIRONMENT": "staging"}, clear=False)
def test_preserves_deployment_environment_from_config(self):
from litellm.integrations.opentelemetry import OpenTelemetryConfig
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(
exporter=MagicMock(), deployment_environment="staging"
),
callback_name="arize_phoenix",
)
resource = logger._get_litellm_resource_for_project("my-proj")
assert resource.attributes.get("deployment.environment") == "staging"
class TestTracerResolutionAndCache:
"""_resolve_tracer_for_kwargs, get_tracer_to_use_for_request, provider cache."""
def test_get_tracer_to_use_for_request_matches_resolve_tracer(self):
from litellm.integrations.opentelemetry import OpenTelemetryConfig
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=MagicMock()),
callback_name="arize_phoenix",
)
kwargs = {
"standard_logging_object": {
"metadata": {"phoenix_project_name": "same-proj"},
}
}
project_name, _ = logger._resolve_tracer_for_kwargs(kwargs)
tracer_from_request = logger.get_tracer_to_use_for_request(kwargs)
assert project_name == "same-proj"
assert "same-proj" in logger._project_providers
assert logger._resolve_project_name(kwargs) == project_name
assert tracer_from_request is not None
def test_cache_reuses_provider_for_same_project(self):
from litellm.integrations.opentelemetry import OpenTelemetryConfig
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=MagicMock()),
callback_name="arize_phoenix",
)
logger._project_providers.clear()
logger._get_tracer_for("cached-proj")
provider_first = logger._project_providers["cached-proj"]
logger._get_tracer_for("cached-proj")
provider_second = logger._project_providers["cached-proj"]
assert provider_first is provider_second
assert len(logger._project_providers) == 1
def test_parallel_cache_miss_for_same_project_inserts_once(self):
import threading
from litellm.integrations.opentelemetry import OpenTelemetryConfig
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=MagicMock()),
callback_name="arize_phoenix",
)
logger._project_providers.clear()
build_calls: list[str] = []
real_build = logger._build_tracer_provider_for_project
def tracking_build(project_name: str):
build_calls.append(project_name)
return real_build(project_name)
barrier = threading.Barrier(10)
errors: list[Exception] = []
def worker() -> None:
try:
barrier.wait()
logger._get_tracer_for("race-proj")
except Exception as exc:
errors.append(exc)
with patch.object(
logger,
"_build_tracer_provider_for_project",
side_effect=tracking_build,
):
threads = [threading.Thread(target=worker) for _ in range(10)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
assert not errors
assert len(logger._project_providers) == 1
assert "race-proj" in logger._project_providers
assert len(build_calls) >= 1
def test_injected_tracer_provider_bypasses_project_cache(self):
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
from litellm.integrations.opentelemetry import OpenTelemetryConfig
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=exporter),
callback_name="arize_phoenix",
tracer_provider=provider,
)
assert getattr(logger, "_use_injected_tracer_provider", False) is True
assert not hasattr(logger, "_project_providers") or not getattr(
logger, "_project_providers", None
)
tracer_a = logger._get_tracer_for("any-project")
tracer_b = logger.get_tracer_to_use_for_request(
{"standard_logging_object": {"metadata": {"phoenix_project_name": "x"}}}
)
assert tracer_a is logger.tracer
assert tracer_b is logger.tracer
def test_flush_tracer_providers_noop_for_injected_provider(self):
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
from litellm.integrations.opentelemetry import OpenTelemetryConfig
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=exporter),
callback_name="arize_phoenix",
tracer_provider=provider,
)
logger.flush_tracer_providers()
exporter.shutdown()
def test_standard_logging_metadata_wins_over_litellm_params(self):
kwargs = {
"standard_logging_object": {
"metadata": {"phoenix_project_name_override": "from-logging"},
},
"litellm_params": {
"metadata": {"phoenix_project_name_override": "from-params"},
},
}
assert ArizePhoenixLogger._resolve_project_name(kwargs) == "from-logging"
class TestPhoenixTraceHandling:
"""_handle_success / _handle_failure span export behavior."""
def test_handle_failure_sets_error_status_on_request_span(self):
from datetime import datetime
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
from opentelemetry.trace import StatusCode
from litellm.integrations.opentelemetry import (
LITELLM_REQUEST_SPAN_NAME,
OpenTelemetryConfig,
)
exporter = InMemorySpanExporter()
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=exporter),
callback_name="arize_phoenix",
)
start = datetime(2024, 1, 1, 12, 0, 0)
end = datetime(2024, 1, 1, 12, 0, 1)
logger._handle_failure(
{
"standard_logging_object": {
"metadata": {"phoenix_project_name": "fail-proj"},
},
"exception": Exception("boom"),
},
response_obj=None,
start_time=start,
end_time=end,
)
spans = exporter.get_finished_spans()
request_spans = [s for s in spans if s.name == LITELLM_REQUEST_SPAN_NAME]
assert len(request_spans) == 1
assert request_spans[0].status.status_code == StatusCode.ERROR
assert (
request_spans[0].resource.attributes.get("openinference.project.name")
== "fail-proj"
)
def test_proxy_mode_parent_and_child_share_trace_id(self):
from datetime import datetime
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
from litellm.integrations.opentelemetry import (
LITELLM_REQUEST_SPAN_NAME,
OpenTelemetryConfig,
)
exporter = InMemorySpanExporter()
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=exporter),
callback_name="arize_phoenix",
)
start = datetime(2024, 1, 1, 12, 0, 0)
end = datetime(2024, 1, 1, 12, 0, 1)
logger._handle_success(
{
"litellm_params": {
"proxy_server_request": {
"url": "/chat/completions",
"method": "POST",
"headers": {},
},
"metadata": {
"user_api_key_auth_metadata": {
"phoenix_project_name_override": "proxy-proj",
},
},
},
},
response_obj={},
start_time=start,
end_time=end,
)
spans = exporter.get_finished_spans()
span_names = {s.name for s in spans}
assert "litellm_proxy_request" in span_names
assert LITELLM_REQUEST_SPAN_NAME in span_names
trace_ids = {s.context.trace_id for s in spans}
assert len(trace_ids) == 1
for span in spans:
assert (
span.resource.attributes.get("openinference.project.name")
== "proxy-proj"
)
def test_override_routes_all_spans_to_one_project_in_single_request(self):
from datetime import datetime
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
)
from litellm.integrations.opentelemetry import OpenTelemetryConfig
exporter = InMemorySpanExporter()
logger = ArizePhoenixLogger(
config=OpenTelemetryConfig(exporter=exporter),
callback_name="arize_phoenix",
)
start = datetime(2024, 1, 1, 12, 0, 0)
end = datetime(2024, 1, 1, 12, 0, 1)
logger._handle_success(
{
"standard_logging_object": {
"metadata": {
"user_api_key_auth_metadata": {
"phoenix_project_name_override": "unified-proj",
},
},
},
"litellm_params": {
"proxy_server_request": {
"url": "/v1/chat/completions",
"method": "POST",
"headers": {},
},
},
},
response_obj={"id": "resp-1"},
start_time=start,
end_time=end,
)
for span in exporter.get_finished_spans():
assert (
span.resource.attributes.get("openinference.project.name")
== "unified-proj"
)
assert span.resource.attributes.get("model_id") == "unified-proj"
class TestGetArizePhoenixConfigProjectName:
@patch.dict(
"os.environ", {"PHOENIX_PROJECT_NAME": "phoenix-config-proj"}, clear=True
)
def test_project_name_from_phoenix_env(self):
config = ArizePhoenixLogger.get_arize_phoenix_config()
assert config.project_name == "phoenix-config-proj"
@patch.dict("os.environ", {}, clear=True)
def test_project_name_defaults_when_env_unset(self):
config = ArizePhoenixLogger.get_arize_phoenix_config()
assert config.project_name == "default"
if __name__ == "__main__":
unittest.main()

67
tests/test_litellm/integrations/datadog/test_datadog_metrics.py
View file

@ -104,6 +104,7 @@ async def test_add_metrics_from_log(clean_env):
logger._add_metrics_from_log(log=payload, kwargs=kwargs, status_code="200")
# Should have 3 series: total_latency, llm_api_latency, request_count
# (no overhead metric because payload has no hidden_params litellm_overhead_time_ms)
assert len(logger.log_queue) == 3
metrics = {s["metric"]: s for s in logger.log_queue}
@ -125,6 +126,72 @@ async def test_add_metrics_from_log(clean_env):
assert "status_code:200" in count["tags"]
@pytest.mark.asyncio
async def test_overhead_latency_metric_emitted(clean_env):
"""Test that litellm.overhead.latency is emitted when hidden_params contains litellm_overhead_time_ms."""
logger = DatadogMetricsLogger(batch_size=100, start_periodic_flush=False)
now = datetime.now()
start_time = now - timedelta(seconds=2)
api_call_start_time = now - timedelta(seconds=1)
payload = StandardLoggingPayload(
custom_llm_provider="openai",
model="gpt-4o",
hidden_params={
"litellm_overhead_time_ms": 250.0, # 250 ms of overhead
},
)
kwargs = {
"start_time": start_time,
"api_call_start_time": api_call_start_time,
"end_time": now,
}
logger._add_metrics_from_log(log=payload, kwargs=kwargs, status_code="200")
metrics = {s["metric"]: s for s in logger.log_queue}
# Overhead metric must be present
assert (
"litellm.overhead.latency" in metrics
), f"Expected 'litellm.overhead.latency' in emitted metrics, got: {list(metrics.keys())}"
overhead = metrics["litellm.overhead.latency"]
assert overhead["type"] == 3 # gauge
# 250 ms → 0.25 s
assert abs(overhead["points"][0]["value"] - 0.25) < 1e-6
# status_code should NOT be in overhead tags (it is a latency metric, not a request count)
assert not any(tag.startswith("status_code:") for tag in overhead["tags"])
@pytest.mark.asyncio
async def test_overhead_latency_metric_absent_when_no_hidden_params(clean_env):
"""Test that litellm.overhead.latency is NOT emitted when hidden_params has no overhead value."""
logger = DatadogMetricsLogger(batch_size=100, start_periodic_flush=False)
now = datetime.now()
start_time = now - timedelta(seconds=2)
api_call_start_time = now - timedelta(seconds=1)
payload = StandardLoggingPayload(
custom_llm_provider="openai",
model="gpt-4o",
# No hidden_params / no litellm_overhead_time_ms
)
kwargs = {
"start_time": start_time,
"api_call_start_time": api_call_start_time,
"end_time": now,
}
logger._add_metrics_from_log(log=payload, kwargs=kwargs, status_code="200")
metrics = {s["metric"]: s for s in logger.log_queue}
assert "litellm.overhead.latency" not in metrics
@pytest.mark.asyncio
async def test_async_log_success_event(clean_env):
"""Test that success events are added to the queue."""

86
tests/test_litellm/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py
View file

@ -10,10 +10,12 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
BedrockConverseMessagesProcessor,
BedrockImageProcessor,
_bedrock_converse_messages_pt,
_bedrock_tools_pt,
_convert_to_bedrock_tool_call_invoke,
_convert_to_bedrock_tool_call_result,
anthropic_messages_pt,
convert_to_gemini_tool_call_result,
make_valid_bedrock_tool_name,
ollama_pt,
sanitize_messages_for_tool_calling,
)
@ -2082,6 +2084,90 @@ def test_bedrock_tool_call_invoke_non_dict_arguments():
assert result[0]["toolUse"]["input"] == {}
def test_make_valid_bedrock_tool_name_preserves_hyphens():
assert make_valid_bedrock_tool_name("my-tool") == "my-tool"
assert (
make_valid_bedrock_tool_name(
"CreateCaseKnowledgeArticle_foTWsqR6yDt-OnSsvR5e6Q"
)
== "CreateCaseKnowledgeArticle_foTWsqR6yDt-OnSsvR5e6Q"
)
def test_bedrock_tool_name_sanitized_consistently_in_tools_and_tool_use():
"""toolSpec and toolUse names must match after sanitization (issue #5007)."""
raw_name = "foo@bar"
tools = [
{
"type": "function",
"function": {
"name": raw_name,
"description": "test",
"parameters": {"type": "object", "properties": {}},
},
}
]
tool_spec_name = _bedrock_tools_pt(tools)[0]["toolSpec"]["name"]
tool_calls = [
{
"id": "call_1",
"type": "function",
"function": {"name": raw_name, "arguments": "{}"},
}
]
tool_use_name = _convert_to_bedrock_tool_call_invoke(tool_calls)[0]["toolUse"][
"name"
]
assert tool_spec_name == "foo_bar"
assert tool_use_name == tool_spec_name
def test_bedrock_converse_messages_pt_tool_use_matches_tool_spec_hyphen_name():
"""Hyphenated tool names are preserved and consistent in multi-turn history."""
tool_name = "my-tool"
messages = [
{"role": "user", "content": "call the tool"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_hyphen",
"type": "function",
"function": {"name": tool_name, "arguments": "{}"},
}
],
},
]
translated = _bedrock_converse_messages_pt(
messages=messages, model="", llm_provider=""
)
tool_use_blocks = [
block
for msg in translated
for block in msg.get("content", [])
if "toolUse" in block
]
assert len(tool_use_blocks) == 1
assert tool_use_blocks[0]["toolUse"]["name"] == tool_name
tool_spec_name = _bedrock_tools_pt(
[
{
"type": "function",
"function": {
"name": tool_name,
"description": "test",
"parameters": {"type": "object", "properties": {}},
},
}
]
)[0]["toolSpec"]["name"]
assert tool_spec_name == tool_name
def test_bedrock_tool_call_invoke_multiple_normal_tools():
"""Multiple separate tool calls (normal parallel calling) work correctly."""
tool_calls = [

54
tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
View file

@ -133,7 +133,9 @@ def test_make_disable_auto_response_message_produces_ga_shape():
"turn_detection" not in session
), "turn_detection must not be at the top-level session (beta shape); use audio.input"
# turn_detection must be nested under audio.input
assert session["audio"]["input"]["turn_detection"]["create_response"] is False
td = session["audio"]["input"]["turn_detection"]
assert td["type"] == "server_vad"
assert td["create_response"] is False
def test_make_disable_auto_response_message_produces_beta_shape_for_beta_clients():
@ -148,7 +150,55 @@ def test_make_disable_auto_response_message_produces_beta_shape_for_beta_clients
assert msg["type"] == "session.update"
session = msg["session"]
assert session == {"turn_detection": {"create_response": False}}
assert session == {
"turn_detection": {"type": "server_vad", "create_response": False}
}
@pytest.mark.asyncio
async def test_backend_to_client_send_text_receives_str_not_bytes():
client_ws = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws = MagicMock()
backend_ws.recv = AsyncMock(
side_effect=[
json.dumps({"type": "session.created", "session": {}}).encode(),
ConnectionClosed(None, None),
]
)
logging_obj = MagicMock()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
await streaming.backend_to_client_send_messages()
assert client_ws.send_text.called
sent = client_ws.send_text.call_args_list[0].args[0]
assert isinstance(sent, str)
@pytest.mark.asyncio
async def test_backend_to_client_skips_non_utf8_binary_frames():
client_ws = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws = MagicMock()
backend_ws.recv = AsyncMock(
side_effect=[
b"\xff\xfe",
json.dumps({"type": "session.created", "session": {}}).encode(),
ConnectionClosed(None, None),
]
)
logging_obj = MagicMock()
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
await streaming.backend_to_client_send_messages()
assert client_ws.send_text.call_count == 1
assert isinstance(client_ws.send_text.call_args_list[0].args[0], str)
@pytest.mark.asyncio

296
tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py
View file

@ -3642,3 +3642,299 @@ class TestOrgMCPPermissions:
user_api_key_auth=auth,
)
assert sorted(result) == ["tool_a", "tool_b"]
# ---------------------------------------------------------------------------
# LIT-3189: key unified access_group_ids extend team MCP scope
# ---------------------------------------------------------------------------
def _patch_proxy_server_globals_for_mcp():
"""Non-None mocks so the helper's None-guard doesn't short-circuit."""
return [
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch("litellm.proxy.proxy_server.user_api_key_cache", MagicMock()),
patch("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock()),
]
def _fake_mcp_access_group(
access_group_id,
access_mcp_server_ids=None,
assigned_team_ids=None,
assigned_key_ids=None,
):
from litellm.proxy._types import LiteLLM_AccessGroupTable
return LiteLLM_AccessGroupTable(
access_group_id=access_group_id,
access_group_name=access_group_id,
access_mcp_server_ids=access_mcp_server_ids or [],
assigned_team_ids=assigned_team_ids or [],
assigned_key_ids=assigned_key_ids or [],
)
def _start_patches(patches):
for p in patches:
p.start()
def _stop_patches(patches):
for p in patches:
p.stop()
@pytest.mark.asyncio
async def test_mcp_key_access_group_extras_when_team_authorized():
"""Group's assigned_team_ids includes key's team and grants an MCP server → server returned."""
valid_token = UserAPIKeyAuth(
token="test-token",
access_group_ids=["mcp-premium"],
team_id="team-a",
)
fake_ag = _fake_mcp_access_group(
access_group_id="mcp-premium",
access_mcp_server_ids=["srv-stripe"],
assigned_team_ids=["team-a"],
)
mock_mgr = MagicMock()
mock_mgr.expand_permission_list.side_effect = lambda x: list(x)
patches = _patch_proxy_server_globals_for_mcp() + [
patch(
"litellm.proxy.auth.auth_checks.get_access_object",
new_callable=AsyncMock,
return_value=fake_ag,
),
patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager",
mock_mgr,
),
]
_start_patches(patches)
try:
result = await MCPRequestHandler._get_key_access_group_mcp_server_extras(
valid_token
)
assert result == ["srv-stripe"]
finally:
_stop_patches(patches)
@pytest.mark.asyncio
async def test_mcp_key_access_group_extras_when_key_directly_authorized():
"""Group's assigned_key_ids includes the key's token → server returned (per-key auth)."""
valid_token = UserAPIKeyAuth(
token="test-token-hashed",
access_group_ids=["mcp-per-key"],
team_id="team-a",
)
fake_ag = _fake_mcp_access_group(
access_group_id="mcp-per-key",
access_mcp_server_ids=["srv-stripe"],
assigned_team_ids=[],
assigned_key_ids=["test-token-hashed"],
)
mock_mgr = MagicMock()
mock_mgr.expand_permission_list.side_effect = lambda x: list(x)
patches = _patch_proxy_server_globals_for_mcp() + [
patch(
"litellm.proxy.auth.auth_checks.get_access_object",
new_callable=AsyncMock,
return_value=fake_ag,
),
patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager",
mock_mgr,
),
]
_start_patches(patches)
try:
result = await MCPRequestHandler._get_key_access_group_mcp_server_extras(
valid_token
)
assert result == ["srv-stripe"]
finally:
_stop_patches(patches)
@pytest.mark.asyncio
async def test_mcp_key_access_group_extras_when_key_has_no_groups():
"""Empty access_group_ids → no extras, no DB read."""
valid_token = UserAPIKeyAuth(
token="test-token",
access_group_ids=[],
team_id="team-a",
)
result = await MCPRequestHandler._get_key_access_group_mcp_server_extras(
valid_token
)
assert result == []
@pytest.mark.asyncio
async def test_mcp_key_access_group_extras_when_group_has_no_servers():
"""Group authorizes the team but its access_mcp_server_ids is empty → no extras."""
valid_token = UserAPIKeyAuth(
token="test-token",
access_group_ids=["mcp-empty"],
team_id="team-a",
)
fake_ag = _fake_mcp_access_group(
access_group_id="mcp-empty",
access_mcp_server_ids=[],
assigned_team_ids=["team-a"],
)
patches = _patch_proxy_server_globals_for_mcp() + [
patch(
"litellm.proxy.auth.auth_checks.get_access_object",
new_callable=AsyncMock,
return_value=fake_ag,
),
]
_start_patches(patches)
try:
result = await MCPRequestHandler._get_key_access_group_mcp_server_extras(
valid_token
)
assert result == []
finally:
_stop_patches(patches)
@pytest.mark.asyncio
async def test_mcp_key_access_group_extras_when_group_authorizes_neither():
"""
Escalation regression: team member attaches a foreign access group to their key.
Group grants servers BUT assigned_team_ids/assigned_key_ids exclude this caller.
No extras contributed.
"""
valid_token = UserAPIKeyAuth(
token="team-a-token",
access_group_ids=["team-b-mcp-group"],
team_id="team-a",
)
fake_ag = _fake_mcp_access_group(
access_group_id="team-b-mcp-group",
access_mcp_server_ids=["srv-finance-only"],
assigned_team_ids=["team-b"],
assigned_key_ids=["team-b-token"],
)
patches = _patch_proxy_server_globals_for_mcp() + [
patch(
"litellm.proxy.auth.auth_checks.get_access_object",
new_callable=AsyncMock,
return_value=fake_ag,
),
]
_start_patches(patches)
try:
result = await MCPRequestHandler._get_key_access_group_mcp_server_extras(
valid_token
)
assert result == []
finally:
_stop_patches(patches)
@pytest.mark.asyncio
async def test_mcp_key_access_group_extras_when_get_access_object_raises():
"""Group lookup failure is treated as no authorization (does not crash)."""
valid_token = UserAPIKeyAuth(
token="test-token",
access_group_ids=["missing-mcp-group"],
team_id="team-a",
)
patches = _patch_proxy_server_globals_for_mcp() + [
patch(
"litellm.proxy.auth.auth_checks.get_access_object",
new_callable=AsyncMock,
side_effect=Exception("not found"),
),
]
_start_patches(patches)
try:
result = await MCPRequestHandler._get_key_access_group_mcp_server_extras(
valid_token
)
assert result == []
finally:
_stop_patches(patches)
@pytest.mark.asyncio
async def test_get_allowed_mcp_servers_unions_key_access_group_extras():
"""End-to-end: team has [srv-team], key access group grants [srv-extra] → both in final list.
Without this fix [srv-extra] would be intersected away because the team doesn't list it.
"""
auth = UserAPIKeyAuth(
token="test-token",
api_key="test-key",
team_id="team-a",
access_group_ids=["mcp-extra-group"],
)
with (
patch.object(
MCPRequestHandler,
"_get_allowed_mcp_servers_for_key",
new_callable=AsyncMock,
return_value=[],
),
patch.object(
MCPRequestHandler,
"_get_allowed_mcp_servers_for_team",
new_callable=AsyncMock,
return_value=["srv-team"],
),
patch.object(
MCPRequestHandler,
"_get_key_access_group_mcp_server_extras",
new_callable=AsyncMock,
return_value=["srv-extra"],
),
):
result = await MCPRequestHandler.get_allowed_mcp_servers(auth)
assert sorted(result) == ["srv-extra", "srv-team"]
@pytest.mark.asyncio
async def test_get_allowed_mcp_servers_no_union_when_no_authorized_extras():
"""End-to-end: no authorized extras → behavior identical to today (team ceiling enforced)."""
auth = UserAPIKeyAuth(
token="test-token",
api_key="test-key",
team_id="team-a",
access_group_ids=["mcp-foreign-group"],
)
with (
patch.object(
MCPRequestHandler,
"_get_allowed_mcp_servers_for_key",
new_callable=AsyncMock,
return_value=["srv-key-only"],
),
patch.object(
MCPRequestHandler,
"_get_allowed_mcp_servers_for_team",
new_callable=AsyncMock,
return_value=["srv-team"],
),
patch.object(
MCPRequestHandler,
"_get_key_access_group_mcp_server_extras",
new_callable=AsyncMock,
return_value=[],
),
):
# key ∩ team = {} (no overlap), extras = [] → final = []
result = await MCPRequestHandler.get_allowed_mcp_servers(auth)
assert result == []

210
tests/test_litellm/proxy/_experimental/mcp_server/test_callback_oauth_error_responses.py Normal file
View file

@ -0,0 +1,210 @@
"""Regression tests for LIT-2750.
The MCP OAuth ``/callback`` endpoint must handle IdP error responses
(e.g. ``?error=access_denied``) gracefully instead of returning a 422
because ``code`` and ``state`` were declared as required FastAPI query
params. Per RFC 6749 §4.1.2.1 the IdP redirects to the configured
redirect URI with ``error`` / ``error_description`` / ``error_uri``
query params and no ``code`` when the user denies access.
These tests cover both the propagate-to-client path (when state decodes
to a trusted ``redirect_uri``) and the in-page fallback (when state is
missing, undecryptable, or carries an untrusted redirect_uri). They also
pin the success path (``code`` + ``state``) against accidental
regressions.
"""
import pytest
@pytest.fixture(autouse=True)
def _mock_mcp_client_ip():
"""Bypass IP-based access control for the in-process TestClient.
Mirrors the autouse fixture in ``test_discoverable_endpoints.py`` so
these tests don't require a real client IP context.
"""
from unittest.mock import patch
with patch(
"litellm.proxy._experimental.mcp_server.discoverable_endpoints.IPAddressUtils.get_mcp_client_ip",
return_value=None,
):
yield
@pytest.fixture
def callback_test_client(monkeypatch):
"""FastAPI TestClient mounted with the MCP discoverable router.
Sets a deterministic ``LITELLM_SALT_KEY`` so encoded states minted
in-test can be decrypted by the handler.
"""
from fastapi import FastAPI
from fastapi.testclient import TestClient
monkeypatch.setenv("LITELLM_SALT_KEY", "sk-test-salt-for-LIT-2750")
from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
router,
)
app = FastAPI()
app.include_router(router)
return TestClient(app)
class TestCallbackOAuthErrorResponses:
"""LIT-2750: IdP error responses to ``/callback`` must not 422."""
def test_idp_error_with_no_state_returns_400_html(self, callback_test_client):
"""Pre-fix: 422 Pydantic. Post-fix: 400 HTML with the IdP's error."""
resp = callback_test_client.get(
"/callback",
params={
"error": "access_denied",
"error_description": "User declined access",
},
follow_redirects=False,
)
assert resp.status_code == 400
assert "text/html" in resp.headers["content-type"]
body = resp.text
assert "access_denied" in body
assert "User declined access" in body
# Sanity: must not leak the Pydantic validation error.
assert "Field required" not in body
def test_idp_error_html_escapes_user_controlled_fields(
self, callback_test_client
):
"""A malicious IdP must not be able to inject HTML/JS via error params."""
resp = callback_test_client.get(
"/callback",
params={
"error": "<script>alert(1)</script>",
"error_description": "<img src=x onerror=alert(2)>",
},
follow_redirects=False,
)
assert resp.status_code == 400
body = resp.text
# Raw tags must be escaped, not present verbatim.
assert "<script>alert(1)</script>" not in body
assert "<img src=x onerror=alert(2)>" not in body
assert "&lt;script&gt;alert(1)&lt;/script&gt;" in body
def test_idp_error_with_trusted_state_propagates_to_client_redirect_uri(
self, callback_test_client
):
"""When state decodes to a trusted (loopback) redirect_uri, propagate
the error back so the MCP client's OAuth library can surface it
instead of timing out waiting on the loopback."""
from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
encode_state_with_base_url,
)
state = encode_state_with_base_url(
base_url="http://localhost:3000/",
original_state="client-original-state-xyz",
client_redirect_uri="http://127.0.0.1:60108/callback",
)
resp = callback_test_client.get(
"/callback",
params={
"error": "access_denied",
"error_description": "User declined access",
"state": state,
},
follow_redirects=False,
)
assert resp.status_code == 302
location = resp.headers["location"]
assert location.startswith("http://127.0.0.1:60108/callback?")
assert "error=access_denied" in location
# Original client state must be round-tripped, not our wrapped state.
assert "state=client-original-state-xyz" in location
# error_description percent-encoded but present.
assert "error_description=User" in location
# Wrapped/encrypted state must NOT leak to the client.
assert state not in location
def test_idp_error_with_untrusted_redirect_uri_does_not_open_redirect(
self, callback_test_client
):
"""If the state minted earlier carries a redirect_uri that the proxy
no longer trusts, we must surface the error inline rather than
302-ing to an attacker-controlled URL (open-redirect)."""
from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
encode_state_with_base_url,
)
state = encode_state_with_base_url(
base_url="http://localhost:3000/",
original_state="x",
client_redirect_uri="https://attacker.example.com/steal",
)
resp = callback_test_client.get(
"/callback",
params={"error": "access_denied", "state": state},
follow_redirects=False,
)
# Must not 3xx — open redirect would defeat the redirect_uri allowlist.
assert resp.status_code == 400
assert "attacker.example.com" not in resp.headers.get("location", "")
assert "access_denied" in resp.text
def test_idp_error_with_undecryptable_state_falls_back_to_html(
self, callback_test_client
):
resp = callback_test_client.get(
"/callback",
params={
"error": "server_error",
"error_description": "boom",
"state": "not-a-valid-encrypted-state",
},
follow_redirects=False,
)
assert resp.status_code == 400
assert "server_error" in resp.text
assert "boom" in resp.text
def test_bare_callback_with_no_params_returns_400_not_422(
self, callback_test_client
):
"""An SSO redirect chain that drops the original /authorize query
params should land on a human-readable 400, not a Pydantic 422."""
resp = callback_test_client.get("/callback", follow_redirects=False)
assert resp.status_code == 400
assert "invalid_request" in resp.text
assert "Field required" not in resp.text
def test_success_path_still_redirects_with_code_and_state(
self, callback_test_client
):
"""Regression: the successful (``code``+``state``) flow must still
redirect back to the trusted client redirect_uri with the original
state preserved."""
from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
encode_state_with_base_url,
)
state = encode_state_with_base_url(
base_url="http://localhost:3000/",
original_state="orig-state-success",
client_redirect_uri="http://127.0.0.1:60108/callback",
)
resp = callback_test_client.get(
"/callback",
params={"code": "auth-code-abc", "state": state},
follow_redirects=False,
)
assert resp.status_code == 302
location = resp.headers["location"]
assert location.startswith("http://127.0.0.1:60108/callback?")
assert "code=auth-code-abc" in location
assert "state=orig-state-success" in location

2
tests/test_litellm/proxy/auth/test_auth_utils.py
View file

@ -1644,6 +1644,7 @@ class TestObservabilityCallbackBans:
"braintrust_api_key",
"braintrust_project",
"phoenix_project_name",
"phoenix_project_name_override",
"wandb_api_key",
"weave_project_id",
"gcs_bucket_name",
@ -1675,6 +1676,7 @@ class TestObservabilityCallbackBans:
"posthog_api_url",
"braintrust_project",
"phoenix_project_name",
"phoenix_project_name_override",
],
)
def test_observability_field_in_metadata_dict_is_rejected(

47
tests/test_litellm/proxy/common_utils/test_callback_utils.py
View file

@ -8,11 +8,14 @@ sys.path.insert(
) # Adds the parent directory to the system path
from litellm.proxy.common_utils.callback_utils import (
add_policy_to_applied_policies_header,
decrypt_callback_vars,
encrypt_callback_vars,
get_logging_caching_headers,
initialize_callbacks_on_proxy,
get_remaining_tokens_and_requests_from_request_data,
normalize_callback_names,
sanitize_openai_provider_metadata,
)
import litellm
@ -92,6 +95,50 @@ def test_normalize_callback_names_lowercases_strings():
]
def test_add_policy_to_applied_policies_header_uses_litellm_metadata_bucket():
request_data = {
"input_file_id": "file-abc123",
"litellm_metadata": {},
}
add_policy_to_applied_policies_header(
request_data=request_data, policy_name="global-baseline"
)
assert request_data["litellm_metadata"]["applied_policies"] == ["global-baseline"]
assert "applied_policies" not in request_data.get("metadata", {})
def test_sanitize_openai_provider_metadata_strips_internal_tracking_fields():
metadata = {
"customer_id": "cust-123",
"applied_policies": ["global-baseline"],
"applied_guardrails": ["pii_blocker"],
"note": 42,
}
sanitized = sanitize_openai_provider_metadata(metadata)
assert sanitized == {"customer_id": "cust-123"}
def test_get_logging_caching_headers_merges_metadata_and_litellm_metadata():
request_data = {
"metadata": {"customer_id": "cust-123"},
"litellm_metadata": {
"applied_policies": ["global-baseline"],
"applied_guardrails": ["pii_blocker"],
"policy_sources": {"global-baseline": "team_default"},
},
}
headers = get_logging_caching_headers(request_data)
assert headers["x-litellm-applied-policies"] == "global-baseline"
assert headers["x-litellm-applied-guardrails"] == "pii_blocker"
assert headers["x-litellm-policy-sources"] == "global-baseline=team_default"
def test_initialize_callbacks_on_proxy_instantiates_compression_interception(
monkeypatch,
):

71
tests/test_litellm/proxy/test_batch_expiry.py
View file

@ -178,6 +178,77 @@ class TestBatchEndpointTeamOverride:
assert kwargs["output_expires_after"] == TEAM_EXPIRY
class TestBatchEndpointPolicyMetadata:
"""Batch create must not forward LiteLLM policy tracking via OpenAI metadata."""
def test_create_batch_does_not_forward_applied_policies_metadata(
self, monkeypatch, llm_router
):
from litellm.proxy.policy_engine.attachment_registry import (
get_attachment_registry,
)
from litellm.proxy.policy_engine.policy_registry import get_policy_registry
from litellm.types.proxy.policy_engine import (
Policy,
PolicyAttachment,
PolicyGuardrails,
)
policy_registry = get_policy_registry()
policy_registry._policies = {
"global-baseline": Policy(
guardrails=PolicyGuardrails(add=["pii_blocker"]),
),
}
policy_registry._initialized = True
attachment_registry = get_attachment_registry()
attachment_registry._attachments = [
PolicyAttachment(policy="global-baseline", scope="*"),
]
attachment_registry._initialized = True
_setup_proxy(monkeypatch, llm_router)
user_key = UserAPIKeyAuth(
api_key="test-key",
team_alias="batch-team",
key_alias="batch-key",
)
app.dependency_overrides[user_api_key_auth] = lambda: user_key
captured_kwargs = {}
async def mock_acreate_batch(**kwargs):
captured_kwargs.update(kwargs)
return _make_batch_response()
monkeypatch.setattr(litellm, "acreate_batch", mock_acreate_batch)
try:
response = client.post(
"/v1/batches",
json={
"input_file_id": "file-abc123",
"endpoint": "/v1/chat/completions",
"completion_window": "24h",
},
headers={"Authorization": "Bearer test-key"},
)
assert response.status_code == 200
finally:
app.dependency_overrides.clear()
policy_registry._policies = {}
policy_registry._initialized = False
attachment_registry._attachments = []
attachment_registry._initialized = False
assert captured_kwargs.get("metadata") in (None, {})
assert (
"global-baseline" in captured_kwargs["litellm_metadata"]["applied_policies"]
)
class TestBatchEndpointTeamValidation:
"""Verify validation errors for malformed team metadata on batch endpoint."""

50
tests/test_litellm/test_cost_calculator.py
View file

@ -13,6 +13,7 @@ from pydantic import BaseModel
import litellm
from litellm.cost_calculator import (
completion_cost,
cost_per_token,
handle_realtime_stream_cost_calculation,
response_cost_calculator,
)
@ -21,6 +22,55 @@ from litellm.types.utils import ModelResponse, PromptTokensDetailsWrapper, Usage
from litellm.utils import TranscriptionResponse
def test_cost_per_token_duplicate_openai_prefix_matches_model_cost(monkeypatch):
"""
Router/proxy configs may use deployment ids like openai/openai/<model>. Cost lookup must
resolve to model_prices keys (e.g. gpt-5.5), not fail or multiply prefixes.
"""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
prompt_usd, completion_usd = cost_per_token(
model="openai/openai/gpt-5.5",
prompt_tokens=100,
completion_tokens=50,
custom_llm_provider="openai",
)
assert prompt_usd + completion_usd > 0
def test_cost_per_token_non_string_model_does_not_hang():
"""
The provider-prefix dedup loop must not spin forever when `model` is a
non-string object (e.g. a MagicMock from a mocked transport). It should
return or raise promptly instead of looping on a truthy `.startswith()`.
"""
import threading
from unittest.mock import MagicMock
result: dict = {}
def _run():
try:
cost_per_token(
model=MagicMock(),
prompt_tokens=10,
completion_tokens=5,
custom_llm_provider="anthropic",
)
result["status"] = "returned"
except Exception:
result["status"] = "raised"
worker = threading.Thread(target=_run, daemon=True)
worker.start()
worker.join(timeout=10)
assert not worker.is_alive(), "cost_per_token hung on a non-string model"
assert result.get("status") in ("returned", "raised")
def test_completion_cost_uses_response_model_for_dynamic_routing():
"""
Test that completion_cost uses the model from the response object

74
tests/test_litellm/test_thinking_enabled.py Normal file
View file

@ -0,0 +1,74 @@
"""
Unit tests for is_thinking_enabled method in BaseConfig.
Tests the fix for issue #28576: handle None thinking param without crashing.
"""
import pytest
from litellm.llms.base_llm.chat.transformation import BaseConfig
class TestIsThinkingEnabled:
"""Test is_thinking_enabled handles various thinking parameter values."""
@pytest.fixture
def transformer(self):
"""Create a BaseConfig instance for testing."""
# BaseConfig is abstract, so we create a minimal concrete subclass
class ConcreteConfig(BaseConfig):
def __init__(self):
pass
def get_complete_url(self, *args, **kwargs):
return ""
def validate_environment(self, *args, **kwargs):
return {}
def transform_request(self, *args, **kwargs):
return {}, {}
def transform_response(self, *args, **kwargs):
return None
def get_supported_openai_params(self, model: str):
return []
def map_openai_params(self, *args, **kwargs):
return {}
def get_error_class(self, *args, **kwargs):
from litellm.llms.base_llm.chat.transformation import BaseLLMException
return BaseLLMException(500, "test error")
return ConcreteConfig()
@pytest.mark.parametrize(
"non_default_params,expected",
[
# thinking=None should not crash, returns False
({"thinking": None}, False),
# thinking={'type': 'enabled'} returns True
({"thinking": {"type": "enabled"}}, True),
# thinking key missing returns False
({}, False),
# thinking={} returns False
({"thinking": {}}, False),
# thinking with different type returns False
({"thinking": {"type": "disabled"}}, False),
# reasoning_effort present returns True
({"reasoning_effort": "medium"}, True),
# both thinking enabled and reasoning_effort returns True
({"thinking": {"type": "enabled"}, "reasoning_effort": "high"}, True),
# falsy thinking values should not crash
({"thinking": False}, False),
({"thinking": 0}, False),
({"thinking": ""}, False),
],
)
def test_is_thinking_enabled(self, transformer, non_default_params, expected):
"""Test is_thinking_enabled with various parameter combinations."""
result = transformer.is_thinking_enabled(non_default_params)
assert result == expected, (
f"Expected {expected} for params {non_default_params}, got {result}"
)