scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-09 22:31:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 87

fix: resolve merge conflicts with staging branch

Browse source 
Keep both structured output tests (ours) and min thinking budget tests
(staging). Accept staging poetry.lock.
This commit is contained in:
Nicholas Gigliotti 2026-02-16 20:27:36 -05:00
commit a946cc4bc9
20 changed files with 1037 additions and 395 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
docs/my-website/docs/adding_provider/generic_guardrail_api.md
View file

@ -237,6 +237,7 @@ litellm_settings:
mode: pre_call # or post_call, during_call
api_base: https://your-guardrail-api.com
api_key: os.environ/YOUR_GUARDRAIL_API_KEY # optional
unreachable_fallback: fail_closed # default: fail_closed. Set to fail_open to proceed if the guardrail endpoint is unreachable (network errors, or HTTP 502/503/504 from an upstream proxy/LB).
additional_provider_specific_params:
# your custom parameters
threshold: 0.8

214
litellm/completion_extras/litellm_responses_transformation/transformation.py
View file

@ -62,9 +62,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
def __init__(self):
pass
def _handle_raw_dict_response_item(
self, item: Dict[str, Any], index: int
) -> Tuple[Optional[Any], int]:
def _handle_raw_dict_response_item(self, item: Dict[str, Any], index: int) -> Tuple[Optional[Any], int]:
"""
Handle raw dict response items from Responses API (e.g., GPT-5 Codex format).
@ -107,13 +105,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
if item_type == "function_call":
# Extract provider_specific_fields if present and pass through as-is
provider_specific_fields = item.get("provider_specific_fields")
if provider_specific_fields and not isinstance(
provider_specific_fields, dict
):
if provider_specific_fields and not isinstance(provider_specific_fields, dict):
provider_specific_fields = (
dict(provider_specific_fields)
if hasattr(provider_specific_fields, "__dict__")
else {}
dict(provider_specific_fields) if hasattr(provider_specific_fields, "__dict__") else {}
)
tool_call_dict = {
@ -129,9 +123,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
if provider_specific_fields:
tool_call_dict["provider_specific_fields"] = provider_specific_fields
# Also add to function's provider_specific_fields for consistency
tool_call_dict["function"][
"provider_specific_fields"
] = provider_specific_fields
tool_call_dict["function"]["provider_specific_fields"] = provider_specific_fields
msg = Message(
content=None,
@ -169,7 +161,8 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
"type": "message",
"role": role,
"content": self._convert_content_to_responses_format(
content, role # type: ignore
content,
role, # type: ignore
),
}
)
@ -186,7 +179,8 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
elif isinstance(content, list):
# Transform list content to Responses API format
tool_output = self._convert_content_to_responses_format(
content, "user" # Use "user" role to get input_* types
content,
"user", # Use "user" role to get input_* types
)
else:
# Fallback: convert unexpected types to input_text
@ -219,9 +213,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
{
"type": "message",
"role": role,
"content": self._convert_content_to_responses_format(
content, cast(str, role)
),
"content": self._convert_content_to_responses_format(content, cast(str, role)),
}
)
@ -344,9 +336,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
previous_response_id = optional_params.get("previous_response_id")
if previous_response_id:
# Use the existing session handler for responses API
verbose_logger.debug(
f"Chat provider: Warning ignoring previous response ID: {previous_response_id}"
)
verbose_logger.debug(f"Chat provider: Warning ignoring previous response ID: {previous_response_id}")
# Convert back to responses API format for the actual request
@ -368,9 +358,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
"client": client,
}
verbose_logger.debug(
f"Chat provider: Final request model={api_model}, input_items={len(input_items)}"
)
verbose_logger.debug(f"Chat provider: Final request model={api_model}, input_items={len(input_items)}")
self._merge_responses_api_request_into_request_data(
request_data, responses_api_request, instructions
@ -450,9 +438,11 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
LiteLLMCompletionResponsesConfig,
)
tool_call_dict = LiteLLMCompletionResponsesConfig.convert_response_function_tool_call_to_chat_completion_tool_call(
tool_call_item=item,
index=tool_call_index,
tool_call_dict = (
LiteLLMCompletionResponsesConfig.convert_response_function_tool_call_to_chat_completion_tool_call(
tool_call_item=item,
index=tool_call_index,
)
)
accumulated_tool_calls.append(tool_call_dict)
tool_call_index += 1
@ -472,9 +462,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
tool_calls=accumulated_tool_calls,
reasoning_content=reasoning_content,
)
choices.append(
Choices(message=msg, finish_reason="tool_calls", index=index)
)
choices.append(Choices(message=msg, finish_reason="tool_calls", index=index))
reasoning_content = None
return choices
@ -510,17 +498,10 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
)
if len(choices) == 0:
if (
raw_response.incomplete_details is not None
and raw_response.incomplete_details.reason is not None
):
raise ValueError(
f"{model} unable to complete request: {raw_response.incomplete_details.reason}"
)
if raw_response.incomplete_details is not None and raw_response.incomplete_details.reason is not None:
raise ValueError(f"{model} unable to complete request: {raw_response.incomplete_details.reason}")
else:
raise ValueError(
f"Unknown items in responses API response: {raw_response.output}"
)
raise ValueError(f"Unknown items in responses API response: {raw_response.output}")
setattr(model_response, "choices", choices)
@ -529,11 +510,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
setattr(
model_response,
"usage",
ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(
raw_response.usage
),
ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(raw_response.usage),
)
# Preserve hidden params from the ResponsesAPIResponse, especially the headers
# which contain important provider information like x-request-id
raw_response_hidden_params = getattr(raw_response, "_hidden_params", {})
@ -550,24 +529,18 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
model_response._hidden_params[key] = merged_headers
else:
model_response._hidden_params[key] = value
return model_response
def get_model_response_iterator(
self,
streaming_response: Union[
Iterator[str], AsyncIterator[str], "ModelResponse", "BaseModel"
],
streaming_response: Union[Iterator[str], AsyncIterator[str], "ModelResponse", "BaseModel"],
sync_stream: bool,
json_mode: Optional[bool] = False,
) -> BaseModelResponseIterator:
return OpenAiResponsesToChatCompletionStreamIterator(
streaming_response, sync_stream, json_mode
)
return OpenAiResponsesToChatCompletionStreamIterator(streaming_response, sync_stream, json_mode)
def _convert_content_str_to_input_text(
self, content: str, role: str
) -> Dict[str, Any]:
def _convert_content_str_to_input_text(self, content: str, role: str) -> Dict[str, Any]:
if role == "user" or role == "system" or role == "tool":
return {"type": "input_text", "text": content}
else:
@ -594,9 +567,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
if actual_image_url is None:
raise ValueError(f"Invalid image URL: {content_image_url}")
image_param = ResponseInputImageParam(
image_url=actual_image_url, detail="auto", type="input_image"
)
image_param = ResponseInputImageParam(image_url=actual_image_url, detail="auto", type="input_image")
if detail:
image_param["detail"] = detail
@ -607,18 +578,14 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
self,
content: Union[
str,
Iterable[
Union["OpenAIMessageContentListBlock", "ChatCompletionThinkingBlock"]
],
Iterable[Union["OpenAIMessageContentListBlock", "ChatCompletionThinkingBlock"]],
],
role: str,
) -> List[Dict[str, Any]]:
"""Convert chat completion content to responses API format"""
from litellm.types.llms.openai import ChatCompletionImageObject
verbose_logger.debug(
f"Chat provider: Converting content to responses format - input type: {type(content)}"
)
verbose_logger.debug(f"Chat provider: Converting content to responses format - input type: {type(content)}")
if isinstance(content, str):
result = [self._convert_content_str_to_input_text(content, role)]
@ -627,9 +594,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
elif isinstance(content, list):
result = []
for i, item in enumerate(content):
verbose_logger.debug(
f"Chat provider: Processing content item {i}: {type(item)} = {item}"
)
verbose_logger.debug(f"Chat provider: Processing content item {i}: {type(item)} = {item}")
if isinstance(item, str):
converted = self._convert_content_str_to_input_text(item, role)
result.append(converted)
@ -638,9 +603,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
# Handle multimodal content
original_type = item.get("type")
if original_type == "text":
converted = self._convert_content_str_to_input_text(
item.get("text", ""), role
)
converted = self._convert_content_str_to_input_text(item.get("text", ""), role)
result.append(converted)
verbose_logger.debug(f"Chat provider: text -> {converted}")
elif original_type == "image_url":
@ -652,18 +615,14 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
),
)
result.append(converted)
verbose_logger.debug(
f"Chat provider: image_url -> {converted}"
)
verbose_logger.debug(f"Chat provider: image_url -> {converted}")
else:
# Try to map other types to responses API format
item_type = original_type or "input_text"
if item_type == "image":
converted = {"type": "input_image", **item}
result.append(converted)
verbose_logger.debug(
f"Chat provider: image -> {converted}"
)
verbose_logger.debug(f"Chat provider: image -> {converted}")
elif item_type in [
"input_text",
"input_image",
@ -675,18 +634,12 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
]:
# Already in responses API format
result.append(item)
verbose_logger.debug(
f"Chat provider: passthrough -> {item}"
)
verbose_logger.debug(f"Chat provider: passthrough -> {item}")
else:
# Default to input_text for unknown types
converted = self._convert_content_str_to_input_text(
str(item.get("text", item)), role
)
converted = self._convert_content_str_to_input_text(str(item.get("text", item)), role)
result.append(converted)
verbose_logger.debug(
f"Chat provider: unknown({original_type}) -> {converted}"
)
verbose_logger.debug(f"Chat provider: unknown({original_type}) -> {converted}")
verbose_logger.debug(f"Chat provider: Final converted content: {result}")
return result
else:
@ -694,17 +647,13 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
verbose_logger.debug(f"Chat provider: Other content type -> {result}")
return result
def _convert_tools_to_responses_format(
self, tools: List[Dict[str, Any]]
) -> List["ALL_RESPONSES_API_TOOL_PARAMS"]:
def _convert_tools_to_responses_format(self, tools: List[Dict[str, Any]]) -> List["ALL_RESPONSES_API_TOOL_PARAMS"]:
"""Convert chat completion tools to responses API tools format"""
responses_tools: List["ALL_RESPONSES_API_TOOL_PARAMS"] = []
for tool in tools:
# convert function tool from chat completion to responses API format
if tool.get("type") == "function":
function_tool = cast(
ChatCompletionToolParamFunctionChunk, tool.get("function")
)
function_tool = cast(ChatCompletionToolParamFunctionChunk, tool.get("function"))
responses_tools.append(
FunctionToolParam(
name=function_tool["name"],
@ -730,9 +679,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
if not extra_body:
return optional_params
supported_responses_api_params = set(
ResponsesAPIOptionalRequestParams.__annotations__.keys()
)
supported_responses_api_params = set(ResponsesAPIOptionalRequestParams.__annotations__.keys())
# Also include params we handle specially
supported_responses_api_params.update(
{
@ -750,9 +697,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
return optional_params
def _map_reasoning_effort(
self, reasoning_effort: Union[str, Dict[str, Any]]
) -> Optional[Reasoning]:
def _map_reasoning_effort(self, reasoning_effort: Union[str, Dict[str, Any]]) -> Optional[Reasoning]:
# If dict is passed, convert it directly to Reasoning object
if isinstance(reasoning_effort, dict):
return Reasoning(**reasoning_effort) # type: ignore[typeddict-item]
@ -760,8 +705,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
# Check if auto-summary is enabled via flag or environment variable
# Priority: litellm.reasoning_auto_summary flag > LITELLM_REASONING_AUTO_SUMMARY env var
auto_summary_enabled = (
litellm.reasoning_auto_summary
or os.getenv("LITELLM_REASONING_AUTO_SUMMARY", "false").lower() == "true"
litellm.reasoning_auto_summary or os.getenv("LITELLM_REASONING_AUTO_SUMMARY", "false").lower() == "true"
)
# If string is passed, map with optional summary based on flag/env var
@ -772,11 +716,15 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
elif reasoning_effort == "xhigh":
return Reasoning(effort="xhigh", summary="detailed") if auto_summary_enabled else Reasoning(effort="xhigh") # type: ignore[typeddict-item]
elif reasoning_effort == "medium":
return Reasoning(effort="medium", summary="detailed") if auto_summary_enabled else Reasoning(effort="medium")
return (
Reasoning(effort="medium", summary="detailed") if auto_summary_enabled else Reasoning(effort="medium")
)
elif reasoning_effort == "low":
return Reasoning(effort="low", summary="detailed") if auto_summary_enabled else Reasoning(effort="low")
elif reasoning_effort == "minimal":
return Reasoning(effort="minimal", summary="detailed") if auto_summary_enabled else Reasoning(effort="minimal")
return (
Reasoning(effort="minimal", summary="detailed") if auto_summary_enabled else Reasoning(effort="minimal")
)
return None
def _add_web_search_tool(
@ -855,7 +803,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
return {"format": {"type": "text"}}
return None
@staticmethod
def _convert_annotations_to_chat_format(
annotations: Optional[List[Any]],
@ -908,9 +856,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
def __init__(
self, streaming_response, sync_stream: bool, json_mode: Optional[bool] = False
):
def __init__(self, streaming_response, sync_stream: bool, json_mode: Optional[bool] = False):
super().__init__(streaming_response, sync_stream, json_mode)
def _handle_string_chunk(
@ -923,9 +869,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
if not str_line or str_line.startswith("event:"):
# ignore.
return GenericStreamingChunk(
text="", tool_use=None, is_finished=False, finish_reason="", usage=None
)
return GenericStreamingChunk(text="", tool_use=None, is_finished=False, finish_reason="", usage=None)
index = str_line.find("data:")
if index != -1:
str_line = str_line[index + 5 :]
@ -988,13 +932,9 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
if output_item.get("type") == "function_call":
# Extract provider_specific_fields if present
provider_specific_fields = output_item.get("provider_specific_fields")
if provider_specific_fields and not isinstance(
provider_specific_fields, dict
):
if provider_specific_fields and not isinstance(provider_specific_fields, dict):
provider_specific_fields = (
dict(provider_specific_fields)
if hasattr(provider_specific_fields, "__dict__")
else {}
dict(provider_specific_fields) if hasattr(provider_specific_fields, "__dict__") else {}
)
function_chunk = ChatCompletionToolCallFunctionChunk(
@ -1003,9 +943,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
)
if provider_specific_fields:
function_chunk["provider_specific_fields"] = (
provider_specific_fields
)
function_chunk["provider_specific_fields"] = provider_specific_fields
tool_call_chunk = ChatCompletionToolCallChunk(
id=output_item.get("call_id"),
@ -1040,9 +978,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
id=None,
index=0,
type="function",
function=ChatCompletionToolCallFunctionChunk(
name=None, arguments=content_part
),
function=ChatCompletionToolCallFunctionChunk(name=None, arguments=content_part),
)
]
),
@ -1051,22 +987,16 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
]
)
else:
raise ValueError(
f"Chat provider: Invalid function argument delta {parsed_chunk}"
)
raise ValueError(f"Chat provider: Invalid function argument delta {parsed_chunk}")
elif event_type == "response.output_item.done":
# New output item added
output_item = parsed_chunk.get("item", {})
if output_item.get("type") == "function_call":
# Extract provider_specific_fields if present
provider_specific_fields = output_item.get("provider_specific_fields")
if provider_specific_fields and not isinstance(
provider_specific_fields, dict
):
if provider_specific_fields and not isinstance(provider_specific_fields, dict):
provider_specific_fields = (
dict(provider_specific_fields)
if hasattr(provider_specific_fields, "__dict__")
else {}
dict(provider_specific_fields) if hasattr(provider_specific_fields, "__dict__") else {}
)
function_chunk = ChatCompletionToolCallFunctionChunk(
@ -1076,9 +1006,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
# Add provider_specific_fields to function if present
if provider_specific_fields:
function_chunk["provider_specific_fields"] = (
provider_specific_fields
)
function_chunk["provider_specific_fields"] = provider_specific_fields
tool_call_chunk = ChatCompletionToolCallChunk(
id=output_item.get("call_id"),
@ -1142,21 +1070,31 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
elif event_type == "response.completed":
# Response is fully complete - now we can signal is_finished=True
# This ensures we don't prematurely end the stream before tool_calls arrive
# Check if response contains function_call items in output
# to determine correct finish_reason
response_data = parsed_chunk.get("response", {})
output_items = response_data.get("output", []) if response_data else []
has_function_calls = any(
item.get("type") == "function_call" for item in output_items if isinstance(item, dict)
)
finish_reason = "tool_calls" if has_function_calls else "stop"
return ModelResponseStream(
choices=[
StreamingChoices(
index=0,
delta=Delta(content=""),
finish_reason="stop",
finish_reason=finish_reason,
)
]
)
else:
pass
# For any unhandled event types, create a minimal valid chunk or skip
verbose_logger.debug(
f"Chat provider: Unhandled event type '{event_type}', creating empty chunk"
)
verbose_logger.debug(f"Chat provider: Unhandled event type '{event_type}', creating empty chunk")
# Return a minimal valid chunk for unknown events
return ModelResponseStream(
@ -1179,9 +1117,5 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
Returns:
ModelResponseStream: OpenAI-formatted streaming chunk
"""
verbose_logger.debug(
f"Chat provider: transform_streaming_response called with chunk: {chunk}"
)
return OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(
chunk
)
verbose_logger.debug(f"Chat provider: transform_streaming_response called with chunk: {chunk}")
return OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(chunk)

