from importlib import import_module import base64 from typing import Final from unittest.mock import MagicMock, patch import pytest import litellm from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token from litellm.responses.litellm_completion_transformation.transformation import LiteLLMCompletionResponsesConfig from litellm.responses.utils import ResponseAPILoggingUtils, ResponsesAPIRequestUtils from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIOptionalRequestParams, ResponsesAPIResponse from litellm.types.utils import ModelInfo, Usage from litellm.utils import _invalidate_model_cost_lowercase_map class TestResponsesAPIRequestUtils: def test_get_optional_params_responses_api(self): """Test that optional parameters are correctly processed for responses API""" # Setup model = "gpt-4o" config = OpenAIResponsesAPIConfig() optional_params = ResponsesAPIOptionalRequestParams( { "temperature": 0.7, "max_output_tokens": 100, "prompt": {"id": "pmpt_123"}, } ) # Execute result = ResponsesAPIRequestUtils.get_optional_params_responses_api( model=model, responses_api_provider_config=config, response_api_optional_params=optional_params, ) # Assert assert result == optional_params assert "temperature" in result assert result["temperature"] == 0.7 assert "max_output_tokens" in result assert result["max_output_tokens"] == 100 assert "prompt" in result assert result["prompt"] == {"id": "pmpt_123"} def test_get_optional_params_responses_api_unsupported_param(self): """Test that unsupported parameters raise an error""" # Setup model = "gpt-4o" config = OpenAIResponsesAPIConfig() optional_params = ResponsesAPIOptionalRequestParams({"temperature": 0.7, "unsupported_param": "value"}) # Execute and Assert with pytest.raises(litellm.UnsupportedParamsError) as excinfo: ResponsesAPIRequestUtils.get_optional_params_responses_api( model=model, responses_api_provider_config=config, response_api_optional_params=optional_params, ) assert "unsupported_param" in str(excinfo.value) assert model in str(excinfo.value) def test_get_optional_params_responses_api_request_level_drop_params(self, monkeypatch): """Request-level drop_params must reach both _check_valid_arg and map_openai_params""" monkeypatch.setattr(litellm, "drop_params", False) config = MagicMock(spec=OpenAIResponsesAPIConfig) config.get_supported_openai_params.return_value = ["temperature"] config.custom_llm_provider = "openai" config.map_openai_params.return_value = {"temperature": 0.7} result = ResponsesAPIRequestUtils.get_optional_params_responses_api( model="gpt-4o", responses_api_provider_config=config, response_api_optional_params=ResponsesAPIOptionalRequestParams( {"temperature": 0.7, "service_tier": "priority"} ), drop_params=True, ) assert config.map_openai_params.call_args.kwargs["drop_params"] is True assert result == {"temperature": 0.7} @pytest.mark.parametrize("request_drop_params", [None, False]) def test_get_optional_params_responses_api_still_raises_without_drop(self, monkeypatch, request_drop_params): """Absent or False request-level drop_params must not suppress the unsupported-param error""" monkeypatch.setattr(litellm, "drop_params", False) config = OpenAIResponsesAPIConfig() with pytest.raises(litellm.UnsupportedParamsError): ResponsesAPIRequestUtils.get_optional_params_responses_api( model="gpt-4o", responses_api_provider_config=config, response_api_optional_params=ResponsesAPIOptionalRequestParams( {"temperature": 0.7, "unsupported_param": "value"} ), drop_params=request_drop_params, ) def test_get_requested_response_api_optional_param(self): """Test filtering parameters to only include those in ResponsesAPIOptionalRequestParams""" # Setup params = { "temperature": 0.7, "max_output_tokens": 100, "prompt": {"id": "pmpt_456"}, "invalid_param": "value", "model": "gpt-4o", # This is not in ResponsesAPIOptionalRequestParams } # Execute result = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(params) # Assert assert "temperature" in result assert "max_output_tokens" in result assert "invalid_param" not in result assert "model" not in result assert result["temperature"] == 0.7 assert result["max_output_tokens"] == 100 assert result["prompt"] == {"id": "pmpt_456"} def test_get_requested_response_api_optional_param_drops_nested_path(self): """Nested additional_drop_params paths like reasoning.summary must be honored""" params = { "temperature": 0.1, "reasoning": {"effort": "high", "summary": "auto"}, "additional_drop_params": ["reasoning.summary"], } result = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(params) assert result["reasoning"] == {"effort": "high"} assert result["temperature"] == 0.1 def test_get_requested_response_api_optional_param_drops_array_path(self): """Array wildcard paths like tools[*].input_examples must be honored""" params = { "tools": [{"type": "function", "name": "t", "input_examples": ["x"]}], "additional_drop_params": ["tools[*].input_examples"], } result = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(params) assert result["tools"] == [{"type": "function", "name": "t"}] def test_get_requested_response_api_optional_param_drops_top_level(self): """Top-level additional_drop_params keys must still be honored""" params = { "reasoning": {"effort": "high", "summary": "auto"}, "additional_drop_params": ["reasoning"], } result = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(params) assert "reasoning" not in result def test_get_requested_response_api_optional_param_non_matching_nested_path(self): """A nested path that does not match anything leaves params untouched""" params = { "reasoning": {"effort": "high", "summary": "auto"}, "additional_drop_params": ["reasoning.nope"], } result = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(params) assert result["reasoning"] == {"effort": "high", "summary": "auto"} def test_get_requested_response_api_optional_param_none_drop_params(self): """additional_drop_params=None is a no-op""" params = { "reasoning": {"effort": "high", "summary": "auto"}, "additional_drop_params": None, } result = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(params) assert result["reasoning"] == {"effort": "high", "summary": "auto"} def test_decode_previous_response_id_to_original_previous_response_id(self): """Test decoding a LiteLLM encoded previous_response_id to the original previous_response_id""" # Setup test_provider = "openai" test_model_id = "gpt-4o" original_response_id = "resp_abc123" # Use the helper method to build an encoded response ID encoded_id = ResponsesAPIRequestUtils._build_responses_api_response_id( custom_llm_provider=test_provider, model_id=test_model_id, response_id=original_response_id, ) # Execute result = ResponsesAPIRequestUtils.decode_previous_response_id_to_original_previous_response_id(encoded_id) # Assert assert result == original_response_id # Test with a non-encoded ID plain_id = "resp_xyz789" result_plain = ResponsesAPIRequestUtils.decode_previous_response_id_to_original_previous_response_id(plain_id) assert result_plain == plain_id def test_update_responses_api_response_id_with_model_id_handles_dict(self): """Ensure _update_responses_api_response_id_with_model_id works with dict input""" responses_api_response = {"id": "resp_abc123"} litellm_metadata = {"model_info": {"id": "gpt-4o"}} updated = ResponsesAPIRequestUtils.update_responses_api_response_id_with_model_id( responses_api_response=responses_api_response, custom_llm_provider="openai", litellm_metadata=litellm_metadata, ) assert updated["id"] != "resp_abc123" decoded = ResponsesAPIRequestUtils.decode_responses_api_response_id(updated["id"]) assert decoded.get("response_id") == "resp_abc123" assert decoded.get("model_id") == "gpt-4o" assert decoded.get("custom_llm_provider") == "openai" def test_update_responses_api_response_id_with_model_id_is_idempotent_for_litellm_ids(self): raw = "resp_" + "a" * 48 litellm_metadata = {"model_info": {"id": "model-123"}} once = ResponsesAPIRequestUtils.update_responses_api_response_id_with_model_id( {"id": raw}, custom_llm_provider="openai", litellm_metadata=litellm_metadata, ) twice = ResponsesAPIRequestUtils.update_responses_api_response_id_with_model_id( {"id": once["id"]}, custom_llm_provider="openai", litellm_metadata=litellm_metadata, ) assert twice == once assert ResponsesAPIRequestUtils.decode_previous_response_id_to_original_previous_response_id(twice["id"]) == raw assert ResponsesAPIRequestUtils.decode_responses_api_response_id(once["id"]).get("response_id") == raw def test_build_decode_container_id_omits_none_model_id(self): """model_id=None must not round-trip as the truthy string 'None'.""" encoded = ResponsesAPIRequestUtils.build_container_id( custom_llm_provider="azure", model_id=None, container_id="cntr_upstream_abc", ) assert "None" not in base64.b64decode(encoded.replace("cntr_", "").encode("utf-8")).decode("utf-8") decoded = ResponsesAPIRequestUtils.decode_container_id(encoded) assert decoded.get("custom_llm_provider") == "azure" assert decoded.get("model_id") is None assert decoded.get("response_id") == "cntr_upstream_abc" def test_decode_container_id_legacy_literal_none_model_id(self): """IDs encoded before the None fix should decode without a bogus model_id.""" legacy_inner = "litellm:custom_llm_provider:azure;model_id:None;container_id:cntr_x" legacy_id = "cntr_" + base64.b64encode(legacy_inner.encode("utf-8")).decode("utf-8") decoded = ResponsesAPIRequestUtils.decode_container_id(legacy_id) assert decoded.get("model_id") is None assert decoded.get("custom_llm_provider") == "azure" assert decoded.get("response_id") == "cntr_x" class TestResponseAPILoggingUtils: def test_is_response_api_usage_true(self): """Test identification of Response API usage format""" # Setup usage = {"input_tokens": 10, "output_tokens": 20} # Execute result = ResponseAPILoggingUtils.is_response_api_usage(usage) # Assert assert result is True def test_is_response_api_usage_false(self): """Test identification of non-Response API usage format""" # Setup usage = {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30} # Execute result = ResponseAPILoggingUtils.is_response_api_usage(usage) # Assert assert result is False def test_transform_response_api_usage_to_chat_usage(self): """Test transformation from Response API usage to Chat usage format""" # Setup usage = { "input_tokens": 10, "output_tokens": 20, "total_tokens": 30, "input_tokens_details": {"cached_tokens": 2}, "output_tokens_details": {"reasoning_tokens": 5}, } # Execute result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) # Assert assert isinstance(result, Usage) assert result.prompt_tokens == 10 assert result.completion_tokens == 20 assert result.total_tokens == 30 assert result.prompt_tokens_details and result.prompt_tokens_details.cached_tokens == 2 def test_transform_response_api_usage_with_none_values(self): """Test transformation handles None values properly""" # Setup usage = { "input_tokens": 0, # Changed from None to 0 "output_tokens": 20, "total_tokens": 20, "output_tokens_details": {"reasoning_tokens": 5}, } # Execute result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) # Assert assert result.prompt_tokens == 0 assert result.completion_tokens == 20 assert result.total_tokens == 20 def test_transform_response_api_usage_calculates_total_from_input_and_output_tokens_if_available( self, ): """Test transformation calculates total_tokens when it's None and input / output tokens are present""" # Setup usage = { "input_tokens": 15, "output_tokens": 25, "total_tokens": None, } # Execute result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) # Assert assert result.prompt_tokens == 15 assert result.completion_tokens == 25 assert result.total_tokens == 40 # 15 + 25 def test_transform_response_api_usage_with_image_tokens(self): """Test transformation handles image_tokens from image generation responses. Note: _transform_response_api_usage_to_chat_usage() is used by multiple endpoints including /images/generations and Response API (/responses), both of which use the input_tokens/output_tokens format. This tests the fix for image generation responses that include image_tokens in both input_tokens_details and output_tokens_details. Example from gpt-image-1.5: - input: text prompt with 13 tokens - output: generated image with 272 image tokens + 100 text tokens """ # Setup - simulating image generation usage from OpenAI usage = { "input_tokens": 13, "output_tokens": 372, "total_tokens": 385, "input_tokens_details": { "image_tokens": 0, "text_tokens": 13, }, "output_tokens_details": { "image_tokens": 272, "text_tokens": 100, }, } # Execute result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) # Assert - verify basic token counts assert isinstance(result, Usage) assert result.prompt_tokens == 13 assert result.completion_tokens == 372 assert result.total_tokens == 385 # Assert - verify prompt_tokens_details includes image_tokens and text_tokens assert result.prompt_tokens_details is not None assert result.prompt_tokens_details.image_tokens == 0 assert result.prompt_tokens_details.text_tokens == 13 # Assert - verify completion_tokens_details includes image_tokens and text_tokens assert result.completion_tokens_details is not None assert result.completion_tokens_details.image_tokens == 272 assert result.completion_tokens_details.text_tokens == 100 def test_transform_response_api_usage_maps_cache_write_tokens(self): """Responses API (/v1/responses) cache-write tokens must survive the usage transform. gpt-5.6 returns usage.input_tokens_details.cache_write_tokens (an extra field not typed on InputTokensDetails). Before the fix the transform rebuilt the token details and dropped it, leaving the cache-creation metric empty (LIT-4633). """ usage = { "input_tokens": 10062, "output_tokens": 16, "total_tokens": 10078, "input_tokens_details": { "cached_tokens": 0, "cache_write_tokens": 10059, }, } result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result.prompt_tokens_details is not None assert result.prompt_tokens_details.cache_write_tokens == 10059 assert result.prompt_tokens_details.cache_creation_tokens == 10059 assert result.prompt_tokens_details.cached_tokens == 0 @pytest.mark.parametrize("as_model", [False, True]) def test_transform_response_api_usage_prices_12h_cache_writes(self, as_model: bool) -> None: usage_data: Final = { "input_tokens": 5_000, "output_tokens": 0, "total_tokens": 5_000, "input_tokens_details": { "cache_write_tokens": 5_000, "cache_write_tokens_details": {"ttl_30m_tokens": 3_000, "ttl_12h_tokens": 2_000}, }, } usage_input: Final = ResponseAPIUsage(**usage_data) if as_model else usage_data usage: Final = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage_input) model_info: Final[ModelInfo] = { "input_cost_per_token": 1e-6, "output_cost_per_token": 2e-6, "cache_creation_input_token_cost": 3e-6, "cache_creation_input_token_cost_above_12hr": 5e-6, } assert usage.prompt_tokens_details is not None assert usage.prompt_tokens_details.cache_write_tokens == 5_000 assert usage.prompt_tokens_details.cache_creation_token_details is not None assert usage.prompt_tokens_details.cache_creation_token_details.ephemeral_12h_input_tokens == 2_000 assert usage.prompt_tokens_details.cache_creation_token_details.ephemeral_30m_input_tokens == 3_000 cost: Final = generic_cost_per_token( model="test-12h-cache-write", usage=usage, custom_llm_provider="openai", model_info=model_info, )[0] assert cost == pytest.approx( 3_000 * model_info["cache_creation_input_token_cost"] + 2_000 * model_info["cache_creation_input_token_cost_above_12hr"] ) def test_transform_response_api_usage_mixed_details(self): """Test transformation handles mixed token details (cached + image + audio).""" # Setup - hypothetical usage with mixed token types usage = { "input_tokens": 100, "output_tokens": 200, "total_tokens": 300, "input_tokens_details": { "cached_tokens": 50, "audio_tokens": 10, "image_tokens": 20, "text_tokens": 20, }, "output_tokens_details": { "reasoning_tokens": 30, "image_tokens": 100, "text_tokens": 50, "audio_tokens": 20, }, } # Execute result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) # Assert - all token detail types should be preserved assert result.prompt_tokens_details is not None assert result.prompt_tokens_details.cached_tokens == 50 assert result.prompt_tokens_details.audio_tokens == 10 assert result.prompt_tokens_details.image_tokens == 20 assert result.prompt_tokens_details.text_tokens == 20 assert result.completion_tokens_details is not None assert result.completion_tokens_details.reasoning_tokens == 30 assert result.completion_tokens_details.image_tokens == 100 assert result.completion_tokens_details.text_tokens == 50 assert result.completion_tokens_details.audio_tokens == 20 def test_transform_response_api_usage_with_realtime_keys(self): """Realtime input_token_details / output_token_details normalize for Usage.""" usage = { "input_tokens": 10, "output_tokens": 20, "total_tokens": 30, "input_token_details": { "text_tokens": 8, "audio_tokens": 2, "cached_tokens": 0, }, "output_token_details": { "text_tokens": 12, "audio_tokens": 8, }, } result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result.prompt_tokens_details is not None assert result.prompt_tokens_details.text_tokens == 8 assert result.prompt_tokens_details.audio_tokens == 2 assert result.completion_tokens_details is not None assert result.completion_tokens_details.text_tokens == 12 assert result.completion_tokens_details.audio_tokens == 8 def test_transform_response_api_usage_tokens_details_keep_values(self): """Keeps input_tokens_details / output_tokens_details when singular keys are also present.""" usage = { "input_tokens": 10, "output_tokens": 20, "total_tokens": 30, "input_tokens_details": {"text_tokens": 10}, "output_tokens_details": {"text_tokens": 20}, "input_token_details": {"text_tokens": 1, "audio_tokens": 99}, "output_token_details": {"text_tokens": 2, "audio_tokens": 98}, } result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result.prompt_tokens_details is not None assert result.prompt_tokens_details.text_tokens == 10 assert result.prompt_tokens_details.audio_tokens is None assert result.completion_tokens_details is not None assert result.completion_tokens_details.text_tokens == 20 assert result.completion_tokens_details.audio_tokens is None def test_transform_realtime_usage_partitions_reasoning_out_of_text_tokens(self): """Realtime nests reasoning_tokens inside text_tokens; the stored text share excludes them.""" usage = { "input_tokens": 237, "output_tokens": 70, "total_tokens": 307, "input_token_details": {"text_tokens": 43, "audio_tokens": 0, "image_tokens": 194, "cached_tokens": 0}, "output_token_details": {"text_tokens": 70, "audio_tokens": 0, "reasoning_tokens": 52}, } result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result.completion_tokens == 70 assert result.completion_tokens_details is not None assert result.completion_tokens_details.text_tokens == 18 assert result.completion_tokens_details.reasoning_tokens == 52 assert result.completion_tokens_details.audio_tokens == 0 def test_transform_realtime_usage_partitions_reasoning_beside_audio_output(self): """Audio output stays as reported; only the text share sheds the nested reasoning tokens.""" usage = { "input_tokens": 100, "output_tokens": 70, "total_tokens": 170, "input_token_details": {"text_tokens": 100, "audio_tokens": 0, "cached_tokens": 0}, "output_token_details": {"text_tokens": 39, "audio_tokens": 31, "reasoning_tokens": 23}, } result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result.completion_tokens_details is not None assert result.completion_tokens_details.text_tokens == 16 assert result.completion_tokens_details.audio_tokens == 31 assert result.completion_tokens_details.reasoning_tokens == 23 def test_transform_response_api_usage_keeps_partitioned_text_tokens(self): """A provider already reporting text_tokens beside reasoning_tokens is stored as sent.""" usage = { "input_tokens": 10, "output_tokens": 20, "total_tokens": 30, "output_tokens_details": {"text_tokens": 12, "reasoning_tokens": 5}, } result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result.completion_tokens_details is not None assert result.completion_tokens_details.text_tokens == 12 assert result.completion_tokens_details.reasoning_tokens == 5 def test_transform_response_api_usage_carries_extra_provider_fields(self): """Non-standard usage fields (e.g. xAI tool details) must survive chat normalization.""" details = {"web_search_calls": 2, "x_search_calls": 0} usage = ResponseAPIUsage( input_tokens=100, output_tokens=20, total_tokens=120, server_side_tool_usage_details=details, ) result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert isinstance(result, Usage) assert result.prompt_tokens == 100 assert result.completion_tokens == 20 assert getattr(result, "server_side_tool_usage_details") == details def test_transform_response_api_usage_ignores_chat_shaped_extras(self): """Gemini image usage carries chat-shaped keys as extras; they must not collide with explicit kwargs.""" usage = ResponseAPIUsage( input_tokens=35, output_tokens=1716, total_tokens=1751, prompt_tokens=35, prompt_tokens_details={"image_tokens": 5, "text_tokens": 30}, completion_tokens=1716, completion_tokens_details={"image_tokens": 1120, "text_tokens": 596}, server_side_tool_usage_details={"web_search_calls": 1}, ) result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result.prompt_tokens == 35 assert result.completion_tokens == 1716 assert getattr(result, "server_side_tool_usage_details") == {"web_search_calls": 1} def test_transform_already_chat_usage_passthrough_keeps_tool_details(self): """Re-running the bridge on an already-converted chat Usage must not drop fields.""" details = {"web_search_calls": 2, "x_search_calls": 0} usage = Usage( prompt_tokens=100, completion_tokens=20, total_tokens=120, prompt_tokens_details={"web_search_requests": 2}, ) setattr(usage, "server_side_tool_usage_details", details) result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result is usage assert getattr(result, "server_side_tool_usage_details") == details assert result.prompt_tokens_details is not None assert result.prompt_tokens_details.web_search_requests == 2 def test_transform_chat_shaped_usage_dict_keeps_tool_details(self): """Streaming chat bridge dumps already-converted Usage as a prompt_tokens dict.""" details = { "web_search_calls": 3, "x_search_calls": 0, "code_interpreter_calls": 0, "file_search_calls": 0, "mcp_calls": 0, "document_search_calls": 0, "image_generation_calls": 0, } usage = { "prompt_tokens": 50, "completion_tokens": 10, "total_tokens": 60, "prompt_tokens_details": {"web_search_requests": 3, "cached_tokens": 8}, "completion_tokens_details": {"reasoning_tokens": 4}, "server_side_tool_usage_details": details, } result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert isinstance(result, Usage) assert result.prompt_tokens == 50 assert result.completion_tokens == 10 assert result.total_tokens == 60 assert getattr(result, "server_side_tool_usage_details") == details assert result.prompt_tokens_details is not None assert result.prompt_tokens_details.web_search_requests == 3 assert result.prompt_tokens_details.cached_tokens == 8 assert result.completion_tokens_details is not None assert result.completion_tokens_details.reasoning_tokens == 4 def test_transform_realtime_usage_dict_keeps_cached_tokens_details(self): usage = { "input_tokens": 283, "output_tokens": 0, "total_tokens": 283, "input_token_details": { "text_tokens": 116, "audio_tokens": 167, "cached_tokens": 192, "cached_tokens_details": {"text_tokens": 64, "audio_tokens": 128}, }, } result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result.prompt_tokens_details is not None assert result.prompt_tokens_details.cached_tokens == 192 assert result.prompt_tokens_details.cached_tokens_details is not None assert result.prompt_tokens_details.cached_tokens_details.audio_tokens == 128 assert result.prompt_tokens_details.cached_tokens_details.text_tokens == 64 def test_transform_response_api_usage_object_keeps_cached_tokens_details(self): usage = ResponseAPIUsage( input_tokens=283, output_tokens=0, total_tokens=283, input_tokens_details={ "text_tokens": 116, "audio_tokens": 167, "cached_tokens": 192, "cached_tokens_details": {"text_tokens": 64, "audio_tokens": 128}, }, ) result = ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage) assert result.prompt_tokens_details is not None assert result.prompt_tokens_details.cached_tokens_details is not None assert result.prompt_tokens_details.cached_tokens_details.audio_tokens == 128 assert result.prompt_tokens_details.cached_tokens_details.text_tokens == 64 class TestResponsesAPIProviderSpecificParams: """ Tests for fix #19782: provider-specific params (aws_*, vertex_*) should work without explicitly passing custom_llm_provider. """ def test_provider_specific_params_no_crash_with_bedrock(self): """Test that processing aws_* params with bedrock provider doesn't crash.""" params = { "temperature": 0.7, "custom_llm_provider": "bedrock", "kwargs": {"aws_region_name": "eu-central-1"}, } # Should not raise any exception result = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(params) assert "temperature" in result def test_provider_specific_params_no_crash_with_openai(self): """Test that processing aws_* params with openai provider doesn't crash.""" params = { "temperature": 0.7, "custom_llm_provider": "openai", "kwargs": {"aws_region_name": "eu-central-1"}, } # Should not raise any exception result = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(params) assert "temperature" in result def test_provider_specific_params_no_crash_with_vertex_ai(self): """Test that processing vertex_* params with vertex_ai provider doesn't crash.""" params = { "temperature": 0.7, "custom_llm_provider": "vertex_ai", "kwargs": {"vertex_project": "my-project"}, } # Should not raise any exception result = ResponsesAPIRequestUtils.get_requested_response_api_optional_param(params) assert "temperature" in result def test_responses_extra_body_forwarded_to_completion_transformation_handler(): """ Regression test: extra_body must be forwarded to response_api_handler when responses_api_provider_config is None (completion transformation path). Before the fix, extra_body was a named parameter of responses() but was not passed to litellm_completion_transformation_handler.response_api_handler(), so it was silently dropped. """ with ( patch.object( import_module("litellm.responses.main").ProviderConfigManager, "get_provider_responses_api_config", return_value=None, ), patch.object( import_module("litellm.responses.main").litellm_completion_transformation_handler, "response_api_handler", ) as mock_handler, ): mock_handler.return_value = MagicMock() litellm.responses( model="openai/gpt-4o", input="Hello", extra_body={"custom_key": "custom_value"}, ) mock_handler.assert_called_once() call_kwargs = mock_handler.call_args # extra_body can be a positional or keyword arg; check both assert call_kwargs.kwargs.get("extra_body") == {"custom_key": "custom_value"} def test_responses_maps_reasoning_effort_from_litellm_params_to_reasoning(): """ Test that when reasoning_effort is passed in kwargs (e.g. from proxy litellm_params) and reasoning is None, it is mapped to reasoning before the request. Supports per-model reasoning_effort/summary config in proxy for clients like Open WebUI that cannot set extra_body. """ with ( patch.object( import_module("litellm.responses.main").ProviderConfigManager, "get_provider_responses_api_config", return_value=None, ), patch.object( import_module("litellm.responses.main").litellm_completion_transformation_handler, "response_api_handler", ) as mock_handler, ): mock_handler.return_value = MagicMock() litellm.responses( model="openai/gpt-4o", input="Hello", reasoning_effort={"effort": "high", "summary": "detailed"}, ) mock_handler.assert_called_once() call_kwargs = mock_handler.call_args responses_api_request = call_kwargs.kwargs.get("responses_api_request", {}) assert "reasoning" in responses_api_request assert responses_api_request["reasoning"] == { "effort": "high", "summary": "detailed", } class TestMergePromptManagementInputReshape: """Chat-shaped text parts produced by prompt management hooks become input_text parts (#37509).""" EXPLICIT = {"mode": "explicit"} def _run_cache_hook(self, client_input, points, model="openai/gpt-5.6"): from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook _, merged, _ = AnthropicCacheControlHook().get_chat_completion_prompt( model=model, messages=client_input, non_default_params={"cache_control_injection_points": points}, prompt_id=None, prompt_variables=None, dynamic_callback_params={}, ) return merged def test_string_system_item_becomes_input_text_with_marker(self): original_input = [{"role": "system", "content": "You are terse."}, {"role": "user", "content": "hi"}] merged = self._run_cache_hook(list(original_input), [{"location": "message", "role": "system"}]) result = ResponsesAPIRequestUtils.merge_prompt_management_input( original_input=original_input, client_input=list(original_input), merged_input=merged ) assert result[0]["content"] == [ {"type": "input_text", "text": "You are terse.", "prompt_cache_breakpoint": self.EXPLICIT} ] assert result[1] == {"role": "user", "content": "hi"} def test_reshape_returns_copies_and_leaves_hook_output_untouched(self): user_part = {"type": "text", "text": "follow-up"} user_message = {"role": "user", "content": [user_part]} merged = [user_message] result = ResponsesAPIRequestUtils.merge_prompt_management_input( original_input="ignored", client_input=[], merged_input=merged ) assert result == [{"role": "user", "content": [{"type": "input_text", "text": "follow-up"}]}] assert user_part == {"type": "text", "text": "follow-up"} assert user_message == {"role": "user", "content": [user_part]} assert result[0] is not user_message def test_reshape_keeps_non_message_items_when_hook_returns_client_objects(self): user_message = {"role": "user", "content": [{"type": "text", "text": "question"}]} reference = {"type": "item_reference", "id": "msg_123"} original_input = [reference, user_message] result = ResponsesAPIRequestUtils.merge_prompt_management_input( original_input=original_input, client_input=[user_message], merged_input=[user_message] ) assert result == [reference, {"role": "user", "content": [{"type": "input_text", "text": "question"}]}] assert result[0] is reference assert user_message["content"] == [{"type": "text", "text": "question"}] def test_assistant_text_parts_are_left_alone(self): merged = [ {"role": "assistant", "content": [{"type": "text", "text": "earlier answer"}]}, {"role": "user", "content": [{"type": "text", "text": "follow-up"}]}, ] result = ResponsesAPIRequestUtils.merge_prompt_management_input( original_input="ignored", client_input=[], merged_input=merged ) assert result[0]["content"] == [{"type": "text", "text": "earlier answer"}] assert result[1]["content"] == [{"type": "input_text", "text": "follow-up"}] def test_parts_already_in_responses_shape_are_unchanged(self): merged = [ { "role": "user", "content": [ {"type": "input_text", "text": "a", "prompt_cache_breakpoint": self.EXPLICIT}, {"type": "input_image", "image_url": "https://example.com/a.png"}, ], } ] result = ResponsesAPIRequestUtils.merge_prompt_management_input( original_input="ignored", client_input=[], merged_input=merged ) assert result == merged class TestResponsesInputToChatMessages: def test_none_input_returns_empty_list(self): assert ResponsesAPIRequestUtils.responses_input_to_chat_messages(None) == [] def test_str_input_becomes_user_message(self): assert ResponsesAPIRequestUtils.responses_input_to_chat_messages("hi") == [ {"role": "user", "content": "hi"} ] def test_list_input_keeps_only_role_items(self): reasoning_item = {"type": "reasoning", "id": "rs_1", "summary": []} user_message = {"role": "user", "content": "hi"} assert ResponsesAPIRequestUtils.responses_input_to_chat_messages( [reasoning_item, user_message, "stray"] ) == [user_message] class TestMergeProviderSpecificHeaders: def test_scoped_headers_merge_only_for_matching_provider(self): scoped = {"custom_llm_provider": "chatgpt", "extra_headers": {"Authorization": "Bearer client-token"}} merged = ResponsesAPIRequestUtils.merge_provider_specific_headers( extra_headers={"originator": "codex_cli_rs"}, provider_specific_header=scoped, custom_llm_provider="chatgpt", ) untouched = ResponsesAPIRequestUtils.merge_provider_specific_headers( extra_headers={"originator": "codex_cli_rs"}, provider_specific_header=scoped, custom_llm_provider="openai", ) assert merged == {"originator": "codex_cli_rs", "Authorization": "Bearer client-token"} assert untouched == {"originator": "codex_cli_rs"} def test_sequence_of_scoped_entries_each_keep_their_own_scope(self): entries = [ {"custom_llm_provider": "anthropic", "extra_headers": {"Authorization": "Bearer anthropic-token"}}, {"custom_llm_provider": "chatgpt", "extra_headers": {"Authorization": "Bearer chatgpt-token"}}, ] merged = ResponsesAPIRequestUtils.merge_provider_specific_headers( extra_headers=None, provider_specific_header=entries, custom_llm_provider="chatgpt" ) assert merged == {"Authorization": "Bearer chatgpt-token"} def test_no_scoped_headers_returns_extra_headers_unchanged(self): assert ResponsesAPIRequestUtils.merge_provider_specific_headers( extra_headers=None, provider_specific_header=None, custom_llm_provider="chatgpt" ) is None def test_responses_object_bills_an_explicit_cache_hit_at_the_explicit_rate(): model = "dashscope/qwen-responses-explicit-cache-test" previous_row = litellm.model_cost.get(model) litellm.model_cost[model] = { "litellm_provider": "dashscope", "mode": "chat", "input_cost_per_token": 2.5e-06, "output_cost_per_token": 7.5e-06, "cache_read_input_token_cost": 5e-07, "explicit_cache_read_input_token_cost": 2.5e-07, } _invalidate_model_cost_lowercase_map() try: chat_usage = Usage( prompt_tokens=1000, completion_tokens=0, prompt_tokens_details={"cached_tokens": 1000, "cache_type": "ephemeral"}, ) responses_object = ResponsesAPIResponse( id="resp_explicit_cache", created_at=0, model="qwen-responses-explicit-cache-test", object="response", status="completed", output=[], parallel_tool_calls=False, tool_choice="auto", tools=[], usage=LiteLLMCompletionResponsesConfig.transform_chat_completion_usage_to_responses_usage(chat_usage), ) assert litellm.completion_cost(completion_response=responses_object, model=model) == pytest.approx( 1000 * 2.5e-07 ) finally: if previous_row is None: litellm.model_cost.pop(model, None) else: litellm.model_cost[model] = previous_row _invalidate_model_cost_lowercase_map()