fix(chatgpt): normalize streaming tool_call indices and deduplicate closing chunks

The ChatGPT backend API sends non-spec-compliant streaming tool call chunks
where index is always 0 for parallel tool calls and id/name get repeated in
duplicate closing chunks. Add ChatGPTToolCallNormalizer to fix indices and
filter duplicates before they reach the consumer.

Fixes #21482

litellm/completion_extras/litellm_responses_transformation/handler.py

@@ -221,7 +221,9 @@ class ResponsesToCompletionBridgeHandler:
                 custom_llm_provider=custom_llm_provider,
                 logging_obj=logging_obj,
             )
-            return streamwrapper
+            return self._apply_post_stream_processing(
+                streamwrapper, model, custom_llm_provider
+            )
 
     async def acompletion(
         self, *args, **kwargs
@@ -300,7 +302,30 @@ class ResponsesToCompletionBridgeHandler:
                 custom_llm_provider=custom_llm_provider,
                 logging_obj=logging_obj,
             )
-            return streamwrapper
+            return self._apply_post_stream_processing(
+                streamwrapper, model, custom_llm_provider
+            )
+
+    @staticmethod
+    def _apply_post_stream_processing(
+        stream: "CustomStreamWrapper",
+        model: str,
+        custom_llm_provider: str,
+    ) -> Any:
+        """Apply provider-specific post-stream processing if available."""
+        from litellm.types.utils import LlmProviders
+        from litellm.utils import ProviderConfigManager
+
+        try:
+            provider_config = ProviderConfigManager.get_provider_chat_config(
+                model=model, provider=LlmProviders(custom_llm_provider)
+            )
+        except (ValueError, KeyError):
+            return stream
+
+        if provider_config is not None:
+            return provider_config.post_stream_processing(stream)
+        return stream
 
 
 responses_api_bridge = ResponsesToCompletionBridgeHandler()

litellm/llms/base_llm/chat/transformation.py

