Welcome to this simple webpage!
\\n\\n > index.html\\n","metadata":{"exit_code":0,"duration_seconds":0}}', + } + ], + tools=[shell_tool], + ) + + print("follow_up_response=", follow_up_response) diff --git a/tests/harness_suites/responses_api_live/test_azure_responses_live.py b/tests/harness_suites/responses_api_live/test_azure_responses_live.py new file mode 100644 index 00000000000..9344e3371bf --- /dev/null +++ b/tests/harness_suites/responses_api_live/test_azure_responses_live.py @@ -0,0 +1,38 @@ +import os +import sys +import pytest +from typing import Optional + +sys.path.insert(0, os.path.abspath("../..")) +import litellm +from base_responses_api import BaseResponsesAPITest + +class TestAzureResponsesAPITest(BaseResponsesAPITest): + def get_base_completion_call_args(self): + return { + "model": "azure/gpt-4.1-mini", + "truncation": "auto", + "api_base": os.getenv("AZURE_AI_API_BASE"), + "api_key": os.getenv("AZURE_AI_API_KEY"), + "api_version": "2025-03-01-preview", + } + + def get_advanced_model_for_shell_tool(self) -> Optional[str]: + """If specified, overrides the model used by test_responses_api_shell_tool_streaming_sees_shell_output (e.g. openai/gpt-5.2 for shell support).""" + return "azure/gpt-5-mini" + + +@pytest.mark.asyncio +async def test_azure_responses_api_preview_api_version(): + """ + Ensure new azure preview api version is working + """ + litellm._turn_on_debug() + response = await litellm.aresponses( + model="azure/gpt-5-mini", + truncation="auto", + api_version="preview", + api_base=os.getenv("AZURE_AI_API_BASE"), + api_key=os.getenv("AZURE_AI_API_KEY"), + input="Hello, can you tell me a short joke?", + ) diff --git a/tests/harness_suites/responses_api_live/test_google_ai_studio_responses_live.py b/tests/harness_suites/responses_api_live/test_google_ai_studio_responses_live.py new file mode 100644 index 00000000000..523280fd043 --- /dev/null +++ b/tests/harness_suites/responses_api_live/test_google_ai_studio_responses_live.py @@ -0,0 +1,234 @@ +import os +import sys +import pytest + +sys.path.insert(0, os.path.abspath("../..")) +import litellm +import json +from base_responses_api import BaseResponsesAPITest + +@pytest.mark.asyncio +async def test_basic_google_ai_studio_responses_api_with_tools(): + litellm._turn_on_debug() + litellm.set_verbose = True + request_model = "gemini/gemini-2.5-flash" + response = await litellm.aresponses( + model=request_model, + input="what is the latest version of supabase python package and when was it released?", + tools=[{"type": "web_search_preview", "search_context_size": "low"}], + ) + print("litellm response=", json.dumps(response, indent=4, default=str)) + + +@pytest.mark.asyncio +async def test_gemini_3_responses_api_with_thought_signatures(): + """ + Test that Gemini 3 Responses API preserves thought signatures in function calls. + This test verifies that provider_specific_fields with thought_signature are correctly + preserved when using the Responses API with Gemini 3. + """ + if not os.getenv("GEMINI_API_KEY"): + pytest.skip("GEMINI_API_KEY not set") + + litellm.set_verbose = False + request_model = "gemini/gemini-3.1-pro-preview" + + tools = [ + { + "type": "function", + "name": "get_weather", + "description": "Get the current weather for a location", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "City and country e.g. Mumbai, India", + }, + "units": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "Units the temperature will be returned in.", + }, + }, + "required": ["location", "units"], + "additionalProperties": False, + }, + "strict": True, + } + ] + + # Step 1: Initial request with tools + response = await litellm.aresponses( + model=request_model, + input="What is the weather in Mumbai?", + tools=tools, + reasoning_effort="low", + ) + + # Validate response structure + from litellm.types.llms.openai import ResponsesAPIResponse + + assert isinstance( + response, ResponsesAPIResponse + ), "Response should be a ResponsesAPIResponse" + assert ( + hasattr(response, "output") or "output" in response + ), "Response should have 'output' field" + assert isinstance(response.output, list), "Output should be a list" + + # Find function call in output + function_call_item = None + for item in response.output: + # Convert to dict if it's a Pydantic model for easier access + if hasattr(item, "model_dump"): + item_dict = item.model_dump() + elif hasattr(item, "__dict__"): + item_dict = dict(item) if not isinstance(item, dict) else item + else: + item_dict = item if isinstance(item, dict) else {} + + if isinstance(item_dict, dict) and item_dict.get("type") == "function_call": + function_call_item = item_dict + break + + # Verify function call exists + assert ( + function_call_item is not None + ), "Response should contain a function_call item" + assert ( + function_call_item.get("name") == "get_weather" + ), "Function call should be for get_weather" + + # Verify thought signature is present in provider_specific_fields + provider_specific_fields = function_call_item.get("provider_specific_fields") + assert ( + provider_specific_fields is not None + ), "Function call should have provider_specific_fields" + assert ( + "thought_signature" in provider_specific_fields + ), "provider_specific_fields should contain thought_signature" + assert isinstance( + provider_specific_fields["thought_signature"], str + ), "thought_signature should be a string" + assert ( + len(provider_specific_fields["thought_signature"]) > 0 + ), "thought_signature should not be empty" + + print( + f"✅ Thought signature preserved: {provider_specific_fields['thought_signature'][:50]}..." + ) + + +@pytest.mark.asyncio +async def test_gemini_3_responses_api_streaming_with_thought_signatures(): + """ + Test that Gemini 3 Responses API preserves thought signatures in streaming mode. + This test verifies that provider_specific_fields with thought_signature are correctly + preserved when using streaming Responses API with Gemini 3. + """ + if not os.getenv("GEMINI_API_KEY"): + pytest.skip("GEMINI_API_KEY not set") + + litellm.set_verbose = False + request_model = "gemini/gemini-3.1-pro-preview" + + tools = [ + { + "type": "function", + "name": "get_weather", + "description": "Get the current weather for a location", + "parameters": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "City and country e.g. Mumbai, India", + }, + "units": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "Units the temperature will be returned in.", + }, + }, + "required": ["location", "units"], + "additionalProperties": False, + }, + "strict": True, + } + ] + + # Step 1: Streaming request with tools + response_stream = await litellm.aresponses( + model=request_model, + input="What is the weather in Mumbai?", + tools=tools, + stream=True, + reasoning_effort="low", + ) + + # Collect all chunks + chunks = [] + completed_response = None + + async for chunk in response_stream: + chunks.append(chunk) + # Check if this is the completed response event + if hasattr(chunk, "type") and chunk.type == "response.completed": + completed_response = chunk.response + elif isinstance(chunk, dict) and chunk.get("type") == "response.completed": + completed_response = chunk.get("response") + + # Verify we got chunks + assert len(chunks) > 0, "Should receive at least one chunk" + + # If we have a completed response, check for thought signatures + if completed_response: + output = completed_response.get("output", []) + function_call_item = None + for item in output: + if isinstance(item, dict) and item.get("type") == "function_call": + function_call_item = item + break + + if function_call_item: + provider_specific_fields = function_call_item.get( + "provider_specific_fields" + ) + if provider_specific_fields: + thought_signature = provider_specific_fields.get("thought_signature") + if thought_signature: + assert isinstance( + thought_signature, str + ), "thought_signature should be a string" + assert ( + len(thought_signature) > 0 + ), "thought_signature should not be empty" + print( + f"✅ Streaming thought signature preserved: {thought_signature[:50]}..." + ) + + print(f"✅ Collected {len(chunks)} streaming chunks") + + +class TestGoogleAIStudioResponsesAPITest(BaseResponsesAPITest): + def get_base_completion_call_args(self): + # litellm._turn_on_debug() + return {"model": "gemini/gemini-2.5-flash-lite"} + + async def test_basic_openai_responses_delete_endpoint(self, sync_mode=False): + pytest.skip("DELETE responses is not supported for Google AI Studio") + + async def test_basic_openai_responses_streaming_delete_endpoint( + self, sync_mode=False + ): + pytest.skip("DELETE responses is not supported for Google AI Studio") + + async def test_basic_openai_responses_get_endpoint(self, sync_mode=False): + pytest.skip("GET responses is not supported for Google AI Studio") + + async def test_basic_openai_responses_cancel_endpoint(self, sync_mode=False): + pytest.skip("CANCEL responses is not supported for Google AI Studio") + + async def test_cancel_responses_invalid_response_id(self, sync_mode=False): + pytest.skip("CANCEL responses is not supported for Google AI Studio") diff --git a/tests/harness_suites/responses_api_live/test_openai_responses_live.py b/tests/harness_suites/responses_api_live/test_openai_responses_live.py new file mode 100644 index 00000000000..4ce38402dbc --- /dev/null +++ b/tests/harness_suites/responses_api_live/test_openai_responses_live.py @@ -0,0 +1,847 @@ +import os +import sys +import pytest +import asyncio +from typing import Optional, cast +import time +import json + +sys.path.insert(0, os.path.abspath("../..")) +import litellm +from litellm.integrations.custom_logger import CustomLogger +from litellm.types.utils import StandardLoggingPayload +from litellm.types.llms.openai import ( + ResponseCompletedEvent, + ResponsesAPIResponse, + ResponseAPIUsage, +) +from base_responses_api import BaseResponsesAPITest, validate_responses_api_response + +class TestCustomLogger(CustomLogger): + def __init__( + self, + ): + self.standard_logging_object: Optional[StandardLoggingPayload] = None + + async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): + print("in async_log_success_event") + print("kwargs=", json.dumps(kwargs, indent=4, default=str)) + self.standard_logging_object = kwargs["standard_logging_object"] + pass + + +def validate_standard_logging_payload( + slp: StandardLoggingPayload, response: ResponsesAPIResponse, request_model: str +): + """ + Validate that a StandardLoggingPayload object matches the expected response + + Args: + slp (StandardLoggingPayload): The standard logging payload object to validate + response (dict): The litellm response to compare against + request_model (str): The model name that was requested + """ + # Validate payload exists + assert slp is not None, "Standard logging payload should not be None" + + # Validate token counts + print( + "VALIDATING STANDARD LOGGING PAYLOAD. response=", + json.dumps(response, indent=4, default=str), + ) + print("FIELDS IN SLP=", json.dumps(slp, indent=4, default=str)) + print("SLP PROMPT TOKENS=", slp["prompt_tokens"]) + print("RESPONSE PROMPT TOKENS=", response["usage"]["input_tokens"]) + assert ( + slp["prompt_tokens"] == response["usage"]["input_tokens"] + ), "Prompt tokens mismatch" + assert ( + slp["completion_tokens"] == response["usage"]["output_tokens"] + ), "Completion tokens mismatch" + assert ( + slp["total_tokens"] + == response["usage"]["input_tokens"] + response["usage"]["output_tokens"] + ), "Total tokens mismatch" + + # Validate spend and response metadata + assert slp["response_cost"] > 0, "Response cost should be greater than 0" + assert slp["id"] == response["id"], "Response ID mismatch" + assert slp["model"] == request_model, "Model name mismatch" + + # Validate messages + assert slp["messages"] == [{"content": "hi", "role": "user"}], "Messages mismatch" + + # Validate complete response structure + validate_responses_match(slp["response"], response) + + +def validate_responses_match(slp_response, litellm_response): + """Validate that the standard logging payload OpenAI response matches the litellm response""" + # Validate core fields + assert slp_response["id"] == litellm_response["id"], "ID mismatch" + assert slp_response["model"] == litellm_response["model"], "Model mismatch" + assert ( + slp_response["created_at"] == litellm_response["created_at"] + ), "Created at mismatch" + + # Validate usage + assert ( + slp_response["usage"]["prompt_tokens"] + == litellm_response["usage"]["input_tokens"] + ), "Input tokens mismatch" + assert ( + slp_response["usage"]["completion_tokens"] + == litellm_response["usage"]["output_tokens"] + ), "Output tokens mismatch" + assert ( + slp_response["usage"]["total_tokens"] + == litellm_response["usage"]["total_tokens"] + ), "Total tokens mismatch" + + # Validate output/messages + assert len(slp_response["output"]) == len( + litellm_response["output"] + ), "Output length mismatch" + for slp_msg, litellm_msg in zip(slp_response["output"], litellm_response["output"]): + assert slp_msg["role"] == litellm_msg.role, "Message role mismatch" + # Access the content's text field for the litellm response + litellm_content = litellm_msg.content[0].text if litellm_msg.content else "" + assert ( + slp_msg["content"][0]["text"] == litellm_content + ), f"Message content mismatch. Expected {litellm_content}, Got {slp_msg['content']}" + assert slp_msg["status"] == litellm_msg.status, "Message status mismatch" + + +def validate_stream_event(event): + """ + Validate that a streaming event from litellm.responses() or litellm.aresponses() + with stream=True conforms to the expected structure based on its event type. + + Args: + event: The streaming event object to validate + + Raises: + AssertionError: If the event doesn't match the expected structure for its type + """ + # Common validation for all event types + assert hasattr(event, "type"), "Event should have a 'type' attribute" + + # Type-specific validation + if event.type == "response.created" or event.type == "response.in_progress": + assert hasattr( + event, "response" + ), f"{event.type} event should have a 'response' attribute" + validate_responses_api_response(event.response, final_chunk=False) + + elif event.type == "response.completed": + assert hasattr( + event, "response" + ), "response.completed event should have a 'response' attribute" + validate_responses_api_response(event.response, final_chunk=True) + # Usage is guaranteed only on the completed event + assert ( + "usage" in event.response + ), "response.completed event should have usage information" + print("Usage in event.response=", event.response["usage"]) + assert isinstance(event.response["usage"], ResponseAPIUsage) + elif event.type == "response.failed" or event.type == "response.incomplete": + assert hasattr( + event, "response" + ), f"{event.type} event should have a 'response' attribute" + + elif ( + event.type == "response.output_item.added" + or event.type == "response.output_item.done" + ): + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "item" + ), f"{event.type} event should have an 'item' attribute" + + elif ( + event.type == "response.content_part.added" + or event.type == "response.content_part.done" + ): + assert hasattr( + event, "item_id" + ), f"{event.type} event should have an 'item_id' attribute" + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "content_index" + ), f"{event.type} event should have a 'content_index' attribute" + assert hasattr( + event, "part" + ), f"{event.type} event should have a 'part' attribute" + + elif event.type == "response.output_text.delta": + assert hasattr( + event, "item_id" + ), f"{event.type} event should have an 'item_id' attribute" + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "content_index" + ), f"{event.type} event should have a 'content_index' attribute" + assert hasattr( + event, "delta" + ), f"{event.type} event should have a 'delta' attribute" + + elif event.type == "response.output_text.annotation.added": + assert hasattr( + event, "item_id" + ), f"{event.type} event should have an 'item_id' attribute" + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "content_index" + ), f"{event.type} event should have a 'content_index' attribute" + assert hasattr( + event, "annotation_index" + ), f"{event.type} event should have an 'annotation_index' attribute" + assert hasattr( + event, "annotation" + ), f"{event.type} event should have an 'annotation' attribute" + + elif event.type == "response.output_text.done": + assert hasattr( + event, "item_id" + ), f"{event.type} event should have an 'item_id' attribute" + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "content_index" + ), f"{event.type} event should have a 'content_index' attribute" + assert hasattr( + event, "text" + ), f"{event.type} event should have a 'text' attribute" + + elif event.type == "response.refusal.delta": + assert hasattr( + event, "item_id" + ), f"{event.type} event should have an 'item_id' attribute" + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "content_index" + ), f"{event.type} event should have a 'content_index' attribute" + assert hasattr( + event, "delta" + ), f"{event.type} event should have a 'delta' attribute" + + elif event.type == "response.refusal.done": + assert hasattr( + event, "item_id" + ), f"{event.type} event should have an 'item_id' attribute" + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "content_index" + ), f"{event.type} event should have a 'content_index' attribute" + assert hasattr( + event, "refusal" + ), f"{event.type} event should have a 'refusal' attribute" + + elif event.type == "response.function_call_arguments.delta": + assert hasattr( + event, "item_id" + ), f"{event.type} event should have an 'item_id' attribute" + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "delta" + ), f"{event.type} event should have a 'delta' attribute" + + elif event.type == "response.function_call_arguments.done": + assert hasattr( + event, "item_id" + ), f"{event.type} event should have an 'item_id' attribute" + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "arguments" + ), f"{event.type} event should have an 'arguments' attribute" + + elif event.type in [ + "response.file_search_call.in_progress", + "response.file_search_call.searching", + "response.file_search_call.completed", + "response.web_search_call.in_progress", + "response.web_search_call.searching", + "response.web_search_call.completed", + ]: + assert hasattr( + event, "output_index" + ), f"{event.type} event should have an 'output_index' attribute" + assert hasattr( + event, "item_id" + ), f"{event.type} event should have an 'item_id' attribute" + + elif event.type == "error": + assert hasattr( + event, "message" + ), "Error event should have a 'message' attribute" + return True # Return True if validation passes + + +class TestOpenAIResponsesAPITest(BaseResponsesAPITest): + def get_base_completion_call_args(self): + return { + "model": "openai/gpt-5.5", + } + + def get_base_completion_reasoning_call_args(self): + return { + "model": "openai/gpt-5-mini", + } + + def get_advanced_model_for_shell_tool(self): + return "openai/gpt-5.2" + + +@pytest.mark.asyncio +def test_basic_openai_responses_api_streaming_with_logging(): + litellm._turn_on_debug() + litellm.set_verbose = True + test_custom_logger = TestCustomLogger() + litellm.callbacks = [test_custom_logger] + request_model = "gpt-5.5" + response = litellm.responses( + model=request_model, + input="hi", + stream=True, + ) + final_response: Optional[ResponseCompletedEvent] = None + for event in response: + if event.type == "response.completed": + final_response = event + print("litellm response=", json.dumps(event, indent=4, default=str)) + + print("sleeping for 2 seconds...") + time.sleep(2) + print( + "standard logging payload=", + json.dumps(test_custom_logger.standard_logging_object, indent=4, default=str), + ) + + assert final_response is not None + assert test_custom_logger.standard_logging_object is not None + + validate_standard_logging_payload( + slp=test_custom_logger.standard_logging_object, + response=final_response.response, + request_model=request_model, + ) + + +@pytest.mark.asyncio +async def test_basic_openai_responses_api_non_streaming_with_logging(): + litellm._turn_on_debug() + litellm.set_verbose = True + test_custom_logger = TestCustomLogger() + litellm.callbacks = [test_custom_logger] + request_model = "gpt-5.5" + response = await litellm.aresponses( + model=request_model, + input="hi", + ) + + print("litellm response=", json.dumps(response, indent=4, default=str)) + print("response hidden params=", response._hidden_params) + + print("sleeping for 2 seconds...") + await asyncio.sleep(5) + print( + "standard logging payload=", + json.dumps(test_custom_logger.standard_logging_object, indent=4, default=str), + ) + print("response usage=", response.usage) + + assert response is not None + assert test_custom_logger.standard_logging_object is not None + + validate_standard_logging_payload( + test_custom_logger.standard_logging_object, response, request_model + ) + + +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_openai_responses_api_returns_headers(sync_mode): + """ + Test that OpenAI responses API returns OpenAI headers in _hidden_params. + This ensures the proxy can forward these headers to clients. + + Related issue: LiteLLM responses API should return OpenAI headers like chat completions does + """ + litellm._turn_on_debug() + litellm.set_verbose = True + + if sync_mode: + response = litellm.responses( + model="gpt-5.5", + input="Say hello", + max_output_tokens=20, + ) + else: + response = await litellm.aresponses( + model="gpt-5.5", + input="Say hello", + max_output_tokens=20, + ) + + # Verify response is valid + assert response is not None + assert isinstance(response, ResponsesAPIResponse) + + # Verify _hidden_params exists + assert hasattr( + response, "_hidden_params" + ), "Response should have _hidden_params attribute" + assert response._hidden_params is not None, "_hidden_params should not be None" + + # Verify additional_headers exists in _hidden_params + assert ( + "additional_headers" in response._hidden_params + ), "_hidden_params should contain 'additional_headers' key" + + additional_headers = response._hidden_params["additional_headers"] + assert isinstance( + additional_headers, dict + ), "additional_headers should be a dictionary" + assert len(additional_headers) > 0, "additional_headers should not be empty" + + # Check for expected OpenAI rate limit headers + # These can be either direct (x-ratelimit-*) or prefixed (llm_provider-x-ratelimit-*) + rate_limit_headers = [ + "x-ratelimit-remaining-tokens", + "x-ratelimit-limit-tokens", + "x-ratelimit-remaining-requests", + "x-ratelimit-limit-requests", + ] + + found_headers = [] + for header_name in rate_limit_headers: + if header_name in additional_headers: + found_headers.append(header_name) + elif f"llm_provider-{header_name}" in additional_headers: + found_headers.append(f"llm_provider-{header_name}") + + assert ( + len(found_headers) > 0 + ), f"Should find at least one OpenAI rate limit header. Headers found: {list(additional_headers.keys())}" + + # Verify headers key also exists (raw headers) + assert ( + "headers" in response._hidden_params + ), "_hidden_params should contain 'headers' key with raw response headers" + + print( + f"✓ Successfully validated OpenAI headers in {'sync' if sync_mode else 'async'} mode" + ) + print(f" Found {len(additional_headers)} headers total") + print(f" Rate limit headers found: {found_headers}") + + +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_openai_responses_api_streaming_validation(sync_mode): + """Test that validates each streaming event from the responses API""" + litellm._turn_on_debug() + + event_types_seen = set() + + if sync_mode: + response = litellm.responses( + model="gpt-5.5", + input="Tell me about artificial intelligence in 3 sentences.", + stream=True, + ) + for event in response: + print(f"Validating event type: {event.type}") + validate_stream_event(event) + event_types_seen.add(event.type) + else: + response = await litellm.aresponses( + model="gpt-5.5", + input="Tell me about artificial intelligence in 3 sentences.", + stream=True, + ) + async for event in response: + print(f"Validating event type: {event.type}") + validate_stream_event(event) + event_types_seen.add(event.type) + + # At minimum, we should see these core event types + required_events = {"response.created", "response.completed"} + + missing_events = required_events - event_types_seen + assert not missing_events, f"Missing required event types: {missing_events}" + + print(f"Successfully validated all event types: {event_types_seen}") + + +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_openai_responses_litellm_router(sync_mode): + """ + Test the OpenAI responses API with LiteLLM Router in both sync and async modes + """ + litellm._turn_on_debug() + router = litellm.Router( + model_list=[ + { + "model_name": "gpt4o-special-alias", + "litellm_params": { + "model": "gpt-5.5", + "api_key": os.getenv("OPENAI_API_KEY"), + }, + } + ] + ) + + # Call the handler + if sync_mode: + response = router.responses( + model="gpt4o-special-alias", + input="Hello, can you tell me a short joke?", + max_output_tokens=100, + ) + print("SYNC MODE RESPONSE=", response) + else: + response = await router.aresponses( + model="gpt4o-special-alias", + input="Hello, can you tell me a short joke?", + max_output_tokens=100, + ) + + print( + f"Router {'sync' if sync_mode else 'async'} response=", + json.dumps(response, indent=4, default=str), + ) + + # Use the helper function to validate the response + validate_responses_api_response(response, final_chunk=True) + + return response + + +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_openai_responses_litellm_router_streaming(sync_mode): + """ + Test the OpenAI responses API with streaming through LiteLLM Router + """ + litellm._turn_on_debug() + router = litellm.Router( + model_list=[ + { + "model_name": "gpt4o-special-alias", + "litellm_params": { + "model": "gpt-5.5", + "api_key": os.getenv("OPENAI_API_KEY"), + }, + } + ] + ) + + event_types_seen = set() + + if sync_mode: + response = router.responses( + model="gpt4o-special-alias", + input="Tell me about artificial intelligence in 2 sentences.", + stream=True, + ) + for event in response: + print(f"Validating event type: {event.type}") + validate_stream_event(event) + event_types_seen.add(event.type) + else: + response = await router.aresponses( + model="gpt4o-special-alias", + input="Tell me about artificial intelligence in 2 sentences.", + stream=True, + ) + async for event in response: + print(f"Validating event type: {event.type}") + validate_stream_event(event) + event_types_seen.add(event.type) + + # At minimum, we should see these core event types + required_events = {"response.created", "response.completed"} + + missing_events = required_events - event_types_seen + assert not missing_events, f"Missing required event types: {missing_events}" + + print(f"Successfully validated all event types: {event_types_seen}") + + +def test_bad_request_bad_param_error(): + """Raise a BadRequestError when an invalid parameter value is provided""" + try: + litellm.responses(model="gpt-5.5", input="This should fail", temperature=2000) + pytest.fail("Expected BadRequestError but no exception was raised") + except litellm.BadRequestError as e: + print(f"Exception raised: {e}") + print(f"Exception type: {type(e)}") + print(f"Exception args: {e.args}") + print(f"Exception details: {e.__dict__}") + except Exception as e: + pytest.fail(f"Unexpected exception raised: {e}") + + +@pytest.mark.asyncio() +async def test_async_bad_request_bad_param_error(): + """Raise a BadRequestError when an invalid parameter value is provided""" + try: + await litellm.aresponses( + model="gpt-5.5", input="This should fail", temperature=2000 + ) + pytest.fail("Expected BadRequestError but no exception was raised") + except litellm.BadRequestError as e: + print(f"Exception raised: {e}") + print(f"Exception type: {type(e)}") + print(f"Exception args: {e.args}") + print(f"Exception details: {e.__dict__}") + except Exception as e: + pytest.fail(f"Unexpected exception raised: {e}") + + +def test_mcp_tools_with_responses_api(): + litellm._turn_on_debug() + MCP_TOOLS = [ + { + "type": "mcp", + "server_label": "zapier", + "server_url": "https://mcp.zapier.com/api/mcp/mcp", + "headers": { + "Authorization": f"Bearer {os.getenv('ZAPIER_CI_CD_MCP_TOKEN')}" + }, + } + ] + MODEL = "openai/gpt-4.1" + USER_QUERY = "how does tiktoken work?" + ######################################################### + # Step 1: OpenAI will use MCP LIST, and return a list of MCP calls for our approval + try: + response = litellm.responses(model=MODEL, tools=MCP_TOOLS, input=USER_QUERY) + print(response) + + response = cast(ResponsesAPIResponse, response) + + mcp_approval_id: Optional[str] = None + for output in response.output: + if output.type == "mcp_approval_request": + mcp_approval_id = output.id + break + + # Step 2: Send followup with approval for the MCP call + if mcp_approval_id: + response_with_mcp_call = litellm.responses( + model=MODEL, + tools=MCP_TOOLS, + input=[ + { + "type": "mcp_approval_response", + "approve": True, + "approval_request_id": mcp_approval_id, + } + ], + previous_response_id=response.id, + ) + print(response_with_mcp_call) + except litellm.APIError as e: + if ( + "424" in str(e) + or "Failed Dependency" in str(e) + or "external_connector_error" in str(e) + ): + pytest.skip(f"Skipping test due to external MCP server error: {e}") + else: + raise e + except litellm.InternalServerError as e: + if "500" in str(e) or "server_error" in str(e): + pytest.skip( + f"Skipping test due to OpenAI server error (likely MCP server unavailable): {e}" + ) + else: + raise e + + +@pytest.mark.asyncio +async def test_openai_responses_api_field_types(): + """Test that specific fields in the response have the correct types""" + litellm._turn_on_debug() + litellm.set_verbose = True + + # Test with store=True + response = await litellm.aresponses( + model="gpt-5.5", + input="hi", + ) + + # Verify created_at is an integer + assert isinstance(response.created_at, int), "created_at should be an integer" + + # Verify store field is present and matches input + assert hasattr(response, "store"), "store field should be present" + assert response.store is True, "store field should match input value" + + # Test without store parameter + response_without_store = await litellm.aresponses(model="gpt-5.5", input="hi") + + # Verify created_at is still an integer + assert isinstance( + response_without_store.created_at, int + ), "created_at should be an integer" + + # Verify store field is present but None when not specified + assert hasattr(response_without_store, "store"), "store field should be present" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stream", [True, False]) +async def test_basic_openai_responses_with_websearch(stream): + litellm._turn_on_debug() + request_model = "gpt-5.5" + response = await litellm.aresponses( + model=request_model, + stream=stream, + input="hi", + tools=[{"type": "web_search", "search_context_size": "low"}], + ) + if stream: + async for chunk in response: + print("chunk=", json.dumps(chunk, indent=4, default=str)) + else: + print("response=", json.dumps(response, indent=4, default=str)) + + +@pytest.mark.asyncio +async def test_openai_responses_api_token_limit_error(): + """ + Relevant issue: https://github.com/BerriAI/litellm/issues/15785 + + + When this fails you'll see: + "pydantic_core._pydantic_core.ValidationError: 3 validation errors for ErrorEvent" + in the console. + """ + litellm._turn_on_debug() + + # Generate text with >400k tokens to trigger token limit error + oversized_text = "This is a test sentence. " * 50000 # ~400k tokens + + # This will raise ValidationError instead of showing the real error + response = await litellm.aresponses( + model="gpt-5-mini", input=oversized_text, stream=True + ) + + async for event in response: + print(event) # Never reaches here - ValidationError is raised + + +async def test_openai_streaming_logging(): + """Test that OpenAI Responses API streaming logging is working correctly.""" + litellm._turn_on_debug() + from litellm.integrations.custom_logger import CustomLogger + from litellm.types.utils import Usage + + class TestCustomLogger(CustomLogger): + validate_usage = False + + def __init__(self): + self.standard_logging_object: Optional[StandardLoggingPayload] = None + + async def async_log_success_event( + self, kwargs, response_obj, start_time, end_time + ): + print(f"response_obj: {response_obj.usage}") + assert isinstance( + response_obj.usage, (Usage, dict) + ), f"Expected response_obj.usage to be of type Usage or dict, but got {type(response_obj.usage)}" + # Verify it has the chat completion format fields + if isinstance(response_obj.usage, dict): + assert ( + "prompt_tokens" in response_obj.usage + ), "Usage dict should have prompt_tokens" + assert ( + "completion_tokens" in response_obj.usage + ), "Usage dict should have completion_tokens" + print("\n\nVALIDATED USAGE\n\n") + self.validate_usage = True + + tcl = TestCustomLogger() + litellm.callbacks = [tcl] + request_model = "gpt-5-mini" + response = await litellm.aresponses( + model=request_model, + input="What is the capital of France?", + stream=True, + ) + print("response=", json.dumps(response, indent=4, default=str)) + + async for event in response: + if event.type == "response.completed": + final_response = event + print("litellm response=", json.dumps(event, indent=4, default=str)) + + await asyncio.sleep(2) + assert tcl.validate_usage, "Usage should be validated" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sync_mode", [True, False]) +async def test_openai_compact_responses_api(sync_mode): + """ + Test the compact_responses API for OpenAI. + + This test verifies that the compact_responses endpoint works correctly + for compressing conversation history. + """ + litellm._turn_on_debug() + litellm.set_verbose = True + + input_messages = [ + {"role": "user", "content": "Hello, how are you?"}, + {"role": "assistant", "content": "I'm doing well, thank you for asking!"}, + {"role": "user", "content": "What is the weather like today?"}, + ] + + try: + if sync_mode: + response = litellm.compact_responses( + model="openai/gpt-5.5", + input=input_messages, + instructions="Be helpful and concise", + ) + else: + response = await litellm.acompact_responses( + model="openai/gpt-5.5", + input=input_messages, + instructions="Be helpful and concise", + ) + except litellm.InternalServerError: + pytest.skip("Skipping test due to InternalServerError") + except litellm.BadRequestError as e: + # compact_responses may not be available for all models/accounts + pytest.skip(f"Skipping test due to BadRequestError: {e}") + + print("compact_responses response=", json.dumps(response, indent=4, default=str)) + + # Validate response structure + assert response is not None + assert "id" in response, "Response should have an 'id' field" + assert "output" in response, "Response should have an 'output' field" + assert isinstance(response["output"], list), "Output should be a list" diff --git a/tests/llm_responses_api_testing/test_azure_responses_api.py b/tests/llm_responses_api_testing/test_azure_responses_api.py index fed9e9e11f0..20876057948 100644 --- a/tests/llm_responses_api_testing/test_azure_responses_api.py +++ b/tests/llm_responses_api_testing/test_azure_responses_api.py @@ -1,54 +1,12 @@ import os import sys import pytest -import asyncio -from typing import Optional -from unittest.mock import patch, AsyncMock +from unittest.mock import patch sys.path.insert(0, os.path.abspath("../..")) import litellm -from litellm.integrations.custom_logger import CustomLogger import json -from litellm.types.utils import StandardLoggingPayload -from litellm.types.llms.openai import ( - ResponseCompletedEvent, - ResponsesAPIResponse, - ResponseAPIUsage, - IncompleteDetails, -) from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler -from base_responses_api import BaseResponsesAPITest - - -class TestAzureResponsesAPITest(BaseResponsesAPITest): - def get_base_completion_call_args(self): - return { - "model": "azure/gpt-4.1-mini", - "truncation": "auto", - "api_base": os.getenv("AZURE_AI_API_BASE"), - "api_key": os.getenv("AZURE_AI_API_KEY"), - "api_version": "2025-03-01-preview", - } - - def get_advanced_model_for_shell_tool(self) -> Optional[str]: - """If specified, overrides the model used by test_responses_api_shell_tool_streaming_sees_shell_output (e.g. openai/gpt-5.2 for shell support).""" - return "azure/gpt-5-mini" - - -@pytest.mark.asyncio -async def test_azure_responses_api_preview_api_version(): - """ - Ensure new azure preview api version is working - """ - litellm._turn_on_debug() - response = await litellm.aresponses( - model="azure/gpt-5-mini", - truncation="auto", - api_version="preview", - api_base=os.getenv("AZURE_AI_API_BASE"), - api_key=os.getenv("AZURE_AI_API_KEY"), - input="Hello, can you tell me a short joke?", - ) @pytest.mark.asyncio @@ -57,7 +15,6 @@ async def test_azure_responses_api_status_error(): Test that 'status' field is not sent in the final request body to Azure API. The status field should be filtered out from input messages before making the API call. """ - from unittest.mock import AsyncMock, MagicMock import json request_data = { @@ -198,7 +155,6 @@ async def test_azure_responses_api_headers_with_llm_provider_prefix(): in response._hidden_params["headers"] instead of additional_headers, making them accessible via completion.headers in the same way as the completion API. """ - import json import httpx mock_response_data = { diff --git a/tests/llm_responses_api_testing/test_google_ai_studio_responses_api.py b/tests/llm_responses_api_testing/test_google_ai_studio_responses_api.py index 70c818f0a0a..601f1c10f5a 100644 --- a/tests/llm_responses_api_testing/test_google_ai_studio_responses_api.py +++ b/tests/llm_responses_api_testing/test_google_ai_studio_responses_api.py @@ -6,20 +6,6 @@ from unittest.mock import patch, AsyncMock sys.path.insert(0, os.path.abspath("../..")) import litellm import json -from base_responses_api import BaseResponsesAPITest - - -@pytest.mark.asyncio -async def test_basic_google_ai_studio_responses_api_with_tools(): - litellm._turn_on_debug() - litellm.set_verbose = True - request_model = "gemini/gemini-2.5-flash" - response = await litellm.aresponses( - model=request_model, - input="what is the latest version of supabase python package and when was it released?", - tools=[{"type": "web_search_preview", "search_context_size": "low"}], - ) - print("litellm response=", json.dumps(response, indent=4, default=str)) @pytest.mark.asyncio @@ -86,215 +72,3 @@ async def test_mock_basic_google_ai_studio_responses_api_with_tools(): ) # web search tools are converted to web_search_options, not kept as tools -@pytest.mark.asyncio -async def test_gemini_3_responses_api_with_thought_signatures(): - """ - Test that Gemini 3 Responses API preserves thought signatures in function calls. - This test verifies that provider_specific_fields with thought_signature are correctly - preserved when using the Responses API with Gemini 3. - """ - if not os.getenv("GEMINI_API_KEY"): - pytest.skip("GEMINI_API_KEY not set") - - litellm.set_verbose = False - request_model = "gemini/gemini-3.1-pro-preview" - - tools = [ - { - "type": "function", - "name": "get_weather", - "description": "Get the current weather for a location", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "City and country e.g. Mumbai, India", - }, - "units": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "Units the temperature will be returned in.", - }, - }, - "required": ["location", "units"], - "additionalProperties": False, - }, - "strict": True, - } - ] - - # Step 1: Initial request with tools - response = await litellm.aresponses( - model=request_model, - input="What is the weather in Mumbai?", - tools=tools, - reasoning_effort="low", - ) - - # Validate response structure - from litellm.types.llms.openai import ResponsesAPIResponse - - assert isinstance( - response, ResponsesAPIResponse - ), "Response should be a ResponsesAPIResponse" - assert ( - hasattr(response, "output") or "output" in response - ), "Response should have 'output' field" - assert isinstance(response.output, list), "Output should be a list" - - # Find function call in output - function_call_item = None - for item in response.output: - # Convert to dict if it's a Pydantic model for easier access - if hasattr(item, "model_dump"): - item_dict = item.model_dump() - elif hasattr(item, "__dict__"): - item_dict = dict(item) if not isinstance(item, dict) else item - else: - item_dict = item if isinstance(item, dict) else {} - - if isinstance(item_dict, dict) and item_dict.get("type") == "function_call": - function_call_item = item_dict - break - - # Verify function call exists - assert ( - function_call_item is not None - ), "Response should contain a function_call item" - assert ( - function_call_item.get("name") == "get_weather" - ), "Function call should be for get_weather" - - # Verify thought signature is present in provider_specific_fields - provider_specific_fields = function_call_item.get("provider_specific_fields") - assert ( - provider_specific_fields is not None - ), "Function call should have provider_specific_fields" - assert ( - "thought_signature" in provider_specific_fields - ), "provider_specific_fields should contain thought_signature" - assert isinstance( - provider_specific_fields["thought_signature"], str - ), "thought_signature should be a string" - assert ( - len(provider_specific_fields["thought_signature"]) > 0 - ), "thought_signature should not be empty" - - print( - f"✅ Thought signature preserved: {provider_specific_fields['thought_signature'][:50]}..." - ) - - -@pytest.mark.asyncio -async def test_gemini_3_responses_api_streaming_with_thought_signatures(): - """ - Test that Gemini 3 Responses API preserves thought signatures in streaming mode. - This test verifies that provider_specific_fields with thought_signature are correctly - preserved when using streaming Responses API with Gemini 3. - """ - if not os.getenv("GEMINI_API_KEY"): - pytest.skip("GEMINI_API_KEY not set") - - litellm.set_verbose = False - request_model = "gemini/gemini-3.1-pro-preview" - - tools = [ - { - "type": "function", - "name": "get_weather", - "description": "Get the current weather for a location", - "parameters": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "City and country e.g. Mumbai, India", - }, - "units": { - "type": "string", - "enum": ["celsius", "fahrenheit"], - "description": "Units the temperature will be returned in.", - }, - }, - "required": ["location", "units"], - "additionalProperties": False, - }, - "strict": True, - } - ] - - # Step 1: Streaming request with tools - response_stream = await litellm.aresponses( - model=request_model, - input="What is the weather in Mumbai?", - tools=tools, - stream=True, - reasoning_effort="low", - ) - - # Collect all chunks - chunks = [] - completed_response = None - - async for chunk in response_stream: - chunks.append(chunk) - # Check if this is the completed response event - if hasattr(chunk, "type") and chunk.type == "response.completed": - completed_response = chunk.response - elif isinstance(chunk, dict) and chunk.get("type") == "response.completed": - completed_response = chunk.get("response") - - # Verify we got chunks - assert len(chunks) > 0, "Should receive at least one chunk" - - # If we have a completed response, check for thought signatures - if completed_response: - output = completed_response.get("output", []) - function_call_item = None - for item in output: - if isinstance(item, dict) and item.get("type") == "function_call": - function_call_item = item - break - - if function_call_item: - provider_specific_fields = function_call_item.get( - "provider_specific_fields" - ) - if provider_specific_fields: - thought_signature = provider_specific_fields.get("thought_signature") - if thought_signature: - assert isinstance( - thought_signature, str - ), "thought_signature should be a string" - assert ( - len(thought_signature) > 0 - ), "thought_signature should not be empty" - print( - f"✅ Streaming thought signature preserved: {thought_signature[:50]}..." - ) - - print(f"✅ Collected {len(chunks)} streaming chunks") - - -class TestGoogleAIStudioResponsesAPITest(BaseResponsesAPITest): - def get_base_completion_call_args(self): - # litellm._turn_on_debug() - return {"model": "gemini/gemini-2.5-flash-lite"} - - async def test_basic_openai_responses_delete_endpoint(self, sync_mode=False): - pytest.skip("DELETE responses is not supported for Google AI Studio") - - async def test_basic_openai_responses_streaming_delete_endpoint( - self, sync_mode=False - ): - pytest.skip("DELETE responses is not supported for Google AI Studio") - - async def test_basic_openai_responses_get_endpoint(self, sync_mode=False): - pytest.skip("GET responses is not supported for Google AI Studio") - - async def test_basic_openai_responses_cancel_endpoint(self, sync_mode=False): - pytest.skip("CANCEL responses is not supported for Google AI Studio") - - async def test_cancel_responses_invalid_response_id(self, sync_mode=False): - pytest.skip("CANCEL responses is not supported for Google AI Studio") diff --git a/tests/llm_responses_api_testing/test_openai_responses_api.py b/tests/llm_responses_api_testing/test_openai_responses_api.py index f4a940722be..985aaba5145 100644 --- a/tests/llm_responses_api_testing/test_openai_responses_api.py +++ b/tests/llm_responses_api_testing/test_openai_responses_api.py @@ -1,8 +1,6 @@ import os import sys import pytest -import asyncio -from typing import Optional, cast from unittest.mock import patch, AsyncMock import httpx from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig @@ -12,584 +10,7 @@ import json sys.path.insert(0, os.path.abspath("../..")) import litellm -from litellm.integrations.custom_logger import CustomLogger -from litellm.types.utils import StandardLoggingPayload -from litellm.types.llms.openai import ( - ResponseCompletedEvent, - ResponsesAPIResponse, - ResponseAPIUsage, -) -from base_responses_api import BaseResponsesAPITest, validate_responses_api_response - - -class TestOpenAIResponsesAPITest(BaseResponsesAPITest): - def get_base_completion_call_args(self): - return { - "model": "openai/gpt-5.5", - } - - def get_base_completion_reasoning_call_args(self): - return { - "model": "openai/gpt-5-mini", - } - - def get_advanced_model_for_shell_tool(self): - return "openai/gpt-5.2" - - -class TestCustomLogger(CustomLogger): - def __init__( - self, - ): - self.standard_logging_object: Optional[StandardLoggingPayload] = None - - async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): - print("in async_log_success_event") - print("kwargs=", json.dumps(kwargs, indent=4, default=str)) - self.standard_logging_object = kwargs["standard_logging_object"] - pass - - -def validate_standard_logging_payload( - slp: StandardLoggingPayload, response: ResponsesAPIResponse, request_model: str -): - """ - Validate that a StandardLoggingPayload object matches the expected response - - Args: - slp (StandardLoggingPayload): The standard logging payload object to validate - response (dict): The litellm response to compare against - request_model (str): The model name that was requested - """ - # Validate payload exists - assert slp is not None, "Standard logging payload should not be None" - - # Validate token counts - print( - "VALIDATING STANDARD LOGGING PAYLOAD. response=", - json.dumps(response, indent=4, default=str), - ) - print("FIELDS IN SLP=", json.dumps(slp, indent=4, default=str)) - print("SLP PROMPT TOKENS=", slp["prompt_tokens"]) - print("RESPONSE PROMPT TOKENS=", response["usage"]["input_tokens"]) - assert ( - slp["prompt_tokens"] == response["usage"]["input_tokens"] - ), "Prompt tokens mismatch" - assert ( - slp["completion_tokens"] == response["usage"]["output_tokens"] - ), "Completion tokens mismatch" - assert ( - slp["total_tokens"] - == response["usage"]["input_tokens"] + response["usage"]["output_tokens"] - ), "Total tokens mismatch" - - # Validate spend and response metadata - assert slp["response_cost"] > 0, "Response cost should be greater than 0" - assert slp["id"] == response["id"], "Response ID mismatch" - assert slp["model"] == request_model, "Model name mismatch" - - # Validate messages - assert slp["messages"] == [{"content": "hi", "role": "user"}], "Messages mismatch" - - # Validate complete response structure - validate_responses_match(slp["response"], response) - - -@pytest.mark.asyncio -def test_basic_openai_responses_api_streaming_with_logging(): - litellm._turn_on_debug() - litellm.set_verbose = True - test_custom_logger = TestCustomLogger() - litellm.callbacks = [test_custom_logger] - request_model = "gpt-5.5" - response = litellm.responses( - model=request_model, - input="hi", - stream=True, - ) - final_response: Optional[ResponseCompletedEvent] = None - for event in response: - if event.type == "response.completed": - final_response = event - print("litellm response=", json.dumps(event, indent=4, default=str)) - - print("sleeping for 2 seconds...") - time.sleep(2) - print( - "standard logging payload=", - json.dumps(test_custom_logger.standard_logging_object, indent=4, default=str), - ) - - assert final_response is not None - assert test_custom_logger.standard_logging_object is not None - - validate_standard_logging_payload( - slp=test_custom_logger.standard_logging_object, - response=final_response.response, - request_model=request_model, - ) - - -def validate_responses_match(slp_response, litellm_response): - """Validate that the standard logging payload OpenAI response matches the litellm response""" - # Validate core fields - assert slp_response["id"] == litellm_response["id"], "ID mismatch" - assert slp_response["model"] == litellm_response["model"], "Model mismatch" - assert ( - slp_response["created_at"] == litellm_response["created_at"] - ), "Created at mismatch" - - # Validate usage - assert ( - slp_response["usage"]["prompt_tokens"] - == litellm_response["usage"]["input_tokens"] - ), "Input tokens mismatch" - assert ( - slp_response["usage"]["completion_tokens"] - == litellm_response["usage"]["output_tokens"] - ), "Output tokens mismatch" - assert ( - slp_response["usage"]["total_tokens"] - == litellm_response["usage"]["total_tokens"] - ), "Total tokens mismatch" - - # Validate output/messages - assert len(slp_response["output"]) == len( - litellm_response["output"] - ), "Output length mismatch" - for slp_msg, litellm_msg in zip(slp_response["output"], litellm_response["output"]): - assert slp_msg["role"] == litellm_msg.role, "Message role mismatch" - # Access the content's text field for the litellm response - litellm_content = litellm_msg.content[0].text if litellm_msg.content else "" - assert ( - slp_msg["content"][0]["text"] == litellm_content - ), f"Message content mismatch. Expected {litellm_content}, Got {slp_msg['content']}" - assert slp_msg["status"] == litellm_msg.status, "Message status mismatch" - - -@pytest.mark.asyncio -async def test_basic_openai_responses_api_non_streaming_with_logging(): - litellm._turn_on_debug() - litellm.set_verbose = True - test_custom_logger = TestCustomLogger() - litellm.callbacks = [test_custom_logger] - request_model = "gpt-5.5" - response = await litellm.aresponses( - model=request_model, - input="hi", - ) - - print("litellm response=", json.dumps(response, indent=4, default=str)) - print("response hidden params=", response._hidden_params) - - print("sleeping for 2 seconds...") - await asyncio.sleep(5) - print( - "standard logging payload=", - json.dumps(test_custom_logger.standard_logging_object, indent=4, default=str), - ) - print("response usage=", response.usage) - - assert response is not None - assert test_custom_logger.standard_logging_object is not None - - validate_standard_logging_payload( - test_custom_logger.standard_logging_object, response, request_model - ) - - -@pytest.mark.parametrize("sync_mode", [True, False]) -@pytest.mark.asyncio -async def test_openai_responses_api_returns_headers(sync_mode): - """ - Test that OpenAI responses API returns OpenAI headers in _hidden_params. - This ensures the proxy can forward these headers to clients. - - Related issue: LiteLLM responses API should return OpenAI headers like chat completions does - """ - litellm._turn_on_debug() - litellm.set_verbose = True - - if sync_mode: - response = litellm.responses( - model="gpt-5.5", - input="Say hello", - max_output_tokens=20, - ) - else: - response = await litellm.aresponses( - model="gpt-5.5", - input="Say hello", - max_output_tokens=20, - ) - - # Verify response is valid - assert response is not None - assert isinstance(response, ResponsesAPIResponse) - - # Verify _hidden_params exists - assert hasattr( - response, "_hidden_params" - ), "Response should have _hidden_params attribute" - assert response._hidden_params is not None, "_hidden_params should not be None" - - # Verify additional_headers exists in _hidden_params - assert ( - "additional_headers" in response._hidden_params - ), "_hidden_params should contain 'additional_headers' key" - - additional_headers = response._hidden_params["additional_headers"] - assert isinstance( - additional_headers, dict - ), "additional_headers should be a dictionary" - assert len(additional_headers) > 0, "additional_headers should not be empty" - - # Check for expected OpenAI rate limit headers - # These can be either direct (x-ratelimit-*) or prefixed (llm_provider-x-ratelimit-*) - rate_limit_headers = [ - "x-ratelimit-remaining-tokens", - "x-ratelimit-limit-tokens", - "x-ratelimit-remaining-requests", - "x-ratelimit-limit-requests", - ] - - found_headers = [] - for header_name in rate_limit_headers: - if header_name in additional_headers: - found_headers.append(header_name) - elif f"llm_provider-{header_name}" in additional_headers: - found_headers.append(f"llm_provider-{header_name}") - - assert ( - len(found_headers) > 0 - ), f"Should find at least one OpenAI rate limit header. Headers found: {list(additional_headers.keys())}" - - # Verify headers key also exists (raw headers) - assert ( - "headers" in response._hidden_params - ), "_hidden_params should contain 'headers' key with raw response headers" - - print( - f"✓ Successfully validated OpenAI headers in {'sync' if sync_mode else 'async'} mode" - ) - print(f" Found {len(additional_headers)} headers total") - print(f" Rate limit headers found: {found_headers}") - - -def validate_stream_event(event): - """ - Validate that a streaming event from litellm.responses() or litellm.aresponses() - with stream=True conforms to the expected structure based on its event type. - - Args: - event: The streaming event object to validate - - Raises: - AssertionError: If the event doesn't match the expected structure for its type - """ - # Common validation for all event types - assert hasattr(event, "type"), "Event should have a 'type' attribute" - - # Type-specific validation - if event.type == "response.created" or event.type == "response.in_progress": - assert hasattr( - event, "response" - ), f"{event.type} event should have a 'response' attribute" - validate_responses_api_response(event.response, final_chunk=False) - - elif event.type == "response.completed": - assert hasattr( - event, "response" - ), "response.completed event should have a 'response' attribute" - validate_responses_api_response(event.response, final_chunk=True) - # Usage is guaranteed only on the completed event - assert ( - "usage" in event.response - ), "response.completed event should have usage information" - print("Usage in event.response=", event.response["usage"]) - assert isinstance(event.response["usage"], ResponseAPIUsage) - elif event.type == "response.failed" or event.type == "response.incomplete": - assert hasattr( - event, "response" - ), f"{event.type} event should have a 'response' attribute" - - elif ( - event.type == "response.output_item.added" - or event.type == "response.output_item.done" - ): - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "item" - ), f"{event.type} event should have an 'item' attribute" - - elif ( - event.type == "response.content_part.added" - or event.type == "response.content_part.done" - ): - assert hasattr( - event, "item_id" - ), f"{event.type} event should have an 'item_id' attribute" - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "content_index" - ), f"{event.type} event should have a 'content_index' attribute" - assert hasattr( - event, "part" - ), f"{event.type} event should have a 'part' attribute" - - elif event.type == "response.output_text.delta": - assert hasattr( - event, "item_id" - ), f"{event.type} event should have an 'item_id' attribute" - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "content_index" - ), f"{event.type} event should have a 'content_index' attribute" - assert hasattr( - event, "delta" - ), f"{event.type} event should have a 'delta' attribute" - - elif event.type == "response.output_text.annotation.added": - assert hasattr( - event, "item_id" - ), f"{event.type} event should have an 'item_id' attribute" - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "content_index" - ), f"{event.type} event should have a 'content_index' attribute" - assert hasattr( - event, "annotation_index" - ), f"{event.type} event should have an 'annotation_index' attribute" - assert hasattr( - event, "annotation" - ), f"{event.type} event should have an 'annotation' attribute" - - elif event.type == "response.output_text.done": - assert hasattr( - event, "item_id" - ), f"{event.type} event should have an 'item_id' attribute" - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "content_index" - ), f"{event.type} event should have a 'content_index' attribute" - assert hasattr( - event, "text" - ), f"{event.type} event should have a 'text' attribute" - - elif event.type == "response.refusal.delta": - assert hasattr( - event, "item_id" - ), f"{event.type} event should have an 'item_id' attribute" - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "content_index" - ), f"{event.type} event should have a 'content_index' attribute" - assert hasattr( - event, "delta" - ), f"{event.type} event should have a 'delta' attribute" - - elif event.type == "response.refusal.done": - assert hasattr( - event, "item_id" - ), f"{event.type} event should have an 'item_id' attribute" - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "content_index" - ), f"{event.type} event should have a 'content_index' attribute" - assert hasattr( - event, "refusal" - ), f"{event.type} event should have a 'refusal' attribute" - - elif event.type == "response.function_call_arguments.delta": - assert hasattr( - event, "item_id" - ), f"{event.type} event should have an 'item_id' attribute" - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "delta" - ), f"{event.type} event should have a 'delta' attribute" - - elif event.type == "response.function_call_arguments.done": - assert hasattr( - event, "item_id" - ), f"{event.type} event should have an 'item_id' attribute" - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "arguments" - ), f"{event.type} event should have an 'arguments' attribute" - - elif event.type in [ - "response.file_search_call.in_progress", - "response.file_search_call.searching", - "response.file_search_call.completed", - "response.web_search_call.in_progress", - "response.web_search_call.searching", - "response.web_search_call.completed", - ]: - assert hasattr( - event, "output_index" - ), f"{event.type} event should have an 'output_index' attribute" - assert hasattr( - event, "item_id" - ), f"{event.type} event should have an 'item_id' attribute" - - elif event.type == "error": - assert hasattr( - event, "message" - ), "Error event should have a 'message' attribute" - return True # Return True if validation passes - - -@pytest.mark.parametrize("sync_mode", [True, False]) -@pytest.mark.asyncio -async def test_openai_responses_api_streaming_validation(sync_mode): - """Test that validates each streaming event from the responses API""" - litellm._turn_on_debug() - - event_types_seen = set() - - if sync_mode: - response = litellm.responses( - model="gpt-5.5", - input="Tell me about artificial intelligence in 3 sentences.", - stream=True, - ) - for event in response: - print(f"Validating event type: {event.type}") - validate_stream_event(event) - event_types_seen.add(event.type) - else: - response = await litellm.aresponses( - model="gpt-5.5", - input="Tell me about artificial intelligence in 3 sentences.", - stream=True, - ) - async for event in response: - print(f"Validating event type: {event.type}") - validate_stream_event(event) - event_types_seen.add(event.type) - - # At minimum, we should see these core event types - required_events = {"response.created", "response.completed"} - - missing_events = required_events - event_types_seen - assert not missing_events, f"Missing required event types: {missing_events}" - - print(f"Successfully validated all event types: {event_types_seen}") - - -@pytest.mark.parametrize("sync_mode", [True, False]) -@pytest.mark.asyncio -async def test_openai_responses_litellm_router(sync_mode): - """ - Test the OpenAI responses API with LiteLLM Router in both sync and async modes - """ - litellm._turn_on_debug() - router = litellm.Router( - model_list=[ - { - "model_name": "gpt4o-special-alias", - "litellm_params": { - "model": "gpt-5.5", - "api_key": os.getenv("OPENAI_API_KEY"), - }, - } - ] - ) - - # Call the handler - if sync_mode: - response = router.responses( - model="gpt4o-special-alias", - input="Hello, can you tell me a short joke?", - max_output_tokens=100, - ) - print("SYNC MODE RESPONSE=", response) - else: - response = await router.aresponses( - model="gpt4o-special-alias", - input="Hello, can you tell me a short joke?", - max_output_tokens=100, - ) - - print( - f"Router {'sync' if sync_mode else 'async'} response=", - json.dumps(response, indent=4, default=str), - ) - - # Use the helper function to validate the response - validate_responses_api_response(response, final_chunk=True) - - return response - - -@pytest.mark.parametrize("sync_mode", [True, False]) -@pytest.mark.asyncio -async def test_openai_responses_litellm_router_streaming(sync_mode): - """ - Test the OpenAI responses API with streaming through LiteLLM Router - """ - litellm._turn_on_debug() - router = litellm.Router( - model_list=[ - { - "model_name": "gpt4o-special-alias", - "litellm_params": { - "model": "gpt-5.5", - "api_key": os.getenv("OPENAI_API_KEY"), - }, - } - ] - ) - - event_types_seen = set() - - if sync_mode: - response = router.responses( - model="gpt4o-special-alias", - input="Tell me about artificial intelligence in 2 sentences.", - stream=True, - ) - for event in response: - print(f"Validating event type: {event.type}") - validate_stream_event(event) - event_types_seen.add(event.type) - else: - response = await router.aresponses( - model="gpt4o-special-alias", - input="Tell me about artificial intelligence in 2 sentences.", - stream=True, - ) - async for event in response: - print(f"Validating event type: {event.type}") - validate_stream_event(event) - event_types_seen.add(event.type) - - # At minimum, we should see these core event types - required_events = {"response.created", "response.completed"} - - missing_events = required_events - event_types_seen - assert not missing_events, f"Missing required event types: {missing_events}" - - print(f"Successfully validated all event types: {event_types_seen}") +from base_responses_api import validate_responses_api_response @pytest.mark.asyncio @@ -685,37 +106,6 @@ async def test_openai_responses_litellm_router_no_metadata(): mock_post.assert_called_once() -def test_bad_request_bad_param_error(): - """Raise a BadRequestError when an invalid parameter value is provided""" - try: - litellm.responses(model="gpt-5.5", input="This should fail", temperature=2000) - pytest.fail("Expected BadRequestError but no exception was raised") - except litellm.BadRequestError as e: - print(f"Exception raised: {e}") - print(f"Exception type: {type(e)}") - print(f"Exception args: {e.args}") - print(f"Exception details: {e.__dict__}") - except Exception as e: - pytest.fail(f"Unexpected exception raised: {e}") - - -@pytest.mark.asyncio() -async def test_async_bad_request_bad_param_error(): - """Raise a BadRequestError when an invalid parameter value is provided""" - try: - await litellm.aresponses( - model="gpt-5.5", input="This should fail", temperature=2000 - ) - pytest.fail("Expected BadRequestError but no exception was raised") - except litellm.BadRequestError as e: - print(f"Exception raised: {e}") - print(f"Exception type: {type(e)}") - print(f"Exception args: {e.args}") - print(f"Exception details: {e.__dict__}") - except Exception as e: - pytest.fail(f"Unexpected exception raised: {e}") - - @pytest.mark.asyncio @pytest.mark.parametrize("sync_mode", [True, False]) async def test_openai_o1_pro_response_api(sync_mode): @@ -934,98 +324,6 @@ async def test_openai_o1_pro_response_api_streaming(sync_mode): assert "stream" not in request_body -def test_mcp_tools_with_responses_api(): - litellm._turn_on_debug() - MCP_TOOLS = [ - { - "type": "mcp", - "server_label": "zapier", - "server_url": "https://mcp.zapier.com/api/mcp/mcp", - "headers": { - "Authorization": f"Bearer {os.getenv('ZAPIER_CI_CD_MCP_TOKEN')}" - }, - } - ] - MODEL = "openai/gpt-4.1" - USER_QUERY = "how does tiktoken work?" - ######################################################### - # Step 1: OpenAI will use MCP LIST, and return a list of MCP calls for our approval - try: - response = litellm.responses(model=MODEL, tools=MCP_TOOLS, input=USER_QUERY) - print(response) - - response = cast(ResponsesAPIResponse, response) - - mcp_approval_id: Optional[str] = None - for output in response.output: - if output.type == "mcp_approval_request": - mcp_approval_id = output.id - break - - # Step 2: Send followup with approval for the MCP call - if mcp_approval_id: - response_with_mcp_call = litellm.responses( - model=MODEL, - tools=MCP_TOOLS, - input=[ - { - "type": "mcp_approval_response", - "approve": True, - "approval_request_id": mcp_approval_id, - } - ], - previous_response_id=response.id, - ) - print(response_with_mcp_call) - except litellm.APIError as e: - if ( - "424" in str(e) - or "Failed Dependency" in str(e) - or "external_connector_error" in str(e) - ): - pytest.skip(f"Skipping test due to external MCP server error: {e}") - else: - raise e - except litellm.InternalServerError as e: - if "500" in str(e) or "server_error" in str(e): - pytest.skip( - f"Skipping test due to OpenAI server error (likely MCP server unavailable): {e}" - ) - else: - raise e - - -@pytest.mark.asyncio -async def test_openai_responses_api_field_types(): - """Test that specific fields in the response have the correct types""" - litellm._turn_on_debug() - litellm.set_verbose = True - - # Test with store=True - response = await litellm.aresponses( - model="gpt-5.5", - input="hi", - ) - - # Verify created_at is an integer - assert isinstance(response.created_at, int), "created_at should be an integer" - - # Verify store field is present and matches input - assert hasattr(response, "store"), "store field should be present" - assert response.store is True, "store field should match input value" - - # Test without store parameter - response_without_store = await litellm.aresponses(model="gpt-5.5", input="hi") - - # Verify created_at is still an integer - assert isinstance( - response_without_store.created_at, int - ), "created_at should be an integer" - - # Verify store field is present but None when not specified - assert hasattr(response_without_store, "store"), "store field should be present" - - @pytest.mark.asyncio async def test_store_field_transformation(): """Test store field transformation with mocked API responses""" @@ -1141,179 +439,3 @@ async def test_store_field_transformation(): ), "created_at should maintain the same value after conversion" -@pytest.mark.asyncio -@pytest.mark.parametrize("stream", [True, False]) -async def test_basic_openai_responses_with_websearch(stream): - litellm._turn_on_debug() - request_model = "gpt-5.5" - response = await litellm.aresponses( - model=request_model, - stream=stream, - input="hi", - tools=[{"type": "web_search", "search_context_size": "low"}], - ) - if stream: - async for chunk in response: - print("chunk=", json.dumps(chunk, indent=4, default=str)) - else: - print("response=", json.dumps(response, indent=4, default=str)) - - -@pytest.mark.asyncio -async def test_openai_responses_api_token_limit_error(): - """ - Relevant issue: https://github.com/BerriAI/litellm/issues/15785 - - - When this fails you'll see: - "pydantic_core._pydantic_core.ValidationError: 3 validation errors for ErrorEvent" - in the console. - """ - litellm._turn_on_debug() - - # Generate text with >400k tokens to trigger token limit error - oversized_text = "This is a test sentence. " * 50000 # ~400k tokens - - # This will raise ValidationError instead of showing the real error - response = await litellm.aresponses( - model="gpt-5-mini", input=oversized_text, stream=True - ) - - async for event in response: - print(event) # Never reaches here - ValidationError is raised - - -async def test_openai_streaming_logging(): - """Test that OpenAI Responses API streaming logging is working correctly.""" - litellm._turn_on_debug() - from litellm.integrations.custom_logger import CustomLogger - from litellm.types.utils import Usage - - class TestCustomLogger(CustomLogger): - validate_usage = False - - def __init__(self): - self.standard_logging_object: Optional[StandardLoggingPayload] = None - - async def async_log_success_event( - self, kwargs, response_obj, start_time, end_time - ): - print(f"response_obj: {response_obj.usage}") - assert isinstance( - response_obj.usage, (Usage, dict) - ), f"Expected response_obj.usage to be of type Usage or dict, but got {type(response_obj.usage)}" - # Verify it has the chat completion format fields - if isinstance(response_obj.usage, dict): - assert ( - "prompt_tokens" in response_obj.usage - ), "Usage dict should have prompt_tokens" - assert ( - "completion_tokens" in response_obj.usage - ), "Usage dict should have completion_tokens" - print("\n\nVALIDATED USAGE\n\n") - self.validate_usage = True - - tcl = TestCustomLogger() - litellm.callbacks = [tcl] - request_model = "gpt-5-mini" - response = await litellm.aresponses( - model=request_model, - input="What is the capital of France?", - stream=True, - ) - print("response=", json.dumps(response, indent=4, default=str)) - - async for event in response: - if event.type == "response.completed": - final_response = event - print("litellm response=", json.dumps(event, indent=4, default=str)) - - await asyncio.sleep(2) - assert tcl.validate_usage, "Usage should be validated" - - -# Tests for extra_body parameter passing -class MockResponse: - def __init__(self, json_data, status_code): - self._json_data = json_data - self.status_code = status_code - self.text = str(json_data) - self.headers = httpx.Headers({}) - - def json(self): - return self._json_data - - -@pytest.fixture -def extra_body_mock_response_data(): - return { - "id": "resp_test123", - "object": "response", - "created_at": 1234567890, - "status": "completed", - "model": "gpt-5.5", - "output": [ - { - "type": "message", - "id": "msg_123", - "status": "completed", - "role": "assistant", - "content": [ - {"type": "output_text", "text": "Hello!", "annotations": []} - ], - } - ], - "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}, - "parallel_tool_calls": True, - "text": {"format": {"type": "text"}}, - "error": None, - "metadata": {}, - "temperature": 1.0, - "reasoning": {"effort": None, "summary": None}, - } - - -@pytest.mark.asyncio -@pytest.mark.parametrize("sync_mode", [True, False]) -async def test_openai_compact_responses_api(sync_mode): - """ - Test the compact_responses API for OpenAI. - - This test verifies that the compact_responses endpoint works correctly - for compressing conversation history. - """ - litellm._turn_on_debug() - litellm.set_verbose = True - - input_messages = [ - {"role": "user", "content": "Hello, how are you?"}, - {"role": "assistant", "content": "I'm doing well, thank you for asking!"}, - {"role": "user", "content": "What is the weather like today?"}, - ] - - try: - if sync_mode: - response = litellm.compact_responses( - model="openai/gpt-5.5", - input=input_messages, - instructions="Be helpful and concise", - ) - else: - response = await litellm.acompact_responses( - model="openai/gpt-5.5", - input=input_messages, - instructions="Be helpful and concise", - ) - except litellm.InternalServerError: - pytest.skip("Skipping test due to InternalServerError") - except litellm.BadRequestError as e: - # compact_responses may not be available for all models/accounts - pytest.skip(f"Skipping test due to BadRequestError: {e}") - - print("compact_responses response=", json.dumps(response, indent=4, default=str)) - - # Validate response structure - assert response is not None - assert "id" in response, "Response should have an 'id' field" - assert "output" in response, "Response should have an 'output' field" - assert isinstance(response["output"], list), "Output should be a list"