From df4213c619b4c977e430574bda4e9326f3a64767 Mon Sep 17 00:00:00 2001 From: ItsRoy69 Date: Sun, 12 Apr 2026 20:22:28 +0530 Subject: [PATCH] fix: emit input_json_delta for atomic tool call chunks in Anthropic streaming adapter Signed-off-by: ItsRoy69 --- .../adapters/streaming_iterator.py | 33 +- .../test_anthropic_streaming_iterator.py | 335 ++++++++++++++++++ 2 files changed, 364 insertions(+), 4 deletions(-) create mode 100644 tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_streaming_iterator.py 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 6bddad09f21..c4de3dc7b95 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -129,8 +129,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if should_start_new_block and not self.sent_content_block_finish: # Queue the sequence: content_block_stop -> content_block_start - # The trigger chunk itself is not emitted as a delta since the - # content_block_start already carries the relevant information. self.chunk_queue.append( { "type": "content_block_stop", @@ -144,6 +142,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): "content_block": self.current_content_block_start, } ) + # Also queue the processed chunk if it carries meaningful streaming chunk, so dropping it loses the arguments. + if self._has_meaningful_delta(processed_chunk): + self.chunk_queue.append(processed_chunk) self.sent_content_block_finish = False return self.chunk_queue.popleft() @@ -305,8 +306,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if not self.queued_usage_chunk: if should_start_new_block and not self.sent_content_block_finish: # Queue the sequence: content_block_stop -> content_block_start - # The trigger chunk itself is not emitted as a delta since the - # content_block_start already carries the relevant information. # 1. Stop current content block self.chunk_queue.append( @@ -325,6 +324,11 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): } ) + # 3. Also queue the processed chunk if it carries meaningful delta data (e.g. tool-call arguments). + # AI Providers deliver name+arguments in a single streaming chunk, so dropping it loses the arguments. + if self._has_meaningful_delta(processed_chunk): + self.chunk_queue.append(processed_chunk) + # Reset state for new block self.sent_content_block_finish = False @@ -424,6 +428,27 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): # For non-dict chunks, forward the original value unchanged yield chunk + def _has_meaningful_delta(self, processed_chunk: Any) -> bool: + """ + Check if a processed chunk contains meaningful delta content that + should be emitted alongside a content_block_start event. + + AI Providers deliver complete tool calls (name + arguments) in a single streaming chunk. Without emitting the delta, the arguments are silently dropped. + """ + if processed_chunk.get("type") != "content_block_delta": + return False + delta = processed_chunk.get("delta", {}) + delta_type = delta.get("type", "") + if delta_type == "input_json_delta": + return bool(delta.get("partial_json")) + elif delta_type == "text_delta": + return bool(delta.get("text")) + elif delta_type == "thinking_delta": + return bool(delta.get("thinking")) + elif delta_type == "signature_delta": + return bool(delta.get("signature")) + return False + def _increment_content_block_index(self): self.current_content_block_index += 1 diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_streaming_iterator.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_streaming_iterator.py new file mode 100644 index 00000000000..62212fae979 --- /dev/null +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_streaming_iterator.py @@ -0,0 +1,335 @@ +import os +import sys +from typing import Any, Dict, List + +sys.path.insert(0, os.path.abspath("../../../../..")) + +from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import ( + AnthropicStreamWrapper, +) +from litellm.types.utils import ( + ChatCompletionDeltaToolCall, + Delta, + Function, + ModelResponseStream, + StreamingChoices, + Usage, +) + + +def _make_text_chunk(text: str, finish_reason=None) -> ModelResponseStream: + """Helper to build a streaming chunk with text content.""" + return ModelResponseStream( + id="chatcmpl-test", + choices=[ + StreamingChoices( + finish_reason=finish_reason, + index=0, + delta=Delta( + content=text, + role="assistant", + function_call=None, + tool_calls=None, + audio=None, + ), + ) + ], + model="test-model", + ) + + +def _make_tool_call_chunk( + name: str, arguments: str, tool_call_id: str = "call_abc123", finish_reason=None +) -> ModelResponseStream: + """ + Helper to build a streaming chunk with a tool call that carries BOTH + name and arguments in the same chunk (atomic delivery). + """ + return ModelResponseStream( + id="chatcmpl-test", + choices=[ + StreamingChoices( + finish_reason=finish_reason, + index=0, + delta=Delta( + content=None, + role="assistant", + function_call=None, + tool_calls=[ + ChatCompletionDeltaToolCall( + id=tool_call_id, + function=Function(arguments=arguments, name=name), + type="function", + index=0, + ) + ], + audio=None, + ), + ) + ], + model="test-model", + ) + + +def _make_finish_chunk(stop_reason: str = "tool_use") -> ModelResponseStream: + """Helper to build a streaming chunk with only a finish reason.""" + return ModelResponseStream( + id="chatcmpl-test", + choices=[ + StreamingChoices( + finish_reason=stop_reason, + index=0, + delta=Delta( + content=None, + role="assistant", + function_call=None, + tool_calls=None, + audio=None, + ), + ) + ], + model="test-model", + usage=Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150), + ) + + +# --------------------------------------------------------------------------- +# Sync iterator tests +# --------------------------------------------------------------------------- + + +def test_emit_input_json_delta_for_atomic_tool_call(): + """ + When a provider delivers name + arguments in a single streaming chunk + (atomic delivery), the adapter must emit an input_json_delta event + between content_block_start and content_block_stop. + + Regression test for https://github.com/BerriAI/litellm/issues/25561 + """ + chunks = [ + _make_text_chunk("Let me read that file for you."), + _make_tool_call_chunk("Read", '{"file_path": "/etc/hosts"}'), + _make_finish_chunk("tool_use"), + ] + + wrapper = AnthropicStreamWrapper( + completion_stream=iter(chunks), + model="test-model", + ) + + events: List[Dict[str, Any]] = list(wrapper) + + # Gather the event types + event_types = [e.get("type") for e in events] + + # There must be at least one content_block_delta with input_json_delta + input_json_deltas: List[Dict[str, Any]] = [ + e + for e in events + if e.get("type") == "content_block_delta" + and e.get("delta", {}).get("type") == "input_json_delta" + ] + assert len(input_json_deltas) >= 1, ( + f"Expected at least one input_json_delta event, got 0. " + f"Event types: {event_types}" + ) + + # The concatenated partial_json must produce the correct arguments + combined_json = "".join( + d["delta"]["partial_json"] for d in input_json_deltas + ) + assert combined_json == '{"file_path": "/etc/hosts"}' + + +def test_correct_event_sequence_for_atomic_tool_call(): + """ + Verify the full event sequence: message_start, text block (start/delta/stop), + tool_use block (start/delta/stop), message_delta, message_stop. + """ + chunks = [ + _make_text_chunk("Hello"), + _make_tool_call_chunk("Read", '{"file_path": "/etc/hosts"}'), + _make_finish_chunk("tool_use"), + ] + + wrapper = AnthropicStreamWrapper( + completion_stream=iter(chunks), + model="test-model", + ) + + events: List[Dict[str, Any]] = list(wrapper) + event_types = [e.get("type") for e in events] + + # Expected sequence pattern + assert event_types[0] == "message_start" + + # Text content block + assert "content_block_start" in event_types + assert "content_block_stop" in event_types + + # tool_use content block + tool_block_starts: List[Dict[str, Any]] = [ + e for e in events + if e.get("type") == "content_block_start" + and e.get("content_block", {}).get("type") == "tool_use" + ] + assert len(tool_block_starts) == 1 + assert tool_block_starts[0]["content_block"]["name"] == "Read" + + # input_json_delta must appear between the tool_use start and stop + tool_start_idx = events.index(tool_block_starts[0]) + input_deltas_after_tool_start: List[Dict[str, Any]] = [ + e for e in events[tool_start_idx:] + if e.get("type") == "content_block_delta" + and e.get("delta", {}).get("type") == "input_json_delta" + ] + assert len(input_deltas_after_tool_start) >= 1 + + # message_stop at the end + assert event_types[-1] == "message_stop" + + +def test_split_tool_call_across_chunks(): + """ + For providers that split name and arguments across multiple chunks, + the existing behavior should still work correctly. + """ + # Chunk 1: tool call with name, empty arguments + name_chunk = ModelResponseStream( + id="chatcmpl-test", + choices=[ + StreamingChoices( + finish_reason=None, + index=0, + delta=Delta( + content=None, + role="assistant", + function_call=None, + tool_calls=[ + ChatCompletionDeltaToolCall( + id="call_abc123", + function=Function(arguments="", name="Read"), + type="function", + index=0, + ) + ], + audio=None, + ), + ) + ], + model="gpt-4", + ) + + # Chunk 2: arguments only (no name) + args_chunk = ModelResponseStream( + id="chatcmpl-test", + choices=[ + StreamingChoices( + finish_reason=None, + index=0, + delta=Delta( + content=None, + role="assistant", + function_call=None, + tool_calls=[ + ChatCompletionDeltaToolCall( + id=None, + function=Function( + arguments='{"file_path": "/etc/hosts"}', name=None + ), + type="function", + index=0, + ) + ], + audio=None, + ), + ) + ], + model="gpt-4", + ) + + chunks = [ + _make_text_chunk("Let me look."), + name_chunk, + args_chunk, + _make_finish_chunk("tool_use"), + ] + + wrapper = AnthropicStreamWrapper( + completion_stream=iter(chunks), + model="gpt-4", + ) + + events: List[Dict[str, Any]] = list(wrapper) + + # Should still produce input_json_delta events + input_json_deltas: List[Dict[str, Any]] = [ + e + for e in events + if e.get("type") == "content_block_delta" + and e.get("delta", {}).get("type") == "input_json_delta" + ] + assert len(input_json_deltas) >= 1 + + combined_json = "".join( + d["delta"]["partial_json"] for d in input_json_deltas + ) + assert combined_json == '{"file_path": "/etc/hosts"}' + + +def test_has_meaningful_delta(): + """Verify the _has_meaningful_delta helper classifies delta types correctly.""" + wrapper = AnthropicStreamWrapper( + completion_stream=iter([]), + model="test-model", + ) + + # input_json_delta with content -> meaningful + assert wrapper._has_meaningful_delta( + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "input_json_delta", "partial_json": '{"a": 1}'}, + } + ) is True + + # input_json_delta with empty string -> not meaningful + assert wrapper._has_meaningful_delta( + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "input_json_delta", "partial_json": ""}, + } + ) is False + + # text_delta with content -> meaningful + assert wrapper._has_meaningful_delta( + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "hello"}, + } + ) is True + + # text_delta with empty string -> not meaningful + assert wrapper._has_meaningful_delta( + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": ""}, + } + ) is False + + # message_delta (not a content_block_delta) -> not meaningful + assert wrapper._has_meaningful_delta( + {"type": "message_delta", "delta": {"stop_reason": "end_turn"}} + ) is False + + # thinking_delta with content -> meaningful + assert wrapper._has_meaningful_delta( + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "thinking_delta", "thinking": "Let me think..."}, + } + ) is True