Add bridge for /chat/completion -> /responses API (#11632)

* refactor(responses/): refactor to move responses_to_completion in separate folder future work to support completion_to_responses bridge allow calling codex mini via chat completions (and other endpoints) * Revert "refactor(responses/): refactor to move responses_to_completion in separate folder" This reverts commit ff87cb895812283d107f47e8e528bcebe93d8015. * feat: initial responses api bridge write it like a custom llm - requires lesser 'new' components * style: add __init__'s and bubble up the responses api bridge * feat(responses/transformation): working sync completion -> responses and back bridge (non-streaming) * feat(responses/): working async (non-streaming) completion <-> responses bridge Allows calling codex mini via proxy * feat(responses/): working sync + async streaming for base model response iterator * fix: reduce function size maintain <50 LOC * fix(main.py): safely handle responses api model check * fix: fix linting errors
2026-08-24 08:30:09 +00:00 · 2025-06-11 22:20:18 -07:00
parent e46ef9d642
commit 7f2e188ee2
13 changed files with 962 additions and 42 deletions
@@ -0,0 +1,4 @@
+Logic specific for `litellm.completion`. 
+
+Includes:
+- Bridge for transforming completion requests to responses api requests
@@ -0,0 +1,3 @@
+from .litellm_responses_transformation import responses_api_bridge
+
+__all__ = ["responses_api_bridge"]
@@ -0,0 +1,3 @@
+from .handler import responses_api_bridge
+
+__all__ = ["responses_api_bridge"]
@@ -0,0 +1,198 @@
+"""
+Handler for transforming /chat/completions api requests to litellm.responses requests
+"""
+from typing import TYPE_CHECKING, Any, Coroutine, TypedDict, Union
+
+if TYPE_CHECKING:
+    from litellm import CustomStreamWrapper, LiteLLMLoggingObj, ModelResponse
+
+
+class ResponsesToCompletionBridgeHandlerInputKwargs(TypedDict):
+    model: str
+    messages: list
+    optional_params: dict
+    litellm_params: dict
+    headers: dict
+    model_response: "ModelResponse"
+    logging_obj: "LiteLLMLoggingObj"
+    custom_llm_provider: str
+
+
+class ResponsesToCompletionBridgeHandler:
+    def __init__(self):
+        from .transformation import LiteLLMResponsesTransformationHandler
+
+        super().__init__()
+        self.transformation_handler = LiteLLMResponsesTransformationHandler()
+
+    def validate_input_kwargs(
+        self, kwargs: dict
+    ) -> ResponsesToCompletionBridgeHandlerInputKwargs:
+        from litellm import LiteLLMLoggingObj
+        from litellm.types.utils import ModelResponse
+
+        model = kwargs.get("model")
+        if model is None or not isinstance(model, str):
+            raise ValueError("model is required")
+
+        custom_llm_provider = kwargs.get("custom_llm_provider")
+        if custom_llm_provider is None or not isinstance(custom_llm_provider, str):
+            raise ValueError("custom_llm_provider is required")
+
+        messages = kwargs.get("messages")
+        if messages is None or not isinstance(messages, list):
+            raise ValueError("messages is required")
+
+        optional_params = kwargs.get("optional_params")
+        if optional_params is None or not isinstance(optional_params, dict):
+            raise ValueError("optional_params is required")
+
+        litellm_params = kwargs.get("litellm_params")
+        if litellm_params is None or not isinstance(litellm_params, dict):
+            raise ValueError("litellm_params is required")
+
+        headers = kwargs.get("headers")
+        if headers is None or not isinstance(headers, dict):
+            raise ValueError("headers is required")
+
+        model_response = kwargs.get("model_response")
+        if model_response is None or not isinstance(model_response, ModelResponse):
+            raise ValueError("model_response is required")
+
+        logging_obj = kwargs.get("logging_obj")
+        if logging_obj is None or not isinstance(logging_obj, LiteLLMLoggingObj):
+            raise ValueError("logging_obj is required")
+
+        return ResponsesToCompletionBridgeHandlerInputKwargs(
+            model=model,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            headers=headers,
+            model_response=model_response,
+            logging_obj=logging_obj,
+            custom_llm_provider=custom_llm_provider,
+        )
+
+    def completion(
+        self, *args, **kwargs
+    ) -> Union[
+        Coroutine[Any, Any, Union["ModelResponse", "CustomStreamWrapper"]],
+        "ModelResponse",
+        "CustomStreamWrapper",
+    ]:
+        if kwargs.get("acompletion") is True:
+            return self.acompletion(**kwargs)
+
+        from litellm import responses
+        from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
+        from litellm.types.llms.openai import ResponsesAPIResponse
+
+        validated_kwargs = self.validate_input_kwargs(kwargs)
+        model = validated_kwargs["model"]
+        messages = validated_kwargs["messages"]
+        optional_params = validated_kwargs["optional_params"]
+        litellm_params = validated_kwargs["litellm_params"]
+        headers = validated_kwargs["headers"]
+        model_response = validated_kwargs["model_response"]
+        logging_obj = validated_kwargs["logging_obj"]
+        custom_llm_provider = validated_kwargs["custom_llm_provider"]
+
+        request_data = self.transformation_handler.transform_request(
+            model=model,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            headers=headers,
+        )
+        result = responses(
+            **request_data,
+        )
+
+        if isinstance(result, ResponsesAPIResponse):
+            return self.transformation_handler.transform_response(
+                model=model,
+                raw_response=result,
+                model_response=model_response,
+                logging_obj=logging_obj,
+                request_data=request_data,
+                messages=messages,
+                optional_params=optional_params,
+                litellm_params=litellm_params,
+                encoding=kwargs.get("encoding"),
+                api_key=kwargs.get("api_key"),
+                json_mode=kwargs.get("json_mode"),
+            )
+        else:
+            completion_stream = self.transformation_handler.get_model_response_iterator(
+                streaming_response=result,  # type: ignore
+                sync_stream=True,
+                json_mode=kwargs.get("json_mode"),
+            )
+            streamwrapper = CustomStreamWrapper(
+                completion_stream=completion_stream,
+                model=model,
+                custom_llm_provider=custom_llm_provider,
+                logging_obj=logging_obj,
+            )
+            return streamwrapper
+
+    async def acompletion(
+        self, *args, **kwargs
+    ) -> Union["ModelResponse", "CustomStreamWrapper"]:
+        from litellm import aresponses
+        from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
+        from litellm.types.llms.openai import ResponsesAPIResponse
+
+        validated_kwargs = self.validate_input_kwargs(kwargs)
+        model = validated_kwargs["model"]
+        messages = validated_kwargs["messages"]
+        optional_params = validated_kwargs["optional_params"]
+        litellm_params = validated_kwargs["litellm_params"]
+        headers = validated_kwargs["headers"]
+        model_response = validated_kwargs["model_response"]
+        logging_obj = validated_kwargs["logging_obj"]
+        custom_llm_provider = validated_kwargs["custom_llm_provider"]
+
+        request_data = self.transformation_handler.transform_request(
+            model=model,
+            messages=messages,
+            optional_params=optional_params,
+            litellm_params=litellm_params,
+            headers=headers,
+        )
+        result = await aresponses(
+            **request_data,
+            aresponses=True,
+        )
+
+        if isinstance(result, ResponsesAPIResponse):
+            return self.transformation_handler.transform_response(
+                model=model,
+                raw_response=result,
+                model_response=model_response,
+                logging_obj=logging_obj,
+                request_data=request_data,
+                messages=messages,
+                optional_params=optional_params,
+                litellm_params=litellm_params,
+                encoding=kwargs.get("encoding"),
+                api_key=kwargs.get("api_key"),
+                json_mode=kwargs.get("json_mode"),
+            )
+        else:
+            completion_stream = self.transformation_handler.get_model_response_iterator(
+                streaming_response=result,  # type: ignore
+                sync_stream=False,
+                json_mode=kwargs.get("json_mode"),
+            )
+            streamwrapper = CustomStreamWrapper(
+                completion_stream=completion_stream,
+                model=model,
+                custom_llm_provider=custom_llm_provider,
+                logging_obj=logging_obj,
+            )
+            return streamwrapper
+
+
+responses_api_bridge = ResponsesToCompletionBridgeHandler()
@@ -0,0 +1,554 @@
+"""
+Handler for transforming /chat/completions api requests to litellm.responses requests
+"""
+import json
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncIterator,
+    Dict,
+    Iterable,
+    Iterator,
+    List,
+    Optional,
+    Tuple,
+    Union,
+    cast,
+)
+
+from litellm import ModelResponse
+from litellm._logging import verbose_logger
+from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
+from litellm.llms.base_llm.bridges.completion_transformation import (
+    CompletionTransformationBridge,
+)
+
+if TYPE_CHECKING:
+    from pydantic import BaseModel
+
+    from litellm import LiteLLMLoggingObj, ModelResponse
+    from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
+    from litellm.types.llms.openai import (
+        ALL_RESPONSES_API_TOOL_PARAMS,
+        AllMessageValues,
+        ChatCompletionThinkingBlock,
+        OpenAIMessageContentListBlock,
+    )
+    from litellm.types.utils import GenericStreamingChunk, ModelResponseStream
+
+
+class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
+    """
+    Handler for transforming /chat/completions api requests to litellm.responses requests
+    """
+
+    def __init__(self):
+        pass
+
+    def convert_chat_completion_messages_to_responses_api(
+        self, messages: List["AllMessageValues"]
+    ) -> Tuple[List[Any], Optional[str]]:
+        input_items: List[Any] = []
+        instructions: Optional[str] = None
+
+        for msg in messages:
+            role = msg.get("role")
+            content = msg.get("content", "")
+            tool_calls = msg.get("tool_calls")
+            tool_call_id = msg.get("tool_call_id")
+
+            if role == "system":
+                # Extract system message as instructions
+                if isinstance(content, str):
+                    instructions = content
+                else:
+                    raise ValueError(f"System message must be a string: {content}")
+            elif role == "tool":
+                # Convert tool message to function call output format
+                input_items.append(
+                    {
+                        "type": "function_call_output",
+                        "call_id": tool_call_id,
+                        "output": content,
+                    }
+                )
+            elif role == "assistant" and tool_calls and isinstance(tool_calls, list):
+                for tool_call in tool_calls:
+                    function = tool_call.get("function")
+                    if function:
+                        input_tool_call = {
+                            "type": "function_call",
+                            "call_id": tool_call["id"],
+                        }
+                        if "name" in function:
+                            input_tool_call["name"] = function["name"]
+                        if "arguments" in function:
+                            input_tool_call["arguments"] = function["arguments"]
+                        input_items.append(input_tool_call)
+                    else:
+                        raise ValueError(f"tool call not supported: {tool_call}")
+            elif content is not None:
+                # Regular user/assistant message
+                input_items.append(
+                    {
+                        "type": "message",
+                        "role": role,
+                        "content": self._convert_content_to_responses_format(content),
+                    }
+                )
+
+        return input_items, instructions
+
+    def transform_request(
+        self,
+        model: str,
+        messages: List["AllMessageValues"],
+        optional_params: dict,
+        litellm_params: dict,
+        headers: dict,
+    ) -> dict:
+        from litellm.types.llms.openai import ResponsesAPIOptionalRequestParams
+
+        (
+            input_items,
+            instructions,
+        ) = self.convert_chat_completion_messages_to_responses_api(messages)
+
+        # Build responses API request using the reverse transformation logic
+        responses_api_request = ResponsesAPIOptionalRequestParams()
+
+        # Set instructions if we found a system message
+        if instructions:
+            responses_api_request["instructions"] = instructions
+
+        # Map optional parameters
+        for key, value in optional_params.items():
+            if value is None:
+                continue
+            if key in ("max_tokens", "max_completion_tokens"):
+                responses_api_request["max_output_tokens"] = value
+            elif key == "tools" and value is not None:
+                # Convert chat completion tools to responses API tools format
+                responses_api_request[
+                    "tools"
+                ] = self._convert_tools_to_responses_format(
+                    cast(List[Dict[str, Any]], value)
+                )
+            elif key in ResponsesAPIOptionalRequestParams.__annotations__.keys():
+                responses_api_request[key] = value  # type: ignore
+            elif key == "metadata":
+                responses_api_request["metadata"] = value
+            elif key == "previous_response_id":
+                # Support for responses API session management
+                responses_api_request["previous_response_id"] = value
+
+        # Get stream parameter from litellm_params if not in optional_params
+        stream = optional_params.get("stream") or litellm_params.get("stream", False)
+        verbose_logger.debug(f"Chat provider: Stream parameter: {stream}")
+
+        # Ensure stream is properly set in the request
+        if stream:
+            responses_api_request["stream"] = True
+
+        # Handle session management if previous_response_id is provided
+        previous_response_id = optional_params.get("previous_response_id")
+        if previous_response_id:
+            # Use the existing session handler for responses API
+            verbose_logger.debug(
+                f"Chat provider: Warning ignoring previous response ID: {previous_response_id}"
+            )
+
+        # Convert back to responses API format for the actual request
+
+        api_model = model
+
+        request_data = {
+            "model": api_model,
+            "input": input_items,
+        }
+
+        verbose_logger.debug(
+            f"Chat provider: Final request model={api_model}, input_items={len(input_items)}"
+        )
+
+        # Add non-None values from responses_api_request
+        for key, value in responses_api_request.items():
+            if value is not None:
+                if key == "instructions" and instructions:
+                    request_data["instructions"] = instructions
+                else:
+                    request_data[key] = value
+
+        return request_data
+
+    def transform_response(
+        self,
+        model: str,
+        raw_response: "BaseModel",
+        model_response: "ModelResponse",
+        logging_obj: "LiteLLMLoggingObj",
+        request_data: dict,
+        messages: List["AllMessageValues"],
+        optional_params: dict,
+        litellm_params: dict,
+        encoding: Any,
+        api_key: Optional[str] = None,
+        json_mode: Optional[bool] = None,
+    ) -> "ModelResponse":
+        """Transform Responses API response to chat completion response"""
+        from openai.types.responses import (
+            ResponseFunctionToolCall,
+            ResponseOutputMessage,
+            ResponseReasoningItem,
+        )
+
+        from litellm.responses.utils import ResponseAPILoggingUtils
+        from litellm.types.llms.openai import ResponsesAPIResponse
+        from litellm.types.responses.main import (
+            GenericResponseOutputItem,
+            OutputFunctionToolCall,
+        )
+        from litellm.types.utils import Choices, Message
+
+        if not isinstance(raw_response, ResponsesAPIResponse):
+            raise ValueError(f"Unexpected response type: {type(raw_response)}")
+
+        choices: List[Choices] = []
+        index = 0
+        for item in raw_response.output:
+            if isinstance(item, ResponseReasoningItem):
+                pass  # ignore for now.
+            elif isinstance(item, ResponseOutputMessage):
+                for content in item.content:
+                    response_text = getattr(content, "text", "")
+                    msg = Message(
+                        role=item.role, content=response_text if response_text else ""
+                    )
+
+                    choices.append(
+                        Choices(message=msg, finish_reason="stop", index=index)
+                    )
+                    index += 1
+            elif isinstance(item, ResponseFunctionToolCall):
+                msg = Message(
+                    content=None,
+                    tool_calls=[
+                        {
+                            "id": item.call_id,
+                            "function": {
+                                "name": item.name,
+                                "arguments": item.arguments,
+                            },
+                            "type": "function",
+                        }
+                    ],
+                )
+
+                choices.append(
+                    Choices(message=msg, finish_reason="tool_calls", index=index)
+                )
+                index += 1
+            elif isinstance(item, GenericResponseOutputItem):
+                raise ValueError("GenericResponseOutputItem not supported")
+            elif isinstance(item, OutputFunctionToolCall):
+                # function/tool calls pass through as-is
+                raise ValueError("Function calling not supported yet.")
+            else:
+                raise ValueError(f"Unknown item type: {item}")
+
+        setattr(model_response, "choices", choices)
+
+        setattr(
+            model_response,
+            "usage",
+            ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(
+                raw_response.usage
+            ),
+        )
+        return model_response
+
+    def get_model_response_iterator(
+        self,
+        streaming_response: Union[
+            Iterator[str], AsyncIterator[str], "ModelResponse", "BaseModel"
+        ],
+        sync_stream: bool,
+        json_mode: Optional[bool] = False,
+    ) -> BaseModelResponseIterator:
+        return OpenAiResponsesToChatCompletionStreamIterator(
+            streaming_response, sync_stream, json_mode
+        )
+
+    def _convert_content_to_responses_format(
+        self,
+        content: Union[
+            str,
+            Iterable[
+                Union["OpenAIMessageContentListBlock", "ChatCompletionThinkingBlock"]
+            ],
+        ],
+    ) -> List[Dict[str, Any]]:
+        """Convert chat completion content to responses API format"""
+        verbose_logger.debug(
+            f"Chat provider: Converting content to responses format - input type: {type(content)}"
+        )
+
+        if isinstance(content, str):
+            result = [{"type": "input_text", "text": content}]
+            verbose_logger.debug(f"Chat provider: String content -> {result}")
+            return result
+        elif isinstance(content, list):
+            result = []
+            for i, item in enumerate(content):
+                verbose_logger.debug(
+                    f"Chat provider: Processing content item {i}: {type(item)} = {item}"
+                )
+                if isinstance(item, str):
+                    converted = {"type": "input_text", "text": item}
+                    result.append(converted)
+                    verbose_logger.debug(f"Chat provider:   -> {converted}")
+                elif isinstance(item, dict):
+                    # Handle multimodal content
+                    original_type = item.get("type")
+                    if original_type == "text":
+                        converted = {"type": "input_text", "text": item.get("text", "")}
+                        result.append(converted)
+                        verbose_logger.debug(f"Chat provider:   text -> {converted}")
+                    elif original_type == "image_url":
+                        # Map to responses API image format
+                        converted = {
+                            "type": "input_image",
+                            "image_url": item.get("image_url", {}),
+                        }
+                        result.append(converted)
+                        verbose_logger.debug(
+                            f"Chat provider:   image_url -> {converted}"
+                        )
+                    else:
+                        # Try to map other types to responses API format
+                        item_type = original_type or "input_text"
+                        if item_type == "image":
+                            converted = {"type": "input_image", **item}
+                            result.append(converted)
+                            verbose_logger.debug(
+                                f"Chat provider:   image -> {converted}"
+                            )
+                        elif item_type in [
+                            "input_text",
+                            "input_image",
+                            "output_text",
+                            "refusal",
+                            "input_file",
+                            "computer_screenshot",
+                            "summary_text",
+                        ]:
+                            # Already in responses API format
+                            result.append(item)
+                            verbose_logger.debug(
+                                f"Chat provider:   passthrough -> {item}"
+                            )
+                        else:
+                            # Default to input_text for unknown types
+                            converted = {
+                                "type": "input_text",
+                                "text": str(item.get("text", item)),
+                            }
+                            result.append(converted)
+                            verbose_logger.debug(
+                                f"Chat provider:   unknown({original_type}) -> {converted}"
+                            )
+            verbose_logger.debug(f"Chat provider: Final converted content: {result}")
+            return result
+        else:
+            result = [{"type": "input_text", "text": str(content)}]
+            verbose_logger.debug(f"Chat provider: Other content type -> {result}")
+            return result
+
+    def _convert_tools_to_responses_format(
+        self, tools: List[Dict[str, Any]]
+    ) -> List["ALL_RESPONSES_API_TOOL_PARAMS"]:
+        """Convert chat completion tools to responses API tools format"""
+        responses_tools = []
+        for tool in tools:
+            if tool.get("type") == "function":
+                function = tool.get("function", {})
+                responses_tools.append(
+                    {
+                        "type": "function",
+                        "name": function.get("name", ""),
+                        "description": function.get("description", ""),
+                        "parameters": function.get("parameters", {}),
+                        "strict": function.get("strict", False),
+                    }
+                )
+        return cast(List["ALL_RESPONSES_API_TOOL_PARAMS"], responses_tools)
+
+    def _map_responses_status_to_finish_reason(self, status: Optional[str]) -> str:
+        """Map responses API status to chat completion finish_reason"""
+        if not status:
+            return "stop"
+
+        status_mapping = {
+            "completed": "stop",
+            "incomplete": "length",
+            "failed": "stop",
+            "cancelled": "stop",
+        }
+
+        return status_mapping.get(status, "stop")
+
+
+class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
+    def __init__(
+        self, streaming_response, sync_stream: bool, json_mode: Optional[bool] = False
+    ):
+        super().__init__(streaming_response, sync_stream, json_mode)
+
+    def _handle_string_chunk(
+        self, str_line: Union[str, "BaseModel"]
+    ) -> Union["GenericStreamingChunk", "ModelResponseStream"]:
+        from pydantic import BaseModel
+
+        if isinstance(str_line, BaseModel):
+            return self.chunk_parser(str_line.model_dump())
+
+        if not str_line or str_line.startswith("event:"):
+            # ignore.
+            return GenericStreamingChunk(
+                text="", tool_use=None, is_finished=False, finish_reason="", usage=None
+            )
+        index = str_line.find("data:")
+        if index != -1:
+            str_line = str_line[index + 5 :]
+
+        return self.chunk_parser(json.loads(str_line))
+
+    def chunk_parser(
+        self, chunk: dict
+    ) -> Union["GenericStreamingChunk", "ModelResponseStream"]:
+        # Transform responses API streaming chunk to chat completion format
+        from litellm.types.llms.openai import ChatCompletionToolCallFunctionChunk
+        from litellm.types.utils import (
+            ChatCompletionToolCallChunk,
+            GenericStreamingChunk,
+        )
+
+        verbose_logger.debug(
+            f"Chat provider: transform_streaming_response called with chunk: {chunk}"
+        )
+        parsed_chunk = chunk
+
+        if not parsed_chunk:
+            raise ValueError("Chat provider: Empty parsed_chunk")
+
+        if not isinstance(parsed_chunk, dict):
+            raise ValueError(f"Chat provider: Invalid chunk type {type(parsed_chunk)}")
+
+        # Handle different event types from responses API
+        event_type = parsed_chunk.get("type")
+        verbose_logger.debug(f"Chat provider: Processing event type: {event_type}")
+
+        if event_type == "response.created":
+            # Initial response creation event
+            verbose_logger.debug(f"Chat provider: response.created -> {chunk}")
+            return GenericStreamingChunk(
+                text="", tool_use=None, is_finished=False, finish_reason="", usage=None
+            )
+        elif event_type == "response.output_item.added":
+            # New output item added
+            output_item = parsed_chunk.get("item", {})
+            if output_item.get("type") == "function_call":
+                return GenericStreamingChunk(
+                    text="",
+                    tool_use=ChatCompletionToolCallChunk(
+                        id=output_item.get("call_id"),
+                        index=0,
+                        type="function",
+                        function=ChatCompletionToolCallFunctionChunk(
+                            name=parsed_chunk.get("name", None),
+                            arguments=parsed_chunk.get("arguments", ""),
+                        ),
+                    ),
+                    is_finished=False,
+                    finish_reason="",
+                    usage=None,
+                )
+            elif output_item.get("type") == "message":
+                pass
+            elif output_item.get("type") == "reasoning":
+                pass
+            else:
+                raise ValueError(f"Chat provider: Invalid output_item  {output_item}")
+        elif event_type == "response.function_call_arguments.delta":
+            content_part: Optional[str] = parsed_chunk.get("delta", None)
+            if content_part:
+                return GenericStreamingChunk(
+                    text="",
+                    tool_use=ChatCompletionToolCallChunk(
+                        id=None,
+                        index=0,
+                        type="function",
+                        function=ChatCompletionToolCallFunctionChunk(
+                            name=None, arguments=content_part
+                        ),
+                    ),
+                    is_finished=False,
+                    finish_reason="",
+                    usage=None,
+                )
+            else:
+                raise ValueError(
+                    f"Chat provider: Invalid function argument delta {parsed_chunk}"
+                )
+        elif event_type == "response.output_item.done":
+            # New output item added
+            output_item = parsed_chunk.get("item", {})
+            if output_item.get("type") == "function_call":
+                return GenericStreamingChunk(
+                    text="",
+                    tool_use=ChatCompletionToolCallChunk(
+                        id=output_item.get("call_id"),
+                        index=0,
+                        type="function",
+                        function=ChatCompletionToolCallFunctionChunk(
+                            name=parsed_chunk.get("name", None),
+                            arguments="",  # responses API sends everything again, we don't
+                        ),
+                    ),
+                    is_finished=True,
+                    finish_reason="tool_calls",
+                    usage=None,
+                )
+            elif output_item.get("type") == "message":
+                return GenericStreamingChunk(
+                    finish_reason="stop", is_finished=True, usage=None, text=""
+                )
+            elif output_item.get("type") == "reasoning":
+                pass
+            else:
+                raise ValueError(f"Chat provider: Invalid output_item  {output_item}")
+
+        elif event_type == "response.output_text.delta":
+            # Content part added to output
+            content_part = parsed_chunk.get("delta", None)
+            if content_part is not None:
+                return GenericStreamingChunk(
+                    text=content_part,
+                    tool_use=None,
+                    is_finished=False,
+                    finish_reason="",
+                    usage=None,
+                )
+            else:
+                raise ValueError(f"Chat provider: Invalid text delta {parsed_chunk}")
+        else:
+            pass
+        # For any unhandled event types, create a minimal valid chunk or skip
+        verbose_logger.debug(
+            f"Chat provider: Unhandled event type '{event_type}', creating empty chunk"
+        )
+
+        # Return a minimal valid chunk for unknown events
+        return GenericStreamingChunk(
+            text="", tool_use=None, is_finished=False, finish_reason="", usage=None
+        )
@@ -0,0 +1,54 @@
+"""
+Bridge for transforming API requests to another API requests
+"""
+
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any, AsyncIterator, Iterator, List, Optional, Union
+
+if TYPE_CHECKING:
+    from pydantic import BaseModel
+
+    from litellm import LiteLLMLoggingObj, ModelResponse
+    from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
+    from litellm.types.llms.openai import AllMessageValues
+
+
+class CompletionTransformationBridge(ABC):
+    @abstractmethod
+    def transform_request(
+        self,
+        model: str,
+        messages: List["AllMessageValues"],
+        optional_params: dict,
+        litellm_params: dict,
+        headers: dict,
+    ) -> dict:
+        """Transform /chat/completions api request to another request"""
+        pass
+
+    @abstractmethod
+    def transform_response(
+        self,
+        model: str,
+        raw_response: "BaseModel",  # the response from the other API
+        model_response: "ModelResponse",
+        logging_obj: "LiteLLMLoggingObj",
+        request_data: dict,
+        messages: List["AllMessageValues"],
+        optional_params: dict,
+        litellm_params: dict,
+        encoding: Any,
+        api_key: Optional[str] = None,
+        json_mode: Optional[bool] = None,
+    ) -> "ModelResponse":
+        """Transform another response to /chat/completions api response"""
+        pass
+
+    @abstractmethod
+    def get_model_response_iterator(
+        self,
+        streaming_response: Union[Iterator[str], AsyncIterator[str], "ModelResponse"],
+        sync_stream: bool,
+        json_mode: Optional[bool] = False,
+    ) -> "BaseModelResponseIterator":
+        pass
@@ -2447,10 +2447,7 @@ class BaseLLMHTTPHandler:
        _is_async: bool = False,
        fake_stream: bool = False,
        litellm_metadata: Optional[Dict[str, Any]] = None,
