fix(anthropic): emit tool_use content_block_start without awaiting the next chunk (#37310)

On /v1/messages, AnthropicStreamWrapper synthesizes the content_block_start for
the first content block, queues it, then hits a bare `continue` when that same
upstream chunk's translated delta is empty. The queue is only drained at the top
of the next __next__ / __anext__, so the queued content_block_start waits for a
further upstream chunk to arrive.

An empty delta on the opening chunk is the normal tool-call shape: Bedrock
Converse's contentBlockStart carries the tool id and name with no arguments, and
OpenAI-format streams send arguments: "" on the chunk that names the function.
So a client learns a tool call started one upstream event late, and when the
provider delivers argument fragments as a trailing burst it sees nothing at all
after message_start for the whole generation.

Flush the queued event before continuing, in both the sync and async paths. The
sibling block-transition path already returns from the queue, so only the
first-block-open case changed.

litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py

@@ -624,6 +624,12 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
                     return self.chunk_queue.popleft()
 
                 if processed_chunk["type"] == "content_block_delta" and not self._delta_has_content(processed_chunk):
+                    # A tool_use block opens with empty arguments (Bedrock Converse's
+                    # ``contentBlockStart``, OpenAI's ``arguments: ""``), so flush the
+                    # block start queued above instead of waiting for the next upstream
+                    # chunk, which on a trailing-burst provider is the whole generation.
+                    if self.chunk_queue:
+                        return self.chunk_queue.popleft()
                     continue
 
                 if processed_chunk["type"] == "message_delta" and self.sent_content_block_finish is False:
@@ -847,6 +853,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
                     if processed_chunk["type"] == "content_block_delta" and not self._delta_has_content(
                         processed_chunk
                     ):
+                        # See ``__next__``: flush the queued block start (issue #32004).
+                        if self.chunk_queue:
+                            return self.chunk_queue.popleft()
                         continue
 
                     if processed_chunk["type"] == "message_delta" and self.sent_content_block_finish is False:

tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py

@@ -916,3 +916,117 @@ def test_mixed_finish_chunk_emits_usage_once_sync():
     assert message_deltas[0]["usage"]["output_tokens"] == 7
     assert _text_deltas(events) == ["Hi"]
     _assert_deltas_match_their_block_type(events)
+
+
+class _CountingSyncStream:
+    """Sync stream recording how many upstream chunks have been pulled."""
+
+    def __init__(self, items: List[MagicMock]):
+        self._items = list(items)
+        self.pulled = 0
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self.pulled >= len(self._items):
+            raise StopIteration
+        item = self._items[self.pulled]
+        self.pulled += 1
+        return item
+
+
+class _CountingAsyncStream(_CountingSyncStream):
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        try:
+            return next(self)
+        except StopIteration:
+            raise StopAsyncIteration
+
+
+def _bedrock_tool_open_then_args() -> List[MagicMock]:
+    """The Bedrock Converse shape: ``contentBlockStart`` names the tool and
+    carries empty arguments, the arguments arrive in later events.
+    """
+    return [
+        _tool_chunk("call_1", "Write", ""),
+        _tool_chunk("call_1", None, '{"file_text":'),
+        _tool_chunk("call_1", None, ' "hello"}'),
+        _make_chunk(Delta(content=None), finish_reason="tool_calls"),
+    ]
+
+
+def test_tool_block_start_emitted_without_awaiting_the_next_chunk_sync():
+    """Regression test for issue #32004.
+
+    A tool_use block opened by a chunk whose delta is empty (Bedrock Converse
+    sends the tool id/name and its arguments in separate events) must emit
+    ``content_block_start`` off that chunk alone. Holding it until the next
+    upstream chunk arrives means a provider that delivers tool arguments as a
+    trailing burst leaves the client with nothing after ``message_start`` for
+    the whole generation, tripping client and load-balancer idle timeouts.
+    """
+    stream = _CountingSyncStream(_bedrock_tool_open_then_args())
+    wrapper = AnthropicStreamWrapper(completion_stream=stream, model="claude-x")
+
+    assert next(wrapper)["type"] == "message_start"
+    assert stream.pulled == 0
+
+    start = next(wrapper)
+    assert start["type"] == "content_block_start"
+    assert start["content_block"] == {
+        "type": "tool_use",
+        "id": "call_1",
+        "name": "Write",
+        "input": {},
+    }
+    assert stream.pulled == 1, (
+        f"content_block_start was withheld until {stream.pulled} upstream chunks had arrived"
+    )
+
+
+@pytest.mark.asyncio
+async def test_tool_block_start_emitted_without_awaiting_the_next_chunk_async():
+    """Async twin of the sync regression test above (issue #32004)."""
+    stream = _CountingAsyncStream(_bedrock_tool_open_then_args())
+    wrapper = AnthropicStreamWrapper(completion_stream=stream, model="claude-x")
+
+    assert (await wrapper.__anext__())["type"] == "message_start"
+    assert stream.pulled == 0
+
+    start = await wrapper.__anext__()
+    assert start["type"] == "content_block_start"
+    assert start["content_block"]["name"] == "Write"
+    assert stream.pulled == 1, (
+        f"content_block_start was withheld until {stream.pulled} upstream chunks had arrived"
+    )
+
+
+@pytest.mark.parametrize("is_async", [False, True])
+@pytest.mark.asyncio
+async def test_tool_block_start_flush_does_not_duplicate_or_drop_events(is_async: bool):
+    """Flushing the queued ``content_block_start`` early must not duplicate it,
+    lose the empty opening delta's successors, or break event ordering.
+    """
+    chunks = _bedrock_tool_open_then_args()
+    if is_async:
+        wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="claude-x")
+        events = await _drain_async(wrapper)
+    else:
+        wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x")
+        events = _drain_sync(wrapper)
+
+    assert [e["type"] for e in events] == [
+        "message_start",
+        "content_block_start",
+        "content_block_delta",
+        "content_block_delta",
+        "content_block_stop",
+        "message_delta",
+        "message_stop",
+    ]
+    assert _input_json_deltas(events) == ['{"file_text":', ' "hello"}']
+    _assert_deltas_match_their_block_type(events)