From 49b72e14da973fd25ca551e6cda834905d7babb3 Mon Sep 17 00:00:00 2001 From: Yassin Kortam Date: Tue, 18 Aug 2026 14:42:42 -0700 Subject: [PATCH] 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. --- .../adapters/streaming_iterator.py | 9 ++ .../test_streaming_iterator_first_delta.py | 114 ++++++++++++++++++ 2 files changed, 123 insertions(+) diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py index 1660f56378f..30b5df1e4ee 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/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: diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py index 2eb8e077320..bd02c61752e 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py +++ b/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)