74
litellm/integrations/prometheus.py
View file

@ -1055,16 +1055,16 @@ class PrometheusLogger(CustomLogger):
enum_values=enum_values,
)
if (
standard_logging_payload["stream"] is True
): # log successful streaming requests from logging event hook.
_labels = prometheus_label_factory(
supported_enum_labels=self.get_labels_for_metric(
metric_name="litellm_proxy_total_requests_metric"
),
enum_values=enum_values,
)
self.litellm_proxy_total_requests_metric.labels(**_labels).inc()
# increment litellm_proxy_total_requests_metric for all successful requests
# (both streaming and non-streaming) in this single location to prevent
# double-counting that occurs when async_post_call_success_hook also increments
_labels = prometheus_label_factory(
supported_enum_labels=self.get_labels_for_metric(
metric_name="litellm_proxy_total_requests_metric"
),
enum_values=enum_values,
)
self.litellm_proxy_total_requests_metric.labels(**_labels).inc()
def _increment_token_metrics(
self,
@ -1086,13 +1086,6 @@ class PrometheusLogger(CustomLogger):
):
_tags = standard_logging_payload["request_tags"]
_labels = prometheus_label_factory(
supported_enum_labels=self.get_labels_for_metric(
metric_name="litellm_proxy_total_requests_metric"
),
enum_values=enum_values,
)
_labels = prometheus_label_factory(
supported_enum_labels=self.get_labels_for_metric(
metric_name="litellm_total_tokens_metric"
@ -1655,49 +1648,12 @@ class PrometheusLogger(CustomLogger):
):
"""
Proxy level tracking - triggered when the proxy responds with a success response to the client
Note: litellm_proxy_total_requests_metric is NOT incremented here to avoid
double-counting. It is incremented in async_log_success_event which fires
for all successful requests (both streaming and non-streaming).
"""
try:
from litellm.litellm_core_utils.litellm_logging import (
StandardLoggingPayloadSetup,
)
if self._should_skip_metrics_for_invalid_key(
user_api_key_dict=user_api_key_dict
):
return
_metadata = data.get("metadata", {}) or {}
enum_values = UserAPIKeyLabelValues(
end_user=user_api_key_dict.end_user_id,
hashed_api_key=user_api_key_dict.api_key,
api_key_alias=user_api_key_dict.key_alias,
requested_model=data.get("model", ""),
team=user_api_key_dict.team_id,
team_alias=user_api_key_dict.team_alias,
user=user_api_key_dict.user_id,
user_email=user_api_key_dict.user_email,
status_code="200",
route=user_api_key_dict.request_route,
tags=StandardLoggingPayloadSetup._get_request_tags(
litellm_params=data,
proxy_server_request=data.get("proxy_server_request", {}),
),
client_ip=_metadata.get("requester_ip_address"),
user_agent=_metadata.get("user_agent"),
)
_labels = prometheus_label_factory(
supported_enum_labels=self.get_labels_for_metric(
metric_name="litellm_proxy_total_requests_metric"
),
enum_values=enum_values,
)
self.litellm_proxy_total_requests_metric.labels(**_labels).inc()
except Exception as e:
verbose_logger.exception(
"prometheus Layer Error(): Exception occured - {}".format(str(e))
)
pass
pass
def _safe_get(self, obj: Any, key: str, default: Any = None) -> Any:
"""Get value from dict or Pydantic model."""

3
litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/__init__.py
View file

@ -18,6 +18,9 @@ def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"
additional_provider_specific_params=getattr(
litellm_params, "additional_provider_specific_params", {}
),
unreachable_fallback=getattr(
litellm_params, "unreachable_fallback", "fail_closed"
),
guardrail_name=guardrail.get("guardrail_name", ""),
event_hook=litellm_params.mode,
default_on=litellm_params.default_on,

1
litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/example_config.yaml
View file

@ -14,6 +14,7 @@ litellm_settings:
mode: pre_call # Options: pre_call, post_call, during_call, [pre_call, post_call]
api_key: os.environ/GENERIC_GUARDRAIL_API_KEY # Optional if using Bearer auth
api_base: http://localhost:8080 # Required. Endpoint /beta/litellm_basic_guardrail_api is automatically appended
unreachable_fallback: fail_closed # Options: fail_closed (default, raise), fail_open (proceed if endpoint unreachable or upstream returns 502/503/504)
default_on: false # Set to true to apply to all requests by default
additional_provider_specific_params:
# Any additional parameters your guardrail API needs

152
litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/generic_guardrail_api.py
View file

@ -9,9 +9,11 @@ import fnmatch
import os
from typing import TYPE_CHECKING, Any, Dict, Literal, Optional
import httpx
from litellm._logging import verbose_proxy_logger
from litellm._version import version as litellm_version
from litellm.exceptions import GuardrailRaisedException
from litellm.exceptions import GuardrailRaisedException, Timeout
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
log_guardrail_information,
@ -34,17 +36,19 @@ if TYPE_CHECKING:
GUARDRAIL_NAME = "generic_guardrail_api"
# Headers whose values are forwarded as-is (case-insensitive). Glob patterns supported (e.g. x-stainless-*, x-litellm*).
_HEADER_VALUE_ALLOWLIST = frozenset({
"host",
"accept-encoding",
"connection",
"accept",
"content-type",
"user-agent",
"x-stainless-*",
"x-litellm-*",
"content-length",
})
_HEADER_VALUE_ALLOWLIST = frozenset(
{
"host",
"accept-encoding",
"connection",
"accept",
"content-type",
"user-agent",
"x-stainless-*",
"x-litellm-*",
"content-length",
}
)
# Placeholder for headers that exist but are not on the allowlist (we don't expose their value).
_HEADER_PRESENT_PLACEHOLDER = "[present]"
@ -166,6 +170,7 @@ class GenericGuardrailAPI(CustomGuardrail):
api_base: Optional[str] = None,
api_key: Optional[str] = None,
additional_provider_specific_params: Optional[Dict[str, Any]] = None,
unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
**kwargs,
):
self.async_handler = get_async_httpx_client(
@ -196,6 +201,10 @@ class GenericGuardrailAPI(CustomGuardrail):
additional_provider_specific_params or {}
)
self.unreachable_fallback: Literal["fail_closed", "fail_open"] = (
unreachable_fallback
)
# Set supported event hooks
if "supported_event_hooks" not in kwargs:
kwargs["supported_event_hooks"] = [
@ -259,6 +268,54 @@ class GenericGuardrailAPI(CustomGuardrail):
return result_metadata
def _fail_open_passthrough(
self,
*,
inputs: GenericGuardrailAPIInputs,
input_type: Literal["request", "response"],
logging_obj: Optional["LiteLLMLoggingObj"],
error: Exception,
http_status_code: Optional[int] = None,
) -> GenericGuardrailAPIInputs:
status_suffix = f" http_status_code={http_status_code}" if http_status_code else ""
verbose_proxy_logger.critical(
"Generic Guardrail API unreachable (fail-open). Proceeding without guardrail.%s "
"guardrail_name=%s api_base=%s input_type=%s litellm_call_id=%s litellm_trace_id=%s",
status_suffix,
getattr(self, "guardrail_name", None),
getattr(self, "api_base", None),
input_type,
getattr(logging_obj, "litellm_call_id", None) if logging_obj else None,
getattr(logging_obj, "litellm_trace_id", None) if logging_obj else None,
exc_info=error,
)
# Keep flow going - treat as action=NONE (no modifications)
return_inputs: GenericGuardrailAPIInputs = {}
return_inputs.update(inputs)
return return_inputs
def _build_guardrail_return_inputs(
self,
*,
texts: list,
images: Any,
tools: Any,
guardrail_response: GenericGuardrailAPIResponse,
) -> GenericGuardrailAPIInputs:
# Action is NONE or no modifications needed
return_inputs = GenericGuardrailAPIInputs(texts=texts)
if guardrail_response.texts:
return_inputs["texts"] = guardrail_response.texts
if guardrail_response.images:
return_inputs["images"] = guardrail_response.images
elif images:
return_inputs["images"] = images
if guardrail_response.tools:
return_inputs["tools"] = guardrail_response.tools
elif tools:
return_inputs["tools"] = tools
return return_inputs
@log_guardrail_information
async def apply_guardrail(
self,
@ -313,7 +370,9 @@ class GenericGuardrailAPI(CustomGuardrail):
# Extract user API key metadata
user_metadata = self._extract_user_api_key_metadata(request_data)
inbound_headers = _extract_inbound_headers(request_data=request_data, logging_obj=logging_obj)
inbound_headers = _extract_inbound_headers(
request_data=request_data, logging_obj=logging_obj
)
# Create request payload
guardrail_request = GenericGuardrailAPIRequest(
@ -370,23 +429,64 @@ class GenericGuardrailAPI(CustomGuardrail):
should_wrap_with_default_message=False,
)
# Action is NONE or no modifications needed
return_inputs = GenericGuardrailAPIInputs(texts=texts)
if guardrail_response.texts:
return_inputs["texts"] = guardrail_response.texts
if guardrail_response.images:
return_inputs["images"] = guardrail_response.images
elif images:
return_inputs["images"] = images
if guardrail_response.tools:
return_inputs["tools"] = guardrail_response.tools
elif tools:
return_inputs["tools"] = tools
return return_inputs
return self._build_guardrail_return_inputs(
texts=texts,
images=images,
tools=tools,
guardrail_response=guardrail_response,
)
except GuardrailRaisedException:
# Re-raise guardrail exceptions as-is
raise
except Timeout as e:
# AsyncHTTPHandler wraps httpx.TimeoutException into litellm.Timeout
if self.unreachable_fallback == "fail_open":
return self._fail_open_passthrough(
inputs=inputs,
input_type=input_type,
logging_obj=logging_obj,
error=e,
)
verbose_proxy_logger.error(
"Generic Guardrail API: failed to make request: %s", str(e)
)
raise Exception(f"Generic Guardrail API failed: {str(e)}")
except httpx.HTTPStatusError as e:
# Common reverse-proxy/LB failures can present as HTTP errors even when the backend is unreachable.
status_code = getattr(getattr(e, "response", None), "status_code", None)
if self.unreachable_fallback == "fail_open" and status_code in (
502,
503,
504,
):
return self._fail_open_passthrough(
inputs=inputs,
input_type=input_type,
logging_obj=logging_obj,
error=e,
http_status_code=status_code,
)
verbose_proxy_logger.error(
"Generic Guardrail API: failed to make request: %s", str(e)
)
raise Exception(f"Generic Guardrail API failed: {str(e)}")
except httpx.RequestError as e:
# Guardrail endpoint is unreachable (DNS/connect/timeout/etc)
if self.unreachable_fallback == "fail_open":
return self._fail_open_passthrough(
inputs=inputs,
input_type=input_type,
logging_obj=logging_obj,
error=e,
)
verbose_proxy_logger.error(
"Generic Guardrail API: failed to make request: %s", str(e)
)
raise Exception(f"Generic Guardrail API failed: {str(e)}")
except Exception as e:
verbose_proxy_logger.error(
"Generic Guardrail API: failed to make request: %s", str(e)

48
litellm/proxy/management_endpoints/key_management_endpoints.py
View file

@ -19,6 +19,7 @@ from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Literal, Optional, Tuple, cast
import fastapi
import prisma
import yaml
from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, status
@ -3093,13 +3094,17 @@ async def _rotate_master_key(
should_create_model_in_db=False,
)
if new_model:
new_models.append(jsonify_object(new_model.model_dump()))
_dumped = new_model.model_dump(exclude_none=True)
_dumped["litellm_params"] = prisma.Json(_dumped["litellm_params"])
_dumped["model_info"] = prisma.Json(_dumped["model_info"])
new_models.append(_dumped)
verbose_proxy_logger.debug("Resetting proxy model table")
await prisma_client.db.litellm_proxymodeltable.delete_many()
verbose_proxy_logger.debug("Creating %s models", len(new_models))
await prisma_client.db.litellm_proxymodeltable.create_many(
data=new_models,
)
async with prisma_client.db.tx() as tx:
await tx.litellm_proxymodeltable.delete_many()
verbose_proxy_logger.debug("Creating %s models", len(new_models))
await tx.litellm_proxymodeltable.create_many(
data=new_models,
)
# 3. process config table
try:
config = await prisma_client.db.litellm_config.find_many()
@ -3125,15 +3130,20 @@ async def _rotate_master_key(
if encrypted_env_vars:
await prisma_client.db.litellm_config.update(
where={"param_name": "environment_variables"},
data={"param_value": jsonify_object(encrypted_env_vars)},
data={"param_value": prisma.Json(encrypted_env_vars)},
)
# 4. process MCP server table
await rotate_mcp_server_credentials_master_key(
prisma_client=prisma_client,
touched_by=user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME,
new_master_key=new_master_key,
)
try:
await rotate_mcp_server_credentials_master_key(
prisma_client=prisma_client,
touched_by=user_api_key_dict.user_id or LITELLM_PROXY_ADMIN_NAME,
new_master_key=new_master_key,
)
except Exception as e:
verbose_proxy_logger.warning(
"Failed to rotate MCP server credentials: %s", str(e)
)
# 5. process credentials table
try:
@ -3151,13 +3161,19 @@ async def _rotate_master_key(
updated_patch=decrypted_cred,
new_encryption_key=new_master_key,
)
credential_object_jsonified = jsonify_object(
encrypted_cred.model_dump()
)
_cred_data = encrypted_cred.model_dump(exclude_none=True)
if "credential_values" in _cred_data:
_cred_data["credential_values"] = prisma.Json(
_cred_data["credential_values"]
)
if "credential_info" in _cred_data:
_cred_data["credential_info"] = prisma.Json(
_cred_data["credential_info"]
)
await prisma_client.db.litellm_credentialstable.update(
where={"credential_name": cred.credential_name},
data={
**credential_object_jsonified,
**_cred_data,
"updated_by": user_api_key_dict.user_id,
},
)

18
litellm/proxy/utils.py
View file

@ -456,18 +456,26 @@ class ProxyLogging:
def _init_litellm_callbacks(self, llm_router: Optional[Router] = None):
self._add_proxy_hooks(llm_router)
litellm.logging_callback_manager.add_litellm_callback(self.service_logging_obj) # type: ignore
for callback in litellm.callbacks:
# Track string callbacks and their initialized instances so we can
# replace them in-place, preventing duplicates (string + instance) in
# litellm.callbacks which caused double-counting of metrics.
string_callbacks_to_replace: Dict[int, CustomLogger] = {}
for idx, callback in enumerate(litellm.callbacks):
if isinstance(callback, str):
callback = litellm.litellm_core_utils.litellm_logging._init_custom_logger_compatible_class( # type: ignore
initialized_callback = litellm.litellm_core_utils.litellm_logging._init_custom_logger_compatible_class(
cast(_custom_logger_compatible_callbacks_literal, callback),
internal_usage_cache=self.internal_usage_cache.dual_cache,
llm_router=llm_router,
)
if callback is None:
continue
if initialized_callback is not None:
string_callbacks_to_replace[idx] = initialized_callback
litellm.logging_callback_manager.add_litellm_callback(callback)
# Replace string entries in litellm.callbacks with initialized instances
for idx, initialized_callback in string_callbacks_to_replace.items():
litellm.callbacks[idx] = initialized_callback
async def update_request_status(
self, litellm_call_id: str, status: Literal["success", "fail"]

17
litellm/responses/main.py
View file

@ -600,8 +600,12 @@ def responses(
# Update input and tools with provider-specific file IDs if managed files are used
#########################################################
model_file_id_mapping = kwargs.get("model_file_id_mapping")
model_info_id = kwargs.get("model_info", {}).get("id") if isinstance(kwargs.get("model_info"), dict) else None
model_info_id = (
kwargs.get("model_info", {}).get("id")
if isinstance(kwargs.get("model_info"), dict)
else None
)
input = cast(
Union[str, ResponseInputParam],
update_responses_input_with_model_file_ids(
@ -611,7 +615,7 @@ def responses(
),
)
local_vars["input"] = input
# Update tools with provider-specific file IDs if needed
if tools:
tools = cast(
@ -696,7 +700,10 @@ def responses(
)
)
# Pre Call logging
# Pre Call logging - preserve metadata for custom callbacks
# When called from completion bridge (codex models), metadata is in litellm_metadata
metadata_for_callbacks = metadata or kwargs.get("litellm_metadata") or {}
litellm_logging_obj.update_environment_variables(
model=model,
user=user,
@ -705,7 +712,7 @@ def responses(
**responses_api_request_params,
"aresponses": _is_async,
"litellm_call_id": litellm_call_id,
"metadata": metadata,
"metadata": metadata_for_callbacks,
},
custom_llm_provider=custom_llm_provider,
)

10
litellm/types/guardrails.py
View file

@ -651,6 +651,15 @@ class BaseLitellmParams(
description="Additional provider-specific parameters for generic guardrail APIs",
)
unreachable_fallback: Literal["fail_closed", "fail_open"] = Field(
default="fail_closed",
description=(
"Behavior when a guardrail endpoint is unreachable due to network errors. "
"NOTE: This is currently only implemented by guardrail='generic_guardrail_api'. "
"'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed."
),
)
# Custom code guardrail params
custom_code: Optional[str] = Field(
default=None,
@ -692,6 +701,7 @@ class LitellmParams(
"mode",
"default_action",
"on_disallowed_action",
"unreachable_fallback",
mode="before",
check_fields=False,
)

14
litellm/types/proxy/guardrails/guardrail_hooks/generic_guardrail_api.py
View file

@ -31,6 +31,14 @@ class GenericGuardrailAPIOptionalParams(BaseModel):
description="Additional provider-specific parameters to send with the guardrail request",
)
unreachable_fallback: Optional[Literal["fail_closed", "fail_open"]] = Field(
default="fail_closed",
description=(
"Behavior when the guardrail endpoint is unreachable due to network errors. "
"'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed."
),
)
class GenericGuardrailAPIConfigModel(
GuardrailConfigModel[GenericGuardrailAPIOptionalParams],
@ -52,9 +60,9 @@ class GenericGuardrailAPIRequest(BaseModel):
input_type: Literal["request", "response"]
litellm_call_id: Optional[str] = None # the call id of the individual LLM call
litellm_trace_id: Optional[
str
] = None # the trace id of the LLM call - useful if there are multiple LLM calls for the same conversation
litellm_trace_id: Optional[str] = (
None # the trace id of the LLM call - useful if there are multiple LLM calls for the same conversation
)
structured_messages: Optional[List[AllMessageValues]] = None
images: Optional[List[str]] = None
tools: Optional[List[ChatCompletionToolParam]] = None

8
poetry.lock generated
View file

@ -3207,15 +3207,15 @@ files = [
[[package]]
name = "litellm-proxy-extras"
version = "0.4.40"
version = "0.4.39"
description = "Additional files for the LiteLLM Proxy. Reduces the size of the main litellm package."
optional = true
python-versions = "!=2.7.*,!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,!=3.7.*,>=3.8"
groups = ["main"]
markers = "extra == \"proxy\""
files = [
{file = "litellm_proxy_extras-0.4.40-py3-none-any.whl", hash = "sha256:291cc5556b739d7b17b1ff79cd8881505cc560c6d5c0706302075550ae02c4bb"},
{file = "litellm_proxy_extras-0.4.40.tar.gz", hash = "sha256:964c151ec56a40c5d7b3532888dd19db9203306d4a70a3c54fb2eb0a1bcff154"},
{file = "litellm_proxy_extras-0.4.39-py3-none-any.whl", hash = "sha256:e5f72b0b74d32e7217d049de604442520ab8c4ef6da94a7e2135e8d469ab9913"},
{file = "litellm_proxy_extras-0.4.39.tar.gz", hash = "sha256:86afd3a7db023d9524a69bb7dd20059c370fb3d132a9909d6af7c350734244f2"},
]
[[package]]
@ -7934,4 +7934,4 @@ utils = ["numpydoc"]
[metadata]
lock-version = "2.1"
python-versions = ">=3.9,<4.0"
content-hash = "dfaf1eabfd17db5e30a8dda813872507aa38664fe7681ece2f8fa06ba035d3cf"
content-hash = "20ca098d83da3b9364b05930a74e9ff8512e31d626018fc9f056b6fbd50a69af"

22
tests/enterprise/litellm_enterprise/enterprise_callbacks/test_prometheus_logging_callbacks.py
View file

@ -792,7 +792,8 @@ async def test_async_post_call_success_hook(prometheus_logger):
"""
Test for the async_post_call_success_hook method
it should increment the litellm_proxy_total_requests_metric
litellm_proxy_total_requests_metric is NOT incremented here to avoid double-counting.
It is incremented in async_log_success_event instead.
"""
# Mock the prometheus metric
prometheus_logger.litellm_proxy_total_requests_metric = MagicMock()
@ -817,23 +818,8 @@ async def test_async_post_call_success_hook(prometheus_logger):
data=data, user_api_key_dict=user_api_key_dict, response=response
)
# Assert total requests metric was incremented with correct labels
prometheus_logger.litellm_proxy_total_requests_metric.labels.assert_called_once_with(
end_user=None,
hashed_api_key="test_key",
api_key_alias="test_alias",
requested_model="gpt-3.5-turbo",
team="test_team",
team_alias="test_team_alias",
user="test_user",
status_code="200",
user_email=None,
route=user_api_key_dict.request_route,
model_id=None,
client_ip=None,
user_agent=None,
)
prometheus_logger.litellm_proxy_total_requests_metric.labels().inc.assert_called_once()
# Assert total requests metric was NOT incremented (moved to async_log_success_event)
prometheus_logger.litellm_proxy_total_requests_metric.labels.assert_not_called()
def test_set_llm_deployment_success_metrics(prometheus_logger):

21
tests/enterprise/litellm_enterprise/integrations/test_prometheus.py
View file

@ -545,6 +545,9 @@ async def test_request_counter_semantic_validation(mock_prometheus_logger):
CRITICAL TEST: Validates that request counters are incremented by 1, not by token count.
This test specifically catches the bug where litellm_proxy_total_requests_metric
is incorrectly incremented by total_tokens instead of 1.
The metric is now ONLY incremented in async_log_success_event (for both streaming
and non-streaming) to prevent double-counting.
"""
from datetime import datetime, timedelta
from unittest.mock import MagicMock
@ -583,18 +586,18 @@ async def test_request_counter_semantic_validation(mock_prometheus_logger):
},
}
# Call the success event
# Call the success event - should increment for both streaming and non-streaming
await mock_prometheus_logger.async_log_success_event(
kwargs, None, kwargs["start_time"], kwargs["end_time"]
)
# CRITICAL ASSERTION: Request counter should not be incremented
# CRITICAL ASSERTION: Request counter should be incremented by 1
total_requests_metric = mock_prometheus_logger.litellm_proxy_total_requests_metric
assert (
len(total_requests_metric.inc_calls) == 0
), "Request metric should not be incremented"
len(total_requests_metric.inc_calls) == 1
), "Request metric should be incremented once in async_log_success_event"
# Call the post-call logging hook
# Call the post-call logging hook - should NOT increment (to prevent double-counting)
await mock_prometheus_logger.async_post_call_success_hook(
data={},
user_api_key_dict=UserAPIKeyAuth(
@ -607,11 +610,11 @@ async def test_request_counter_semantic_validation(mock_prometheus_logger):
response=MagicMock(),
)
# CRITICAL ASSERTION: Request counter be incremented by 1
# CRITICAL ASSERTION: Request counter should still be 1 (not incremented again)
total_requests_metric = mock_prometheus_logger.litellm_proxy_total_requests_metric
assert (
len(total_requests_metric.inc_calls) == 1
), "Request metric should not be incremented"
), "Request metric should not be incremented again in async_post_call_success_hook"
# Check that ALL request counter increments are by 1 (not by token count)
for inc_value in total_requests_metric.inc_calls:
@ -684,8 +687,8 @@ async def test_multiple_requests_counter_semantics(mock_prometheus_logger):
expected_total_tokens = num_requests * tokens_per_request # 3 * 500 = 1500
# With the bug, total_request_increments would be 1500 instead of 3
assert total_request_increments == 0, (
f"SEMANTIC BUG: Request counter total increments = 0, "
assert total_request_increments == num_requests, (
f"SEMANTIC BUG: Request counter total increments = {total_request_increments}, "
f"expected {num_requests}. This suggests request counters are being incremented "
f"by token counts instead of request counts."
)

175
tests/litellm/proxy/test_init_litellm_callbacks.py Normal file
View file

@ -0,0 +1,175 @@
"""
Unit tests for ProxyLogging._init_litellm_callbacks.
Validates that string callbacks in litellm.callbacks are replaced in-place
with their initialized instances, preventing duplicate entries (string + instance)
that caused double-counting of metrics like litellm_proxy_total_requests_metric.
"""
from typing import List, Union
from unittest.mock import MagicMock, patch
import pytest
import litellm
from litellm.integrations.custom_logger import CustomLogger
class FakeCustomLogger(CustomLogger):
"""A minimal CustomLogger subclass for testing."""
pass
class TestInitLitellmCallbacks:
"""Tests for ProxyLogging._init_litellm_callbacks."""
def _make_proxy_logging(self):
"""Create a ProxyLogging instance with mocked dependencies."""
from litellm.proxy.utils import ProxyLogging
mock_cache = MagicMock()
proxy_logging = ProxyLogging(user_api_key_cache=mock_cache)
return proxy_logging
@patch(
"litellm.proxy.utils.ProxyLogging._add_proxy_hooks",
new_callable=lambda: lambda self, *a, **kw: None,
)
def test_should_replace_string_callback_with_instance(self, _mock_hooks):
"""
When litellm.callbacks contains a string callback (e.g. "lago"),
_init_litellm_callbacks should replace the string with the initialized
CustomLogger instance, not leave both the string and instance in the list.
"""
fake_logger = FakeCustomLogger()
# Start with a string callback in litellm.callbacks
litellm.callbacks = ["lago"] # type: ignore
proxy_logging = self._make_proxy_logging()
with patch(
"litellm.litellm_core_utils.litellm_logging._init_custom_logger_compatible_class",
return_value=fake_logger,
):
proxy_logging._init_litellm_callbacks(llm_router=None)
# The string "lago" should be replaced by the instance, not appended
string_entries = [c for c in litellm.callbacks if isinstance(c, str)]
instance_entries = [
c for c in litellm.callbacks if isinstance(c, FakeCustomLogger)
]
assert len(string_entries) == 0, (
f"String callbacks should have been replaced, but found: {string_entries}"
)
assert len(instance_entries) == 1, (
f"Expected exactly one FakeCustomLogger instance, found {len(instance_entries)}"
)
assert instance_entries[0] is fake_logger
# Clean up
litellm.callbacks = [] # type: ignore
@patch(
"litellm.proxy.utils.ProxyLogging._add_proxy_hooks",
new_callable=lambda: lambda self, *a, **kw: None,
)
def test_should_not_duplicate_existing_instance_callbacks(self, _mock_hooks):
"""
When litellm.callbacks already contains a CustomLogger instance (not a string),
_init_litellm_callbacks should not create a duplicate.
"""
existing_logger = FakeCustomLogger()
litellm.callbacks = [existing_logger] # type: ignore
proxy_logging = self._make_proxy_logging()
proxy_logging._init_litellm_callbacks(llm_router=None)
# Count how many FakeCustomLogger instances are in litellm.callbacks
instance_count = sum(
1 for c in litellm.callbacks if isinstance(c, FakeCustomLogger)
)
assert instance_count == 1, (
f"Expected exactly 1 FakeCustomLogger instance, found {instance_count}. "
f"litellm.callbacks = {litellm.callbacks}"
)
# Clean up
litellm.callbacks = [] # type: ignore
@patch(
"litellm.proxy.utils.ProxyLogging._add_proxy_hooks",
new_callable=lambda: lambda self, *a, **kw: None,
)
def test_should_handle_unrecognized_string_callback(self, _mock_hooks):
"""
When _init_custom_logger_compatible_class returns None for a string callback,
the string should remain in litellm.callbacks (not crash).
"""
litellm.callbacks = ["unknown_callback"] # type: ignore
proxy_logging = self._make_proxy_logging()
with patch(
"litellm.litellm_core_utils.litellm_logging._init_custom_logger_compatible_class",
return_value=None,
):
proxy_logging._init_litellm_callbacks(llm_router=None)
# The unknown string callback should still be there (not replaced, not crashed)
assert "unknown_callback" in litellm.callbacks
# Clean up
litellm.callbacks = [] # type: ignore
@patch(
"litellm.proxy.utils.ProxyLogging._add_proxy_hooks",
new_callable=lambda: lambda self, *a, **kw: None,
)
def test_should_replace_multiple_string_callbacks(self, _mock_hooks):
"""
When litellm.callbacks contains multiple string callbacks,
each should be replaced with its corresponding initialized instance.
"""
fake_logger_a = FakeCustomLogger()
fake_logger_b = FakeCustomLogger()
litellm.callbacks = ["callback_a", "callback_b"] # type: ignore
proxy_logging = self._make_proxy_logging()
call_count = 0
def mock_init_class(callback_name, **kwargs):
nonlocal call_count
call_count += 1
if call_count == 1:
return fake_logger_a
return fake_logger_b
with patch(
"litellm.litellm_core_utils.litellm_logging._init_custom_logger_compatible_class",
side_effect=mock_init_class,
):
proxy_logging._init_litellm_callbacks(llm_router=None)
string_entries = [c for c in litellm.callbacks if isinstance(c, str)]
instance_entries = [
c for c in litellm.callbacks if isinstance(c, FakeCustomLogger)
]
assert len(string_entries) == 0, (
f"All string callbacks should have been replaced: {string_entries}"
)
assert len(instance_entries) == 2, (
f"Expected 2 FakeCustomLogger instances, found {len(instance_entries)}"
)
assert instance_entries[0] is fake_logger_a
assert instance_entries[1] is fake_logger_b
# Clean up
litellm.callbacks = [] # type: ignore

256
tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py
View file

@ -9,9 +9,7 @@ from unittest.mock import ANY, MagicMock, Mock, patch
import httpx
import pytest
sys.path.insert(
0, os.path.abspath("../../..")
) # Adds the parent directory to the system-path
sys.path.insert(0, os.path.abspath("../../..")) # Adds the parent directory to the system-path
import litellm
@ -119,9 +117,7 @@ def test_convert_chat_completion_messages_to_responses_api_tool_result_with_imag
function_call_output = item
break
assert (
function_call_output is not None
), "function_call_output not found in response"
assert function_call_output is not None, "function_call_output not found in response"
assert function_call_output["call_id"] == "call_abc123"
# Check that the output is correctly transformed
@ -131,12 +127,8 @@ def test_convert_chat_completion_messages_to_responses_api_tool_result_with_imag
image_item = output[0]
# Should be transformed to Responses API format
assert (
image_item["type"] == "input_image"
), f"Expected type 'input_image', got '{image_item.get('type')}'"
assert (
image_item["image_url"] == test_image_base64
), "image_url should be a flat string, not a nested object"
assert image_item["type"] == "input_image", f"Expected type 'input_image', got '{image_item.get('type')}'"
assert image_item["image_url"] == test_image_base64, "image_url should be a flat string, not a nested object"
assert "detail" in image_item, "detail field should be present"
print("✓ Tool result with image correctly transformed to Responses API format")
@ -198,9 +190,7 @@ def test_convert_chat_completion_messages_to_responses_api_tool_result_with_text
function_call_output = item
break
assert (
function_call_output is not None
), "function_call_output not found in response"
assert function_call_output is not None, "function_call_output not found in response"
assert function_call_output["call_id"] == "call_abc123"
# Check that the output is correctly transformed to use input_text, not output_text
@ -210,12 +200,10 @@ def test_convert_chat_completion_messages_to_responses_api_tool_result_with_text
text_item = output[0]
# Should be transformed to use input_text for tool results in Responses API format
assert (
text_item["type"] == "input_text"
), f"Expected type 'input_text' for tool result, got '{text_item.get('type')}'"
assert (
text_item["text"] == "15 degrees"
), f"Expected text '15 degrees', got '{text_item.get('text')}'"
assert text_item["type"] == "input_text", (
f"Expected type 'input_text' for tool result, got '{text_item.get('type')}'"
)
assert text_item["text"] == "15 degrees", f"Expected text '15 degrees', got '{text_item.get('text')}'"
print("✓ Tool result with text correctly transformed to use input_text for Responses API format")
@ -226,9 +214,7 @@ def test_openai_responses_chunk_parser_reasoning_summary():
)
from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
iterator = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
chunk = {
"delta": "**Compar",
@ -260,9 +246,7 @@ def test_chunk_parser_string_output_text_delta_produces_text():
)
from litellm.types.utils import ModelResponseStream
iterator = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
chunk = {"type": "response.output_text.delta", "delta": "literal text"}
@ -283,9 +267,7 @@ def test_chunk_parser_enum_output_text_delta_produces_text():
from litellm.types.llms.openai import ResponsesAPIStreamEvents
from litellm.types.utils import ModelResponseStream
iterator = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
chunk = {"type": ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, "delta": "enum text"}
@ -306,9 +288,7 @@ def test_chunk_parser_function_call_added_produces_tool_use():
from litellm.types.llms.openai import ResponsesAPIStreamEvents
from litellm.types.utils import ModelResponseStream
iterator = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
chunk = {
"type": ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
@ -393,9 +373,7 @@ Tomorrow will bring its petitions and promises,
but for now the city breathes slow and wide,
and I learn to carry this small calm home."""
output_text = ResponseOutputText(
annotations=[], text=poem_text, type="output_text", logprobs=[]
)
output_text = ResponseOutputText(annotations=[], text=poem_text, type="output_text", logprobs=[])
output_message = ResponseOutputMessage(
id="msg_04c8021b8b3188a00068e9ae0b92f4819dac64d85b4abb67ec",
content=[output_text],
@ -407,9 +385,7 @@ and I learn to carry this small calm home."""
# Create usage information
usage = ResponseAPIUsage(
input_tokens=16,
input_tokens_details=InputTokensDetails(
audio_tokens=None, cached_tokens=0, text_tokens=None
),
input_tokens_details=InputTokensDetails(audio_tokens=None, cached_tokens=0, text_tokens=None),
output_tokens=195,
output_tokens_details=OutputTokensDetails(reasoning_tokens=0, text_tokens=None),
total_tokens=211,
@ -621,9 +597,7 @@ def test_transform_request_single_char_keys_not_matched():
assert result_correct.get("metadata") == {"user_id": "123"}
assert result_correct.get("previous_response_id") == "resp_abc"
print(
"✓ Single-character keys are not incorrectly matched to metadata/previous_response_id"
)
print("✓ Single-character keys are not incorrectly matched to metadata/previous_response_id")
# =============================================================================
@ -643,9 +617,7 @@ def test_message_done_does_not_emit_is_finished():
OpenAiResponsesToChatCompletionStreamIterator,
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
chunk = {
"type": "response.output_item.done",
@ -657,9 +629,9 @@ def test_message_done_does_not_emit_is_finished():
# After the fix, message completion should NOT set finish_reason
# ModelResponseStream doesn't have is_finished - check finish_reason instead
assert len(result.choices) > 0, "result should have choices"
assert (
result.choices[0].finish_reason is None or result.choices[0].finish_reason == ""
), "message completion should not emit finish_reason"
assert result.choices[0].finish_reason is None or result.choices[0].finish_reason == "", (
"message completion should not emit finish_reason"
)
def test_response_completed_emits_is_finished():
@ -671,9 +643,7 @@ def test_response_completed_emits_is_finished():
OpenAiResponsesToChatCompletionStreamIterator,
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
chunk = {"type": "response.completed"}
@ -681,9 +651,91 @@ def test_response_completed_emits_is_finished():
# response.completed should emit finish_reason='stop'
assert len(result.choices) > 0, "result should have choices"
assert (
result.choices[0].finish_reason == "stop"
), "response.completed should emit finish_reason='stop'"
assert result.choices[0].finish_reason == "stop", "response.completed should emit finish_reason='stop'"
def test_response_completed_with_function_calls_emits_tool_calls_finish_reason():
"""
Test that response.completed with function_call items in output emits finish_reason='tool_calls'.
This is a regression test for an issue where response.completed always returned
finish_reason='stop' even when the response contained tool calls, causing agents
like OpenCode to incorrectly conclude the stream ended without tools to execute.
When the response.completed event includes function_call items in its output,
the finish_reason should be 'tool_calls' to signal the client that tools need
to be executed.
"""
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
# Simulate a response.completed event with function_call in output
# This matches what Azure/OpenAI sends for gpt-5.1-codex-mini and similar models
chunk = {
"type": "response.completed",
"response": {
"id": "resp_123",
"status": "completed",
"output": [
{
"type": "function_call",
"id": "call_abc123",
"call_id": "call_abc123",
"name": "read_file",
"arguments": '{"path": "/tmp/test.py"}',
"status": "completed",
}
],
},
}
result = iterator.chunk_parser(chunk)
# response.completed with function_call should emit finish_reason='tool_calls'
assert len(result.choices) > 0, "result should have choices"
assert result.choices[0].finish_reason == "tool_calls", (
"response.completed with function_call output should emit finish_reason='tool_calls'"
)
def test_response_completed_with_message_only_emits_stop_finish_reason():
"""
Test that response.completed with only message output (no function_call) emits finish_reason='stop'.
"""
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
# Simulate a response.completed event with only message output
chunk = {
"type": "response.completed",
"response": {
"id": "resp_456",
"status": "completed",
"output": [
{
"type": "message",
"id": "msg_xyz",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hello, world!"}],
"status": "completed",
}
],
},
}
result = iterator.chunk_parser(chunk)
# response.completed with only message should emit finish_reason='stop'
assert len(result.choices) > 0, "result should have choices"
assert result.choices[0].finish_reason == "stop", (
"response.completed with only message output should emit finish_reason='stop'"
)
def test_function_call_done_emits_is_finished():
@ -695,9 +747,7 @@ def test_function_call_done_emits_is_finished():
OpenAiResponsesToChatCompletionStreamIterator,
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
chunk = {
"type": "response.output_item.done",
@ -713,13 +763,10 @@ def test_function_call_done_emits_is_finished():
# function_call completion should emit finish_reason='tool_calls'
assert len(result.choices) > 0, "result should have choices"
assert (
result.choices[0].finish_reason == "tool_calls"
), "function_call should emit finish_reason='tool_calls'"
assert (
result.choices[0].delta.tool_calls is not None
and len(result.choices[0].delta.tool_calls) > 0
), "function_call should include tool_calls"
assert result.choices[0].finish_reason == "tool_calls", "function_call should emit finish_reason='tool_calls'"
assert result.choices[0].delta.tool_calls is not None and len(result.choices[0].delta.tool_calls) > 0, (
"function_call should include tool_calls"
)
def test_text_plus_tool_calls_sequence():
@ -734,9 +781,7 @@ def test_text_plus_tool_calls_sequence():
OpenAiResponsesToChatCompletionStreamIterator,
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True
)
iterator = OpenAiResponsesToChatCompletionStreamIterator(streaming_response=None, sync_stream=True)
# Simulate the sequence from OpenAI Responses API
chunks = [
@ -775,26 +820,21 @@ def test_text_plus_tool_calls_sequence():
# Check message done (index 2) does NOT have finish_reason set
message_done_result = results[2]
assert len(message_done_result.choices) > 0, "message done should have choices"
assert (
message_done_result.choices[0].finish_reason is None
or message_done_result.choices[0].finish_reason == ""
), "message done should not have finish_reason"
assert message_done_result.choices[0].finish_reason is None or message_done_result.choices[0].finish_reason == "", (
"message done should not have finish_reason"
)
# Check function_call done (index 5) DOES have finish_reason='tool_calls'
function_done_result = results[5]
assert (
len(function_done_result.choices) > 0
), "function_call done should have choices"
assert (
function_done_result.choices[0].finish_reason == "tool_calls"
), "function_call done should have finish_reason='tool_calls'"
assert len(function_done_result.choices) > 0, "function_call done should have choices"
assert function_done_result.choices[0].finish_reason == "tool_calls", (
"function_call done should have finish_reason='tool_calls'"
)
# Check response.completed (index 6) has finish_reason='stop'
completed_result = results[6]
assert len(completed_result.choices) > 0, "response.completed should have choices"
assert (
completed_result.choices[0].finish_reason == "stop"
), "response.completed should have finish_reason='stop'"
assert completed_result.choices[0].finish_reason == "stop", "response.completed should have finish_reason='stop'"
# =============================================================================
@ -1012,11 +1052,11 @@ def test_multiple_tool_calls_in_single_choice():
def test_map_reasoning_effort_adds_summary_detailed():
"""
Test that _map_reasoning_effort behavior with reasoning_auto_summary flag.
By default (flag=False), summary should NOT be added to avoid:
1. Breaking for users without verified OpenAI orgs (400 errors)
2. Making requests more expensive by including summary reasoning tokens
When flag is enabled (flag=True or env var), summary="detailed" is added.
"""
import os
@ -1030,64 +1070,68 @@ def test_map_reasoning_effort_adds_summary_detailed():
# Test all string effort levels - DEFAULT BEHAVIOR (no summary)
effort_levels = ["none", "low", "medium", "high", "xhigh", "minimal"]
# Save original flag value
original_flag = litellm.reasoning_auto_summary
original_env = os.environ.get("LITELLM_REASONING_AUTO_SUMMARY")
try:
# Test 1: Default behavior (flag=False, no env var) - NO summary
litellm.reasoning_auto_summary = False
if "LITELLM_REASONING_AUTO_SUMMARY" in os.environ:
del os.environ["LITELLM_REASONING_AUTO_SUMMARY"]
for effort in effort_levels:
result = handler._map_reasoning_effort(effort)
assert result is not None, f"Result should not be None for effort={effort}"
assert result["effort"] == effort, f"Effort should be {effort}"
assert "summary" not in result, f"Summary should NOT be present by default for effort={effort}"
print(f"✓ reasoning_effort='{effort}' correctly maps to effort='{effort}' (no summary by default)")
# Test 2: With flag enabled - summary IS added
litellm.reasoning_auto_summary = True
for effort in effort_levels:
result = handler._map_reasoning_effort(effort)
assert result is not None, f"Result should not be None for effort={effort}"
assert result["effort"] == effort, f"Effort should be {effort}"
assert result["summary"] == "detailed", f"Summary should be 'detailed' when flag is enabled for effort={effort}"
print(f"✓ reasoning_effort='{effort}' correctly maps to effort='{effort}', summary='detailed' (flag enabled)")
assert result["summary"] == "detailed", (
f"Summary should be 'detailed' when flag is enabled for effort={effort}"
)
print(
f"✓ reasoning_effort='{effort}' correctly maps to effort='{effort}', summary='detailed' (flag enabled)"
)
# Test 3: With env var enabled (flag disabled) - summary IS added
litellm.reasoning_auto_summary = False
os.environ["LITELLM_REASONING_AUTO_SUMMARY"] = "true"
result = handler._map_reasoning_effort("high")
assert result["summary"] == "detailed", "Summary should be 'detailed' when env var is enabled"
print("✓ LITELLM_REASONING_AUTO_SUMMARY env var works correctly")
# Test 4: Dict input is passed through as-is (no modification)
litellm.reasoning_auto_summary = False
if "LITELLM_REASONING_AUTO_SUMMARY" in os.environ:
del os.environ["LITELLM_REASONING_AUTO_SUMMARY"]
dict_input = {"effort": "high", "summary": "custom_summary"}
result_dict = handler._map_reasoning_effort(dict_input)
assert result_dict["effort"] == "high"
assert result_dict["summary"] == "custom_summary"
print("✓ Dict input is passed through without modification")
# Test 5: None/unknown values return None
result_unknown = handler._map_reasoning_effort("unknown_value")
assert result_unknown is None
print("✓ Unknown reasoning_effort values return None")
print("✓ All reasoning_effort behaviors work correctly with flag/env var control")
finally:
# Restore original values
litellm.reasoning_auto_summary = original_flag
@ -1100,10 +1144,10 @@ def test_map_reasoning_effort_adds_summary_detailed():
def test_transform_response_preserves_annotations():
"""
Test that annotations from Responses API are preserved when transforming to Chat Completions format.
This is a regression test for the bug where annotations (like url_citation) were being
dropped during the transformation from ResponsesAPIResponse to ModelResponse.
The fix ensures annotations are extracted from ResponseOutputText content items and
passed through to the Message object in the Chat Completions response.
"""
@ -1162,13 +1206,9 @@ def test_transform_response_preserves_annotations():
# Create usage information
usage = ResponseAPIUsage(
input_tokens=10,
input_tokens_details=InputTokensDetails(
audio_tokens=None, cached_tokens=0, text_tokens=None
),
input_tokens_details=InputTokensDetails(audio_tokens=None, cached_tokens=0, text_tokens=None),
output_tokens=20,
output_tokens_details=OutputTokensDetails(
reasoning_tokens=0, text_tokens=None
),
output_tokens_details=OutputTokensDetails(reasoning_tokens=0, text_tokens=None),
total_tokens=30,
cost=None,
)

1
tests/test_litellm/llms/bedrock/chat/test_converse_transformation.py
View file

@ -3376,6 +3376,7 @@ def test_output_config_applies_additional_properties():
assert parsed["properties"]["nested"]["additionalProperties"] is False
class TestBedrockMinThinkingBudgetTokens:
"""Test that thinking.budget_tokens is clamped to the Bedrock minimum (1024)."""

102
tests/test_litellm/proxy/guardrails/guardrail_hooks/test_generic_guardrail_api.py
View file

@ -13,7 +13,7 @@ import pytest
import litellm
from litellm import ModelResponse
from litellm.exceptions import GuardrailRaisedException
from litellm.exceptions import GuardrailRaisedException, Timeout
from litellm._version import version as litellm_version
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api import (
@ -704,6 +704,104 @@ class TestErrorHandling:
assert "Generic Guardrail API failed" in str(exc_info.value)
@pytest.mark.asyncio
async def test_network_error_defaults_to_fail_closed_when_unreachable_fallback_not_set(
self, mock_request_data_input
):
"""Test default behavior is fail_closed when unreachable_fallback is omitted"""
guardrail = GenericGuardrailAPI(
api_base="https://api.test.guardrail.com",
headers={"Authorization": "Bearer test-key"},
)
with patch.object(
guardrail.async_handler,
"post",
side_effect=httpx.RequestError("Connection failed", request=MagicMock()),
):
with pytest.raises(Exception) as exc_info:
await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data=mock_request_data_input,
input_type="request",
)
assert "Generic Guardrail API failed" in str(exc_info.value)
@pytest.mark.asyncio
async def test_network_error_fail_open_allows_flow(self, mock_request_data_input):
"""Test network error handling allows flow when unreachable_fallback=fail_open"""
guardrail = GenericGuardrailAPI(
api_base="https://api.test.guardrail.com",
headers={"Authorization": "Bearer test-key"},
unreachable_fallback="fail_open",
)
with patch.object(
guardrail.async_handler,
"post",
side_effect=httpx.RequestError("Connection failed", request=MagicMock()),
):
result = await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data=mock_request_data_input,
input_type="request",
)
assert result.get("texts") == ["test"]
@pytest.mark.asyncio
async def test_503_fail_open_allows_flow(self, mock_request_data_input):
"""Test HTTP 503 allows flow when unreachable_fallback=fail_open"""
guardrail = GenericGuardrailAPI(
api_base="https://api.test.guardrail.com",
headers={"Authorization": "Bearer test-key"},
unreachable_fallback="fail_open",
)
with patch.object(
guardrail.async_handler,
"post",
side_effect=httpx.HTTPStatusError(
"Service Unavailable",
request=MagicMock(),
response=MagicMock(status_code=503),
),
):
result = await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data=mock_request_data_input,
input_type="request",
)
assert result.get("texts") == ["test"]
@pytest.mark.asyncio
async def test_timeout_fail_open_allows_flow(self, mock_request_data_input):
"""Test litellm.Timeout allows flow when unreachable_fallback=fail_open"""
guardrail = GenericGuardrailAPI(
api_base="https://api.test.guardrail.com",
headers={"Authorization": "Bearer test-key"},
unreachable_fallback="fail_open",
)
with patch.object(
guardrail.async_handler,
"post",
side_effect=Timeout(
message="Connection timed out",
model="default-model-name",
llm_provider="litellm-httpx-handler",
),
):
result = await guardrail.apply_guardrail(
inputs={"texts": ["test"]},
request_data=mock_request_data_input,
input_type="request",
)
assert result.get("texts") == ["test"]
class TestMultimodalSupport:
"""Test multimodal (image) message handling and serialization"""
@ -830,4 +928,4 @@ class TestMultimodalSupport:
# Verify serialization succeeded
call_args = mock_post.call_args
json_payload = call_args.kwargs["json"]
assert isinstance(json_payload["structured_messages"], list)
assert isinstance(json_payload["structured_messages"], list)

119
tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py
View file

@ -5606,3 +5606,122 @@ async def test_validate_key_list_check_key_hash_not_found():
assert exc_info.value.code == "403" or exc_info.value.code == 403
assert "Key Hash not found" in exc_info.value.message
@pytest.mark.asyncio
@patch(
"litellm.proxy.management_endpoints.key_management_endpoints.rotate_mcp_server_credentials_master_key"
)
async def test_rotate_master_key_model_data_valid_for_prisma(
mock_rotate_mcp,
):
"""
Test that _rotate_master_key produces valid data for Prisma create_many().
Regression test for: master key rotation fails with Prisma validation error
because created_at/updated_at are None (non-nullable DateTime) and
litellm_params/model_info are JSON strings (create_many expects dicts).
"""
from unittest.mock import AsyncMock, MagicMock
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.key_management_endpoints import (
_rotate_master_key,
)
# Setup mock prisma client
mock_prisma_client = AsyncMock()
mock_prisma_client.db = MagicMock()
# Mock model table — return one model
mock_model = MagicMock()
mock_model.model_id = "model-1"
mock_model.model_name = "test-model"
mock_model.litellm_params = '{"model": "openai/gpt-4", "api_key": "sk-encrypted-old"}'
mock_model.model_info = '{"id": "model-1"}'
mock_model.created_by = "admin"
mock_model.updated_by = "admin"
mock_prisma_client.db.litellm_proxymodeltable.find_many = AsyncMock(
return_value=[mock_model]
)
# Mock transaction context manager
mock_tx = AsyncMock()
mock_tx.litellm_proxymodeltable = MagicMock()
mock_tx.litellm_proxymodeltable.delete_many = AsyncMock()
mock_tx.litellm_proxymodeltable.create_many = AsyncMock()
mock_prisma_client.db.tx = MagicMock(return_value=AsyncMock(
__aenter__=AsyncMock(return_value=mock_tx),
__aexit__=AsyncMock(return_value=False),
))
# Mock config table — no env vars
mock_prisma_client.db.litellm_config.find_many = AsyncMock(return_value=[])
# Mock credentials table — no credentials
mock_prisma_client.db.litellm_credentialstable.find_many = AsyncMock(
return_value=[]
)
# Mock MCP rotation
mock_rotate_mcp.return_value = None
# Mock proxy_config
mock_proxy_config = MagicMock()
mock_proxy_config.decrypt_model_list_from_db.return_value = [
{
"model_name": "test-model",
"litellm_params": {
"model": "openai/gpt-4",
"api_key": "sk-decrypted-key",
},
"model_info": {"id": "model-1"},
}
]
user_api_key_dict = UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN,
api_key="sk-1234",
user_id="test-user",
)
with patch(
"litellm.proxy.proxy_server.proxy_config",
mock_proxy_config,
):
await _rotate_master_key(
prisma_client=mock_prisma_client,
user_api_key_dict=user_api_key_dict,
current_master_key="sk-old-master-key",
new_master_key="sk-new-master-key",
)
# Verify create_many was called
mock_tx.litellm_proxymodeltable.create_many.assert_called_once()
# Get the data passed to create_many
call_args = mock_tx.litellm_proxymodeltable.create_many.call_args
created_models = call_args.kwargs.get("data") or call_args[1].get("data")
assert len(created_models) == 1
model_data = created_models[0]
# Verify timestamps are NOT present (Prisma @default(now()) should apply)
assert "created_at" not in model_data, (
"created_at should be excluded so Prisma @default(now()) applies"
)
assert "updated_at" not in model_data, (
"updated_at should be excluded so Prisma @default(now()) applies"
)
# Verify litellm_params and model_info are prisma.Json wrappers, NOT JSON strings
import prisma
assert isinstance(model_data["litellm_params"], prisma.Json), (
f"litellm_params should be prisma.Json for create_many(), got {type(model_data['litellm_params'])}"
)
assert isinstance(model_data["model_info"], prisma.Json), (
f"model_info should be prisma.Json for create_many(), got {type(model_data['model_info'])}"
)
# Verify delete_many was called inside the transaction (before create_many)
mock_tx.litellm_proxymodeltable.delete_many.assert_called_once()

176
tests/test_litellm/responses/test_metadata_codex_callback.py Normal file
View file

@ -0,0 +1,176 @@
"""
Test that metadata is passed to custom callbacks during chat completion calls to codex models.
Fixes issue: Metadata is no longer passed to custom callback during chat completion
calls to codex models (#21204)
Codex models (gpt-5.1-codex, gpt-5.2-codex) use mode=responses and route through
responses_api_bridge. The bridge converts metadata to litellm_metadata. This test
verifies metadata is preserved for custom callbacks via kwargs['litellm_params']['metadata'].
"""
import asyncio
import os
import sys
from typing import Optional
from unittest.mock import AsyncMock, patch
sys.path.insert(0, os.path.abspath("../../.."))
import pytest
import litellm
from litellm.integrations.custom_logger import CustomLogger
def _make_mock_http_response(response_dict: dict):
"""Create a mock HTTP response that returns response_dict from .json()."""
class MockResponse:
def __init__(self, json_data, status_code=200):
self._json_data = json_data
self.status_code = status_code
self.text = str(json_data)
self.headers = {}
def json(self):
return self._json_data
return MockResponse(response_dict, 200)
class MetadataCaptureCallback(CustomLogger):
"""Custom callback that captures kwargs passed to async_log_success_event."""
def __init__(self):
self.captured_kwargs: Optional[dict] = None
async def async_log_success_event(
self, kwargs, response_obj, start_time, end_time
):
self.captured_kwargs = kwargs
@pytest.mark.asyncio
async def test_metadata_passed_to_custom_callback_codex_models():
"""
Test that metadata passed to completion() is available in custom callback
when using codex models (responses API bridge path).
Codex models have mode=responses and route through responses_api_bridge,
which passes litellm_metadata. The fix ensures this is preserved as
litellm_params.metadata for callback compatibility.
"""
from litellm.types.llms.openai import ResponsesAPIResponse
mock_response = ResponsesAPIResponse.model_construct(
id="resp-test",
created_at=0,
output=[
{
"type": "message",
"id": "msg-1",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hello!"}],
}
],
object="response",
model="gpt-5.1-codex",
status="completed",
usage={
"input_tokens": 5,
"output_tokens": 10,
"total_tokens": 15,
},
)
test_metadata = {"foo": "bar", "trace_id": "test-123"}
callback = MetadataCaptureCallback()
original_callbacks = litellm.callbacks.copy() if litellm.callbacks else []
litellm.callbacks = [callback]
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new_callable=AsyncMock,
) as mock_post:
mock_post.return_value = _make_mock_http_response(
mock_response.model_dump()
)
# gpt-5.1-codex has mode=responses - routes through responses bridge
await litellm.acompletion(
model="gpt-5.1-codex",
messages=[{"role": "user", "content": "Hello"}],
metadata=test_metadata,
)
await asyncio.sleep(1)
assert callback.captured_kwargs is not None, "Callback should have been invoked"
litellm_params = callback.captured_kwargs.get("litellm_params", {})
metadata = litellm_params.get("metadata") or {}
assert "foo" in metadata, "metadata['foo'] should be accessible in callback"
assert metadata["foo"] == "bar"
assert metadata.get("trace_id") == "test-123"
@pytest.mark.asyncio
async def test_metadata_passed_via_litellm_metadata_responses_api():
"""
Test that when calling responses() directly with litellm_metadata,
metadata is preserved for custom callbacks.
Uses HTTP mock since mock_response returns early before update_environment_variables.
"""
from litellm.types.llms.openai import ResponsesAPIResponse
mock_response = ResponsesAPIResponse.model_construct(
id="resp-test-2",
created_at=0,
output=[
{
"type": "message",
"id": "msg-2",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hi there!"}],
}
],
object="response",
model="gpt-4o",
status="completed",
usage={
"input_tokens": 2,
"output_tokens": 3,
"total_tokens": 5,
},
)
test_metadata = {"request_id": "req-456"}
callback = MetadataCaptureCallback()
litellm.callbacks = [callback]
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new_callable=AsyncMock,
) as mock_post:
mock_post.return_value = _make_mock_http_response(
mock_response.model_dump()
)
await litellm.aresponses(
model="gpt-4o",
input="hi",
litellm_metadata=test_metadata,
)
await asyncio.sleep(1)
assert callback.captured_kwargs is not None
litellm_params = callback.captured_kwargs.get("litellm_params", {})
metadata = litellm_params.get("metadata") or {}
assert "request_id" in metadata
assert metadata["request_id"] == "req-456"