scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 200

Merge commit '9d8fe1ddb5' into litellm_gemini_tool_call_id_length

Browse source 
# Conflicts:
#	litellm/responses/litellm_completion_transformation/session_handler.py
#	litellm/responses/litellm_completion_transformation/streaming_iterator.py
#	litellm/responses/litellm_completion_transformation/transformation.py
This commit is contained in:
mateo-berri 2026-10-21 05:33:46 -07:00
commit 10d8fd29c0
548 changed files with 21949 additions and 4752 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.circleci/config.yml
View file

@ -1291,7 +1291,7 @@ jobs:
name: Install MCP SDK1 peer
command: |
uv venv --python 3.12 .venv-mcp-peer
uv pip install --python .venv-mcp-peer 'mcp==1.28.1' 'langchain-mcp-adapters==0.2.1'
uv pip install --no-config --exclude-newer "3 days" --python .venv-mcp-peer 'mcp==1.28.1' 'langchain-mcp-adapters==0.2.1'
echo "export MCP_TEST_PEER_PYTHON=$PWD/.venv-mcp-peer/bin/python" >> "$BASH_ENV"
- when:
condition:

2
.circleci/scripts/classify_changes.sh
View file

@ -33,7 +33,7 @@ while IFS= read -r file || [ -n "$file" ]; do
*) has_backend=true ;;
esac
case "$file" in
model_prices_and_context_window.json | litellm/model_prices_and_context_window_backup.json | model_prices_and_context_window.schema.json)
model_prices_and_context_window.json | litellm/model_prices_and_context_window_backup.json)
has_cost_map=true ;;
tests/unit/proxy/*) : ;;
*) outside_cost_map_set=true ;;

2
.github/workflows/test-unit.yml vendored
View file

@ -216,10 +216,12 @@ jobs:
- shard: llms-providers
test-path: |-
tests/unit/llms/test_api_base_trailing_slash_routes.py
tests/unit/llms/test_cache_control_and_reasoning.py
tests/unit/llms/test_custom_llm.py
tests/unit/llms/test_file_content_block.py
tests/unit/llms/test_file_search_responses.py
tests/unit/llms/test_get_complete_url_trailing_slash.py
tests/unit/llms/test_lifecycle_fix.py
tests/unit/llms/test_oss_decision.py
tests/unit/llms/test_polling_url_origin_match.py

6
ci_cd/cost_map_guard.py
View file

@ -4,8 +4,7 @@ Every pull request whose diff against its merge base touches one of the two cost
checks: the files parse and the backup copy matches the root file. A pull request that leaves both untouched
skips them, since merging it keeps the base branch's copies and its head tree only carries whatever state the
branch was cut from. Pull requests from the cost map sync bot (branches named litellm_cost_map_sync_*) always
get the file checks and additionally may only touch those two files or the legacy schema
path and may only add or update models.
get the file checks and additionally may only touch those two files and may only add or update models.
"""
from __future__ import annotations
@ -21,7 +20,6 @@ from typing import Final
COST_MAP_PATH: Final = "model_prices_and_context_window.json"
BACKUP_PATH: Final = "litellm/model_prices_and_context_window_backup.json"
LEGACY_SCHEMA_PATH: Final = "model_prices_and_context_window.schema.json"
GUARDED_PATHS: Final = (COST_MAP_PATH, BACKUP_PATH)
BOT_BRANCH_PREFIX: Final = "litellm_cost_map_sync_"
SPECIAL_ROOT_KEYS: Final = frozenset({"sample_spec", "fallback_generalizations"})
@ -110,7 +108,7 @@ def _bot_failures(base: Snapshot, head_map: CostMap, changed_files: Sequence[str
*(
f"bot PRs may only change the cost map files, not {path}"
for path in changed_files
if path not in GUARDED_PATHS and path != LEGACY_SCHEMA_PATH
if path not in GUARDED_PATHS
),
*(f"bot PRs may not remove models: {key}" for key in base_map if key not in head_map),
*(f"bot PRs may not remove fields: {ref}" for ref in removed_fields),

10
litellm/caching/in_memory_cache.py
View file

@ -95,7 +95,10 @@ class InMemoryCache(BaseCache):
"""
Check if a specific key is expired
"""
return key in self.ttl_dict and self._clock() > self.ttl_dict[key]
try:
return self._clock() > self.ttl_dict[key]
except KeyError:
return False
def _cache_keys(self) -> tuple[str, ...]:
return tuple(self.cache_dict) # pyright: ignore[reportUnknownVariableType] # cache_dict is untyped legacy state
@ -227,7 +230,10 @@ class InMemoryCache(BaseCache):
if key in self.cache_dict:
if self.evict_element_if_expired(key):
return None
original_cached_response: Final = self.cache_dict[key]
try:
original_cached_response: Final = self.cache_dict[key]
except KeyError:
return None
try:
cached_response = json.loads(original_cached_response)
except Exception:

4
litellm/constants.py
View file

@ -156,6 +156,9 @@ DEFAULT_REPLICATE_POLLING_RETRIES: Final = get_env_int("DEFAULT_REPLICATE_POLLIN
DEFAULT_REPLICATE_POLLING_DELAY_SECONDS: Final = get_env_int("DEFAULT_REPLICATE_POLLING_DELAY_SECONDS", 1)
DEFAULT_IMAGE_TOKEN_COUNT: Final = get_env_int("DEFAULT_IMAGE_TOKEN_COUNT", 250)
HF_CONFIG_FETCH_TIMEOUT_SECONDS: Final = 10.0
HUGGINGFACE_TOKENIZER_LOAD_RETRY_SECONDS: Final = get_env_int_in_range(
"HUGGINGFACE_TOKENIZER_LOAD_RETRY_SECONDS", 300, 1, 86400
)
# Maximum wall-clock seconds a streaming response is allowed to run.
# Streams exceeding this duration are terminated with a Timeout error.
@ -608,6 +611,7 @@ NON_LLM_CONNECTION_TIMEOUT: Final = get_env_int(
) # timeout for adjacent services (e.g. jwt auth)
MAX_EXCEPTION_MESSAGE_LENGTH: Final = get_env_int("MAX_EXCEPTION_MESSAGE_LENGTH", 2000)
MAX_STRING_LENGTH_PROMPT_IN_DB: Final = get_env_int("MAX_STRING_LENGTH_PROMPT_IN_DB", 2048)
MAX_STRING_LENGTH_COMPACTION_SUMMARY_IN_DB: Final = 64 * 1024
BEDROCK_MAX_POLICY_SIZE: Final = get_env_int("BEDROCK_MAX_POLICY_SIZE", 75)
# One entry per distinct AWS credential-argument set. Per-user cost attribution passes the attributed
# identity as aws_session_name, so this bounds how many attributed identities keep a cached STS session.

6
litellm/cost_calculator.py
View file

@ -1673,8 +1673,10 @@ def completion_cost(
verbose_logger.debug("selected model name for cost calculation: %s", model)
if completion_response is not None and (
isinstance(completion_response, BaseModel) or isinstance(completion_response, dict)
): # tts returns a custom class
isinstance(completion_response, (BaseModel, dict))
or isinstance(completion_response, HttpxBinaryResponseContent)
and completion_response.usage is not None
):
if isinstance(completion_response, dict):
usage_obj: dict | Usage | None = completion_response.get("usage", {})
else:

1
litellm/endpoints/speech/speech_to_completion_bridge/handler.py
View file

@ -122,6 +122,7 @@ class SpeechToCompletionBridgeHandler:
if isinstance(result, ModelResponse):
return self.transformation_handler.transform_response(
model_response=result,
model=model,
response_format=requested_response_format if isinstance(requested_response_format, str) else None,
)
else:

67
litellm/endpoints/speech/speech_to_completion_bridge/transformation.py
View file

@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Final, cast
from typing_extensions import NotRequired, ReadOnly, TypedDict
from litellm.constants import OPENAI_CHAT_COMPLETION_PARAMS
from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
from litellm.litellm_core_utils.hidden_params import get_hidden_params, served_from_cache, set_hidden_param
if TYPE_CHECKING:
@ -38,6 +39,11 @@ class ChatAudioParam(TypedDict):
format: ReadOnly[NotRequired[str]]
class ChatSpeechConfigAudioParam(TypedDict):
speech_config: ReadOnly[Mapping[str, object]]
format: ReadOnly[str]
class SpeechToCompletionBridgeTransformationHandler:
def _validate_response_format(
self, model: str, custom_llm_provider: str, optional_params: Mapping[str, object]
@ -49,6 +55,20 @@ class SpeechToCompletionBridgeTransformationHandler:
return
from litellm.exceptions import BadRequestError
if custom_llm_provider.startswith("vertex_ai"):
from litellm.llms.vertex_ai.text_to_speech.transformation import VertexAITextToSpeechConfig
vertex_supported: Final = ", ".join(
sorted({*GEMINI_TTS_SUPPORTED_RESPONSE_FORMATS, *VertexAITextToSpeechConfig.GEMINI_FORMAT_MAPPINGS})
)
raise BadRequestError(
message=(
f"Vertex AI Gemini TTS does not support response_format='{response_format}'."
f" Supported response formats: {vertex_supported}."
),
model=model,
llm_provider=custom_llm_provider,
)
supported: Final = ", ".join(sorted(GEMINI_TTS_SUPPORTED_RESPONSE_FORMATS))
raise BadRequestError(
message=(
@ -75,8 +95,27 @@ class SpeechToCompletionBridgeTransformationHandler:
return response_format if isinstance(response_format, str) else None
def _chat_audio_param(
self, model: str, voice: str | Mapping[str, object] | None, optional_params: Mapping[str, object]
) -> ChatAudioParam | None:
self,
model: str,
voice: str | Mapping[str, object] | None,
optional_params: Mapping[str, object],
drop_params: bool,
custom_llm_provider: str,
) -> ChatAudioParam | ChatSpeechConfigAudioParam | None:
import litellm
if isinstance(voice, Mapping) and self._is_gemini_tts_model(litellm.model_alias_map.get(model, model)):
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
normalize_gemini_speech_config,
reject_unknown_gemini_tts_voice_keys,
)
reject_unknown_gemini_tts_voice_keys(model, voice, drop_params, custom_llm_provider)
speech_audio: Final[ChatSpeechConfigAudioParam] = {
"speech_config": normalize_gemini_speech_config(voice, drops_unsupported_params(drop_params)),
"format": GEMINI_TTS_CHAT_AUDIO_FORMAT,
}
return speech_audio
if not isinstance(voice, str):
return None
audio_format: Final = self._chat_audio_format(model, optional_params)
@ -90,9 +129,9 @@ class SpeechToCompletionBridgeTransformationHandler:
self,
model: str,
input: str,
voice: str | dict | None,
optional_params: dict,
litellm_params: dict,
voice: str | Mapping[str, object] | None,
optional_params: Mapping[str, object],
litellm_params: Mapping[str, object],
headers: dict,
litellm_logging_obj: "LiteLLMLoggingObj",
custom_llm_provider: str,
@ -104,7 +143,13 @@ class SpeechToCompletionBridgeTransformationHandler:
"messages": [user_message],
"modalities": ["audio"],
**self._chat_completion_params(optional_params),
"audio": self._chat_audio_param(model, voice, optional_params),
"audio": self._chat_audio_param(
model,
voice,
optional_params,
drops_unsupported_params(litellm_params.get("drop_params")),
custom_llm_provider,
),
**litellm_params,
"headers": headers,
"litellm_logging_obj": litellm_logging_obj,
@ -153,8 +198,9 @@ class SpeechToCompletionBridgeTransformationHandler:
return wav_header + pcm_data
def _is_gemini_tts_model(self, model: str) -> bool:
"""Check if the model is a Gemini TTS model that returns PCM16 data."""
return "gemini" in model.lower() and ("tts" in model.lower() or "preview-tts" in model.lower())
from litellm.utils import is_gemini_tts_model
return is_gemini_tts_model(model)
def _gemini_tts_response_body(self, decoded_audio: bytes, response_format: str | None) -> tuple[bytes, str]:
if response_format == GEMINI_TTS_RAW_RESPONSE_FORMAT:
@ -162,7 +208,7 @@ class SpeechToCompletionBridgeTransformationHandler:
return self._convert_pcm16_to_wav(decoded_audio), "audio/wav"
def transform_response(
self, model_response: "ModelResponse", response_format: str | None
self, model_response: "ModelResponse", model: str, response_format: str | None
) -> "HttpxBinaryResponseContent":
import base64
@ -176,10 +222,9 @@ class SpeechToCompletionBridgeTransformationHandler:
raise ValueError("No audio part found in the response")
decoded_audio: Final = base64.b64decode(audio_part.data)
model: Final = getattr(model_response, "model", "")
content, content_type = (
self._gemini_tts_response_body(decoded_audio, response_format)
if self._is_gemini_tts_model(model)
if self._is_gemini_tts_model(model) or self._is_gemini_tts_model(model_response.model or "")
else (decoded_audio, "audio/mpeg")
)
response: Final = httpx.Response(

18
litellm/harness/endpoint.py
View file

@ -37,6 +37,7 @@ from litellm.harness.context import GatewayTarget
from litellm.harness.errors import HarnessError, HarnessInstallFailed
from litellm.harness.types import Harness, Usage
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.types.litellm_params import STREAM_USAGE_INJECTED_FIELD
from litellm.types.llms.custom_http import httpxSpecialProvider
if TYPE_CHECKING:
@ -690,18 +691,21 @@ class ModelEndpoint:
async def _invoke_sdk(
self,
route: str,
kwargs: dict[str, Any], # mutable-ok: injects stream_options into the SDK kwargs
kwargs: dict[str, Any], # mutable-ok: SDK call kwargs built by _sdk_kwargs
) -> object:
if route == ROUTE_MESSAGES:
return await litellm.anthropic.messages.acreate(**kwargs)
if route == ROUTE_CHAT:
if kwargs.get("stream"):
stream_options = kwargs.get("stream_options") or {} # mutable-ok: empty default for a JSON field
kwargs["stream_options"] = { # mutable-ok: JSON field sent to litellm.acompletion
"include_usage": True,
**stream_options,
if not kwargs.get("stream"):
return await litellm.acompletion(**kwargs)
stream_options = kwargs.get("stream_options") or {} # mutable-ok: empty default for a JSON field
return await litellm.acompletion(
**{
**kwargs,
"stream_options": {"include_usage": True, **stream_options},
STREAM_USAGE_INJECTED_FIELD: "include_usage" not in stream_options,
}
return await litellm.acompletion(**kwargs)
)
return await litellm.aresponses(**kwargs)
async def _call_sdk(self, route: str, body: Mapping[str, object]) -> Response:

4
litellm/integrations/deepeval/types.py
View file

@ -1,6 +1,6 @@
# Duplicate -> https://github.com/confident-ai/deepeval/blob/main/deepeval/tracing/api.py
from enum import Enum
from typing import Any, ClassVar, Literal
from typing import ClassVar, Literal
from pydantic import ConfigDict, Field
@ -55,7 +55,7 @@ class TraceApi(LiteLLMBaseModel):
tool_spans: list[BaseApiSpan] = Field(alias="toolSpans")
start_time: str = Field(alias="startTime")
end_time: str = Field(alias="endTime")
metadata: dict[str, Any] | None = Field(None)
metadata: dict[str, object] | None = Field(None)
tags: list[str] | None = Field(None)
environment: str | None = Field(None)

6
litellm/interactions/litellm_responses_transformation/streaming_iterator.py
View file

@ -3,8 +3,8 @@ Streaming iterator for transforming Responses API stream to Interactions API str
"""
from collections import deque
from collections.abc import AsyncIterator, Iterator
from typing import Any, Final, cast
from collections.abc import AsyncIterator, Iterator, Mapping
from typing import Final, cast
from litellm.responses.streaming_iterator import (
BaseResponsesAPIStreamingIterator,
@ -44,7 +44,7 @@ class LiteLLMResponsesInteractionsStreamingIterator:
request_input: InteractionInput | None,
optional_params: InteractionsAPIOptionalRequestParams,
custom_llm_provider: str | None = None,
litellm_metadata: dict[str, Any] | None = None,
litellm_metadata: Mapping[str, object] | None = None,
):
self.model = model
self.responses_stream_iterator = litellm_custom_stream_wrapper

7
litellm/litellm_core_utils/core_helpers.py
View file

@ -329,11 +329,14 @@ def add_missing_spend_metadata_to_litellm_metadata(litellm_metadata: dict, metad
Helper to get litellm metadata for spend tracking
PATCH for issue where both `litellm_metadata` and `metadata` are present in the kwargs
and user_api_key values are in 'metadata'.
and user_api_key values are in 'metadata'. The proxy receive stamp rides along for the same
reason: a guardrail hook can plant `litellm_metadata` on a chat request after the proxy stamped
`metadata`, and the response timing headers anchor on that stamp.
"""
potential_spend_tracking_metadata_substring: Final = "user_api_key"
for key, value in dict(metadata).items():
if potential_spend_tracking_metadata_substring in key:
carry_receive_stamp = key == "litellm_received_at" and key not in litellm_metadata
if potential_spend_tracking_metadata_substring in key or carry_receive_stamp:
litellm_metadata[key] = value
return litellm_metadata

169
litellm/litellm_core_utils/litellm_logging.py
View file

@ -12,7 +12,7 @@ import subprocess
import sys
import time
import traceback
from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping, Sequence
from collections.abc import Awaitable, Callable, Coroutine, Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime as dt_object
from functools import lru_cache
@ -138,6 +138,7 @@ from litellm.types.interactions import (
InteractionsAPIResponse,
InteractionsAPIStreamingResponse,
)
from litellm.types.litellm_params import STREAM_USAGE_INJECTED_FIELD
from litellm.types.llms.openai import (
AllMessageValues,
Batch,
@ -574,6 +575,27 @@ class UncachedPrice:
cost_breakdown: CostBreakdown | None
@dataclass(frozen=True, slots=True)
class BilledRoundsSpend:
usage: Usage
cost: float
cost_breakdown: CostBreakdown | None
@dataclass(frozen=True, slots=True)
class BilledRoundsDispatch:
"""A parked success log for a stream bridged through chat completions that billed several rounds, and the spend
its failure log bills when a post-call hook rejects the response, as it does for a rejected one-round bridged
stream."""
dispatch: Coroutine[object, object, None]
spend: BilledRoundsSpend
def _fixed_margin(cost_breakdown: CostBreakdown | None) -> float:
return 0.0 if cost_breakdown is None else cost_breakdown.get("margin_fixed_amount", 0.0)
def _timestamp_seconds(moment: object) -> float | None:
if isinstance(moment, datetime.datetime):
return moment.timestamp()
@ -582,6 +604,10 @@ def _timestamp_seconds(moment: object) -> float | None:
return None
def same_attempt_marker(first: object, second: object) -> bool:
return first is second or first == second
class Logging(LiteLLMLoggingBaseClass):
global \
supabaseClient, \
@ -595,6 +621,7 @@ class Logging(LiteLLMLoggingBaseClass):
prometheusLogger, \
slack_app
custom_pricing: bool = False
replaced_global_callback_keys: frozenset[str] = frozenset()
stream_options = None
litellm_request_debug: bool = False
streamed_anthropic_message_id: str | None = None
@ -602,6 +629,7 @@ class Logging(LiteLLMLoggingBaseClass):
classifier_input: Mapping[str, JsonValue] | None = None
baseline_cache_context: "BaselineCacheContext | None" = None
baseline_observation: "CapturedBaselineObservation | None" = None
_on_failure_payload_built: "Callable[[], None] | None" = None
@property
def _defer_async_logging(self) -> bool:
@ -649,11 +677,11 @@ class Logging(LiteLLMLoggingBaseClass):
litellm_call_id: str,
function_id: str,
litellm_trace_id: str | None = None,
dynamic_input_callbacks: list[str | Callable | CustomLogger] | None = None,
dynamic_input_callbacks: list[str | Callable[..., object] | CustomLogger] | None = None,
dynamic_success_callbacks: list[str | Callable[..., object] | CustomLogger] | None = None,
dynamic_async_success_callbacks: list[str | Callable[..., object] | CustomLogger] | None = None,
dynamic_failure_callbacks: list[str | Callable | CustomLogger] | None = None,
dynamic_async_failure_callbacks: list[str | Callable | CustomLogger] | None = None,
dynamic_failure_callbacks: list[str | Callable[..., object] | CustomLogger] | None = None,
dynamic_async_failure_callbacks: list[str | Callable[..., object] | CustomLogger] | None = None,
applied_guardrails: list[str] | None = None,
kwargs: dict | None = None,
log_raw_request_response: bool = False,
@ -698,6 +726,7 @@ class Logging(LiteLLMLoggingBaseClass):
self._pre_call_session_id: str = session_id_var.get()
_sid: Final = kwargs.get("litellm_session_id") if kwargs else None
self.litellm_session_id: str = str(_sid) if _sid else ""
self.stream_usage_injected: Final = kwargs is not None and kwargs.get(STREAM_USAGE_INJECTED_FIELD) is True
# supports_correlation_logging is False for calls originating from the
# sync client entry point (wrapper() in utils.py): a plain OS thread
# has no per-call context isolation the way an asyncio Task does, and
@ -726,15 +755,17 @@ class Logging(LiteLLMLoggingBaseClass):
self.raw_request_only = raw_request_only
# Initialize dynamic callbacks
self.dynamic_input_callbacks: list[str | Callable | CustomLogger] | None = dynamic_input_callbacks
self.dynamic_input_callbacks: list[str | Callable[..., object] | CustomLogger] | None = dynamic_input_callbacks
self.dynamic_success_callbacks: list[str | Callable[..., object] | CustomLogger] | None = (
dynamic_success_callbacks
)
self.dynamic_async_success_callbacks: list[str | Callable[..., object] | CustomLogger] | None = (
dynamic_async_success_callbacks
)
self.dynamic_failure_callbacks: list[str | Callable | CustomLogger] | None = dynamic_failure_callbacks
self.dynamic_async_failure_callbacks: list[str | Callable | CustomLogger] | None = (
self.dynamic_failure_callbacks: list[str | Callable[..., object] | CustomLogger] | None = (
dynamic_failure_callbacks
)
self.dynamic_async_failure_callbacks: list[str | Callable[..., object] | CustomLogger] | None = (
dynamic_async_failure_callbacks
)
@ -826,6 +857,7 @@ class Logging(LiteLLMLoggingBaseClass):
"litellm_call_id": litellm_call_id,
"start_time": start_time,
"first_api_call_start_time": None,
"current_attempt_api_call_start_time": None,
"litellm_params": litellm_params,
}
new_call.streaming_chunks = []
@ -851,10 +883,32 @@ class Logging(LiteLLMLoggingBaseClass):
self.dynamic_async_failure_callbacks, callback
)
def replace_global_callback(self, callback: CustomLogger) -> None:
self.add_dynamic_callback(callback)
self.replaced_global_callback_keys = self.replaced_global_callback_keys | {
litellm.logging_callback_manager.get_custom_logger_key(callback)
}
def add_dynamic_function_callbacks(
self,
*,
async_success: Callable[..., object],
failure: Callable[..., object],
async_failure: Callable[..., object],
) -> None:
self.dynamic_async_success_callbacks = self._with_dynamic_callback(
self.dynamic_async_success_callbacks, async_success
)
self.dynamic_failure_callbacks = self._with_dynamic_callback(self.dynamic_failure_callbacks, failure)
self.dynamic_async_failure_callbacks = self._with_dynamic_callback(
self.dynamic_async_failure_callbacks, async_failure
)
@staticmethod
def _with_dynamic_callback(
callbacks: Sequence[str | Callable | CustomLogger] | None, callback: CustomLogger
) -> list[str | Callable | CustomLogger]:
callbacks: Sequence[str | Callable[..., object] | CustomLogger] | None,
callback: Callable[..., object] | CustomLogger,
) -> list[str | Callable[..., object] | CustomLogger]:
existing: Final = tuple(callbacks or ())
return [*existing, *(() if callback in existing else (callback,))]
@ -891,9 +945,9 @@ class Logging(LiteLLMLoggingBaseClass):
def _process_dynamic_callback_list(
self,
callback_list: list[str | Callable | CustomLogger] | None,
callback_list: list[str | Callable[..., object] | CustomLogger] | None,
dynamic_callbacks_type: Literal["input", "success", "failure", "async_success", "async_failure"],
) -> list[str | Callable | CustomLogger] | None:
) -> list[str | Callable[..., object] | CustomLogger] | None:
"""
Helper function to initialize CustomLogger compatible callbacks in self.dynamic_* callbacks
@ -905,7 +959,7 @@ class Logging(LiteLLMLoggingBaseClass):
if callback_list is None:
return None
processed_list: Final[list[str | Callable | CustomLogger]] = []
processed_list: Final[list[str | Callable[..., object] | CustomLogger]] = []
for callback in callback_list:
if isinstance(callback, str) and callback in litellm._known_custom_logger_compatible_callbacks:
for callback_instance in self._resolve_dynamic_callback_string(callback):
@ -1554,7 +1608,9 @@ class Logging(LiteLLMLoggingBaseClass):
# A proxy-gate failure (auth, rate limit, client disconnect) fires pre_call for its callbacks
# without any provider handoff, so it must not stamp the handoff instants.
if LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL not in self.model_call_details:
self.model_call_details["api_call_start_time"] = datetime.datetime.now()
api_call_start_time: Final = datetime.datetime.now()
self.model_call_details["api_call_start_time"] = api_call_start_time
self.model_call_details["current_attempt_api_call_start_time"] = api_call_start_time
# Set-once first provider-handoff instant. api_call_start_time
# is overwritten on every retry, so it can't measure one-time
# preprocessing; pinning the first attempt excludes retry loops
@ -1562,11 +1618,12 @@ class Logging(LiteLLMLoggingBaseClass):
# litellm_params["metadata"] (caller request metadata, typed
# Dict[str, str], echoed downstream; a datetime breaks it).
if self.model_call_details.get("first_api_call_start_time") is None:
self.model_call_details["first_api_call_start_time"] = self.model_call_details[
"api_call_start_time"
]
self.model_call_details["first_api_call_start_time"] = api_call_start_time
# Input Integration Logging -> If you want to log the fact that an attempt to call the model was made
callbacks: Final = litellm.input_callback + (self.dynamic_input_callbacks or [])
callbacks: Final = (
*self._without_replaced_global_callbacks(litellm.input_callback),
*(self.dynamic_input_callbacks or ()),
)
for callback in callbacks:
try:
if callback == "supabase" and supabaseClient is not None:
@ -1739,7 +1796,10 @@ class Logging(LiteLLMLoggingBaseClass):
)
# Input Integration Logging -> If you want to log the fact that an attempt to call the model was made
callbacks: Final = litellm.input_callback + (self.dynamic_input_callbacks or [])
callbacks: Final = (
*self._without_replaced_global_callbacks(litellm.input_callback),
*(self.dynamic_input_callbacks or ()),
)
for callback in callbacks:
try:
if callback == "sentry" and add_breadcrumb:
@ -2388,6 +2448,31 @@ class Logging(LiteLLMLoggingBaseClass):
self.model_call_details["combined_usage_object"] = usage
self.model_call_details["response_cost"] = response_cost
def record_failure_after_earlier_rounds(
self, earlier: BilledRoundsSpend, failed_usage: Usage, failed_cost: float
) -> None:
"""Bill a failed round on the failure log after the rounds of its request before it were billed on the success
log.
The fixed margin is a per-request fee, so it is charged once: the failed round's margin is dropped when the
earlier rounds already paid one."""
from litellm.responses.mcp.round_billing import breakdown_less_repeated_fixed_margin
failed_breakdown: Final = None if self.cost_breakdown is earlier.cost_breakdown else self.cost_breakdown
repeated_fixed_margin: Final = (
0.0
if failed_breakdown is None
else min(_fixed_margin(earlier.cost_breakdown), _fixed_margin(failed_breakdown), earlier.cost, failed_cost)
)
self.model_call_details["combined_usage_object"] = failed_usage
self.model_call_details["response_cost"] = failed_cost - repeated_fixed_margin
self.cost_breakdown = breakdown_less_repeated_fixed_margin(failed_breakdown, repeated_fixed_margin)
def record_billed_rounds_for_failure(self, spend: BilledRoundsSpend) -> None:
"""Bill every round of a fully streamed request on the failure log when a post-call hook rejects it."""
self.record_partial_usage_for_failure(spend.usage, spend.cost)
self.cost_breakdown = spend.cost_breakdown
def attempt_scoped_copy(self) -> "Logging":
"""A copy with its own model_call_details, so logging one attempt of a request through it leaves
the state the request's other attempts share untouched."""
@ -2395,12 +2480,37 @@ class Logging(LiteLLMLoggingBaseClass):
attempt.model_call_details = dict(self.model_call_details)
return attempt
def dispatch_attempt_failure_handlers(
self, exception: Exception, traceback_exception: str
) -> Coroutine[None, None, None]:
shared_payload: Final = self.model_call_details.get("standard_logging_object")
attempt_marker: Final = self.model_call_details.get("current_attempt_api_call_start_time")
attempt: Final = self.attempt_scoped_copy()
def publish() -> None:
attempt_payload: Final = attempt.model_call_details.get("standard_logging_object")
current_marker: Final = self.model_call_details.get("current_attempt_api_call_start_time")
if (
attempt_payload is not None
and self.model_call_details.get("standard_logging_object") is shared_payload
and same_attempt_marker(current_marker, attempt_marker)
):
self.model_call_details["standard_logging_object"] = attempt_payload
attempt._on_failure_payload_built = publish
async def dispatch() -> None:
await attempt.dispatch_failure_handlers(exception, traceback_exception, prefer_async_handlers=True)
return dispatch()
def carry_failed_attempt_usage(self) -> None:
"""Move what a failed earlier attempt stashed off the retry sharing this logging object, so the retry logs its
own usage while the request's failure log still bills every attempt."""
usage: Final = self.model_call_details.pop("combined_usage_object", None)
response_cost: Final = self.model_call_details.pop("response_cost", None)
if isinstance(usage, Usage):
self.model_call_details["current_attempt_api_call_start_time"] = None
self.model_call_details["failed_attempts_usage"] = (
*self.model_call_details.get("failed_attempts_usage", ()),
(usage, float(response_cost) if isinstance(response_cost, (int, float)) else 0.0),
@ -2514,6 +2624,10 @@ class Logging(LiteLLMLoggingBaseClass):
"""
return
self.model_call_details[f"has_logged_{event_type}"] = True
if event_type == "async_failure":
self.model_call_details["async_failure_logged_attempt"] = self.model_call_details.get(
"current_attempt_api_call_start_time"
)
return
def has_run_logging(
@ -4331,7 +4445,8 @@ class Logging(LiteLLMLoggingBaseClass):
start_time=start_time,
end_time=end_time,
)
if self._on_failure_payload_built is not None:
self._on_failure_payload_built()
callbacks: Final = self.get_combined_callback_list(
dynamic_success_callbacks=self.dynamic_async_failure_callbacks,
global_callbacks=litellm._async_failure_callback,
@ -4516,14 +4631,26 @@ class Logging(LiteLLMLoggingBaseClass):
_filtered_failure_callbacks = self._remove_internal_litellm_callbacks(_filtered_failure_callbacks)
return len(_filtered_failure_callbacks) > 0
def _without_replaced_global_callbacks(
self, global_callbacks: Iterable[str | Callable[..., object] | CustomLogger]
) -> tuple[str | Callable[..., object] | CustomLogger, ...]:
if not self.replaced_global_callback_keys:
return tuple(global_callbacks)
return tuple(
litellm.logging_callback_manager.without_custom_logger_keys(
global_callbacks, self.replaced_global_callback_keys
)
)
def get_combined_callback_list(
self,
dynamic_success_callbacks: list[str | Callable[..., object] | CustomLogger] | None,
global_callbacks: list[str | Callable[..., object] | CustomLogger],
) -> list[str | Callable[..., object] | CustomLogger]:
kept_global_callbacks: Final = self._without_replaced_global_callbacks(global_callbacks)
if dynamic_success_callbacks is None:
return list(global_callbacks)
return list(dict.fromkeys(dynamic_success_callbacks + global_callbacks))
return list(kept_global_callbacks)
return list(dict.fromkeys((*dynamic_success_callbacks, *kept_global_callbacks)))
def _remove_internal_litellm_callbacks(self, callbacks: list) -> list:
"""

149
litellm/litellm_core_utils/logging_callback_manager.py
View file

@ -1,4 +1,6 @@
from collections.abc import Callable
import threading
from collections.abc import Callable, Iterable, Iterator
from types import MethodType
from typing import TYPE_CHECKING, Final
import litellm
@ -15,6 +17,7 @@ else:
_custom_logger_compatible_callbacks_literal = str
_generic_api_logger_cache: Final[dict[str, GenericAPILogger]] = {}
_callback_lists_lock: Final = threading.RLock()
class LoggingCallbackManager:
@ -43,16 +46,18 @@ class LoggingCallbackManager:
Add a input callback to litellm.input_callback.
Auto-routes async callbacks to litellm._async_input_callback.
"""
if not isinstance(callback, str) and self._is_async_callable(callback):
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_input_callback)
else:
self._safe_add_callback_to_list(callback=callback, parent_list=litellm.input_callback)
with _callback_lists_lock:
if not isinstance(callback, str) and self._is_async_callable(callback):
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_input_callback)
else:
self._safe_add_callback_to_list(callback=callback, parent_list=litellm.input_callback)
def add_litellm_service_callback(self, callback: CustomLogger | str | Callable[..., object]):
"""
Add a service callback to litellm.service_callback
"""
self._safe_add_callback_to_list(callback=callback, parent_list=litellm.service_callback)
with _callback_lists_lock:
self._safe_add_callback_to_list(callback=callback, parent_list=litellm.service_callback)
def add_litellm_callback(self, callback: CustomLogger | str | Callable[..., object]):
"""
@ -60,10 +65,11 @@ class LoggingCallbackManager:
Ensures no duplicates are added.
"""
self._safe_add_callback_to_list(
callback=callback,
parent_list=litellm.callbacks,
)
with _callback_lists_lock:
self._safe_add_callback_to_list(
callback=callback,
parent_list=litellm.callbacks,
)
def add_litellm_success_callback(self, callback: CustomLogger | str | Callable[..., object]):
"""
@ -71,37 +77,41 @@ class LoggingCallbackManager:
Auto-routes async callbacks to litellm._async_success_callback.
Special-cases 'dynamodb' and 'openmeter' as async callbacks.
"""
if (
isinstance(callback, str)
and callback in ("dynamodb", "openmeter")
or not isinstance(callback, str)
and self._is_async_callable(callback)
):
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_success_callback)
else:
self._safe_add_callback_to_list(callback=callback, parent_list=litellm.success_callback)
with _callback_lists_lock:
if (
isinstance(callback, str)
and callback in ("dynamodb", "openmeter")
or not isinstance(callback, str)
and self._is_async_callable(callback)
):
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_success_callback)
else:
self._safe_add_callback_to_list(callback=callback, parent_list=litellm.success_callback)
def add_litellm_failure_callback(self, callback: CustomLogger | str | Callable[..., object]):
"""
Add a failure callback to `litellm.failure_callback`.
Auto-routes async callbacks to litellm._async_failure_callback.
"""
if not isinstance(callback, str) and self._is_async_callable(callback):
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_failure_callback)
else:
self._safe_add_callback_to_list(callback=callback, parent_list=litellm.failure_callback)
with _callback_lists_lock:
if not isinstance(callback, str) and self._is_async_callable(callback):
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_failure_callback)
else:
self._safe_add_callback_to_list(callback=callback, parent_list=litellm.failure_callback)
def add_litellm_async_success_callback(self, callback: CustomLogger | Callable[..., object] | str):
"""
Add a success callback to litellm._async_success_callback
"""
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_success_callback)
with _callback_lists_lock:
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_success_callback)
def add_litellm_async_failure_callback(self, callback: CustomLogger | Callable[..., object] | str):
"""
Add a failure callback to litellm._async_failure_callback
"""
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_failure_callback)
with _callback_lists_lock:
self._safe_add_callback_to_list(callback=callback, parent_list=litellm._async_failure_callback)
def remove_callback_from_list_by_object(self, callback_list, obj, require_self=True):
"""
@ -110,13 +120,14 @@ class LoggingCallbackManager:
if not isinstance(callback_list, list): # Not list -> do nothing
return
if require_self:
remove_list = [c for c in callback_list if hasattr(c, "__self__") and c.__self__ == obj]
else:
remove_list = [c for c in callback_list if c == obj]
with _callback_lists_lock:
if require_self:
remove_list = [c for c in callback_list if hasattr(c, "__self__") and c.__self__ == obj]
else:
remove_list = [c for c in callback_list if c == obj]
for c in remove_list:
callback_list.remove(c)
for c in remove_list:
callback_list.remove(c)
def remove_callbacks_by_type(self, callback_list, callback_type):
"""
@ -134,10 +145,11 @@ class LoggingCallbackManager:
if not isinstance(callback_list, list):
return
remove_list: Final = [c for c in callback_list if isinstance(c, callback_type)]
with _callback_lists_lock:
remove_list: Final = [c for c in callback_list if isinstance(c, callback_type)]
for c in remove_list:
callback_list.remove(c)
for c in remove_list:
callback_list.remove(c)
def _add_string_callback_to_list(
self, callback: str, parent_list: list[CustomLogger | Callable[..., object] | str]
@ -336,6 +348,55 @@ class LoggingCallbackManager:
return "-".join(key_parts)
get_custom_logger_key = _get_custom_logger_key
callback_lists_lock: Final = _callback_lists_lock
def drop_callbacks(self, callbacks: Iterable[object], method_owners: Iterable[object] = ()) -> frozenset[str]:
dropped_ids: Final = frozenset(map(id, callbacks))
owner_ids: Final = frozenset(map(id, method_owners))
def keep(callback: object) -> bool:
if id(callback) in dropped_ids:
return False
return not (isinstance(callback, MethodType) and id(callback.__self__) in owner_ids)
with _callback_lists_lock:
vacated_keys: Final = frozenset(
self._get_custom_logger_key(callback)
for callback in litellm.callbacks
if isinstance(callback, CustomLogger) and id(callback) in dropped_ids
)
litellm.callbacks = [callback for callback in litellm.callbacks if keep(callback)]
litellm.input_callback = [callback for callback in litellm.input_callback if keep(callback)]
litellm.success_callback = [callback for callback in litellm.success_callback if keep(callback)]
litellm.failure_callback = [callback for callback in litellm.failure_callback if keep(callback)]
litellm.service_callback = [callback for callback in litellm.service_callback if keep(callback)]
litellm._async_success_callback = [
callback for callback in litellm._async_success_callback if keep(callback)
]
litellm._async_failure_callback = [
callback for callback in litellm._async_failure_callback if keep(callback)
]
return vacated_keys
def with_custom_logger_keys(
self, candidates: Iterable[CustomLogger], keys: frozenset[str]
) -> tuple[CustomLogger, ...]:
if not keys:
return ()
class_names: Final = frozenset(key.split("-", 1)[0] for key in keys)
same_class: Final = (candidate for candidate in candidates if type(candidate).__name__ in class_names)
return tuple(candidate for candidate in same_class if self._get_custom_logger_key(candidate) in keys)
def without_custom_logger_keys(
self, callbacks: Iterable[str | Callable[..., object] | CustomLogger], keys: frozenset[str]
) -> Iterator[str | Callable[..., object] | CustomLogger]:
return (
callback
for callback in callbacks
if not (isinstance(callback, CustomLogger) and self._get_custom_logger_key(callback) in keys)
)
def _reset_all_callbacks(self):
"""
Reset all callbacks to an empty list
@ -363,19 +424,21 @@ class LoggingCallbackManager:
_get_all_callbacks = get_all_callbacks
def remove_callback_from_all_lists(self, obj, require_self=False) -> None:
def remove_callback_from_all_lists(self, obj: object, require_self: bool = False) -> None:
"""
Remove a callback object from every callback list it may have been
promoted into, so a re-initialized callback leaves no stale instance behind.
"""
for callback_list in (
litellm.callbacks,
litellm.success_callback,
litellm.failure_callback,
litellm._async_success_callback,
litellm._async_failure_callback,
):
self.remove_callback_from_list_by_object(callback_list, obj, require_self=require_self)
with _callback_lists_lock:
for callback_list in (
litellm.callbacks,
litellm.input_callback,
litellm.success_callback,
litellm.failure_callback,
litellm._async_success_callback,
litellm._async_failure_callback,
):
self.remove_callback_from_list_by_object(callback_list, obj, require_self=require_self)
def get_active_additional_logging_utils_from_custom_logger(
self,

57
litellm/litellm_core_utils/redact_messages.py
View file

@ -24,6 +24,7 @@ from litellm.litellm_core_utils.core_helpers import (
get_metadata_variable_name_from_kwargs,
)
from litellm.litellm_core_utils.served_output_texts import SERVED_OUTPUT_TEXTS_KEY
from litellm.llms.base_llm.guardrail_translation.utils import stream_item_field, stream_item_items
from litellm.llms.vertex_ai.common_utils import (
redact_vertex_ai_metadata_from_litellm_params,
redact_vertex_ai_metadata_from_logged_object,
@ -41,6 +42,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObject = Any
MESSAGE_REDACTION_ENABLE_HEADERS: Final = ("litellm-enable-message-redaction", "x-litellm-enable-message-redaction")
def _opted_out_of_message_logging(custom_logger: CustomLogger) -> bool:
return hasattr(custom_logger, "message_logging") and custom_logger.message_logging is not True
@ -195,33 +198,45 @@ def _redacted_provider_specific_fields(fields: Mapping[str, object]) -> Mapping[
return {
**fields,
**({"reasoning_content": REDACTED_BY_LITELLM} if fields.get("reasoning_content") is not None else {}),
**{key: None for key in ("thinking_blocks", "reasoningContentBlocks", "citations") if key in fields},
**{
key: None
for key in (
"thinking_blocks",
"reasoningContentBlocks",
"citations",
"compaction_blocks",
"compaction_delta",
)
if key in fields
},
}
def _redact_field_when_present(item: object, field: str) -> None:
if getattr(item, field, None) is not None:
setattr(item, field, REDACTED_BY_LITELLM)
def _redact_responses_api_output(output_items):
"""Helper to redact ResponsesAPIResponse output items."""
for output_item in output_items:
output_type = stream_item_field(output_item, "type")
if getattr(output_item, "text", None) is not None:
output_item.text = REDACTED_BY_LITELLM
if hasattr(output_item, "content") and isinstance(output_item.content, list):
for content_part in output_item.content:
if getattr(content_part, "text", None) is not None:
content_part.text = REDACTED_BY_LITELLM
if getattr(content_part, "refusal", None) is not None:
content_part.refusal = REDACTED_BY_LITELLM
for content_part in stream_item_items(output_item, "content"):
_redact_field_when_present(content_part, "text")
_redact_field_when_present(content_part, "refusal")
# Redact reasoning items in output array
if hasattr(output_item, "type") and output_item.type == "reasoning":
if hasattr(output_item, "summary") and isinstance(output_item.summary, list):
for summary_item in output_item.summary:
if getattr(summary_item, "text", None) is not None:
summary_item.text = REDACTED_BY_LITELLM
if output_type == "reasoning":
for summary_item in stream_item_items(output_item, "summary"):
_redact_field_when_present(summary_item, "text")
if hasattr(output_item, "type") and output_item.type == "function_call" and hasattr(output_item, "arguments"):
if output_type == "compaction" and hasattr(output_item, "encrypted_content"):
output_item.encrypted_content = REDACTED_BY_LITELLM
if output_type == "function_call" and hasattr(output_item, "arguments"):
output_item.arguments = REDACTED_BY_LITELLM
if hasattr(output_item, "type") and output_item.type == "custom_tool_call" and hasattr(output_item, "input"):
if output_type == "custom_tool_call" and hasattr(output_item, "input"):
output_item.input = REDACTED_BY_LITELLM
@ -262,6 +277,11 @@ def _redacted_responses_api_output_item(item: object, redacted_str: str) -> obje
if item_mapping.get("type") == "reasoning" and isinstance(summary, list)
else {}
),
**(
{"encrypted_content": redacted_str}
if item_mapping.get("type") == "compaction" and "encrypted_content" in item_mapping
else {}
),
**(
{"arguments": redacted_str}
if item_mapping.get("type") == "function_call" and "arguments" in item_mapping
@ -556,13 +576,8 @@ def should_redact_message_logging(model_call_details: dict) -> bool:
# User explicitly disabled redaction via header
return False
possible_enable_headers: Final = [
"litellm-enable-message-redaction", # old header. maintain backwards compatibility
"x-litellm-enable-message-redaction", # new header
]
is_redaction_enabled_via_header = False
for header in possible_enable_headers:
for header in MESSAGE_REDACTION_ENABLE_HEADERS:
if bool(request_headers.get(header, False)):
is_redaction_enabled_via_header = True
break

117
litellm/litellm_core_utils/streaming_handler.py
View file

@ -10,7 +10,7 @@ import traceback
from collections.abc import AsyncIterator, Callable, Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import Any, Final, NoReturn, Protocol, TypeVar, cast
from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypeVar, cast
import anyio
import httpx
@ -60,6 +60,9 @@ from .exception_mapping_utils import exception_type, with_deployment_api_base
from .llm_response_utils.get_api_base import get_api_base
from .rules import Rules
if TYPE_CHECKING:
from .litellm_logging import BilledRoundsSpend
# Constants for special delta attribute names
AUDIO_ATTRIBUTE: Final = "audio"
IMAGE_ATTRIBUTE: Final = "images"
@ -259,6 +262,8 @@ def _provider_hidden_params(
class CustomStreamWrapper:
_earlier_rounds_spend: "BilledRoundsSpend | None" = None
@property
def hidden_params(self) -> dict[str, object]: # mutable-ok: API requires mutation
return self._hidden_params
@ -342,6 +347,10 @@ class CustomStreamWrapper:
self.messages = getattr(logging_obj, "messages", None)
self.sent_stream_usage = False
self.send_stream_usage = True if self.check_send_stream_usage(self.stream_options) else False
self.caller_requested_stream_usage: Final = (
self.check_send_stream_usage(getattr(logging_obj, "stream_options", None))
and getattr(logging_obj, "stream_usage_injected", False) is not True
)
self.tool_call = False
# keep track of the returned chunks - used for calculating the input/output tokens for stream options
self.chunks: list[ModelResponseStream] = []
@ -2070,7 +2079,7 @@ class CustomStreamWrapper:
except StopIteration:
if self.sent_last_chunk is True:
complete_streaming_response: Final = self._build_complete_streaming_response()
response = self.model_response_creator()
response = self._new_usage_chunk()
if complete_streaming_response is not None:
self._propagate_usage_cost_to_hidden_params(complete_streaming_response, self.custom_llm_provider)
@ -2222,27 +2231,10 @@ class CustomStreamWrapper:
)
if _has_usage:
# Store a copy ONLY when usage stripping below will mutate
# the chunk. For non-usage chunks (vast majority), store
# directly to avoid expensive model_copy() per chunk.
self.chunks.append(processed_chunk.model_copy())
# Strip usage from the outgoing chunk so it's not sent twice
# (once in the chunk, once in _hidden_params).
obj_dict = processed_chunk.model_dump()
if "usage" in obj_dict:
del obj_dict["usage"]
processed_chunk = self.model_response_creator(
chunk=obj_dict,
hidden_params=cast( # cast-ok: preserve mapping operations on dynamic storage
Mapping[str, object], getattr(processed_chunk, HIDDEN_PARAMS_ATTR)
),
)
is_empty = is_model_response_stream_empty(
model_response=cast(ModelResponseStream, processed_chunk)
)
if is_empty:
usage_free_chunk = self._record_and_strip_usage_chunk(processed_chunk)
if usage_free_chunk is None:
continue
processed_chunk = usage_free_chunk
else:
# No usage data — safe to store directly without copying
self.chunks.append(processed_chunk)
@ -2273,25 +2265,25 @@ class CustomStreamWrapper:
if processed_chunk is None:
continue
choice = processed_chunk.choices[0]
if isinstance(choice, StreamingChoices):
self._accumulate_streamed_delta(choice.delta)
if processed_chunk.choices:
choice = processed_chunk.choices[0]
if isinstance(choice, StreamingChoices):
self._accumulate_streamed_delta(choice.delta)
self.rules.post_call_rules(input=self.response_uptil_now, model=self.model)
# RETURN RESULT
self.chunks.append(processed_chunk)
return processed_chunk
if not self.send_stream_usage or getattr(processed_chunk, "usage", None) is None:
self.chunks.append(processed_chunk)
return processed_chunk
usage_free_chunk = self._record_and_strip_usage_chunk(processed_chunk)
if usage_free_chunk is None:
continue
return usage_free_chunk
except (StopAsyncIteration, StopIteration):
return await self._finalize_completed_stream(cache_hit=cache_hit)
except httpx.TimeoutException as e: # if httpx read timeout error occues
traceback_exception = traceback.format_exc()
## ADD DEBUG INFORMATION - E.G. LITELLM REQUEST TIMEOUT
traceback_exception += f"\nLiteLLM Default Request Timeout - {litellm.request_timeout}"
if self.logging_obj is not None:
self._record_partial_usage_for_failure()
## LOGGING
self._failure_logging = asyncio.create_task(
self.logging_obj.dispatch_failure_handlers(e, traceback_exception, prefer_async_handlers=True)
)
self._dispatch_stream_failure(e, traceback_exception)
await self.aclose()
self._handle_stream_fallback_error(e)
except (httpx.ReadError, httpx.RemoteProtocolError) as e:
@ -2306,7 +2298,7 @@ class CustomStreamWrapper:
async def _finalize_completed_stream(self, cache_hit: bool) -> "ModelResponseStream":
if self.sent_last_chunk is True:
complete_streaming_response: Final = await self._async_build_complete_streaming_response()
response: Final = self.model_response_creator()
response: Final = self._new_usage_chunk()
if complete_streaming_response is not None:
self._propagate_usage_cost_to_hidden_params(complete_streaming_response, self.custom_llm_provider)
@ -2392,29 +2384,36 @@ class CustomStreamWrapper:
return processed_chunk
def _log_stream_failure_and_raise(self, e: Exception) -> NoReturn:
traceback_exception: Final = traceback.format_exc()
if self.logging_obj is not None:
self._record_partial_usage_for_failure()
## LOGGING
self._failure_logging = asyncio.create_task(
self.logging_obj.dispatch_failure_handlers(e, traceback_exception, prefer_async_handlers=True)
)
self._dispatch_stream_failure(e, traceback.format_exc())
self._handle_stream_fallback_error(e)
async def wait_for_failure_logging(self, timeout: float) -> None:
def _dispatch_stream_failure(self, e: Exception, traceback_exception: str) -> None:
self._record_partial_usage_for_failure()
self._failure_logging = asyncio.create_task(
self.logging_obj.dispatch_attempt_failure_handlers(e, traceback_exception)
)
@property
def has_dispatched_failure_logging(self) -> bool:
return self._failure_logging is not None
async def wait_for_failure_logging(self, timeout: float | None) -> None:
"""Let this stream's failure log read its own deployment before a fallback reuses the logging object."""
if self._failure_logging is None:
return
await asyncio.wait((self._failure_logging,), timeout=timeout)
def bill_failure_after_earlier_rounds(self, earlier: "BilledRoundsSpend") -> None:
self._earlier_rounds_spend = earlier
def _record_partial_usage_for_failure(self) -> None:
"""
A stream that breaks mid-flight still billed the provider for the chunks
already delivered. Recover that partial usage from the chunks seen so
far and stash it, with its cost, on the logging object so the failure
handler records the real partial spend instead of zero. A request that
later recovers via a router fallback overwrites this with the combined
success log on the same request id, so this never double counts.
handler records the real partial spend instead of zero. The router
moves it into terminal failure usage before opening the next attempt,
so a recovered attempt logs its own usage.
"""
if self.logging_obj is None or not self.chunks:
return
@ -2433,10 +2432,12 @@ class CustomStreamWrapper:
if self.model:
partial_response.model = self.model
backfill_missing_cache_usage_fields(usage)
self.logging_obj.model_call_details["combined_usage_object"] = usage
self.logging_obj.model_call_details["response_cost"] = (
self.logging_obj.response_cost_calculator(result=partial_response) or 0.0
)
partial_cost: Final = self.logging_obj.response_cost_calculator(result=partial_response) or 0.0
if self._earlier_rounds_spend is None:
self.logging_obj.model_call_details["combined_usage_object"] = usage
self.logging_obj.model_call_details["response_cost"] = partial_cost
return
self.logging_obj.record_failure_after_earlier_rounds(self._earlier_rounds_spend, usage, partial_cost)
except Exception as recover_error:
verbose_logger.debug(
"could not recover partial usage for interrupted stream: %s",
@ -2463,6 +2464,22 @@ class CustomStreamWrapper:
return False
return any(get(field) for field in cls._CONTINUATION_DISQUALIFYING_DELTA_FIELDS)
def _new_usage_chunk(self) -> ModelResponseStream:
usage_chunk: Final = self.model_response_creator()
if self.caller_requested_stream_usage:
usage_chunk.choices = []
return usage_chunk
def _record_and_strip_usage_chunk(self, chunk: ModelResponseStream) -> ModelResponseStream | None:
self.chunks.append(chunk.model_copy())
usage_free_chunk: Final = self.model_response_creator(
chunk={key: value for key, value in chunk.model_dump().items() if key != "usage"},
hidden_params=cast( # cast-ok: preserve mapping operations on dynamic storage
Mapping[str, object], getattr(chunk, HIDDEN_PARAMS_ATTR)
),
)
return None if is_model_response_stream_empty(model_response=usage_free_chunk) else usage_free_chunk
def _accumulate_streamed_delta(self, delta: object) -> None:
# Shared by the sync, async, and non-aiohttp iteration sites so answer
# text and the disqualifying-content latch stay in step across all three.

24
litellm/litellm_core_utils/url_utils.py
View file

@ -23,7 +23,7 @@ import asyncio
import socket
from ipaddress import ip_address, ip_network
from typing import Any, Final, Protocol
from urllib.parse import quote, urlparse, urlunparse
from urllib.parse import quote, urlparse, urlsplit, urlunparse
import httpx
from typing_extensions import ReadOnly, TypedDict
@ -129,6 +129,28 @@ def encode_url_path_segments(value: object, *, field_name: str = "path") -> str:
return "/".join(encode_url_path_segment(segment, field_name=field_name) for segment in value_str.split("/"))
def _strip_trailing_slashes_keeping_the_scheme(api_base: str) -> str:
stripped: Final = api_base.rstrip("/")
parts: Final = urlsplit(stripped)
if parts.scheme and not (parts.netloc or parts.path or parts.query or parts.fragment):
return api_base
if not parts.query or not parts.path.endswith("/"):
return stripped
url_before_query, _, query = stripped.partition("?")
return f"{url_before_query.rstrip('/')}?{query}"
def join_api_base(api_base: str, route: str) -> str:
return f"{_strip_trailing_slashes_keeping_the_scheme(api_base)}/{route.lstrip('/')}"
def ensure_api_route(api_base: str, route: str) -> str:
base: Final = _strip_trailing_slashes_keeping_the_scheme(api_base)
if urlsplit(base).path.endswith(f"/{route.lstrip('/')}"):
return base
return join_api_base(base, route)
def _is_blocked_ip(addr: str) -> bool:
"""Return True for any IP not safe to reach from a user-supplied URL.

3
litellm/llms/aiml/image_generation/transformation.py
View file

@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
)
@ -121,7 +122,7 @@ class AimlImageGenerationConfig(BaseImageGenerationConfig):
complete_url = complete_url.rstrip("/")
# Strip /v1 suffix if present since IMAGE_GENERATION_ENDPOINT already includes v1
complete_url = complete_url.removesuffix("/v1")
complete_url = f"{complete_url}/{self.IMAGE_GENERATION_ENDPOINT}"
complete_url = join_api_base(complete_url, self.IMAGE_GENERATION_ENDPOINT)
return complete_url
def validate_environment(

8
litellm/llms/aiohttp_openai/chat/transformation.py
View file

@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, Any, Final
from aiohttp import ClientResponse
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.openai_like.chat.transformation import OpenAILikeChatConfig
from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import Choices, ModelResponse
@ -44,12 +45,7 @@ class AiohttpOpenAIChatConfig(OpenAILikeChatConfig):
Ensure - /v1/chat/completions is at the end of the url
"""
if api_base is None:
api_base = "https://api.openai.com"
if not api_base.endswith("/chat/completions"):
api_base += "/chat/completions"
return api_base
return ensure_api_route("https://api.openai.com" if api_base is None else api_base, "chat/completions")
def validate_environment(
self,

12
litellm/llms/anthropic/batches/transformation.py
View file

@ -7,7 +7,7 @@ import httpx
from httpx import Headers, Response
from typing_extensions import ReadOnly, TypedDict
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, ensure_api_route, join_api_base
from litellm.llms.base_llm.batches.transformation import BaseBatchesConfig
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.types.llms.openai import AllMessageValues, CreateBatchRequest
@ -106,10 +106,8 @@ class AnthropicBatchesConfig(BaseBatchesConfig):
data: CreateBatchRequest,
) -> str:
"""Get the complete URL for batch creation request."""
api_base = api_base or self.anthropic_model_info.get_api_base(api_base)
if not api_base.endswith("/v1/messages/batches"):
api_base = f"{api_base.rstrip('/')}/v1/messages/batches"
return api_base
api_base = self.anthropic_model_info.resolve_anthropic_api_base(api_base)
return ensure_api_route(api_base, "v1/messages/batches")
def transform_create_batch_request(
self,
@ -158,9 +156,9 @@ class AnthropicBatchesConfig(BaseBatchesConfig):
Returns:
Complete URL for Anthropic batch retrieval: {api_base}/v1/messages/batches/{batch_id}
"""
api_base = api_base or self.anthropic_model_info.get_api_base(api_base)
api_base = self.anthropic_model_info.resolve_anthropic_api_base(api_base)
encoded_batch_id: Final = encode_url_path_segment(batch_id, field_name="batch_id")
return f"{api_base.rstrip('/')}/v1/messages/batches/{encoded_batch_id}"
return join_api_base(api_base, f"v1/messages/batches/{encoded_batch_id}")
def transform_retrieve_batch_request(
self,

14
litellm/llms/anthropic/chat/handler.py
View file

@ -856,11 +856,21 @@ class ModelResponseIterator:
if reasoning_content is None:
reasoning_content = ""
elif "content" in content_block["delta"] and content_block["delta"].get("type") == "compaction_delta":
# Handle compaction delta
compaction_delta: Final = content_block["delta"]["content"]
provider_specific_fields["compaction_delta"] = {
"type": "compaction_delta",
"content": content_block["delta"]["content"],
"content": compaction_delta,
}
if self.compaction_blocks and isinstance(compaction_delta, str):
streamed_summary: Final = self.compaction_blocks[-1].get("content")
self.compaction_blocks = [
*self.compaction_blocks[:-1],
{
**self.compaction_blocks[-1],
"content": (streamed_summary if isinstance(streamed_summary, str) else "") + compaction_delta,
},
]
provider_specific_fields["compaction_blocks"] = self.compaction_blocks
return (
text,

13
litellm/llms/anthropic/common_utils.py
View file

@ -35,6 +35,7 @@ from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import
parts_of,
)
from litellm.litellm_core_utils.reasoning_effort_utils import non_numeric_thinking_budget_error
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.anthropic.wif import (
aget_anthropic_wif_token,
anthropic_base_without_chat_suffix,
@ -468,7 +469,7 @@ def _fetch_anthropic_models_page(
# which a read-only one cannot support
query: Final = f"?after_id={quote(after_id)}" if after_id else ""
response: Final = litellm.module_level_client.get(
url=f"{api_base}/v1/models{query}",
url=f"{join_api_base(api_base, 'v1/models')}{query}",
headers=headers,
follow_redirects=False,
)
@ -1406,7 +1407,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
return {**caller_headers, **anthropic_headers}
@staticmethod
def get_api_base(api_base: str | None = None) -> str | None:
def resolve_anthropic_api_base(api_base: str | None = None) -> str:
from litellm.secret_managers.main import get_secret_str
return (
@ -1416,6 +1417,10 @@ class AnthropicModelInfo(BaseLLMModelInfo):
or "https://api.anthropic.com"
)
@staticmethod
def get_api_base(api_base: str | None = None) -> str | None:
return AnthropicModelInfo.resolve_anthropic_api_base(api_base)
@staticmethod
def get_api_key(api_key: str | None = None) -> str | None:
"""An empty or whitespace-only key counts as unset: it can never authenticate anything, and
@ -1533,14 +1538,14 @@ class AnthropicModelInfo(BaseLLMModelInfo):
api_base: str | None,
litellm_params: Mapping[str, object] | None,
) -> list[str]: # mutable-ok: matches get_models' list[str] contract shared by every provider override
resolved_api_base: Final = AnthropicModelInfo.get_api_base(api_base)
resolved_api_base: Final = AnthropicModelInfo.resolve_anthropic_api_base(api_base)
auth_header: Final = AnthropicModelInfo.get_auth_header(
api_key,
resolved_api_base,
litellm_params=litellm_params,
allow_workload_identity=config_allows_workload_identity(self),
)
if resolved_api_base is None or auth_header is None:
if auth_header is None:
raise ValueError(
"ANTHROPIC_API_BASE/ANTHROPIC_BASE_URL or ANTHROPIC_API_KEY/ANTHROPIC_AUTH_TOKEN (or workload "
"identity federation via ANTHROPIC_FEDERATION_RULE_ID/ANTHROPIC_ORGANIZATION_ID/"

3
litellm/llms/anthropic/count_tokens/transformation.py
View file

@ -25,6 +25,7 @@ from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import
anthropic_system_blocks,
split_leading_system_run,
)
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.anthropic.common_utils import merge_anthropic_beta_headers
from litellm.llms.anthropic.wif import resolve_anthropic_base
from litellm.types.llms.openai import AllMessageValues, ChatCompletionImageObject, ValidUserMessageContentTypes
@ -317,7 +318,7 @@ class AnthropicCountTokensConfig:
Returns:
The endpoint URL for the CountTokens API
"""
return resolve_anthropic_base(api_base) + "/v1/messages/count_tokens"
return join_api_base(resolve_anthropic_base(api_base), "v1/messages/count_tokens")
def transform_request_to_count_tokens(
self,

6
litellm/llms/anthropic/files/handler.py
View file

@ -10,7 +10,7 @@ import litellm
from litellm._logging import verbose_logger
from litellm._uuid import uuid
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, join_api_base
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.types.llms.openai import (
FileContentRequest,
@ -74,7 +74,7 @@ class AnthropicFilesHandler:
batch_id = file_id
# Get Anthropic API credentials
api_base = self.anthropic_model_info.get_api_base(api_base)
api_base = self.anthropic_model_info.resolve_anthropic_api_base(api_base)
auth_header: Final = await self.anthropic_model_info.aget_auth_header(
api_key, api_base, litellm_params=litellm_params, allow_workload_identity=True
)
@ -84,7 +84,7 @@ class AnthropicFilesHandler:
# Construct the Anthropic batch results URL
encoded_batch_id: Final = encode_url_path_segment(batch_id, field_name="batch_id")
results_url: Final = f"{api_base.rstrip('/')}/v1/messages/batches/{encoded_batch_id}/results"
results_url: Final = join_api_base(api_base, f"v1/messages/batches/{encoded_batch_id}/results")
# Prepare headers
headers: Final = {

14
litellm/llms/anthropic/files/transformation.py
View file

@ -22,7 +22,7 @@ from openai.types.file_deleted import FileDeleted
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.prompt_templates.common_utils import extract_file_data
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.files.transformation import (
BaseFilesConfig,
@ -80,7 +80,7 @@ class AnthropicFilesConfig(BaseFilesConfig):
stream: bool | None = None,
) -> str:
api_base = AnthropicModelInfo.get_api_base(api_base) or ANTHROPIC_FILES_API_BASE
return f"{api_base.rstrip('/')}/v1/files"
return join_api_base(api_base, "v1/files")
def get_error_class(
self,
@ -226,7 +226,7 @@ class AnthropicFilesConfig(BaseFilesConfig):
) -> tuple[str, dict]:
api_base: Final = AnthropicModelInfo.get_api_base(litellm_params.get("api_base")) or ANTHROPIC_FILES_API_BASE
encoded_file_id: Final = encode_url_path_segment(file_id, field_name="file_id")
return f"{api_base.rstrip('/')}/v1/files/{encoded_file_id}", {}
return join_api_base(api_base, f"v1/files/{encoded_file_id}"), {}
def transform_retrieve_file_response(
self,
@ -245,7 +245,7 @@ class AnthropicFilesConfig(BaseFilesConfig):
) -> tuple[str, dict]:
api_base: Final = AnthropicModelInfo.get_api_base(litellm_params.get("api_base")) or ANTHROPIC_FILES_API_BASE
encoded_file_id: Final = encode_url_path_segment(file_id, field_name="file_id")
return f"{api_base.rstrip('/')}/v1/files/{encoded_file_id}", {}
return join_api_base(api_base, f"v1/files/{encoded_file_id}"), {}
def transform_delete_file_response(
self,
@ -264,7 +264,7 @@ class AnthropicFilesConfig(BaseFilesConfig):
litellm_params: dict,
) -> tuple[str, dict]:
api_base: Final = AnthropicModelInfo.get_api_base(litellm_params.get("api_base")) or ANTHROPIC_FILES_API_BASE
url: Final = f"{api_base.rstrip('/')}/v1/files"
url: Final = join_api_base(api_base, "v1/files")
params: Final[dict[str, str]] = {}
if purpose:
params["purpose"] = purpose
@ -299,8 +299,8 @@ class AnthropicFilesConfig(BaseFilesConfig):
api_base: Final = AnthropicModelInfo.get_api_base(litellm_params.get("api_base")) or ANTHROPIC_FILES_API_BASE
encoded_file_id: Final = encode_url_path_segment(file_id, field_name="file_id")
if file_id.startswith(ANTHROPIC_MESSAGE_BATCH_ID_PREFIX):
return f"{api_base.rstrip('/')}/v1/messages/batches/{encoded_file_id}/results", {}
return f"{api_base.rstrip('/')}/v1/files/{encoded_file_id}/content", {}
return join_api_base(api_base, f"v1/messages/batches/{encoded_file_id}/results"), {}
return join_api_base(api_base, f"v1/files/{encoded_file_id}/content"), {}
def transform_file_content_response(
self,

26
litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py
View file

@ -16,6 +16,7 @@ from typing import (
Protocol,
cast,
get_args,
runtime_checkable,
)
from pydantic import TypeAdapter
@ -65,6 +66,14 @@ class _ChunkStream(Protocol):
def __aiter__(self) -> "AsyncIterator[ModelResponseStream]": ...
@runtime_checkable
class _FailureLoggingWaiter(Protocol):
@property
def has_dispatched_failure_logging(self) -> bool: ...
async def wait_for_failure_logging(self, timeout: float | None) -> None: ...
def _optional_attr(obj: object, name: str) -> object:
return getattr(obj, name, None)
@ -189,6 +198,14 @@ class _CombinedChunkSplitter:
self._held_calls: Mapping[int, tuple[ModelResponseStream, _ToolCallPart]] = MappingProxyType({})
self._live_call_index: int | None = None
@property
def has_dispatched_failure_logging(self) -> bool:
return isinstance(self._stream, _FailureLoggingWaiter) and self._stream.has_dispatched_failure_logging is True
async def wait_for_failure_logging(self, timeout: float | None) -> None:
if isinstance(self._stream, _FailureLoggingWaiter):
await self._stream.wait_for_failure_logging(timeout=timeout)
@property
def chunks(self) -> "list[ModelResponseStream] | None":
return cast( # cast-ok: chunks is a list of ModelResponseStream on the inner stream
@ -522,7 +539,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
# Wrap the upstream stream so chunks that carry both content and a
# finish_reason (fake-streamed providers) are split into two — see
# _CombinedChunkSplitter.
super().__init__(_CombinedChunkSplitter(completion_stream))
combined_chunk_splitter: Final = _CombinedChunkSplitter(completion_stream)
super().__init__(combined_chunk_splitter)
self._combined_chunk_splitter: Final = combined_chunk_splitter
self.model = model
self._message_id: str = f"msg_{uuid.uuid4()}"
if litellm_logging_obj is not None:
@ -1244,7 +1263,10 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
if provider_error is e:
raise
raise provider_error from e
if logging_obj is not None:
stream_has_dispatched_failure_logging: Final = self._combined_chunk_splitter.has_dispatched_failure_logging
if stream_has_dispatched_failure_logging:
await self._combined_chunk_splitter.wait_for_failure_logging(timeout=None)
elif logging_obj is not None:
try:
await logging_obj.dispatch_failure_handlers(
exception=provider_error,

6
litellm/llms/anthropic/pass_through/messages/transformation.py
View file

@ -9,6 +9,7 @@ from litellm.anthropic_beta_headers_manager import get_unsupported_headers, is_b
from litellm.exceptions import AuthenticationError
from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.anthropic.pass_through.convertible_blocks import drop_non_string_tool_result_texts
from litellm.llms.anthropic.pass_through.safeguards import proxy_serves_safeguards
from litellm.llms.base_llm.anthropic_messages.transformation import (
@ -287,10 +288,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
litellm_params: dict,
stream: bool | None = None,
) -> str:
api_base = AnthropicModelInfo.get_api_base(api_base) or "https://api.anthropic.com"
if not api_base.endswith("/v1/messages"):
api_base = f"{api_base}/v1/messages"
return api_base
return ensure_api_route(AnthropicModelInfo.get_api_base(api_base) or "https://api.anthropic.com", "v1/messages")
def validate_anthropic_messages_environment(
self,

10
litellm/llms/anthropic/skills/transformation.py
View file

@ -9,7 +9,7 @@ import httpx
from pydantic import TypeAdapter
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, join_api_base
from litellm.llms.base_llm.skills.transformation import (
BaseSkillsAPIConfig,
LiteLLMLoggingObj,
@ -75,13 +75,11 @@ class AnthropicSkillsConfig(BaseSkillsAPIConfig):
"""Get complete URL for Anthropic Skills API"""
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
if api_base is None:
api_base = AnthropicModelInfo.get_api_base()
resolved_api_base: Final = AnthropicModelInfo.get_api_base(api_base) or "https://api.anthropic.com"
if skill_id:
encoded_skill_id: Final = encode_url_path_segment(skill_id, field_name="skill_id")
return f"{api_base}/v1/skills/{encoded_skill_id}"
return f"{api_base}/v1/{endpoint}"
return join_api_base(resolved_api_base, f"v1/skills/{encoded_skill_id}")
return join_api_base(resolved_api_base, f"v1/{endpoint}")
def transform_create_skill_request(
self,

3
litellm/llms/anthropic/wif.py
View file

@ -15,6 +15,7 @@ from typing_extensions import assert_never
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.auth.client_credentials import keycloak_assertion_source
from litellm.llms.base_llm.auth.identity_source import (
AnthropicIdentitySourceKind,
@ -324,7 +325,7 @@ def anthropic_internal_issuer_jwks(credential_values: Mapping[str, object]) -> A
def build_anthropic_wif_spec(params: AnthropicWifParams, api_base: str) -> TokenExchangeSpec:
return TokenExchangeSpec(
token_url=api_base.rstrip("/") + ANTHROPIC_TOKEN_EXCHANGE_PATH,
token_url=join_api_base(api_base, ANTHROPIC_TOKEN_EXCHANGE_PATH),
assertion_ref=params.assertion_ref,
assertion_field="assertion",
static_body=MappingProxyType(

7
litellm/llms/apiserpent/search/transformation.py
View file

@ -14,6 +14,7 @@ from urllib.parse import urlencode
import httpx
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.apiserpent.search.defaults import (
DEEP_SEARCH_PATH,
NUM_MAX,
@ -77,13 +78,11 @@ class APISerpentSearchConfig(BaseSearchConfig):
Build the search URL. APISerpent uses GET, so the transformed request is
serialized into the query string. The endpoint path (quick vs deep) is
always applied; an ``api_base`` / ``APISERPENT_API_BASE`` override only
changes the host. The ``endswith`` guard keeps this idempotent, since the
changes the host. ``ensure_api_route`` keeps this idempotent, since the
handler re-invokes this method with the already-resolved URL as api_base.
"""
base = (api_base or get_secret_str("APISERPENT_API_BASE") or APISERPENT_BASE).rstrip("/")
path: Final = DEEP_SEARCH_PATH if self._is_deep_search(optional_params) else QUICK_SEARCH_PATH
if not base.endswith(path):
base = f"{base}{path}"
base: Final = ensure_api_route(api_base or get_secret_str("APISERPENT_API_BASE") or APISERPENT_BASE, path)
if data and isinstance(data, dict) and APISERPENT_PARAMS_KEY in data:
query_string: Final = urlencode(data[APISERPENT_PARAMS_KEY], doseq=True)

4
litellm/llms/azure/audio_transcription/transformation.py
View file

@ -13,6 +13,7 @@ import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.audio_transcription.transformation import (
AudioTranscriptionRequestData,
BaseAudioTranscriptionConfig,
@ -110,7 +111,7 @@ class AzureSpeechAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
"language": optional_params.get("language", self.DEFAULT_LANGUAGE),
"format": self._get_azure_response_format(optional_params.get("response_format")),
}
return f"{base_url}{self.STT_ENDPOINT_PATH}?{urlencode(query_params)}"
return f"{join_api_base(base_url, self.STT_ENDPOINT_PATH)}?{urlencode(query_params)}"
def transform_audio_transcription_request(
self,
@ -153,7 +154,6 @@ class AzureSpeechAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
)
def _resolve_stt_base_url(self, api_base: str) -> str:
api_base = api_base.rstrip("/")
parsed_url: Final = urlparse(api_base)
hostname: Final = parsed_url.hostname or ""

3
litellm/llms/azure/common_utils.py
View file

@ -18,6 +18,7 @@ import litellm
from litellm._logging import verbose_logger
from litellm.caching.caching import DualCache
from litellm.constants import DEFAULT_MAX_RETRIES
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.llms.openai.common_utils import BaseOpenAILLM, OpenAIAsyncHTTPClient, OpenAIHTTPClient
@ -616,7 +617,7 @@ class BaseAzureLLM(BaseOpenAILLM):
v1_params: Final[_AzureV1ClientParams] = {
"api_key": v1_api_key,
"base_url": f"{api_base}/openai/v1/",
"base_url": f"{ensure_api_route(api_base or '', 'openai/v1')}/",
}
if "timeout" in azure_client_params:
v1_params["timeout"] = azure_client_params["timeout"]

4
litellm/llms/azure/realtime/handler.py
View file

@ -10,6 +10,7 @@ from typing import Final, Protocol, cast
from litellm._logging import verbose_proxy_logger
from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.types.realtime import RealtimeQueryParams
from ....litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
@ -136,7 +137,8 @@ class AzureOpenAIRealtime(AzureChatCompletion):
query_parts.append(urlencode({"intent": intent}))
qs: Final = "&".join(query_parts)
return f"{api_base}{path}?{qs}" if qs else f"{api_base}{path}"
endpoint_url: Final = join_api_base(api_base, path)
return f"{endpoint_url}?{qs}" if qs else endpoint_url
def construct_url(
self,

5
litellm/llms/azure/realtime/http_transformation.py
View file

@ -3,6 +3,7 @@
from typing import Final
import litellm
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.realtime.http_transformation import BaseRealtimeHTTPConfig
from litellm.secret_managers.main import get_secret_str
@ -39,9 +40,9 @@ class AzureRealtimeHTTPConfig(BaseRealtimeHTTPConfig):
return _azure_realtime_v1_url(self.get_api_base(api_base), "calls")
def get_transcription_session_url(self, api_base: str | None, model: str, api_version: str | None = None) -> str:
base: Final = self.get_api_base(api_base).rstrip("/")
base: Final = self.get_api_base(api_base)
version: Final = api_version or get_secret_str("AZURE_API_VERSION") or "2024-12-17"
return f"{base}/openai/realtime/transcription_sessions?api-version={version}"
return f"{join_api_base(base, 'openai/realtime/transcription_sessions')}?api-version={version}"
def get_realtime_calls_headers(self, ephemeral_key: str) -> dict[str, str]:
return {

6
litellm/llms/azure/search/transformation.py
View file

@ -35,6 +35,7 @@ from typing import TYPE_CHECKING, Final, Literal
import httpx
from pydantic import ConfigDict, ValidationError
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.azure.common_utils import cached_azure_ad_token_refresh_provider
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.search.transformation import (
@ -340,10 +341,7 @@ class BingGroundingSearchConfig(BaseSearchConfig):
f"{PROJECT_ENDPOINT_ENV} is not set. Set it to your Microsoft Foundry project "
f"endpoint, e.g. https://<account>.services.ai.azure.com/api/projects/<project>."
)
trimmed: Final = resolved_base.rstrip("/")
if trimmed.endswith(_RESPONSES_PATH):
return trimmed
return f"{trimmed}{_RESPONSES_PATH}"
return ensure_api_route(resolved_base, _RESPONSES_PATH)
def transform_search_request(
self,

11
litellm/llms/azure/text_to_speech/transformation.py
View file

@ -11,6 +11,7 @@ from urllib.parse import urlparse
import httpx
import litellm
from litellm.litellm_core_utils.url_utils import ensure_api_route, join_api_base
from litellm.llms.base_llm.text_to_speech.transformation import (
BaseTextToSpeechConfig,
TextToSpeechRequestData,
@ -329,8 +330,6 @@ class AzureAVATextToSpeechConfig(BaseTextToSpeechConfig):
f"https://{{region}}.{self.TTS_SPEECH_DOMAIN}"
)
# Remove trailing slash and parse URL
api_base = api_base.rstrip("/")
parsed_url: Final = urlparse(api_base)
hostname: Final = parsed_url.hostname or ""
@ -339,14 +338,10 @@ class AzureAVATextToSpeechConfig(BaseTextToSpeechConfig):
region: Final = self._extract_region_from_hostname(hostname=hostname, domain=self.COGNITIVE_SERVICES_DOMAIN)
return self._build_tts_url(region=region)
# Check if it's already a TTS endpoint
if self._is_tts_endpoint(hostname=hostname):
if not api_base.endswith(self.TTS_ENDPOINT_PATH):
return f"{api_base}{self.TTS_ENDPOINT_PATH}"
return api_base
return ensure_api_route(api_base, self.TTS_ENDPOINT_PATH)
# Assume it's a custom endpoint, append the path
return f"{api_base}{self.TTS_ENDPOINT_PATH}"
return join_api_base(api_base, self.TTS_ENDPOINT_PATH)
def _is_cognitive_services_endpoint(self, hostname: str) -> bool:
"""Check if hostname is a Cognitive Services endpoint"""

17
litellm/llms/azure_ai/agents/handler.py
View file

@ -30,7 +30,7 @@ from typing_extensions import ReadOnly
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.hidden_params import set_hidden_param
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, join_api_base
from litellm.llms.azure_ai.agents.transformation import (
AzureAIAgentsConfig,
AzureAIAgentsError,
@ -130,28 +130,29 @@ class AzureAIAgentsHandler:
# See: https://learn.microsoft.com/en-us/azure/ai-foundry/agents/quickstart
# -------------------------------------------------------------------------
def _build_thread_url(self, api_base: str, api_version: str) -> str:
return f"{api_base}/threads?api-version={api_version}"
return f"{join_api_base(api_base, 'threads')}?api-version={api_version}"
def _build_messages_url(self, api_base: str, thread_id: str, api_version: str) -> str:
encoded_thread_id: Final = encode_url_path_segment(thread_id, field_name="thread_id")
return f"{api_base}/threads/{encoded_thread_id}/messages?api-version={api_version}"
return f"{join_api_base(api_base, f'threads/{encoded_thread_id}/messages')}?api-version={api_version}"
def _build_runs_url(self, api_base: str, thread_id: str, api_version: str) -> str:
encoded_thread_id: Final = encode_url_path_segment(thread_id, field_name="thread_id")
return f"{api_base}/threads/{encoded_thread_id}/runs?api-version={api_version}"
return f"{join_api_base(api_base, f'threads/{encoded_thread_id}/runs')}?api-version={api_version}"
def _build_run_status_url(self, api_base: str, thread_id: str, run_id: str, api_version: str) -> str:
encoded_thread_id: Final = encode_url_path_segment(thread_id, field_name="thread_id")
encoded_run_id: Final = encode_url_path_segment(run_id, field_name="run_id")
return f"{api_base}/threads/{encoded_thread_id}/runs/{encoded_run_id}?api-version={api_version}"
run_status_route: Final = f"threads/{encoded_thread_id}/runs/{encoded_run_id}"
return f"{join_api_base(api_base, run_status_route)}?api-version={api_version}"
def _build_list_messages_url(self, api_base: str, thread_id: str, api_version: str) -> str:
encoded_thread_id: Final = encode_url_path_segment(thread_id, field_name="thread_id")
return f"{api_base}/threads/{encoded_thread_id}/messages?api-version={api_version}"
return f"{join_api_base(api_base, f'threads/{encoded_thread_id}/messages')}?api-version={api_version}"
def _build_create_thread_and_run_url(self, api_base: str, api_version: str) -> str:
"""URL for the create-thread-and-run endpoint (supports streaming)."""
return f"{api_base}/threads/runs?api-version={api_version}"
return f"{join_api_base(api_base, 'threads/runs')}?api-version={api_version}"
# -------------------------------------------------------------------------
# Response Helpers
@ -281,8 +282,6 @@ class AzureAIAgentsHandler:
api_version: Final = optional_params.get("api_version", self.config.DEFAULT_API_VERSION)
agent_id: Final = self.config.get_agent_id(model, optional_params)
thread_id: Final = optional_params.get("thread_id")
api_base = api_base.rstrip("/")
verbose_logger.debug("Azure AI Agents completion - api_base: %s, agent_id: %s", api_base, agent_id)
return headers, api_version, agent_id, thread_id, api_base

18
litellm/llms/azure_ai/anthropic/count_tokens/transformation.py
View file

@ -5,8 +5,10 @@ Extends the base Anthropic CountTokens transformation with Azure authentication.
"""
from typing import Final
from urllib.parse import urlsplit
from litellm.constants import ANTHROPIC_TOKEN_COUNTING_BETA_VERSION
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.anthropic.count_tokens.transformation import (
AnthropicCountTokensConfig,
)
@ -76,14 +78,8 @@ class AzureAIAnthropicCountTokensConfig(AnthropicCountTokensConfig):
Returns:
The endpoint URL for the CountTokens API
"""
# Azure AI Anthropic endpoint format:
# https://<resource>.services.ai.azure.com/anthropic/v1/messages/count_tokens
api_base = api_base.rstrip("/")
# Ensure the URL has /anthropic path
if not api_base.endswith("/anthropic"):
if "/anthropic" not in api_base:
api_base = f"{api_base}/anthropic"
# Add the count_tokens path
return f"{api_base}/v1/messages/count_tokens"
base: Final = api_base.rstrip("/")
if base.endswith("/v1/messages"):
return join_api_base(base, "count_tokens")
anthropic_base: Final = base if "/anthropic" in urlsplit(base).path else join_api_base(base, "anthropic")
return join_api_base(anthropic_base, "v1/messages/count_tokens")

21
litellm/llms/azure_ai/anthropic/messages_transformation.py
View file

@ -9,6 +9,7 @@ from litellm.llms.anthropic.pass_through.messages.transformation import (
AnthropicMessagesConfig,
)
from litellm.llms.azure.common_utils import BaseAzureLLM
from litellm.llms.azure_ai.anthropic.transformation import azure_anthropic_messages_url
from litellm.types.router import GenericLiteLLMParams
@ -96,25 +97,7 @@ class AzureAnthropicMessagesConfig(AnthropicMessagesConfig):
"Expected format: https://<resource-name>.services.ai.azure.com/anthropic"
)
# Ensure the URL ends with /v1/messages
api_base = api_base.rstrip("/")
if api_base.endswith("/v1/messages") or api_base.endswith("/anthropic/v1/messages"):
# Already correct
pass
else:
# Check if /anthropic is already in the path
if "/anthropic" in api_base:
# /anthropic exists, ensure we end with /anthropic/v1/messages
# Extract the base URL up to and including /anthropic
parts: Final = api_base.split("/anthropic", 1)
api_base = parts[0] + "/anthropic"
else:
# /anthropic not in path, add it
api_base = api_base + "/anthropic"
# Add /v1/messages
api_base = api_base + "/v1/messages"
return api_base
return azure_anthropic_messages_url(api_base)
def _remove_scope_from_cache_control(self, anthropic_messages_request: dict) -> None:
"""

11
litellm/llms/azure_ai/anthropic/transformation.py
View file

@ -4,12 +4,23 @@ Azure Anthropic transformation config - extends AnthropicConfig with Azure authe
from typing import Final
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.azure.common_utils import BaseAzureLLM
from litellm.types.llms.openai import AllMessageValues
from litellm.types.router import GenericLiteLLMParams
def azure_anthropic_messages_url(api_base: str) -> str:
base: Final = api_base.rstrip("/")
if base.endswith("/v1/messages"):
return base
path_start: Final = base.find("/", base.find("://") + len("://")) if "://" in base else 0
anthropic_start: Final = base.find("/anthropic", path_start) if path_start >= 0 else -1
anthropic_root: Final = base[:anthropic_start] if anthropic_start >= 0 else base
return join_api_base(anthropic_root, "anthropic/v1/messages")
def _promote_extra_body_to_optional_params(optional_params: dict) -> None:
"""Promote anthropic-native passthrough keys out of ``extra_body``.

10
litellm/llms/azure_ai/embed/handler.py
View file

@ -5,6 +5,7 @@ from urllib.parse import urlsplit, urlunsplit
from openai import OpenAI
import litellm
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.azure_ai.common_utils import is_foundry_model_inference_base
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
@ -74,13 +75,18 @@ class AzureAIEmbedding(OpenAIChatCompletion):
api_base: str | None,
client: HTTPHandler | AsyncHTTPHandler | None = None,
) -> EmbeddingResponse:
if api_base is None:
raise ValueError(
"api_base is None. Please set AZURE_AI_API_BASE or dynamically via `api_base` param, "
"to make the request."
)
if client is None or not isinstance(client, AsyncHTTPHandler):
client = get_async_httpx_client(
llm_provider=litellm.LlmProviders.AZURE_AI,
params={"timeout": timeout},
)
url: Final = f"{api_base}/images/embeddings"
url: Final = join_api_base(api_base, "images/embeddings")
response: Final = await client.post(
url=url,
@ -123,7 +129,7 @@ class AzureAIEmbedding(OpenAIChatCompletion):
if client is None or not isinstance(client, HTTPHandler):
client = HTTPHandler(timeout=timeout, concurrent_limit=1)
url: Final = f"{api_base}/images/embeddings"
url: Final = join_api_base(api_base, "images/embeddings")
response: Final = client.post(
url=url,

4
litellm/llms/azure_ai/image_generation/flux_transformation.py
View file

@ -3,6 +3,7 @@ from types import MappingProxyType
from typing import Final
from litellm.exceptions import BadRequestError, UnsupportedParamsError
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.openai.image_generation import GPTImageGenerationConfig
from litellm.types.llms.openai import OpenAIImageGenerationOptionalParams
@ -52,7 +53,8 @@ class AzureFoundryFluxImageGenerationConfig(GPTImageGenerationConfig):
return f"{api_base}?api-version={api_version}"
provider_model_path: Final = AzureFoundryFluxImageGenerationConfig.get_flux2_provider_model_path(model)
return f"{api_base}/providers/blackforestlabs/v1/{provider_model_path}?api-version={api_version}"
flux_route: Final = f"providers/blackforestlabs/v1/{provider_model_path}"
return f"{join_api_base(api_base, flux_route)}?api-version={api_version}"
@staticmethod
def is_flux2_model(model: str) -> bool:

9
litellm/llms/azure_ai/image_generation/mai_transformation.py
View file

@ -3,6 +3,7 @@ from typing import TYPE_CHECKING, Any, Final
import httpx
from litellm.exceptions import UnsupportedParamsError
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
)
@ -41,9 +42,9 @@ class AzureFoundryMAIImageGenerationConfig(BaseImageGenerationConfig):
if "/mai/" in path:
prefix, _, _ = path.partition("/images/")
path = f"{prefix}/images/generations"
path = join_api_base(prefix, "images/generations")
else:
path = f"{path}/mai/v1/images/generations"
path = join_api_base(path, "mai/v1/images/generations")
if separator:
return f"{path}?{query}"
@ -63,9 +64,9 @@ class AzureFoundryMAIImageGenerationConfig(BaseImageGenerationConfig):
if "/mai/" in path:
prefix, _, _ = path.partition("/images/")
path = f"{prefix}/images/edits"
path = join_api_base(prefix, "images/edits")
else:
path = f"{path}/mai/v1/images/edits"
path = join_api_base(path, "mai/v1/images/edits")
if separator:
return f"{path}?{query}"

4
litellm/llms/azure_ai/vector_stores/transformation.py
View file

@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Any, Final
import httpx
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.azure.common_utils import BaseAzureLLM
from litellm.llms.base_llm.vector_store.transformation import (
BaseQueryEmbeddingVectorStoreConfig,
@ -172,7 +173,8 @@ class AzureAIVectorStoreConfig(BaseQueryEmbeddingVectorStoreConfig, BaseAzureLLM
litellm_logging_obj.model_call_details["input"] = query_text
litellm_logging_obj.model_call_details["embedding_model"] = litellm_params.get("litellm_embedding_model")
litellm_logging_obj.model_call_details["top_k"] = top_k
return f"{api_base}/indexes/{vector_store_id}/docs/search?api-version=2024-07-01", {
search_route: Final = f"indexes/{vector_store_id}/docs/search"
return f"{join_api_base(api_base, search_route)}?api-version=2024-07-01", {
"search": "*",
"vectorQueries": [{"vector": query_vector, "fields": vector_field, "kind": "vector", "k": top_k}],
"select": "id,content",

3
litellm/llms/base_llm/decisions/transformation.py
View file

@ -12,6 +12,7 @@ from pydantic_core import ErrorDetails
from typing_extensions import assert_never
import litellm
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.secret_managers.main import get_secret_str
from litellm.types.decisions import (
@ -582,7 +583,7 @@ class BaseDecisionsConfig(ABC):
)
def get_complete_url(self, api_base: str, model: str) -> str:
return f"{api_base.rstrip('/').removesuffix('/v1')}{self.path}"
return join_api_base(api_base.rstrip("/").removesuffix("/v1"), self.path)
def signs_with_aws(self, api_base: str) -> bool:
return False

3
litellm/llms/base_llm/evals/transformation.py
View file

@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Any
import httpx
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.types.llms.openai_evals import (
CancelEvalResponse,
@ -79,7 +80,7 @@ class BaseEvalsAPIConfig(ABC):
"""
if api_base is None:
raise ValueError("api_base is required")
return f"{api_base}/v1/{endpoint}"
return join_api_base(api_base, f"v1/{endpoint}")
@abstractmethod
def transform_create_eval_request(

6
litellm/llms/base_llm/ocr/transformation.py
View file

@ -4,7 +4,7 @@ Base OCR types shared by the Rust OCR route and Python consumers.
import builtins
from collections.abc import Mapping
from typing import Any, Final, Literal
from typing import Final, Literal
from pydantic import PrivateAttr
@ -45,7 +45,7 @@ class OCRPageImage(LiteLLMPydanticObjectBase):
"""Image extracted from OCR page."""
image_base64: str | None = None
bbox: dict[str, Any] | None = None
bbox: dict[str, builtins.object] | None = None
model_config = {"extra": "allow"}
@ -80,7 +80,7 @@ class OCRResponse(LiteLLMPydanticObjectBase):
pages: list[OCRPage]
model: str
document_annotation: Any | None = None
document_annotation: builtins.object | None = None
usage_info: OCRUsageInfo | None = None
content: str | None = None
tables: list[dict[str, builtins.object]] | None = None

8
litellm/llms/base_llm/realtime/http_transformation.py
View file

@ -7,10 +7,11 @@ These are HTTP (not WebSocket) endpoints used by the WebRTC flow:
"""
from abc import ABC, abstractmethod
from typing import Final
import httpx
from litellm.litellm_core_utils.url_utils import join_api_base
class BaseRealtimeHTTPConfig(ABC):
"""
@ -61,7 +62,7 @@ class BaseRealtimeHTTPConfig(ABC):
"""Return the full URL for POST /realtime/transcription_sessions."""
base = (api_base or "").rstrip("/")
base = base.removesuffix("/v1")
return f"{base}/v1/realtime/transcription_sessions"
return join_api_base(base, "v1/realtime/transcription_sessions")
@abstractmethod
def validate_environment(
@ -83,8 +84,7 @@ class BaseRealtimeHTTPConfig(ABC):
def get_realtime_calls_url(self, api_base: str | None, model: str, api_version: str | None = None) -> str:
"""Return the full URL for POST /realtime/calls (SDP exchange)."""
base: Final = (api_base or "").rstrip("/")
return f"{base}/v1/realtime/calls"
return join_api_base(api_base or "", "v1/realtime/calls")
def get_realtime_calls_headers(self, ephemeral_key: str) -> dict:
"""

6
litellm/llms/base_llm/sandbox/transformation.py
View file

@ -7,7 +7,7 @@ run code -> delete container; `code_interpreter_tool` combines all three.
"""
import builtins
from typing import Any, Final
from typing import Final
import httpx
from pydantic import Field, PrivateAttr
@ -42,8 +42,8 @@ class CodeExecutionResult(LiteLLMPydanticObjectBase):
stdout: str = ""
stderr: str = ""
results: list[dict[str, Any]] = Field(default_factory=list)
error: dict[str, Any] | None = None
results: list[dict[str, object]] = Field(default_factory=list)
error: dict[str, object] | None = None
execution_count: int | None = None
object: str = "code_execution"

3
litellm/llms/base_llm/skills/transformation.py
View file

@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Any
import httpx
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.types.llms.anthropic_skills import (
CreateSkillRequest,
@ -71,7 +72,7 @@ class BaseSkillsAPIConfig(ABC):
"""
if api_base is None:
raise ValueError("api_base is required")
return f"{api_base}/v1/{endpoint}"
return join_api_base(api_base, f"v1/{endpoint}")
@abstractmethod
def transform_create_skill_request(

5
litellm/llms/bedrock/chat/agentcore/transformation.py
View file

@ -18,6 +18,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
convert_content_list_to_str,
)
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.a2a.common_utils import extract_text_from_a2a_response
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
@ -98,14 +99,14 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM):
# Build the base endpoint URL for AgentCore
# Note: We don't use get_runtime_endpoint as AgentCore has its own endpoint structure
if aws_bedrock_runtime_endpoint:
base_url = aws_bedrock_runtime_endpoint
base_url = f"{aws_bedrock_runtime_endpoint}"
else:
base_url = f"https://bedrock-agentcore.{region}.{get_aws_dns_suffix(region)}"
# Based on boto3 client.invoke_agent_runtime, the path is:
# /runtimes/{URL-ENCODED-ARN}/invocations?qualifier=<value>
encoded_arn: Final = quote(agent_runtime_arn, safe="")
endpoint_url = f"{base_url}/runtimes/{encoded_arn}/invocations"
endpoint_url = join_api_base(base_url, f"runtimes/{encoded_arn}/invocations")
# Add qualifier as query parameter if provided
if "qualifier" in optional_params:

10
litellm/llms/bedrock/chat/chat_completions/transformation.py
View file

@ -33,6 +33,7 @@ from litellm.litellm_core_utils.prompt_templates.image_handling import (
inline_remote_image_urls,
inline_remote_media,
)
from litellm.litellm_core_utils.url_utils import ensure_api_route, join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, bedrock_bearer_token
from litellm.llms.bedrock.common_utils import (
@ -371,12 +372,9 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
),
aws_region_name=aws_region_name,
)
base: Final = proxy_endpoint_url.rstrip("/")
if base.endswith("/openai/v1/chat/completions"):
return base
if base.endswith("/openai/v1"):
return f"{base}/chat/completions"
return f"{base}/openai/v1/chat/completions"
if proxy_endpoint_url.rstrip("/").endswith("/openai/v1"):
return join_api_base(proxy_endpoint_url, "chat/completions")
return ensure_api_route(proxy_endpoint_url, "openai/v1/chat/completions")
def _params_with_region_from_path(
self, optional_params: dict, model: str | None

9
litellm/llms/bedrock/chat/converse_handler.py
View file

@ -10,6 +10,7 @@ from litellm.anthropic_beta_headers_manager import (
)
from litellm.litellm_core_utils.get_litellm_params import stored_control_options
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm import base_model_iterator
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
@ -363,11 +364,11 @@ class BedrockConverseLLM(BaseAWSLLM):
aws_region_name=aws_region_name,
)
if (stream is not None and stream is True) and not fake_stream:
endpoint_url = f"{endpoint_url}/model/{modelId}/converse-stream"
proxy_endpoint_url = f"{proxy_endpoint_url}/model/{modelId}/converse-stream"
endpoint_url = join_api_base(endpoint_url, f"model/{modelId}/converse-stream")
proxy_endpoint_url = join_api_base(proxy_endpoint_url, f"model/{modelId}/converse-stream")
else:
endpoint_url = f"{endpoint_url}/model/{modelId}/converse"
proxy_endpoint_url = f"{proxy_endpoint_url}/model/{modelId}/converse"
endpoint_url = join_api_base(endpoint_url, f"model/{modelId}/converse")
proxy_endpoint_url = join_api_base(proxy_endpoint_url, f"model/{modelId}/converse")
## COMPLETION CALL
headers = {"Content-Type": "application/json"}

9
litellm/llms/bedrock/chat/invoke_agent/transformation.py
View file

@ -15,7 +15,7 @@ from litellm._uuid import uuid
from litellm.litellm_core_utils.prompt_templates.common_utils import (
convert_content_list_to_str,
)
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, join_api_base
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import BedrockError
@ -101,9 +101,10 @@ class AmazonInvokeAgentConfig(BaseConfig, BaseAWSLLM):
encoded_agent_alias_id: Final = encode_url_path_segment(agent_alias_id, field_name="agent_alias_id")
encoded_session_id: Final = encode_url_path_segment(session_id, field_name="session_id")
endpoint_url = f"{endpoint_url}/agents/{encoded_agent_id}/agentAliases/{encoded_agent_alias_id}/sessions/{encoded_session_id}/text"
return endpoint_url
return join_api_base(
endpoint_url,
f"agents/{encoded_agent_id}/agentAliases/{encoded_agent_alias_id}/sessions/{encoded_session_id}/text",
)
def sign_request(
self,

10
litellm/llms/bedrock/chat/invoke_transformations/amazon_openai_transformation.py
View file

@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any, Final
import httpx
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.request_metadata import (
@ -97,13 +98,8 @@ class AmazonBedrockOpenAIConfig(OpenAIGPTConfig, BaseAWSLLM):
# Encode model ID for ARNs (e.g., :imported-model/ -> :imported-model%2F)
model_id = CommonUtils.encode_bedrock_runtime_modelid_arn(model_id)
# Build the invoke URL
if stream:
endpoint_url = f"{endpoint_url}/model/{model_id}/invoke-with-response-stream"
else:
endpoint_url = f"{endpoint_url}/model/{model_id}/invoke"
return endpoint_url
invoke_route: Final = "invoke-with-response-stream" if stream else "invoke"
return join_api_base(endpoint_url, f"model/{model_id}/{invoke_route}")
def sign_request(
self,

14
litellm/llms/bedrock/chat/invoke_transformations/base_invoke_transformation.py
View file

@ -18,6 +18,7 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
prompt_factory,
render_custom_prompt_template,
)
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.llms.bedrock.chat.invoke_handler import make_call, make_sync_call
from litellm.llms.bedrock.common_utils import BedrockError
@ -136,20 +137,13 @@ class AmazonInvokeConfig(BaseConfig, BaseAWSLLM):
aws_bedrock_runtime_endpoint: Final = optional_params.get(
"aws_bedrock_runtime_endpoint", None
) # https://bedrock-runtime.{region_name}.amazonaws.com
endpoint_url, proxy_endpoint_url = self.get_runtime_endpoint(
endpoint_url, _ = self.get_runtime_endpoint(
api_base=api_base,
aws_bedrock_runtime_endpoint=aws_bedrock_runtime_endpoint,
aws_region_name=self._get_aws_region_name(optional_params=optional_params, model=model),
)
if (stream is not None and stream is True) and provider != "ai21":
endpoint_url = f"{endpoint_url}/model/{modelId}/invoke-with-response-stream"
proxy_endpoint_url = f"{proxy_endpoint_url}/model/{modelId}/invoke-with-response-stream"
else:
endpoint_url = f"{endpoint_url}/model/{modelId}/invoke"
proxy_endpoint_url = f"{proxy_endpoint_url}/model/{modelId}/invoke"
return endpoint_url
invoke_route: Final = "invoke-with-response-stream" if stream is True and provider != "ai21" else "invoke"
return join_api_base(endpoint_url, f"model/{modelId}/{invoke_route}")
def sign_request(
self,

5
litellm/llms/bedrock/claude_platform/common_utils.py
View file

@ -5,6 +5,7 @@ import httpx
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.secret_managers.main import get_secret_str
@ -140,9 +141,7 @@ class BedrockClaudePlatformMixin(BaseAWSLLM):
if api_base is None:
aws_region_name: Final = self._get_required_aws_region_name(optional_params)
api_base = f"https://{CLAUDE_PLATFORM_SERVICE_NAME}.{aws_region_name}.api.aws"
if not api_base.endswith("/v1/messages"):
api_base = f"{api_base.rstrip('/')}/v1/messages"
return api_base
return ensure_api_route(api_base, "v1/messages")
def sign_request(
self,

6
litellm/llms/bedrock/common_utils.py
View file

@ -31,6 +31,7 @@ from litellm import verbose_logger
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
from litellm.litellm_core_utils.optional_imports import ensure_optional_import
from litellm.litellm_core_utils.prompt_templates.common_utils import schema_property_count
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.base_llm.anthropic_messages.transformation import (
BaseAnthropicMessagesConfig,
)
@ -1305,10 +1306,7 @@ def build_mantle_messages_url(
"""
override: Final = api_base or aws_bedrock_runtime_endpoint or _mantle_api_base_from_env()
if override:
base: Final = override.rstrip("/")
if base.endswith(MANTLE_MESSAGES_PATH):
return base
return f"{base}{MANTLE_MESSAGES_PATH}"
return ensure_api_route(override, MANTLE_MESSAGES_PATH)
return f"https://bedrock-mantle.{region}.api.aws{MANTLE_MESSAGES_PATH}"

3
litellm/llms/bedrock/count_tokens/mantle_handler.py
View file

@ -7,6 +7,7 @@ from typing_extensions import NotRequired, ReadOnly, TypedDict
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.anthropic.count_tokens.transformation import COUNT_TOKEN_OPTION_NAMES, AnthropicCountTokensConfig
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, run_aws_signing
from litellm.llms.bedrock.common_utils import BedrockError, build_mantle_messages_url
@ -43,7 +44,7 @@ class BedrockMantleCountTokensHandler(BaseAWSLLM):
messages_url: Final = build_mantle_messages_url(
api_base=None, aws_bedrock_runtime_endpoint=None, region=aws_region_name
)
return f"{messages_url}{MANTLE_COUNT_TOKENS_SUFFIX}"
return join_api_base(messages_url, MANTLE_COUNT_TOKENS_SUFFIX)
async def handle_count_tokens_request(
self,

5
litellm/llms/bedrock/count_tokens/transformation.py
View file

@ -11,6 +11,7 @@ from typing import Final, Literal
from pydantic import JsonValue
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import get_bedrock_base_model
@ -236,9 +237,7 @@ class BedrockCountTokensConfig(BaseAWSLLM):
aws_bedrock_runtime_endpoint=aws_bedrock_runtime_endpoint,
aws_region_name=aws_region_name,
)
endpoint: Final = f"{base_url}/model/{encoded_model_id}/count-tokens"
return endpoint
return join_api_base(base_url, f"model/{encoded_model_id}/count-tokens")
def transform_bedrock_response_to_anthropic(
self, bedrock_response: Mapping[str, JsonValue]

7
litellm/llms/bedrock/embed/embedding.py
View file

@ -12,6 +12,7 @@ import httpx
import litellm
from litellm.constants import BEDROCK_EMBEDDING_PROVIDERS_LITERAL
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.cohere.embed.handler import embedding as cohere_embedding
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
@ -512,9 +513,9 @@ class BedrockEmbedding(BaseAWSLLM):
aws_region_name=aws_region_name,
)
if has_async_invoke:
endpoint_url = f"{endpoint_url}/async-invoke"
endpoint_url = join_api_base(endpoint_url, "async-invoke")
else:
endpoint_url = f"{endpoint_url}/model/{modelId}/invoke"
endpoint_url = join_api_base(endpoint_url, f"model/{modelId}/invoke")
if batch_data is not None:
if aembedding:
@ -611,7 +612,7 @@ class BedrockEmbedding(BaseAWSLLM):
# Encode the ARN for use in URL path
encoded_arn: Final = quote(invocation_arn, safe="")
status_url: Final = f"{endpoint_url.rstrip('/')}/async-invoke/{encoded_arn}"
status_url: Final = join_api_base(endpoint_url, f"async-invoke/{encoded_arn}")
# Prepare headers for GET request
headers: Final = {"Content-Type": "application/json"}

3
litellm/llms/bedrock/image_edit/handler.py
View file

@ -14,6 +14,7 @@ import httpx
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.bedrock.image_edit.amazon_nova_canvas_image_edit_transformation import (
BedrockAmazonNovaCanvasImageEditConfig,
)
@ -226,7 +227,7 @@ class BedrockImageEdit(BaseAWSLLM):
aws_bedrock_runtime_endpoint=boto3_credentials_info.aws_bedrock_runtime_endpoint,
aws_region_name=boto3_credentials_info.aws_region_name,
)
proxy_endpoint_url = f"{proxy_endpoint_url}/model/{modelId}/invoke"
proxy_endpoint_url = join_api_base(proxy_endpoint_url, f"model/{modelId}/invoke")
data: Final = self._get_request_body(
model=model,
image=image,

3
litellm/llms/bedrock/image_generation/image_handler.py
View file

@ -8,6 +8,7 @@ import httpx
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.bedrock.image_generation.amazon_nova_canvas_transformation import (
AmazonNovaCanvasConfig,
)
@ -248,7 +249,7 @@ class BedrockImageGeneration(BaseAWSLLM):
aws_bedrock_runtime_endpoint=boto3_credentials_info.aws_bedrock_runtime_endpoint,
aws_region_name=boto3_credentials_info.aws_region_name,
)
proxy_endpoint_url = f"{proxy_endpoint_url}/model/{modelId}/invoke"
proxy_endpoint_url = join_api_base(proxy_endpoint_url, f"model/{modelId}/invoke")
data: Final = self._get_request_body(
model=model,
prompt=prompt,

3
litellm/llms/bedrock/rerank/handler.py
View file

@ -6,6 +6,7 @@ from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
HTTPHandler,
@ -160,7 +161,7 @@ class BedrockRerankHandler(BaseAWSLLM):
aws_region_name=boto3_credentials_info.aws_region_name,
)
proxy_endpoint_url = proxy_endpoint_url.replace("bedrock-runtime", "bedrock-agent-runtime")
proxy_endpoint_url = f"{proxy_endpoint_url}/rerank"
proxy_endpoint_url = join_api_base(proxy_endpoint_url, "rerank")
body: Final = json.dumps(data).encode("utf-8")
headers = {"Content-Type": "application/json"}

6
litellm/llms/bedrock/responses/transformation.py
View file

@ -46,6 +46,7 @@ from litellm.litellm_core_utils.prompt_templates.image_handling import (
async_convert_url_to_base64,
convert_url_to_base64,
)
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.responses.codex_compat import drop_unsupported_tools, normalize_codex_input_items
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
@ -223,9 +224,10 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
litellm_params: dict, # mutable-ok: signature fixed by the BaseResponsesAPIConfig override contract
) -> str:
region: Final = self._get_aws_region_name(optional_params=litellm_params, model=None)
configured_endpoint: Final = litellm_params.get("aws_bedrock_runtime_endpoint")
override: Final = (
api_base
or litellm_params.get("aws_bedrock_runtime_endpoint")
or (f"{configured_endpoint}" if configured_endpoint else None)
or get_secret_str("AWS_BEDROCK_RUNTIME_ENDPOINT")
)
# Partition-aware: bedrock-runtime is amazonaws.com.cn in China, and other
@ -237,7 +239,7 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
(host[: -len(suffix)] for suffix in BEDROCK_RUNTIME_OPENAI_BASE_SUFFIXES if host.endswith(suffix)),
host,
)
return f"{base}{BEDROCK_RUNTIME_OPENAI_RESPONSES_PATH}"
return join_api_base(base, BEDROCK_RUNTIME_OPENAI_RESPONSES_PATH)
def supports_native_file_search(self) -> bool:
return False

6
litellm/llms/bedrock/vector_stores/transformation.py
View file

@ -6,7 +6,7 @@ from urllib.parse import urlparse
import httpx
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, join_api_base
from litellm.llms.base_llm.vector_store.transformation import BaseVectorStoreConfig
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import BedrockError
@ -196,7 +196,7 @@ class BedrockVectorStoreConfig(BaseVectorStoreConfig, BaseAWSLLM):
aws_region_name=self.get_aws_region_name_for_non_llm_api_calls(aws_region_name=aws_region_name),
endpoint_type="agent",
)
return f"{endpoint_url}/knowledgebases"
return join_api_base(endpoint_url, "knowledgebases")
def transform_search_vector_store_request(
self,
@ -212,7 +212,7 @@ class BedrockVectorStoreConfig(BaseVectorStoreConfig, BaseAWSLLM):
query = " ".join(query)
encoded_vector_store_id: Final = encode_url_path_segment(vector_store_id, field_name="vector_store_id")
url: Final = f"{api_base}/{encoded_vector_store_id}/retrieve"
url: Final = join_api_base(api_base, f"{encoded_vector_store_id}/retrieve")
request_body: Final[dict[str, object]] = {
"retrievalQuery": BedrockKBRetrievalQuery(text=query),

3
litellm/llms/bedrock_mantle/messages/transformation.py
View file

@ -2,6 +2,7 @@ from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import MANTLE_MESSAGES_PATH
from litellm.llms.bedrock.messages.mantle_transformation import AmazonMantleMessagesConfig
@ -33,7 +34,7 @@ def build_mantle_native_messages_url(api_base: str | None, litellm_params: Mappi
configured,
)
host: Final = f"https://bedrock-mantle.{region}.api.aws" if MANTLE_HOST_RE.match(stripped) else stripped
return f"{host}{MANTLE_MESSAGES_PATH}"
return join_api_base(host, MANTLE_MESSAGES_PATH)
class BedrockMantleAnthropicMessagesConfig(BedrockMantleAuthMixin, AmazonMantleMessagesConfig):

3
litellm/llms/bedrock_mantle/responses/transformation.py
View file

@ -22,6 +22,7 @@ import httpx
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.responses.codex_compat import drop_unsupported_tools, normalize_codex_input_items
from litellm.llms.base_llm.responses.transformation import ReasoningSummaryLevel
@ -100,7 +101,7 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
if MANTLE_HOST_RE.match(base):
base = f"https://bedrock-mantle.{region}.api.aws"
path: Final = "/openai/v1/responses" if self.use_openai_path else "/v1/responses"
return f"{base}{path}"
return join_api_base(base, path)
def validate_environment(self, headers: dict, model: str, litellm_params: GenericLiteLLMParams | None) -> dict:
litellm_params = litellm_params or GenericLiteLLMParams()

7
litellm/llms/black_forest_labs/image_edit/transformation.py
View file

@ -19,7 +19,7 @@ from typing_extensions import NotRequired, ReadOnly, TypedDict
import litellm
from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
from litellm.litellm_core_utils.url_utils import async_safe_get, safe_get
from litellm.litellm_core_utils.url_utils import async_safe_get, join_api_base, safe_get
from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig
from litellm.secret_managers.main import get_secret_str
from litellm.types.images.main import ImageEditOptionalRequestParams
@ -189,11 +189,10 @@ class BlackForestLabsImageEditConfig(BaseImageEditConfig):
"""
Get the complete URL for the Black Forest Labs API request.
"""
base_url: str = api_base or get_secret_str("BFL_API_BASE") or DEFAULT_API_BASE
base_url = base_url.rstrip("/")
base_url: Final = api_base or get_secret_str("BFL_API_BASE") or DEFAULT_API_BASE
endpoint: Final = self._get_model_endpoint(model)
return f"{base_url}{endpoint}"
return join_api_base(base_url, endpoint)
def _read_image_bytes(
self,

6
litellm/llms/black_forest_labs/image_generation/transformation.py
View file

@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
)
@ -202,11 +203,10 @@ class BlackForestLabsImageGenerationConfig(BaseImageGenerationConfig):
"""
Get the complete URL for the Black Forest Labs API request.
"""
base_url: str = api_base or get_secret_str("BFL_API_BASE") or DEFAULT_API_BASE
base_url = base_url.rstrip("/")
base_url: Final = api_base or get_secret_str("BFL_API_BASE") or DEFAULT_API_BASE
endpoint: Final = self._get_model_endpoint(model)
return f"{base_url}{endpoint}"
return join_api_base(base_url, endpoint)
def transform_image_generation_request(
self,

4
litellm/llms/chatgpt/responses/transformation.py
View file

@ -9,6 +9,7 @@ from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, set_hid
from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import (
safe_convert_created_field,
)
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.openai.common_utils import OpenAIError
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.responses.sse_output_recovery import (
@ -253,8 +254,7 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
litellm_params: dict,
) -> str:
api_base = api_base or self.authenticator.get_api_base() or CHATGPT_API_BASE
api_base = api_base.rstrip("/")
return f"{api_base}/responses"
return join_api_base(api_base, "responses")
def supports_native_websocket(self) -> bool:
"""ChatGPT does not support native WebSocket for Responses API"""

3
litellm/llms/cloudflare/chat/transformation.py
View file

@ -3,6 +3,7 @@ from typing import Final
import httpx
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
from litellm.secret_managers.main import (
@ -59,7 +60,7 @@ class CloudflareChatConfig(OpenAIGPTConfig):
verbose_logger.warning(
"Cloudflare api_base ending in '/ai/run' is the legacy Workers AI path and no longer serves OpenAI-compatible requests; rewriting to the '/ai/v1' endpoint"
)
return f"{trimmed[: -len('/ai/run')]}/ai/v1"
return join_api_base(trimmed[: -len("/ai/run")], "ai/v1")
return api_base
def validate_environment(

5
litellm/llms/cloudflare/common_utils.py
View file

@ -3,6 +3,7 @@ from typing import Final
from pydantic import TypeAdapter, ValidationError
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.secret_managers.main import (
get_secret_str,
normalize_nonempty_secret_str,
@ -34,10 +35,10 @@ def workers_ai_request_model(model: str) -> str:
def workers_ai_run_url(api_base: str, model: str) -> str:
normalized_api_base: Final = api_base.rstrip("/")
if normalized_api_base.endswith("/ai/v1"):
return f"{normalized_api_base.removesuffix('/ai/v1')}/ai/run/{model}"
return join_api_base(normalized_api_base.removesuffix("/ai/v1"), f"ai/run/{model}")
if normalized_api_base.endswith("/ai/run"):
return f"{normalized_api_base}/{model}"
return f"{normalized_api_base}/ai/run/{model}"
return join_api_base(normalized_api_base, f"ai/run/{model}")
def unwrap_workers_ai_result(payload: object) -> object:

7
litellm/llms/cohere/rerank/transformation.py
View file

@ -6,6 +6,7 @@ from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.rerank.transformation import BaseRerankConfig
from litellm.secret_managers.main import get_secret_str
@ -31,11 +32,7 @@ class CohereRerankConfig(BaseRerankConfig):
optional_params: dict | None = None,
) -> str:
if api_base:
# Remove trailing slashes and ensure clean base URL
api_base = api_base.rstrip("/")
if not api_base.endswith("/v1/rerank"):
api_base = f"{api_base}/v1/rerank"
return api_base
return ensure_api_route(api_base, "v1/rerank")
return "https://api.cohere.ai/v1/rerank"
def get_supported_cohere_rerank_params(self, model: str) -> list:

7
litellm/llms/cohere/rerank_v2/transformation.py
View file

@ -1,5 +1,6 @@
from typing import Any, Final
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.cohere.rerank.transformation import CohereRerankConfig
from litellm.types.rerank import OptionalRerankParams, RerankRequest
@ -19,11 +20,7 @@ class CohereRerankV2Config(CohereRerankConfig):
optional_params: dict | None = None,
) -> str:
if api_base:
# Remove trailing slashes and ensure clean base URL
api_base = api_base.rstrip("/")
if not api_base.endswith("/v2/rerank"):
api_base = f"{api_base}/v2/rerank"
return api_base
return ensure_api_route(api_base, "v2/rerank")
return "https://api.cohere.ai/v2/rerank"
def get_supported_cohere_rerank_params(self, model: str) -> list:

11
litellm/llms/cometapi/chat/transformation.py
View file

@ -10,6 +10,7 @@ from typing import Any, Final
import httpx
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolParam
@ -112,16 +113,14 @@ class CometAPIConfig(OpenAIGPTConfig):
# Ensure we include /v1 prefix when missing
if api_base.endswith("/v1"):
return f"{api_base}/{endpoint}"
if api_base.endswith("/v1/"):
return f"{api_base}{endpoint}"
return join_api_base(api_base, endpoint)
# If user provided https://api.cometapi.com, add /v1
if api_base == "https://api.cometapi.com":
return f"{api_base}/v1/{endpoint}"
return join_api_base(api_base, f"v1/{endpoint}")
# Generic fallback: if '/v1' not in path, add it
if "/v1" not in api_base.split("//", 1)[-1]:
return f"{api_base}/v1/{endpoint}"
return f"{api_base}/{endpoint}"
return join_api_base(api_base, f"v1/{endpoint}")
return join_api_base(api_base, endpoint)
def get_error_class(self, error_message: str, status_code: int, headers: dict | httpx.Headers) -> BaseLLMException:
"""

5
litellm/llms/cometapi/embed/transformation.py
View file

@ -9,6 +9,7 @@ import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.embedding.transformation import BaseEmbeddingConfig
from litellm.secret_managers.main import get_secret_str
@ -44,9 +45,7 @@ class CometAPIEmbeddingConfig(BaseEmbeddingConfig):
"""
Get the complete URL for the CometAPI embedding endpoint.
"""
api_base = "https://api.cometapi.com/v1" if api_base is None else api_base.rstrip("/")
complete_url: Final = f"{api_base}/embeddings"
return complete_url
return join_api_base("https://api.cometapi.com/v1" if api_base is None else api_base, "embeddings")
def validate_environment(
self,

8
litellm/llms/cometapi/image_generation/transformation.py
View file

@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
)
@ -77,16 +78,13 @@ class CometAPIImageGenerationConfig(BaseImageGenerationConfig):
"""
Get the complete url for the request
"""
complete_url: str = (
base_url: Final = (
api_base
or get_secret_str("COMETAPI_BASE_URL")
or get_secret_str("COMETAPI_API_BASE")
or self.DEFAULT_BASE_URL
)
complete_url = complete_url.rstrip("/")
complete_url = f"{complete_url}/{self.IMAGE_GENERATION_ENDPOINT}"
return complete_url
return join_api_base(base_url, self.IMAGE_GENERATION_ENDPOINT)
def validate_environment(
self,

6
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -59,7 +59,7 @@ from litellm.litellm_core_utils.realtime_errors import (
websocket_close_reason,
)
from litellm.litellm_core_utils.realtime_streaming import RealTimeStreaming
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, join_api_base
from litellm.llms.base_llm.anthropic_messages.transformation import (
BaseAnthropicMessagesConfig,
)
@ -6937,7 +6937,7 @@ class BaseLLMHTTPHandler:
headers={}, model=model or "", api_key=api_key
)
else:
url = f"{api_base.rstrip('/')}/v1/realtime/{endpoint}"
url = join_api_base(api_base, f"v1/realtime/{endpoint}")
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
@ -7006,7 +7006,7 @@ class BaseLLMHTTPHandler:
url = provider_config.get_realtime_calls_url(api_base=api_base, model=model or "", api_version=api_version)
headers: dict[str, object] = provider_config.get_realtime_calls_headers(ephemeral_key=openai_ephemeral_key)
else:
url = f"{api_base.rstrip('/')}/v1/realtime/calls"
url = join_api_base(api_base, "v1/realtime/calls")
headers = {
"Authorization": f"Bearer {openai_ephemeral_key}",
}

6
litellm/llms/dashscope/chat/transformation.py
View file

@ -5,6 +5,7 @@ Translates from OpenAI's `/v1/chat/completions` to DashScope's `/v1/chat/complet
from collections.abc import Coroutine
from typing import Final, Literal, overload
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolParam
@ -82,7 +83,4 @@ class DashScopeChatConfig(OpenAIGPTConfig):
"""
If api_base is not provided, use the default DashScope /chat/completions endpoint.
"""
resolved_api_base: Final = self._resolve_chat_api_base(api_base).rstrip("/")
if resolved_api_base.endswith("/chat/completions"):
return resolved_api_base
return f"{resolved_api_base}/chat/completions"
return ensure_api_route(self._resolve_chat_api_base(api_base), "chat/completions")

6
litellm/llms/dashscope/embed/transformation.py
View file

@ -16,6 +16,7 @@ from typing import Final
import httpx
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.embedding.transformation import BaseEmbeddingConfig
from litellm.secret_managers.main import get_secret_str
@ -98,10 +99,7 @@ class DashScopeEmbeddingConfig(BaseEmbeddingConfig):
litellm_params: dict,
stream: bool | None = None,
) -> str:
base: Final = self._resolve_embedding_api_base(api_base).rstrip("/")
if base.endswith("/embeddings"):
return base
return f"{base}/embeddings"
return ensure_api_route(self._resolve_embedding_api_base(api_base), "embeddings")
def transform_embedding_request(
self,

5
litellm/llms/dashscope/rerank/transformation.py
View file

@ -34,6 +34,7 @@ from pydantic import ConfigDict, TypeAdapter
from litellm._uuid import uuid
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.rerank.transformation import BaseRerankConfig
from litellm.secret_managers.main import get_secret_str
@ -94,10 +95,10 @@ class DashScopeRerankConfig(BaseRerankConfig):
return cleaned
if cleaned.endswith("/v1"):
return f"{cleaned}/reranks"
return join_api_base(cleaned, "reranks")
# Unknown base: append /reranks rather than silently ignoring the caller's api_base.
return f"{cleaned}/reranks"
return join_api_base(cleaned, "reranks")
def validate_environment(
self,

6
litellm/llms/dashscope/videos/transformation.py
View file

@ -16,7 +16,7 @@ from pydantic import BaseModel, Field, StrictBool, StrictInt, StrictStr, TypeAda
import litellm
from litellm.exceptions import UnsupportedParamsError
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.videos.async_task_utils import (
input_reference_url,
@ -708,7 +708,7 @@ class DashScopeVideoConfig(BaseVideoConfig):
if parameters:
request_data["parameters"] = parameters
return request_data, (), f"{api_base}{DASHSCOPE_VIDEO_SYNTHESIS_PATH}"
return request_data, (), join_api_base(api_base, DASHSCOPE_VIDEO_SYNTHESIS_PATH)
@staticmethod
def _reference_input(
@ -773,7 +773,7 @@ class DashScopeVideoConfig(BaseVideoConfig):
def _task_url(video_id: str, api_base: str) -> str:
original_task_id: Final = extract_original_video_id(video_id)
encoded_task_id: Final = encode_url_path_segment(original_task_id, field_name="video_id")
return f"{api_base}{DASHSCOPE_TASKS_PATH}/{encoded_task_id}"
return join_api_base(api_base, f"{DASHSCOPE_TASKS_PATH}/{encoded_task_id}")
def transform_video_status_retrieve_response(
self,

6
litellm/llms/databricks/chat/transformation.py
View file

@ -22,6 +22,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
strip_name_from_message,
strip_prefill_prefix_field,
)
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
from litellm.llms.base_llm.base_utils import (
type_to_response_format_param, # pyright: ignore[reportUnknownVariableType] # base_utils helper returns an untyped dict
@ -339,10 +340,7 @@ class DatabricksConfig(DatabricksBase, OpenAILikeChatConfig, AnthropicConfig):
stream: bool | None = None,
) -> str:
use_ai_gateway: Final = self.uses_ai_gateway(model)
api_base = self._get_api_base(api_base, use_ai_gateway=use_ai_gateway)
url_base: Final = api_base.rstrip("/") if use_ai_gateway else api_base
complete_url: Final = f"{url_base}/chat/completions"
return complete_url
return join_api_base(self._get_api_base(api_base, use_ai_gateway=use_ai_gateway), "chat/completions")
def get_supported_openai_params(self, model: str | None = None) -> list:
return [

12
litellm/llms/databricks/common_utils.py
View file

@ -20,6 +20,7 @@ from urllib.parse import urlsplit, urlunsplit
from pydantic import BaseModel, ConfigDict
import litellm
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.secret_managers.main import get_secret_str
@ -278,10 +279,10 @@ class DatabricksBase:
if normalized_api_base.endswith("/ai-gateway/mlflow/v1"):
return normalized_api_base
if normalized_api_base.endswith("/serving-endpoints"):
return f"{normalized_api_base.removesuffix('/serving-endpoints')}/ai-gateway/mlflow/v1"
return join_api_base(normalized_api_base.removesuffix("/serving-endpoints"), "ai-gateway/mlflow/v1")
api_base_parts: Final = urlsplit(normalized_api_base)
if api_base_parts.path in ("", "/"):
return f"{normalized_api_base}/ai-gateway/mlflow/v1"
return join_api_base(normalized_api_base, "ai-gateway/mlflow/v1")
return api_base
def get_oauth_m2m_token(
@ -319,7 +320,7 @@ class DatabricksBase:
api_base_parts: Final = urlsplit(api_base)
workspace_url: Final = urlunsplit((api_base_parts.scheme, api_base_parts.netloc, "", "", ""))
token_url: Final = f"{workspace_url}/oidc/v1/token"
token_url: Final = join_api_base(workspace_url, "oidc/v1/token")
try:
response: Final = requests.post(
@ -477,5 +478,6 @@ class DatabricksBase:
# Debug logging with redaction (never log actual tokens)
verbose_logger.debug("Databricks request headers: %s", self.redact_headers_for_logging(headers))
endpoint_path: Final = "" if custom_endpoint is True else DATABRICKS_ENDPOINT_PATHS[endpoint_type]
return f"{api_base}{endpoint_path}", headers
if custom_endpoint is True:
return api_base, headers
return join_api_base(api_base, DATABRICKS_ENDPOINT_PATHS[endpoint_type]), headers

5
litellm/llms/databricks/decisions/transformation.py
View file

@ -6,6 +6,7 @@ from typing import Final
from pydantic import TypeAdapter, ValidationError
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.decisions.transformation import BaseDecisionsConfig
from litellm.types.decisions import DecisionsIRRequest, UnsupportedDecisionsRequest
@ -84,8 +85,8 @@ class DatabricksDecisionsConfig(BaseDecisionsConfig):
def get_complete_url(self, api_base: str, model: str) -> str:
if is_ai_decide_model(model):
return f"{databricks_workspace_host(api_base)}{_AI_DECIDE_PATH}"
return f"{api_base.rstrip('/')}/{validate_databricks_decisions_model(model)}/invocations"
return join_api_base(databricks_workspace_host(api_base), _AI_DECIDE_PATH)
return join_api_base(api_base, f"{validate_databricks_decisions_model(model)}/invocations")
def transform_decisions_request(
self,

4
litellm/llms/databricks/responses/transformation.py
View file

@ -10,6 +10,7 @@ Reference: https://docs.databricks.com/aws/en/machine-learning/foundation-model-
import os
from typing import TYPE_CHECKING, Any, Final
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.databricks.common_utils import DatabricksBase
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.types.llms.openai import ResponseInputParam
@ -70,8 +71,7 @@ class DatabricksResponsesAPIConfig(DatabricksBase, OpenAIResponsesAPIConfig):
) -> str:
api_base = api_base or os.getenv("DATABRICKS_API_BASE")
api_base = self._get_api_base(api_base)
api_base = api_base.rstrip("/")
return f"{api_base}/responses"
return join_api_base(api_base, "responses")
def transform_responses_api_request(
self,

4
litellm/llms/deepgram/audio_transcription/transformation.py
View file

@ -10,6 +10,7 @@ from httpx import Headers, Response
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import (
@ -179,7 +180,6 @@ class DeepgramAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
) -> str:
if api_base is None:
api_base = get_secret_str("DEEPGRAM_API_BASE") or "https://api.deepgram.com/v1"
api_base = api_base.rstrip("/") # Remove trailing slash if present
# Build query parameters including the model
all_query_params: Final = {"model": model}
@ -189,7 +189,7 @@ class DeepgramAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
all_query_params.update(additional_params)
# Construct URL with proper query string encoding
base_url: Final = f"{api_base}/listen"
base_url: Final = join_api_base(api_base, "listen")
query_string: Final = urlencode(all_query_params)
url: Final = f"{base_url}?{query_string}"

3
litellm/llms/deepgram/common_utils.py
View file

@ -8,6 +8,7 @@ import httpx
import litellm
from litellm.constants import DEEPGRAM_DEFAULT_API_BASE, DEEPGRAM_LISTEN_DEFAULT_MODEL
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.types.utils import LlmProviders
@ -50,7 +51,7 @@ def _first_occurrences(query_string: str) -> httpx.QueryParams:
def deepgram_listen_websocket_target(api_base: str | None, query_string: str) -> str:
listen_url: Final = httpx.URL(f"{(api_base or DEEPGRAM_DEFAULT_API_BASE).rstrip('/')}/listen")
listen_url: Final = httpx.URL(join_api_base(api_base or DEEPGRAM_DEFAULT_API_BASE, "listen"))
websocket_url: Final = listen_url.copy_with(scheme=_WEBSOCKET_SCHEMES.get(listen_url.scheme, listen_url.scheme))
params: Final = _first_occurrences(query_string)
query: Final = params if params.get("model") else params.remove("model").add("model", DEEPGRAM_LISTEN_DEFAULT_MODEL)

8
litellm/llms/deepseek/chat/transformation.py
View file

@ -10,6 +10,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
convert_content_list_to_str,
extract_search_results_text,
)
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllMessageValues
from litellm.utils import supports_reasoning, supports_vision
@ -363,9 +364,4 @@ class DeepSeekChatConfig(OpenAIGPTConfig):
"""
If api_base is not provided, use the default DeepSeek /chat/completions endpoint.
"""
clean_api_base: Final = (api_base or "https://api.deepseek.com/beta").rstrip("/")
if not clean_api_base.endswith("/chat/completions"):
return f"{clean_api_base}/chat/completions"
return clean_api_base
return ensure_api_route(api_base or "https://api.deepseek.com/beta", "chat/completions")

5
litellm/llms/deepseek/messages/transformation.py
View file

@ -6,6 +6,7 @@ from collections.abc import Sequence
from typing import Any, Final
import litellm
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.anthropic.pass_through.messages.transformation import (
AnthropicMessagesConfig,
)
@ -90,9 +91,9 @@ class DeepSeekAnthropicMessagesConfig(AnthropicMessagesConfig):
base_url = base_url.removesuffix("/beta")
if not base_url.endswith("/anthropic") and "/anthropic/" not in base_url:
base_url = f"{base_url}/anthropic"
base_url = join_api_base(base_url, "anthropic")
return f"{base_url}/v1/messages"
return join_api_base(base_url, "v1/messages")
@staticmethod
def _sanitize_tools_for_deepseek(tools: Any) -> object:

6
litellm/llms/docker_model_runner/chat/transformation.py
View file

@ -10,6 +10,7 @@ from typing import Final, Literal, overload
from litellm.litellm_core_utils.prompt_templates.common_utils import (
handle_messages_with_content_list_to_str_conversion,
)
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllMessageValues
@ -104,12 +105,9 @@ class DockerModelRunnerChatConfig(OpenAIGPTConfig):
if not api_base:
api_base = "http://localhost:22088/engines/llama.cpp"
# Remove trailing slashes from api_base
api_base = api_base.rstrip("/")
# Build the URL: {api_base}/v1/chat/completions
# api_base is expected to already contain the engine path
complete_url: Final = f"{api_base}/v1/chat/completions"
complete_url: Final = join_api_base(api_base, "v1/chat/completions")
return complete_url

3
litellm/llms/duckduckgo/search/transformation.py
View file

@ -12,6 +12,7 @@ import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.search.transformation import (
BaseSearchConfig,
SearchResponse,
@ -91,7 +92,7 @@ class DuckDuckGoSearchConfig(BaseSearchConfig):
if data and isinstance(data, dict) and "_duckduckgo_params" in data:
params: Final = data["_duckduckgo_params"]
query_string: Final = urlencode(params, doseq=True)
return f"{api_base}/?{query_string}"
return f"{join_api_base(api_base, '')}?{query_string}"
return api_base

7
litellm/llms/e2b/sandbox/transformation.py
View file

@ -15,6 +15,7 @@ import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.sandbox.transformation import (
SANDBOX_MAX_OUTPUT_BYTES,
BaseSandboxConfig,
@ -78,7 +79,7 @@ class E2BSandboxConfig(BaseSandboxConfig):
body["metadata"] = metadata
response: Final = await self._http(client).post(
url=f"{base}/sandboxes",
url=join_api_base(base, "sandboxes"),
headers={"X-API-Key": key, "Content-Type": "application/json"},
json=body,
)
@ -151,10 +152,10 @@ class E2BSandboxConfig(BaseSandboxConfig):
dict[str, object], getattr(handle, HIDDEN_PARAMS_ATTR)
)
key: Final = api_key or hidden_params.get("api_key") or self.validate_environment()
base: Final = api_base or hidden_params.get("api_base") or E2B_API_BASE
base: Final = str(api_base or hidden_params.get("api_base") or E2B_API_BASE)
try:
response: Final = await self._http(client).delete(
url=f"{base}/sandboxes/{handle.id}",
url=join_api_base(base, f"sandboxes/{handle.id}"),
headers={"X-API-Key": key},
)
except httpx.HTTPStatusError as e:

4
litellm/llms/edenai/chat/transformation.py
View file

@ -23,7 +23,7 @@ from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import ModelResponse, ModelResponseStream, Usage
from ..common_utils import EdenAIException, reported_cost, resolve_api_base, resolve_api_key
from ..common_utils import EdenAIException, endpoint_url, reported_cost, resolve_api_base, resolve_api_key
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -138,7 +138,7 @@ class EdenAIChatConfig(OpenAIGPTConfig):
def get_models(
self, api_key: str | None = None, api_base: str | None = None
) -> list[str]: # mutable-ok: inherited contract
response: Final = litellm.module_level_client.get(url=f"{self.get_api_base(api_base)}/models")
response: Final = litellm.module_level_client.get(url=endpoint_url(api_base, "models"))
if not response.is_success:
raise EdenAIException(status_code=response.status_code, message=response.text, headers=response.headers)
catalog: Final = _EdenAIModelCatalog.model_validate(response.json())

3
litellm/llms/edenai/common_utils.py
View file

@ -11,6 +11,7 @@ from pydantic import AliasChoices, Field, ValidationError
import litellm
from litellm.exceptions import AuthenticationError
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.base import LiteLLMBaseModel
@ -74,7 +75,7 @@ def json_headers(
def endpoint_url(api_base: str | None, path: str) -> str:
return f"{resolve_api_base(api_base).rstrip('/')}/{path}"
return join_api_base(resolve_api_base(api_base), path)
def pick(params: Mapping[str, object], keys: Container[str]) -> Mapping[str, object]:

8
litellm/llms/elevenlabs/audio_transcription/transformation.py
View file

@ -10,6 +10,7 @@ from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import (
@ -165,12 +166,7 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
) -> str:
if api_base is None:
api_base = get_secret_str("ELEVENLABS_API_BASE") or "https://api.elevenlabs.io"
api_base = api_base.rstrip("/") # Remove trailing slash if present
# ElevenLabs speech-to-text endpoint
url: Final = f"{api_base}/v1/speech-to-text"
return url
return join_api_base(api_base, "v1/speech-to-text")
def validate_environment(
self,

7
litellm/llms/elevenlabs/text_to_speech/transformation.py
View file

@ -11,7 +11,7 @@ import httpx
from httpx import Headers
import litellm
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.litellm_core_utils.url_utils import encode_url_path_segment, join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.text_to_speech.transformation import (
BaseTextToSpeechConfig,
@ -297,15 +297,14 @@ class ElevenLabsTextToSpeechConfig(BaseTextToSpeechConfig):
"""
Construct the ElevenLabs endpoint URL, including path voice_id and query params.
"""
base_url = api_base or get_secret_str("ELEVENLABS_API_BASE") or self.TTS_BASE_URL
base_url = base_url.rstrip("/")
base_url: Final = api_base or get_secret_str("ELEVENLABS_API_BASE") or self.TTS_BASE_URL
voice_id: Final = litellm_params.get(self.ELEVENLABS_VOICE_ID_KEY)
if not isinstance(voice_id, str) or not voice_id.strip():
raise ValueError("ElevenLabs voice_id is required. Pass `voice` when calling `litellm.speech()`.")
encoded_voice_id: Final = encode_url_path_segment(voice_id, field_name="voice_id")
url = f"{base_url}{self.TTS_ENDPOINT_PATH}/{encoded_voice_id}"
url = join_api_base(base_url, f"{self.TTS_ENDPOINT_PATH}/{encoded_voice_id}")
query_params: Final = litellm_params.get(self.ELEVENLABS_QUERY_PARAMS_KEY, {})
if query_params:

7
litellm/llms/exa_ai/search/transformation.py
View file

@ -11,6 +11,7 @@ import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.url_utils import ensure_api_route
from litellm.llms.base_llm.search.transformation import (
BaseSearchConfig,
SearchResponse,
@ -93,11 +94,7 @@ class ExaAISearchConfig(BaseSearchConfig):
"""
Get complete URL for Search endpoint.
"""
resolved_api_base: Final = api_base or get_secret_str("EXA_API_BASE") or self.EXA_AI_API_BASE
clean_api_base: Final = resolved_api_base.rstrip("/")
if clean_api_base.endswith("/search"):
return clean_api_base
return f"{clean_api_base}/search"
return ensure_api_route(api_base or get_secret_str("EXA_API_BASE") or self.EXA_AI_API_BASE, "search")
def transform_search_request(
self,

Some files were not shown because too many files have changed in this diff Show more