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

test(harness): move live Responses API tests to responses_api_live suite

Browse source 
Per audit 5/8c (the scope-doc keeper, migrating as harness cells):
- test_openai_responses_api.py: TestOpenAIResponsesAPITest + the 15 live
  tests (streaming/non-streaming logging, headers, stream validation, router
  + router streaming, bad-request pair, MCP tools, field types, websearch,
  token-limit error, streaming logging, compact) moved with the
  TestCustomLogger/validate_* helpers they alone used; the o1-pro mock
  goldens, store-field transformation and router-no-metadata keepers stay;
  the module-level MockResponse/extra_body fixture pair died with the
  already-deleted extra-body drops
- test_anthropic_responses_api.py: TestAnthropicResponsesAPITest +
  test_multiturn_tool_calls moved; file deleted (nothing remained after the
  drop commit)
- test_azure_responses_api.py: TestAzureResponsesAPITest + preview
  api-version test moved; status-stripping and header-prefix goldens stay
- test_google_ai_studio_responses_api.py: live tools + thought-signature
  tests + base subclass moved; the bridge-transform mock golden stays
This commit is contained in:
mateo-berri 2026-06-11 19:15:00 +00:00
parent d5d69b1569
commit 34d8af9340
7 changed files with 1229 additions and 1150 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

108
tests/harness_suites/responses_api_live/test_anthropic_responses_live.py Normal file
View file

@ -0,0 +1,108 @@
import os
import sys
import pytest
sys.path.insert(0, os.path.abspath("../.."))
import litellm
from base_responses_api import BaseResponsesAPITest
from openai.types.responses.function_tool import FunctionTool
class TestAnthropicResponsesAPITest(BaseResponsesAPITest):
def get_base_completion_call_args(self):
# litellm._turn_on_debug()
return {
"model": "anthropic/claude-sonnet-4-5",
}
async def test_basic_openai_responses_delete_endpoint(self, sync_mode=False):
pytest.skip("DELETE responses is not supported for anthropic")
async def test_basic_openai_responses_streaming_delete_endpoint(
self, sync_mode=False
):
pytest.skip("DELETE responses is not supported for anthropic")
async def test_basic_openai_responses_get_endpoint(self, sync_mode=False):
pytest.skip("GET responses is not supported for anthropic")
async def test_basic_openai_responses_cancel_endpoint(self, sync_mode=False):
pytest.skip("CANCEL responses is not supported for anthropic")
async def test_cancel_responses_invalid_response_id(self, sync_mode=False):
pytest.skip("CANCEL responses is not supported for anthropic")
def test_multiturn_tool_calls():
# Test streaming response with tools for Anthropic
litellm._turn_on_debug()
shell_tool = dict(
FunctionTool(
type="function",
name="shell",
description="Runs a shell command, and returns its output.",
parameters={
"type": "object",
"properties": {
"command": {"type": "array", "items": {"type": "string"}},
"workdir": {
"type": "string",
"description": "The working directory for the command.",
},
},
"required": ["command"],
},
strict=True,
)
)
# Step 1: Initial request with the tool
response = litellm.responses(
input=[
{
"role": "user",
"content": [
{"type": "input_text", "text": "make a hello world html file"}
],
"type": "message",
}
],
model="anthropic/claude-haiku-4-5-20251001",
instructions="You are a helpful coding assistant.",
tools=[shell_tool],
)
print("response=", response)
# Step 2: Send the results of the tool call back to the model
# Get the response ID and tool call ID from the response
response_id = response.id
tool_call_id = None
for item in response.output:
if hasattr(item, "type") and item.type == "function_call":
tool_call_id = getattr(item, "call_id", None)
if tool_call_id:
break
# Validate that we got a tool call with a valid call_id
if not tool_call_id:
raise AssertionError(
f"Expected a function_call with a valid call_id in response.output, but got: {response.output}"
)
# Use await with asyncio.run for the async function
follow_up_response = litellm.responses(
model="anthropic/claude-haiku-4-5-20251001",
previous_response_id=response_id,
input=[
{
"type": "function_call_output",
"call_id": tool_call_id,
"output": '{"output":"<html>\\n<head>\\n <title>Hello Page</title>\\n</head>\\n<body>\\n <h1>Hi</h1>\\n <p>Welcome to this simple webpage!</p>\\n</body>\\n</html> > index.html\\n","metadata":{"exit_code":0,"duration_seconds":0}}',
}
],
tools=[shell_tool],
)
print("follow_up_response=", follow_up_response)

38
tests/harness_suites/responses_api_live/test_azure_responses_live.py Normal file
View file

@ -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?",
)

234
tests/harness_suites/responses_api_live/test_google_ai_studio_responses_live.py Normal file
View file

@ -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")

847
tests/harness_suites/responses_api_live/test_openai_responses_live.py Normal file
View file

@ -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"

46
tests/llm_responses_api_testing/test_azure_responses_api.py
View file

@ -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 = {

226
tests/llm_responses_api_testing/test_google_ai_studio_responses_api.py
View file

@ -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")

880
tests/llm_responses_api_testing/test_openai_responses_api.py
View file

@ -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"