From fbf9da9af0ce246204501810071d6ce32308c7aa Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sun, 26 Jul 2026 08:26:22 +0000 Subject: [PATCH] fix(anthropic): correct streaming tool_use framing for /v1/messages When a provider such as ollama_chat streams a tool call and then a plain stop finish_reason through the Anthropic /v1/messages adapter, the SSE framing was wrong in two ways: the turn was prefixed with a spurious empty text content block, and the final message_delta carried stop_reason end_turn instead of tool_use even though a tool_use block was streamed. Anthropic tool-runners key off stop_reason tool_use to decide whether to execute the tool, so the call was silently dropped. Open the first content block lazily from the first chunk that carries content, and override a translated end_turn to tool_use when a tool_use block was emitted. --- .../adapters/streaming_iterator.py | 155 ++++++++++++++---- .../test_streaming_iterator_first_delta.py | 115 +++++++++++++ .../messages/test_parallel_tool_calls.py | 24 +-- 3 files changed, 238 insertions(+), 56 deletions(-) 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 f02333c34c8..032462f79a4 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -161,6 +161,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): sent_content_block_start: bool = False sent_content_block_finish: bool = False current_content_block_type: Literal["text", "tool_use", "thinking"] = "text" + emitted_tool_use_block: bool = False sent_last_message: bool = False holding_chunk: Optional[Any] = None holding_stop_reason_chunk: Optional[Any] = None @@ -258,6 +259,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): held-chunk flush path stays in sync with the merge path's guarantee when ``self.applied_edits`` is non-empty. """ + message_delta_chunk = self._override_stop_reason_for_tool_use(message_delta_chunk) message_delta_chunk = self._ensure_context_management_attached(message_delta_chunk) if self.iterations_usage is None: return message_delta_chunk @@ -393,22 +395,17 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if compaction_event is not None: return compaction_event - if self.sent_content_block_start is False: - self.sent_content_block_start = True - self.sent_content_block_finish = False - self.chunk_queue.append( - { - "type": "content_block_start", - "index": self.current_content_block_index, - "content_block": {"type": "text", "text": ""}, - } - ) - return self.chunk_queue.popleft() - for chunk in self.completion_stream: if chunk == "None" or chunk is None: raise Exception + if self.sent_content_block_start is False: + first_block_action = self._open_first_content_block(chunk) + if first_block_action == "skip": + continue + if first_block_action == "opened": + return self.chunk_queue.popleft() + should_start_new_block = self._should_start_new_content_block(chunk) if should_start_new_block: self._increment_content_block_index() @@ -615,22 +612,17 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if compaction_event is not None: return compaction_event - if self.sent_content_block_start is False: - self.sent_content_block_start = True - self.sent_content_block_finish = False - self.chunk_queue.append( - { - "type": "content_block_start", - "index": self.current_content_block_index, - "content_block": {"type": "text", "text": ""}, - } - ) - return self.chunk_queue.popleft() - async for chunk in self.completion_stream: if chunk == "None" or chunk is None: raise Exception + if self.sent_content_block_start is False: + first_block_action = self._open_first_content_block(chunk) + if first_block_action == "skip": + continue + if first_block_action == "opened": + return self.chunk_queue.popleft() + # Check if we need to start a new content block should_start_new_block = self._should_start_new_content_block(chunk) if should_start_new_block: @@ -900,18 +892,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): ) # Restore original tool name if it was truncated for OpenAI's 64-char limit + self._restore_tool_name(block_type, content_block_start) if block_type == "tool_use": - # Type narrowing: content_block_start is ToolUseBlock when block_type is "tool_use" - from typing import cast - - from litellm.types.llms.anthropic import ToolUseBlock - - tool_block = cast(ToolUseBlock, content_block_start) - - if tool_block.get("name"): - truncated_name = tool_block["name"] - original_name = self.tool_name_mapping.get(truncated_name, truncated_name) - tool_block["name"] = original_name + self.emitted_tool_use_block = True if block_type != self.current_content_block_type: self.current_content_block_type = block_type @@ -932,3 +915,105 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): return True return False + + def _restore_tool_name( + self, + block_type: Literal["text", "tool_use", "thinking"], + content_block_start: "AnthropicStreamWrapper.ContentBlockContentBlockDict", + ) -> None: + """Restore a tool_use block's original name when it was truncated to fit + OpenAI's 64-char tool-name limit. No-op for non-tool blocks. + """ + if block_type != "tool_use": + return + + from typing import cast + + from litellm.types.llms.anthropic import ToolUseBlock + + tool_block = cast(ToolUseBlock, content_block_start) + truncated_name = tool_block.get("name") + if truncated_name: + tool_block["name"] = self.tool_name_mapping.get(truncated_name, truncated_name) + + def _open_first_content_block(self, chunk: "ModelResponseStream") -> Literal["skip", "opened", "fell_through"]: + """Open the very first Anthropic content block lazily, from the first + chunk that actually carries content, so a turn that starts with a + tool_use (or thinking) block is not prefixed with a spurious empty + ``{"type": "text", "text": ""}`` block. + + Returns one of: + - ``"skip"``: the chunk carries no content yet (e.g. a role-only + delta); no block was opened, the caller should keep iterating. + - ``"opened"``: a ``content_block_start`` (and, when the trigger chunk + bundles the first delta, that delta) was queued; the caller should + return the queued head. + - ``"fell_through"``: the stream ended before any content arrived, so + an empty text block was opened to keep the message well-formed; the + caller should continue into the normal ``message_delta`` handling. + """ + from .transformation import LiteLLMAnthropicMessagesAdapter + + if chunk.choices[0].finish_reason is not None: + self.sent_content_block_start = True + self.sent_content_block_finish = False + self.chunk_queue.append( + { + "type": "content_block_start", + "index": self.current_content_block_index, + "content_block": {"type": "text", "text": ""}, + } + ) + return "fell_through" + + adapter = LiteLLMAnthropicMessagesAdapter() + ( + block_type, + content_block_start, + ) = adapter._translate_streaming_openai_chunk_to_anthropic_content_block(choices=chunk.choices) + processed_chunk = adapter.translate_streaming_openai_response_to_anthropic( + response=chunk, + current_content_block_index=self.current_content_block_index, + ) + if block_type == "text" and not self._delta_has_content(processed_chunk): + return "skip" + + self._restore_tool_name(block_type, content_block_start) + if block_type == "tool_use": + self.emitted_tool_use_block = True + + self.current_content_block_type = block_type + self.current_content_block_start = content_block_start + self.sent_content_block_start = True + self.sent_content_block_finish = False + self.chunk_queue.append( + { + "type": "content_block_start", + "index": self.current_content_block_index, + "content_block": content_block_start, + } + ) + if self._delta_has_content(processed_chunk): + self.chunk_queue.append(processed_chunk) + return "opened" + + def _override_stop_reason_for_tool_use(self, message_delta_chunk: dict[str, Any]) -> dict[str, Any]: + """Force ``stop_reason: "tool_use"`` on the final ``message_delta`` when + the streamed message contains a tool_use block but the upstream + finish_reason mapped to ``end_turn``. + + Some providers (e.g. ``ollama_chat``) stream the tool call in an earlier + chunk and then send a plain ``stop`` finish_reason in the terminating + chunk, which maps to ``end_turn``. Anthropic-Messages tool-runners key + off ``stop_reason == "tool_use"`` to decide whether to execute the + emitted ``tool_use`` block, so this keeps the streaming framing in sync + with the non-streaming response and the Anthropic spec. + """ + if not self.emitted_tool_use_block: + return message_delta_chunk + delta = message_delta_chunk.get("delta") + if not isinstance(delta, dict) or delta.get("stop_reason") != "end_turn": + return message_delta_chunk + augmented = message_delta_chunk.copy() + augmented["delta"] = {**delta, "stop_reason": "tool_use"} + return augmented 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 19ec1a04b45..a056e2eba7e 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 @@ -506,3 +506,118 @@ def test_empty_content_chunk_mid_text_block_is_suppressed_sync(): assert _text_deltas(events) == ["Hi", " there"] _assert_deltas_match_their_block_type(events) + + +# --------------------------------------------------------------------------- +# Regression tests for issue #34692. +# +# When a provider (e.g. ``ollama_chat``) streams a tool call and then a plain +# ``stop`` finish_reason, the Anthropic ``/v1/messages`` framing was wrong in +# two ways: the turn was prefixed with a spurious empty ``{"type": "text", +# "text": ""}`` content block, and the final ``message_delta`` carried +# ``stop_reason: "end_turn"`` instead of ``"tool_use"`` even though a tool_use +# block was streamed. Anthropic tool-runners key off ``stop_reason == +# "tool_use"`` to decide whether to execute the tool, so the tool call was +# silently dropped downstream. +# --------------------------------------------------------------------------- + + +def _content_block_starts(events: List[dict]) -> List[dict]: + return [e for e in events if e.get("type") == "content_block_start"] + + +def _final_stop_reason(events: List[dict]) -> Optional[str]: + for event in events: + if event.get("type") == "message_delta": + return event["delta"].get("stop_reason") + return None + + +def _ollama_style_tool_then_stop_chunks() -> List[MagicMock]: + """``ollama_chat`` streams the whole tool call in a chunk with no + finish_reason, then a terminating chunk whose finish_reason is a plain + ``stop`` (the tool_calls are not repeated on the terminating chunk). + """ + return [ + _tool_chunk("call_1", "get_weather", '{"city": "SF"}'), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + + +def test_tool_first_turn_has_no_leading_empty_text_block_sync(): + wrapper = AnthropicStreamWrapper( + completion_stream=iter(_ollama_style_tool_then_stop_chunks()), model="qwen3" + ) + events = _drain_sync(wrapper) + + starts = _content_block_starts(events) + assert [s["content_block"]["type"] for s in starts] == ["tool_use"] + assert starts[0]["index"] == 0 + assert _input_json_deltas(events) == ['{"city": "SF"}'] + + +def test_stop_reason_is_tool_use_when_tool_streamed_before_plain_stop_sync(): + wrapper = AnthropicStreamWrapper( + completion_stream=iter(_ollama_style_tool_then_stop_chunks()), model="qwen3" + ) + assert _final_stop_reason(_drain_sync(wrapper)) == "tool_use" + + +@pytest.mark.asyncio +async def test_tool_first_turn_has_no_leading_empty_text_block_async(): + wrapper = AnthropicStreamWrapper( + completion_stream=_AsyncStream(_ollama_style_tool_then_stop_chunks()), + model="qwen3", + ) + events = await _drain_async(wrapper) + + starts = _content_block_starts(events) + assert [s["content_block"]["type"] for s in starts] == ["tool_use"] + assert _final_stop_reason(events) == "tool_use" + + +def test_max_tokens_stop_reason_is_preserved_even_with_tool_use_sync(): + """The override only rewrites ``end_turn`` -> ``tool_use``; a genuine + ``max_tokens`` truncation mid-tool-call must survive so clients can tell + the turn was cut short. + """ + chunks = [ + _tool_chunk("call_1", "get_weather", '{"city": "S'), + _make_chunk(Delta(content=None), finish_reason="length"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="qwen3") + assert _final_stop_reason(_drain_sync(wrapper)) == "max_tokens" + + +def test_plain_text_turn_keeps_end_turn_and_leading_text_block_sync(): + """A text-only turn must be unaffected: it opens a single text block and + ends with ``end_turn`` (no tool_use override, no dropped block). + """ + chunks = [ + _make_chunk(Delta(content="Hello")), + _make_chunk(Delta(content=" there")), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="qwen3") + events = _drain_sync(wrapper) + + starts = _content_block_starts(events) + assert [s["content_block"]["type"] for s in starts] == ["text"] + assert _text_deltas(events) == ["Hello", " there"] + assert _final_stop_reason(events) == "end_turn" + + +def test_thinking_first_turn_has_no_leading_empty_text_block_sync(): + """A reasoning-first turn must open a ``thinking`` block directly rather + than a spurious empty text block ahead of it. + """ + chunks = [ + _thinking_chunk("Let me think"), + _make_chunk(Delta(content="Answer")), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="qwen3") + events = _drain_sync(wrapper) + + starts = _content_block_starts(events) + assert [s["content_block"]["type"] for s in starts] == ["thinking", "text"] diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py index e44413cf837..533b3ac95fa 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py @@ -134,13 +134,6 @@ def test_anthropic_stream_wrapper_single_tool_call(): # Verify the expected sequence of chunk types expected_types = [ "message_start", # Initial message start - # TODO: for future contributors: if the initial content_block_start - # respects the upstream's starting chunk, the initial empty text block - # should be removed (and this test should be updated accordingly) - # --------------------------------------------------------------------- - "content_block_start", # Initial empty text block start - "content_block_stop", # End of empty text block - # --------------------------------------------------------------------- "content_block_start", # Start of first tool_use content block "content_block_delta", # {"city": "content_block_delta", # "NY"} @@ -151,6 +144,9 @@ def test_anthropic_stream_wrapper_single_tool_call(): assert expected_types == chunk_types + message_delta = next(c for c in chunks if c.get("type") == "message_delta") + assert message_delta["delta"]["stop_reason"] == "tool_use" + get_weather_calls = 0 for chunk in chunks: @@ -196,13 +192,6 @@ def test_anthropic_stream_wrapper_back_to_back_tool_calls(): # Verify the expected sequence of chunk types expected_types = [ "message_start", # Initial message start - # TODO: for future contributors: if the initial content_block_start - # respects the upstream's starting chunk, the initial empty text block - # should be removed (and this test should be updated accordingly) - # --------------------------------------------------------------------- - "content_block_start", # Initial empty text block start - "content_block_stop", # End of empty text block - # --------------------------------------------------------------------- "content_block_start", # Start of first tool_use content block "content_block_delta", # {"city": "content_block_delta", # "NY"} @@ -267,13 +256,6 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text(): # Verify the expected sequence of chunk types expected_types = [ "message_start", # Initial message start - # TODO: for future contributors: if the initial content_block_start - # respects the upstream's starting chunk, the initial empty text block - # should be removed (and this test should be updated accordingly) - # --------------------------------------------------------------------- - "content_block_start", # Initial empty text block start - "content_block_stop", # End of empty text block - # --------------------------------------------------------------------- "content_block_start", # Start of first tool_use content block "content_block_delta", # {"city": "content_block_delta", # "NY"}