-    ) -> Union[
-        ImageResponse,
-        Coroutine[Any, Any, ImageResponse],
-    ]:
+    ) -> Union[ImageResponse, Coroutine[Any, Any, ImageResponse],]:
        """

        Handles image edit requests.
@@ -8,16 +8,28 @@
 - async_streaming
 """

-from typing import Any, AsyncIterator, Callable, Iterator, Optional, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncIterator,
+    Callable,
+    Coroutine,
+    Iterator,
+    Optional,
+    Union,
+)

 import httpx

 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
 from litellm.types.utils import GenericStreamingChunk
-from litellm.utils import ImageResponse, ModelResponse, EmbeddingResponse
+from litellm.utils import EmbeddingResponse, ImageResponse, ModelResponse

 from .base import BaseLLM

+if TYPE_CHECKING:
+    from litellm import CustomStreamWrapper
+

 class CustomLLMError(Exception):  # use this for all your exceptions
    def __init__(
@@ -54,7 +66,7 @@ class CustomLLM(BaseLLM):
        headers={},
        timeout: Optional[Union[float, httpx.Timeout]] = None,
        client: Optional[HTTPHandler] = None,
-    ) -> ModelResponse:
+    ) -> Union[ModelResponse, "CustomStreamWrapper"]:
        raise CustomLLMError(status_code=500, message="Not implemented yet!")

    def streaming(
@@ -96,7 +108,10 @@ class CustomLLM(BaseLLM):
        headers={},
        timeout: Optional[Union[float, httpx.Timeout]] = None,
        client: Optional[AsyncHTTPHandler] = None,
-    ) -> ModelResponse:
+    ) -> Union[
+        Coroutine[Any, Any, Union[ModelResponse, "CustomStreamWrapper"]],
+        Union[ModelResponse, "CustomStreamWrapper"],
+    ]:
        raise CustomLLMError(status_code=500, message="Not implemented yet!")

    async def astreaming(
@@ -86,6 +86,7 @@ from litellm.utils import (
    CustomStreamWrapper,
    ProviderConfigManager,
    Usage,
+    _get_model_info_helper,
    add_openai_metadata,
    add_provider_specific_params_to_optional_params,
    async_mock_completion_streaming_obj,
@@ -1287,6 +1288,36 @@ def completion(  # type: ignore # noqa: PLR0915
                timeout=timeout,
            )

+        ## RESPONSES API BRIDGE LOGIC ## - check if model has 'mode: responses' in litellm.model_cost map
+        try:
+            model_info = _get_model_info_helper(
+                model=model, custom_llm_provider=custom_llm_provider
+            )
+        except Exception as e:
+            verbose_logger.debug("Error getting model info: {}".format(e))
+            model_info = {}
+
+        if model_info.get("mode") == "responses":
+            from litellm.completion_extras import responses_api_bridge
+
+            return responses_api_bridge.completion(
+                model=model,
+                messages=messages,
+                headers=headers,
+                model_response=model_response,
+                api_key=api_key,
+                api_base=api_base,
+                acompletion=acompletion,
+                logging_obj=logging,
+                optional_params=optional_params,
+                litellm_params=litellm_params,
+                timeout=timeout,  # type: ignore
+                client=client,  # pass AsyncOpenAI, OpenAI client
+                custom_llm_provider=custom_llm_provider,
+                encoding=encoding,
+                stream=stream,
+            )
+
        if custom_llm_provider == "azure":
            # azure configs
            ## check dynamic params ##
@@ -2778,9 +2809,9 @@ def completion(  # type: ignore # noqa: PLR0915
                    "aws_region_name" not in optional_params
                    or optional_params["aws_region_name"] is None
                ):
-                    optional_params["aws_region_name"] = (
-                        aws_bedrock_client.meta.region_name
-                    )
+                    optional_params[
+                        "aws_region_name"
+                    ] = aws_bedrock_client.meta.region_name

            bedrock_route = BedrockModelInfo.get_bedrock_route(model)
            if bedrock_route == "converse":
@@ -4557,9 +4588,9 @@ def adapter_completion(
    new_kwargs = translation_obj.translate_completion_input_params(kwargs=kwargs)

    response: Union[ModelResponse, CustomStreamWrapper] = completion(**new_kwargs)  # type: ignore
-    translated_response: Optional[Union[BaseModel, AdapterCompletionStreamWrapper]] = (
-        None
-    )
+    translated_response: Optional[
+        Union[BaseModel, AdapterCompletionStreamWrapper]
+    ] = None
    if isinstance(response, ModelResponse):
        translated_response = translation_obj.translate_completion_output_params(
            response=response
@@ -5517,9 +5548,9 @@ def stream_chunk_builder(  # noqa: PLR0915
        ]

        if len(content_chunks) > 0:
-            response["choices"][0]["message"]["content"] = (
-                processor.get_combined_content(content_chunks)
-            )
+            response["choices"][0]["message"][
+                "content"
+            ] = processor.get_combined_content(content_chunks)

        reasoning_chunks = [
            chunk
@@ -5530,9 +5561,9 @@ def stream_chunk_builder(  # noqa: PLR0915
        ]

        if len(reasoning_chunks) > 0:
-            response["choices"][0]["message"]["reasoning_content"] = (
-                processor.get_combined_reasoning_content(reasoning_chunks)
-            )
+            response["choices"][0]["message"][
+                "reasoning_content"
+            ] = processor.get_combined_reasoning_content(reasoning_chunks)

        audio_chunks = [
            chunk
@@ -194,11 +194,11 @@ def responses(
        )

        # get provider config
-        responses_api_provider_config: Optional[BaseResponsesAPIConfig] = (
-            ProviderConfigManager.get_provider_responses_api_config(
-                model=model,
-                provider=litellm.LlmProviders(custom_llm_provider),
-            )
+        responses_api_provider_config: Optional[
+            BaseResponsesAPIConfig
+        ] = ProviderConfigManager.get_provider_responses_api_config(
+            model=model,
+            provider=litellm.LlmProviders(custom_llm_provider),
        )

        local_vars.update(kwargs)
@@ -388,11 +388,11 @@ def delete_responses(
            raise ValueError("custom_llm_provider is required but passed as None")

        # get provider config
-        responses_api_provider_config: Optional[BaseResponsesAPIConfig] = (
-            ProviderConfigManager.get_provider_responses_api_config(
-                model=None,
-                provider=litellm.LlmProviders(custom_llm_provider),
-            )
+        responses_api_provider_config: Optional[
+            BaseResponsesAPIConfig
+        ] = ProviderConfigManager.get_provider_responses_api_config(
+            model=None,
+            provider=litellm.LlmProviders(custom_llm_provider),
        )

        if responses_api_provider_config is None:
@@ -567,11 +567,11 @@ def get_responses(
            raise ValueError("custom_llm_provider is required but passed as None")

        # get provider config
-        responses_api_provider_config: Optional[BaseResponsesAPIConfig] = (
-            ProviderConfigManager.get_provider_responses_api_config(
-                model=None,
-                provider=litellm.LlmProviders(custom_llm_provider),
-            )
+        responses_api_provider_config: Optional[
+            BaseResponsesAPIConfig
+        ] = ProviderConfigManager.get_provider_responses_api_config(
+            model=None,
+            provider=litellm.LlmProviders(custom_llm_provider),
        )

        if responses_api_provider_config is None:
@@ -723,11 +723,11 @@ def list_input_items(
        if custom_llm_provider is None:
            raise ValueError("custom_llm_provider is required but passed as None")

-        responses_api_provider_config: Optional[BaseResponsesAPIConfig] = (
-            ProviderConfigManager.get_provider_responses_api_config(
-                model=None,
-                provider=litellm.LlmProviders(custom_llm_provider),
-            )
+        responses_api_provider_config: Optional[
+            BaseResponsesAPIConfig
+        ] = ProviderConfigManager.get_provider_responses_api_config(
+            model=None,
+            provider=litellm.LlmProviders(custom_llm_provider),
        )

        if responses_api_provider_config is None:
@@ -231,9 +231,15 @@ class ResponseAPILoggingUtils:

    @staticmethod
    def _transform_response_api_usage_to_chat_usage(
-        usage: Union[dict, ResponseAPIUsage],
+        usage: Optional[Union[dict, ResponseAPIUsage]],
    ) -> Usage:
        """Tranforms the ResponseAPIUsage object to a Usage object"""
+        if usage is None:
+            return Usage(
+                prompt_tokens=0,
+                completion_tokens=0,
+                total_tokens=0,
+            )
        response_api_usage: ResponseAPIUsage = (
            ResponseAPIUsage(**usage) if isinstance(usage, dict) else usage
        )
@@ -927,6 +927,9 @@ class ComputerToolParam(TypedDict, total=False):
    type: Required[Union[Literal["computer_use_preview"], str]]


+ALL_RESPONSES_API_TOOL_PARAMS = Union[ToolParam, ComputerToolParam]
+
+
 class ResponsesAPIOptionalRequestParams(TypedDict, total=False):
    """TypedDict for Optional parameters supported by the responses API."""

@@ -943,7 +946,7 @@ class ResponsesAPIOptionalRequestParams(TypedDict, total=False):
    temperature: Optional[float]
    text: Optional[ResponseTextConfigParam]
    tool_choice: Optional[ToolChoice]
-    tools: Optional[List[Union[ToolParam, ComputerToolParam]]]
+    tools: Optional[List[ALL_RESPONSES_API_TOOL_PARAMS]]
    top_p: Optional[float]
    truncation: Optional[Literal["auto", "disabled"]]
    user: Optional[str]
@@ -469,3 +469,55 @@ async def test_openai_pdf_url(model):

    assert "file_data" in request["raw_request_body"]["messages"][0]["content"][1]["file"]

+
+@pytest.mark.parametrize("sync_mode", [True, False])
+@pytest.mark.asyncio
+async def test_openai_codex_stream(sync_mode):
+    from litellm.main import stream_chunk_builder
+    kwargs = {
+        "model": "openai/codex-mini-latest",
+        "messages": [{"role": "user", "content": "Hey!"}],
+        "stream": True,
+    }
+
+    chunks = []
+    if sync_mode:
+        response = litellm.completion(
+            **kwargs
+        )
+        for chunk in response:
+            chunks.append(chunk)
+    else:
+        response = await litellm.acompletion(
+            **kwargs
+        )
+        async for chunk in response:
+            chunks.append(chunk)
+    
+    complete_response = stream_chunk_builder(chunks=chunks)
+    print("complete_response: ", complete_response)
+
+    assert complete_response.choices[0].message.content is not None
+
+@pytest.mark.parametrize("sync_mode", [True, False])
+@pytest.mark.asyncio
+async def test_openai_codex(sync_mode):
+    
+    from litellm import acompletion
+
+    kwargs = {
+        "model": "openai/codex-mini-latest",
+        "messages": [{"role": "user", "content": "Hey!"}],
+    }
+
+    if sync_mode:
+        response = litellm.completion(
+            **kwargs
+        )
+    else:
+        response = await litellm.acompletion(
+            **kwargs
+        )
+    print("response: ", response)
+
+    assert response.choices[0].message.content is not None