from collections.abc import Mapping, Sequence from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, cast, get_type_hints import httpx from openai.types.responses import ResponseReasoningItem from pydantic import BaseModel, ValidationError import litellm from litellm._logging import verbose_logger from litellm.litellm_core_utils.core_helpers import process_response_headers from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import ( _safe_convert_created_field, ) from litellm.litellm_core_utils.url_utils import encode_url_path_segment from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig from litellm.llms.openai.chat.gpt_5_transformation import is_gpt_reasoning_series_name from litellm.secret_managers.main import get_secret_str from litellm.types.llms.openai import * from litellm.types.responses.main import * from litellm.types.router import GenericLiteLLMParams from litellm.types.utils import LlmProviders from ..common_utils import OpenAIError OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS: Final = 16 if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj LiteLLMLoggingObj = _LiteLLMLoggingObj else: LiteLLMLoggingObj = Any _NO_TOOL_UPDATE: Final[Mapping[str, object]] = MappingProxyType({}) _MODEL_FAMILIES_REJECTING_TOP_LEVEL_SCHEMA_COMBINATORS: Final = ("gpt-4", "gpt-3.5", "chatgpt-4o", "o1", "o3", "o4") _PROVIDERS_WITH_COMBINATOR_REJECTING_VALIDATOR: Final = frozenset({LlmProviders.AZURE, LlmProviders.OPENAI}) class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig): @property def custom_llm_provider(self) -> LlmProviders: return LlmProviders.OPENAI def supports_native_file_search(self) -> bool: return True @staticmethod def _is_gpt_5_model(model: str) -> bool: """Return True only for actual OpenAI GPT-5 models. Excludes pass-through models from other providers that happen to reference gpt-5 in their name (e.g. perplexity/openai/gpt-5.2). """ parts: Final = model.split("/") if len(parts) > 1 and parts[0] not in ("openai",): return False return is_gpt_reasoning_series_name(model) @staticmethod def _supports_reasoning_effort_none(model: str) -> bool: """Return True if the model supports reasoning.effort='none'.""" from litellm.utils import _supports_factory return _supports_factory( model=model, custom_llm_provider=None, key="supports_none_reasoning_effort", ) @staticmethod def _effort_resolves_to_none(model: str, effort: str | None) -> bool: """Whether this request's reasoning effort ends up as "none", the one condition under which a non-default temperature is accepted. Delegates to the chat-completions gpt-5 config so both surfaces answer from one rule: the Responses API reaches the same models over a different wire, and a second copy of the rule here is what let this surface keep forwarding temperature after the chat surface stopped. """ from litellm.llms.openai.chat.gpt_5_transformation import OpenAIGPT5Config return OpenAIGPT5Config.effort_resolves_to_none(model, effort) @staticmethod def _enforce_min_max_output_tokens(max_output_tokens: "int | None") -> "int | None": """Raise sub-minimum max_output_tokens up to the OpenAI Responses API minimum. OpenAI's Responses API rejects max_output_tokens below 16 for every model (not gpt-5 specific), so a client like Claude Code that sends a max_tokens=1 warmup probe on model switch would otherwise 400. Values that are None or already at/above the minimum are returned unchanged. """ if isinstance(max_output_tokens, int) and max_output_tokens < OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS: return OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS return max_output_tokens def get_supported_openai_params(self, model: str) -> list: """ All OpenAI Responses API params are supported """ supported_params: Final = get_type_hints(ResponsesAPIRequestParams).keys() return list( set( [ "input", "model", "extra_headers", "extra_query", "extra_body", "timeout", ] + list(supported_params) ) ) def map_openai_params( self, response_api_optional_params: ResponsesAPIOptionalRequestParams, model: str, drop_params: bool, ) -> dict: """No mapping applied since inputs are in OpenAI spec already. GPT-5 models have restrictions on temperature (only temperature=1 is accepted unless reasoning_effort='none' on models that support it). Apply the same validation used by the chat completions path. """ params: Final = dict(response_api_optional_params) if "max_output_tokens" in params: params["max_output_tokens"] = self._enforce_min_max_output_tokens(params.get("max_output_tokens")) if self._is_gpt_5_model(model=model): temperature: Final = params.get("temperature") if temperature is not None and temperature != 1: reasoning: Final = params.get("reasoning") or {} effort: Final = reasoning.get("effort") if isinstance(reasoning, dict) else None supports_none: Final = self._supports_reasoning_effort_none(model=model) if supports_none and self._effort_resolves_to_none(model, effort): pass # flexible temperature allowed elif drop_params or litellm.drop_params: params.pop("temperature", None) else: raise litellm.UnsupportedParamsError( message=( f"{model} doesn't support temperature={temperature} while reasoning is " "active. Only temperature=1 is supported unless reasoning.effort resolves " "to 'none', either set explicitly on the request or declared as the " "model's default_reasoning_effort. " "To drop unsupported params set `litellm.drop_params = True`" ), status_code=400, ) return params def transform_responses_api_request( self, model: str, input: str | ResponseInputParam, response_api_optional_request_params: dict, litellm_params: GenericLiteLLMParams, headers: dict, ) -> dict: """Strip Anthropic-only `cache_control` markers before sending to OpenAI. OpenAI's Responses API rejects unknown fields on input content blocks with HTTP 400 ("Unknown parameter: 'input[0].content[0].cache_control'"). Chat Completions strips these in `remove_cache_control_flag_from_messages_and_tools`; mirror that here. """ input = self._validate_input_param(input) tools = response_api_optional_request_params.get("tools") input, tools = self.remove_cache_control_flag_from_input_and_tools(model=model, input=input, tools=tools) sanitized_tools: Final = self._flatten_tool_schema_combinators_for_openai( model=model, tools=tools, litellm_params=litellm_params ) if sanitized_tools is not None: response_api_optional_request_params["tools"] = sanitized_tools final_request_params: Final = dict( ResponsesAPIRequestParams(model=model, input=input, **response_api_optional_request_params) ) return final_request_params def remove_cache_control_flag_from_input_and_tools( self, model: str, # allows overrides to selectively run this input: str | ResponseInputParam, tools: list[ALL_RESPONSES_API_TOOL_PARAMS] | None = None, ) -> tuple[ str | ResponseInputParam, list[ALL_RESPONSES_API_TOOL_PARAMS] | None, ]: """Sibling of `remove_cache_control_flag_from_messages_and_tools` on the chat path. Strips Anthropic-only `cache_control` markers from Responses API input content blocks and tools. `filter_value_from_dict` mutates each dict in place, so the same objects are returned. """ from litellm.litellm_core_utils.prompt_templates.common_utils import ( filter_value_from_dict, ) if isinstance(input, list): for item in input: if isinstance(item, dict): filter_value_from_dict(cast(dict, item), "cache_control") if tools is not None: for tool in tools: if isinstance(tool, dict): filter_value_from_dict(cast(dict, tool), "cache_control") return input, tools def _flatten_tool_schema_combinators_for_openai( self, model: str, tools: list[ALL_RESPONSES_API_TOOL_PARAMS] | None, # mutable-ok: request tools are a JSON list litellm_params: GenericLiteLLMParams, ) -> list[ALL_RESPONSES_API_TOOL_PARAMS] | None: # mutable-ok: request tools are a JSON list """Flatten top-level schema combinators only where OpenAI's validator rejects them. OpenAI-compatible backends reusing this config (and the ChatGPT backend Codex talks to natively) accept them, and so do GPT-5 and later models, which also call tools better with the union intact. Codex wraps MCP tools inside namespace entries, so nested ``tools`` arrays are walked too. Azure OpenAI shares the validator but names deployments arbitrarily, so the router's declared ``model_info.base_model`` wins over the deployment name and an unrecognized name without one is left untouched. """ if tools is None or self.custom_llm_provider not in _PROVIDERS_WITH_COMBINATOR_REJECTING_VALIDATOR: return tools gate_model: Final = self._combinator_gate_model(model=model, litellm_params=litellm_params) if not self._rejects_top_level_schema_combinators(gate_model): return tools flattened: Final = [ # mutable-ok: request tools are a JSON list self._flattened_tool_or_passthrough(tool) for tool in tools ] return cast("list[ALL_RESPONSES_API_TOOL_PARAMS]", flattened) # cast-ok: dict spread keeps each tool's shape @staticmethod def _flattened_tool_or_passthrough(tool: object) -> object: return OpenAIResponsesAPIConfig._flattened_tool_entry(tool) if isinstance(tool, dict) else tool @staticmethod def _rejects_top_level_schema_combinators(model: str) -> bool: bare_model: Final = model.split("/")[-1] base_model: Final = bare_model.split(":")[1] if bare_model.startswith("ft:") else bare_model return base_model.startswith(_MODEL_FAMILIES_REJECTING_TOP_LEVEL_SCHEMA_COMBINATORS) @staticmethod def _combinator_gate_model(model: str, litellm_params: GenericLiteLLMParams) -> str: model_info: Final[object] = getattr(litellm_params, "model_info", None) base_model: Final[object] = model_info.get("base_model") if isinstance(model_info, dict) else None return base_model if isinstance(base_model, str) and base_model else model @staticmethod def _flattened_tool_entry( entry: Mapping[str, object], ) -> dict[str, object]: # mutable-ok: request tools are JSON dicts from litellm.litellm_core_utils.prompt_templates.common_utils import ( flatten_top_level_schema_combinators, ) parameters: Final = entry.get("parameters") nested_tools: Final = entry.get("tools") parameters_update: Final = ( MappingProxyType({"parameters": flatten_top_level_schema_combinators(parameters)}) if isinstance(parameters, dict) else _NO_TOOL_UPDATE ) tools_update: Final = ( MappingProxyType({"tools": OpenAIResponsesAPIConfig._flattened_nested_tools(nested_tools)}) if isinstance(nested_tools, list) else _NO_TOOL_UPDATE ) return {**entry, **parameters_update, **tools_update} # mutable-ok: request tools are JSON dicts @staticmethod def _flattened_nested_tools( nested_tools: Sequence[object], ) -> list[object]: # mutable-ok: namespace tools are a JSON list return [ # mutable-ok: namespace tools are a JSON list OpenAIResponsesAPIConfig._flattened_tool_entry(item) if isinstance(item, dict) else item for item in nested_tools ] def _validate_input_param(self, input: str | ResponseInputParam) -> str | ResponseInputParam: """ Ensure all input fields if pydantic are converted to dict OpenAI API Fails when we try to JSON dumps specific input pydantic fields. This function ensures all input fields are converted to dict. """ if isinstance(input, list): validated_input: Final = [] for item in input: # if it's pydantic, convert to dict if isinstance(item, BaseModel): validated_input.append(item.model_dump(exclude_none=True)) elif isinstance(item, dict): # Handle reasoning items specifically to filter out status=None if item.get("type") == "reasoning": verbose_logger.debug("Handling reasoning item: %s", item) # Type assertion since we know it's a dict at this point dict_item = cast(dict[str, Any], item) filtered_item = self._handle_reasoning_item(dict_item) else: # For other dict items, just pass through filtered_item = cast(dict[str, Any], item) validated_input.append(filtered_item) else: validated_input.append(item) return validated_input # Input is expected to be either str or List, no single BaseModel expected return input def _handle_reasoning_item(self, item: dict[str, Any]) -> dict[str, Any]: """ Handle reasoning items specifically to filter out status=None using OpenAI's model. Issue: https://github.com/BerriAI/litellm/issues/13484 OpenAI API does not accept ReasoningItem(status=None), so we need to: 1. Check if the item is a reasoning type 2. Create a ResponseReasoningItem object with the item data 3. Convert it back to dict with exclude_none=True to filter None values """ if item.get("type") == "reasoning": try: # Ensure required fields are present for ResponseReasoningItem item_data: Final = dict(item) if "summary" not in item_data: item_data["summary"] = ( item_data.get("reasoning_content", "")[:100] + "..." if len(item_data.get("reasoning_content", "")) > 100 else item_data.get("reasoning_content", "") ) # Create ResponseReasoningItem object from the item data reasoning_item: Final = ResponseReasoningItem(**item_data) # Convert back to dict with exclude_none=True to exclude None fields dict_reasoning_item: Final = reasoning_item.model_dump(exclude_none=True) return dict_reasoning_item except Exception as e: verbose_logger.debug("Failed to create ResponseReasoningItem, falling back to manual filtering: %s", e) # Fallback: manually filter out known None fields filtered_item: Final = { k: v for k, v in item.items() if v is not None or k not in {"status", "content", "encrypted_content"} } return filtered_item return item def transform_response_api_response( self, model: str, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, ) -> ResponsesAPIResponse: """No transform applied since outputs are in OpenAI spec already""" try: logging_obj.post_call( original_response=raw_response.text, additional_args={"complete_input_dict": {}}, ) raw_response_json: Final = raw_response.json() raw_response_json["created_at"] = _safe_convert_created_field(raw_response_json["created_at"]) except Exception: raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code) raw_response_headers: Final = dict(raw_response.headers) processed_headers: Final = process_response_headers(raw_response_headers) try: response = ResponsesAPIResponse.model_validate(raw_response_json) except Exception: verbose_logger.debug( "Error constructing ResponsesAPIResponse: %s, using model_construct", raw_response_json ) response = ResponsesAPIResponse.model_construct(**raw_response_json) # Store processed headers in additional_headers so they get returned to the client response._hidden_params["additional_headers"] = processed_headers response._hidden_params["headers"] = raw_response_headers return response def validate_environment(self, headers: dict, model: str, litellm_params: GenericLiteLLMParams | None) -> dict: litellm_params = litellm_params or GenericLiteLLMParams() api_key = litellm_params.api_key or litellm.api_key or litellm.openai_key or get_secret_str("OPENAI_API_KEY") headers.setdefault("Content-Type", "application/json") headers["Authorization"] = f"Bearer {api_key}" return headers def get_complete_url( self, api_base: str | None, litellm_params: dict, ) -> str: """ Get the endpoint for OpenAI responses API """ api_base = ( api_base or litellm.api_base or get_secret_str("OPENAI_BASE_URL") or get_secret_str("OPENAI_API_BASE") or "https://api.openai.com/v1" ) # Remove trailing slashes api_base = api_base.rstrip("/") return f"{api_base}/responses" def transform_streaming_response( self, model: str, parsed_chunk: dict, logging_obj: LiteLLMLoggingObj, ) -> ResponsesAPIStreamingResponse: """ Transform a parsed streaming response chunk into a ResponsesAPIStreamingResponse """ # Convert the dictionary to a properly typed ResponsesAPIStreamingResponse verbose_logger.debug("Raw OpenAI Chunk=%s", parsed_chunk) event_type: Final = str(parsed_chunk.get("type")) event_pydantic_model: Final = OpenAIResponsesAPIConfig.get_event_model_class(event_type=event_type) # Some OpenAI-compatible providers send error.code: null; coalesce so validation succeeds. try: error_obj: Final = parsed_chunk.get("error") if isinstance(error_obj, dict) and error_obj.get("code") is None: parsed_chunk = dict(parsed_chunk) parsed_chunk["error"] = dict(error_obj) parsed_chunk["error"]["code"] = "unknown_error" except Exception: verbose_logger.debug("Failed to coalesce error.code in parsed_chunk") try: return event_pydantic_model(**parsed_chunk) except ValidationError: verbose_logger.debug( "Pydantic validation failed for %s with chunk %s, falling back to model_construct", event_pydantic_model.__name__, parsed_chunk, ) return event_pydantic_model.model_construct(**parsed_chunk) @staticmethod def parse_terminal_response_from_stream_chunks(all_chunks: list[str]) -> ResponsesAPIResponse | None: for chunk_str in reversed(all_chunks): for event_model in (ResponseCompletedEvent, ResponseIncompleteEvent, ResponseFailedEvent): try: return event_model.model_validate_json(chunk_str.removeprefix("data: ")).response except ValueError: continue return None @staticmethod def get_event_model_class(event_type: str) -> Any: """ Returns the appropriate event model class based on the event type. Args: event_type (str): The type of event from the response chunk Returns: Any: The corresponding event model class Raises: ValueError: If the event type is unknown """ event_models: Final = { ResponsesAPIStreamEvents.RESPONSE_CREATED: ResponseCreatedEvent, ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS: ResponseInProgressEvent, ResponsesAPIStreamEvents.RESPONSE_COMPLETED: ResponseCompletedEvent, ResponsesAPIStreamEvents.RESPONSE_FAILED: ResponseFailedEvent, ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE: ResponseIncompleteEvent, ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED: OutputItemAddedEvent, ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: OutputItemDoneEvent, ResponsesAPIStreamEvents.CONTENT_PART_ADDED: ContentPartAddedEvent, ResponsesAPIStreamEvents.CONTENT_PART_DONE: ContentPartDoneEvent, ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA: OutputTextDeltaEvent, ResponsesAPIStreamEvents.OUTPUT_TEXT_ANNOTATION_ADDED: OutputTextAnnotationAddedEvent, ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: OutputTextDoneEvent, ResponsesAPIStreamEvents.REFUSAL_DELTA: RefusalDeltaEvent, ResponsesAPIStreamEvents.REFUSAL_DONE: RefusalDoneEvent, ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA: FunctionCallArgumentsDeltaEvent, ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE: FunctionCallArgumentsDoneEvent, ResponsesAPIStreamEvents.FILE_SEARCH_CALL_IN_PROGRESS: FileSearchCallInProgressEvent, ResponsesAPIStreamEvents.FILE_SEARCH_CALL_SEARCHING: FileSearchCallSearchingEvent, ResponsesAPIStreamEvents.FILE_SEARCH_CALL_COMPLETED: FileSearchCallCompletedEvent, ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS: WebSearchCallInProgressEvent, ResponsesAPIStreamEvents.WEB_SEARCH_CALL_SEARCHING: WebSearchCallSearchingEvent, ResponsesAPIStreamEvents.WEB_SEARCH_CALL_COMPLETED: WebSearchCallCompletedEvent, ResponsesAPIStreamEvents.MCP_LIST_TOOLS_IN_PROGRESS: MCPListToolsInProgressEvent, ResponsesAPIStreamEvents.MCP_LIST_TOOLS_COMPLETED: MCPListToolsCompletedEvent, ResponsesAPIStreamEvents.MCP_LIST_TOOLS_FAILED: MCPListToolsFailedEvent, ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS: MCPCallInProgressEvent, ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DELTA: MCPCallArgumentsDeltaEvent, ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DONE: MCPCallArgumentsDoneEvent, ResponsesAPIStreamEvents.MCP_CALL_COMPLETED: MCPCallCompletedEvent, ResponsesAPIStreamEvents.MCP_CALL_FAILED: MCPCallFailedEvent, ResponsesAPIStreamEvents.IMAGE_GENERATION_PARTIAL_IMAGE: ImageGenerationPartialImageEvent, ResponsesAPIStreamEvents.ERROR: ErrorEvent, # Shell tool events: passthrough as GenericEvent so payload is preserved ResponsesAPIStreamEvents.SHELL_CALL_IN_PROGRESS: GenericEvent, ResponsesAPIStreamEvents.SHELL_CALL_COMPLETED: GenericEvent, ResponsesAPIStreamEvents.SHELL_CALL_OUTPUT: GenericEvent, } model_class: Final = event_models.get(cast(ResponsesAPIStreamEvents, event_type)) if not model_class: return GenericEvent return model_class def should_fake_stream( self, model: str | None, stream: bool | None, custom_llm_provider: str | None = None, ) -> bool: if stream is not True: return False if model is not None: try: if ( litellm.utils.supports_native_streaming( model=model, custom_llm_provider=custom_llm_provider, ) is False ): return True except Exception as e: verbose_logger.debug("Error getting model info in OpenAIResponsesAPIConfig: %s", e) return False def supports_native_websocket(self) -> bool: """OpenAI supports native WebSocket for Responses API""" return True ######################################################### ########## DELETE RESPONSE API TRANSFORMATION ############## ######################################################### def transform_delete_response_api_request( self, response_id: str, api_base: str, litellm_params: GenericLiteLLMParams, headers: dict, ) -> tuple[str, dict]: """ Transform the delete response API request into a URL and data OpenAI API expects the following request - DELETE /v1/responses/{response_id} """ encoded_response_id: Final = encode_url_path_segment(response_id, field_name="response_id") url: Final = f"{api_base}/{encoded_response_id}" data: Final[dict] = {} return url, data def transform_delete_response_api_response( self, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, ) -> DeleteResponseResult: """ Transform the delete response API response into a DeleteResponseResult """ try: raw_response_json: Final = raw_response.json() except Exception: raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code) return DeleteResponseResult(**raw_response_json) ######################################################### ########## GET RESPONSE API TRANSFORMATION ############### ######################################################### def transform_get_response_api_request( self, response_id: str, api_base: str, litellm_params: GenericLiteLLMParams, headers: dict, ) -> tuple[str, dict]: """ Transform the get response API request into a URL and data OpenAI API expects the following request - GET /v1/responses/{response_id} """ encoded_response_id: Final = encode_url_path_segment(response_id, field_name="response_id") url: Final = f"{api_base}/{encoded_response_id}" data: Final[dict] = {} return url, data def transform_get_response_api_response( self, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, ) -> ResponsesAPIResponse: """ Transform the get response API response into a ResponsesAPIResponse """ try: raw_response_json: Final = raw_response.json() except Exception: raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code) raw_response_headers: Final = dict(raw_response.headers) processed_headers: Final = process_response_headers(raw_response_headers) response: Final = ResponsesAPIResponse.model_validate(raw_response_json) response._hidden_params["additional_headers"] = processed_headers response._hidden_params["headers"] = raw_response_headers return response ######################################################### ########## LIST INPUT ITEMS TRANSFORMATION ############# ######################################################### def transform_list_input_items_request( self, response_id: str, api_base: str, litellm_params: GenericLiteLLMParams, headers: dict, after: str | None = None, before: str | None = None, include: list[str] | None = None, limit: int = 20, order: Literal["asc", "desc"] = "desc", ) -> tuple[str, dict]: encoded_response_id: Final = encode_url_path_segment(response_id, field_name="response_id") url: Final = f"{api_base}/{encoded_response_id}/input_items" params: Final[dict[str, Any]] = {} if after is not None: params["after"] = after if before is not None: params["before"] = before if include: params["include"] = ",".join(include) if limit is not None: params["limit"] = limit if order is not None: params["order"] = order return url, params def transform_list_input_items_response( self, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, ) -> dict: try: return raw_response.json() except Exception: raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code) ######################################################### ########## CANCEL RESPONSE API TRANSFORMATION ########## ######################################################### def transform_cancel_response_api_request( self, response_id: str, api_base: str, litellm_params: GenericLiteLLMParams, headers: dict, ) -> tuple[str, dict]: """ Transform the cancel response API request into a URL and data OpenAI API expects the following request - POST /v1/responses/{response_id}/cancel """ encoded_response_id: Final = encode_url_path_segment(response_id, field_name="response_id") url: Final = f"{api_base}/{encoded_response_id}/cancel" data: Final[dict] = {} return url, data def transform_cancel_response_api_response( self, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, ) -> ResponsesAPIResponse: """ Transform the cancel response API response into a ResponsesAPIResponse """ try: raw_response_json: Final = raw_response.json() except Exception: raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code) raw_response_headers: Final = dict(raw_response.headers) processed_headers: Final = process_response_headers(raw_response_headers) response: Final = ResponsesAPIResponse.model_validate(raw_response_json) response._hidden_params["additional_headers"] = processed_headers response._hidden_params["headers"] = raw_response_headers return response ######################################################### ########## COMPACT RESPONSE API TRANSFORMATION ########## ######################################################### def transform_compact_response_api_request( self, model: str, input: str | ResponseInputParam, response_api_optional_request_params: dict, api_base: str, litellm_params: GenericLiteLLMParams, headers: dict, ) -> tuple[str, dict]: """ Transform the compact response API request into a URL and data OpenAI API expects the following request - POST /v1/responses/compact """ # Preserve query params (e.g., api-version) while appending /compact. parsed_url: Final = httpx.URL(api_base) compact_path: Final = parsed_url.path.rstrip("/") + "/compact" url: Final = str(parsed_url.copy_with(path=compact_path)) input = self._validate_input_param(input) tools = response_api_optional_request_params.get("tools") input, tools = self.remove_cache_control_flag_from_input_and_tools(model=model, input=input, tools=tools) sanitized_tools: Final = self._flatten_tool_schema_combinators_for_openai( model=model, tools=tools, litellm_params=litellm_params ) if sanitized_tools is not None: response_api_optional_request_params["tools"] = sanitized_tools data: Final = dict(ResponsesAPIRequestParams(model=model, input=input, **response_api_optional_request_params)) return url, data def transform_compact_response_api_response( self, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, ) -> ResponsesAPIResponse: """ Transform the compact response API response into a ResponsesAPIResponse """ try: logging_obj.post_call( original_response=raw_response.text, additional_args={"complete_input_dict": {}}, ) raw_response_json: Final = raw_response.json() raw_response_json["created_at"] = _safe_convert_created_field(raw_response_json["created_at"]) except Exception: raise OpenAIError(message=raw_response.text, status_code=raw_response.status_code) raw_response_headers: Final = dict(raw_response.headers) processed_headers: Final = process_response_headers(raw_response_headers) try: response = ResponsesAPIResponse.model_validate(raw_response_json) except Exception: verbose_logger.debug( "Error constructing ResponsesAPIResponse: %s, using model_construct", raw_response_json ) response = ResponsesAPIResponse.model_construct(**raw_response_json) response._hidden_params["additional_headers"] = processed_headers response._hidden_params["headers"] = raw_response_headers return response