@@ -438,6 +438,10 @@ class BaseConfig(ABC):
         """
         return True
 
+    def post_stream_processing(self, stream: Any) -> Any:
+        """Hook for providers to post-process streaming responses. Default: pass-through."""
+        return stream
+
     def calculate_additional_costs(
         self, model: str, prompt_tokens: int, completion_tokens: int
     ) -> Optional[dict]:

litellm/llms/chatgpt/chat/streaming_utils.py

@@ -0,0 +1,83 @@
+"""
+Streaming utilities for ChatGPT provider.
+
+Normalizes non-spec-compliant tool_call chunks from the ChatGPT backend API.
+"""
+
+from typing import Any
+
+
+class ChatGPTToolCallNormalizer:
+    """
+    Wraps a streaming response and fixes tool_call index/dedup issues.
+
+    The ChatGPT backend API (chatgpt.com/backend-api) sends non-spec-compliant
+    streaming tool call chunks:
+    1. `index` is always 0, even for multiple parallel tool calls
+    2. `id` and `name` get repeated in "closing" chunks that shouldn't exist
+
+    This wrapper normalizes the stream to match the OpenAI spec before yielding
+    chunks to the consumer.
+    """
+
+    def __init__(self, stream: Any):
+        self._stream = stream
+        self._seen_ids: dict[str, int] = {}  # tool_call_id -> assigned_index
+        self._next_index: int = 0
+        self._last_id: str | None = None  # tracks which tool call the next delta belongs to
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._stream, name)
+
+    def __iter__(self):
+        return self
+
+    def __aiter__(self):
+        return self
+
+    def __next__(self):
+        while True:
+            chunk = next(self._stream)
+            result = self._normalize(chunk)
+            if result is not None:
+                return result
+
+    async def __anext__(self):
+        while True:
+            chunk = await self._stream.__anext__()
+            result = self._normalize(chunk)
+            if result is not None:
+                return result
+
+    def _normalize(self, chunk: Any) -> Any:
+        """Fix tool_calls in the chunk. Returns None to skip duplicate chunks."""
+        if not chunk.choices:
+            return chunk
+
+        delta = chunk.choices[0].delta
+        if delta is None or not delta.tool_calls:
+            return chunk
+
+        normalized = []
+        for tc in delta.tool_calls:
+            if tc.id and tc.id not in self._seen_ids:
+                # New tool call — assign correct index
+                self._seen_ids[tc.id] = self._next_index
+                tc.index = self._next_index
+                self._last_id = tc.id
+                self._next_index += 1
+                normalized.append(tc)
+            elif tc.id and tc.id in self._seen_ids:
+                # Duplicate "closing" chunk — skip it
+                continue
+            else:
+                # Continuation delta (id=None) — fix index
+                if self._last_id:
+                    tc.index = self._seen_ids[self._last_id]
+                normalized.append(tc)
+
+        if not normalized:
+            return None  # all tool_calls were duplicates, skip chunk
+
+        delta.tool_calls = normalized
+        return chunk

litellm/llms/chatgpt/chat/transformation.py

@@ -1,4 +1,4 @@
-from typing import List, Optional, Tuple
+from typing import Any, List, Optional, Tuple
 
 from litellm.exceptions import AuthenticationError
 from litellm.llms.openai.openai import OpenAIConfig
@@ -10,6 +10,7 @@ from ..common_utils import (
     ensure_chatgpt_session_id,
     get_chatgpt_default_headers,
 )
+from .streaming_utils import ChatGPTToolCallNormalizer
 
 
 class ChatGPTConfig(OpenAIConfig):
@@ -61,6 +62,9 @@ class ChatGPTConfig(OpenAIConfig):
         )
         return {**default_headers, **validated_headers}
 
+    def post_stream_processing(self, stream: Any) -> Any:
+        return ChatGPTToolCallNormalizer(stream)
+
     def map_openai_params(
         self,
         non_default_params: dict,

tests/test_litellm/llms/chatgpt/chat/test_streaming_utils.py

@@ -0,0 +1,195 @@
+"""
+Tests for ChatGPTToolCallNormalizer.
+
+Verifies that non-spec-compliant tool_call chunks from the ChatGPT backend API
+are normalized to match the OpenAI streaming spec:
+- Correct index assignment for parallel tool calls
+- Deduplication of "closing" chunks with repeated id/name
+"""
+
+import pytest
+
+from litellm.llms.chatgpt.chat.streaming_utils import ChatGPTToolCallNormalizer
+from litellm.types.utils import (
+    ChatCompletionDeltaToolCall,
+    Delta,
+    Function,
+    ModelResponseStream,
+    StreamingChoices,
+)
+
+
+def _make_chunk(tool_calls=None, content=None):
+    """Helper to build a ModelResponseStream chunk with tool_calls on the delta."""
+    delta = Delta(
+        content=content,
+        role="assistant",
+        tool_calls=tool_calls,
+    )
+    choice = StreamingChoices(delta=delta, index=0)
+    return ModelResponseStream(choices=[choice])
+
+
+def _make_tc(index=0, id=None, name=None, arguments=None):
+    """Helper to build a ChatCompletionDeltaToolCall."""
+    func = Function(name=name, arguments=arguments)
+    return ChatCompletionDeltaToolCall(
+        index=index,
+        id=id,
+        function=func,
+        type="function" if id else None,
+    )
+
+
+class TestChatGPTToolCallNormalizer:
+    """Test that the normalizer fixes ChatGPT-style tool_call streaming issues."""
+
+    def test_single_tool_call_index_preserved(self):
+        """A single tool call should get index=0."""
+        chunks = [
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_1", name="get_weather")]),
+            _make_chunk(tool_calls=[_make_tc(index=0, arguments='{"loc')]),
+            _make_chunk(tool_calls=[_make_tc(index=0, arguments='ation": "NYC"}')]),
+        ]
+        normalizer = ChatGPTToolCallNormalizer(iter(chunks))
+        results = list(normalizer)
+
+        assert len(results) == 3
+        assert results[0].choices[0].delta.tool_calls[0].index == 0
+        assert results[0].choices[0].delta.tool_calls[0].id == "call_1"
+        assert results[1].choices[0].delta.tool_calls[0].index == 0
+        assert results[2].choices[0].delta.tool_calls[0].index == 0
+
+    def test_parallel_tool_calls_get_correct_indices(self):
+        """
+        ChatGPT sends all tool_calls with index=0. The normalizer should assign
+        sequential indices: 0 for the first, 1 for the second.
+        """
+        chunks = [
+            # First tool call: intro chunk with id + name
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_aaa", name="get_weather")]),
+            # First tool call: arguments streaming
+            _make_chunk(tool_calls=[_make_tc(index=0, arguments='{"location": "NYC"}')]),
+            # First tool call: duplicate closing chunk (id repeated) — should be skipped
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_aaa", name="get_weather")]),
+            # Second tool call: intro chunk with id + name (index=0 from ChatGPT)
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_bbb", name="get_time")]),
+            # Second tool call: arguments streaming
+            _make_chunk(tool_calls=[_make_tc(index=0, arguments='{"tz": "EST"}')]),
+            # Second tool call: duplicate closing chunk — should be skipped
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_bbb", name="get_time")]),
+        ]
+
+        normalizer = ChatGPTToolCallNormalizer(iter(chunks))
+        results = list(normalizer)
+
+        # 2 duplicate chunks should be skipped → 4 results
+        assert len(results) == 4
+
+        # First tool call chunks should have index=0
+        assert results[0].choices[0].delta.tool_calls[0].index == 0
+        assert results[0].choices[0].delta.tool_calls[0].id == "call_aaa"
+        assert results[1].choices[0].delta.tool_calls[0].index == 0
+
+        # Second tool call chunks should have index=1
+        assert results[2].choices[0].delta.tool_calls[0].index == 1
+        assert results[2].choices[0].delta.tool_calls[0].id == "call_bbb"
+        assert results[3].choices[0].delta.tool_calls[0].index == 1
+
+    def test_non_tool_call_chunks_pass_through(self):
+        """Chunks without tool_calls should pass through unchanged."""
+        chunks = [
+            _make_chunk(content="Hello"),
+            _make_chunk(content=" world"),
+        ]
+        normalizer = ChatGPTToolCallNormalizer(iter(chunks))
+        results = list(normalizer)
+
+        assert len(results) == 2
+        assert results[0].choices[0].delta.content == "Hello"
+        assert results[1].choices[0].delta.content == " world"
+
+    def test_empty_choices_pass_through(self):
+        """Chunks with empty choices should pass through."""
+        chunk = ModelResponseStream(choices=[])
+        normalizer = ChatGPTToolCallNormalizer(iter([chunk]))
+        results = list(normalizer)
+
+        assert len(results) == 1
+
+    def test_three_parallel_tool_calls(self):
+        """Three parallel tool calls should get indices 0, 1, 2."""
+        chunks = [
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_1", name="fn_a")]),
+            _make_chunk(tool_calls=[_make_tc(index=0, arguments='{"a":1}')]),
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_2", name="fn_b")]),
+            _make_chunk(tool_calls=[_make_tc(index=0, arguments='{"b":2}')]),
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_3", name="fn_c")]),
+            _make_chunk(tool_calls=[_make_tc(index=0, arguments='{"c":3}')]),
+        ]
+
+        normalizer = ChatGPTToolCallNormalizer(iter(chunks))
+        results = list(normalizer)
+
+        assert len(results) == 6
+        # First tool call
+        assert results[0].choices[0].delta.tool_calls[0].index == 0
+        assert results[1].choices[0].delta.tool_calls[0].index == 0
+        # Second tool call
+        assert results[2].choices[0].delta.tool_calls[0].index == 1
+        assert results[3].choices[0].delta.tool_calls[0].index == 1
+        # Third tool call
+        assert results[4].choices[0].delta.tool_calls[0].index == 2
+        assert results[5].choices[0].delta.tool_calls[0].index == 2
+
+    def test_all_duplicates_skipped(self):
+        """If a chunk contains only duplicate tool_calls, the entire chunk is skipped."""
+        chunks = [
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_x", name="fn")]),
+            # Duplicate — same id seen before
+            _make_chunk(tool_calls=[_make_tc(index=0, id="call_x", name="fn")]),
+        ]
+
+        normalizer = ChatGPTToolCallNormalizer(iter(chunks))
+        results = list(normalizer)
+
+        assert len(results) == 1
+        assert results[0].choices[0].delta.tool_calls[0].id == "call_x"
+
+    @pytest.mark.asyncio
+    async def test_async_iteration(self):
+        """The normalizer should work with async iteration."""
+
+        async def async_gen():
+            chunks = [
+                _make_chunk(tool_calls=[_make_tc(index=0, id="call_a", name="fn_a")]),
+                _make_chunk(tool_calls=[_make_tc(index=0, arguments='{"x":1}')]),
+                _make_chunk(tool_calls=[_make_tc(index=0, id="call_b", name="fn_b")]),
+                _make_chunk(tool_calls=[_make_tc(index=0, arguments='{"y":2}')]),
+            ]
+            for c in chunks:
+                yield c
+
+        normalizer = ChatGPTToolCallNormalizer(async_gen())
+        results = []
+        async for chunk in normalizer:
+            results.append(chunk)
+
+        assert len(results) == 4
+        assert results[0].choices[0].delta.tool_calls[0].index == 0
+        assert results[2].choices[0].delta.tool_calls[0].index == 1
+
+    def test_getattr_proxies_to_stream(self):
+        """Attribute access should be proxied to the underlying stream."""
+
+        class FakeStream:
+            custom_attr = "test_value"
+
+            def __iter__(self):
+                return iter([])
+
+            def __next__(self):
+                raise StopIteration
+
+        normalizer = ChatGPTToolCallNormalizer(FakeStream())
+        assert normalizer.custom_attr == "test_value"