diff --git a/litellm/llms/anthropic/chat/handler.py b/litellm/llms/anthropic/chat/handler.py index 51b9c9835a..7d5fa2a559 100644 --- a/litellm/llms/anthropic/chat/handler.py +++ b/litellm/llms/anthropic/chat/handler.py @@ -545,7 +545,7 @@ class ModelResponseIterator: # Track server tool use inputs and results for code_interpreter_results self._server_tool_inputs: Dict[str, Any] = {} self.tool_results: List[Dict[str, Any]] = [] - self._last_code_interpreter_results_count: int = 0 + self._current_server_tool_id: Optional[str] = None def check_empty_tool_call_args(self) -> bool: """ @@ -695,13 +695,14 @@ class ModelResponseIterator: Called during streaming to produce provider-neutral code_interpreter_results alongside the raw tool_results, so the Responses API layer doesn't need Anthropic-specific knowledge. + + Returns the full cumulative list each time (not incremental), matching + how web_search_results works. stream_chunk_builder uses "last value + wins" for list-valued provider_specific_fields keys, so the last + emission must contain every result. """ - # Only convert tool_results added since the last call to avoid - # duplicates when _merge_provider_specific_fields extends the list. - new_results = self.tool_results[self._last_code_interpreter_results_count :] - self._last_code_interpreter_results_count = len(self.tool_results) results = [] - for tr in new_results: + for tr in self.tool_results: call_id = tr.get("tool_use_id", "") content = tr.get("content", {}) if isinstance(content, dict): @@ -793,17 +794,23 @@ class ModelResponseIterator: ), index=self.tool_index, ) - # Track server tool use inputs for code_interpreter_results + # Track server tool use inputs for code_interpreter_results. + # The initial input in content_block_start is typically {} + # for streaming; the full input arrives via input_json_delta + # and is assembled at content_block_stop. if ( content_block_start["content_block"]["type"] == "server_tool_use" ): + self._current_server_tool_id = content_block_start[ + "content_block" + ]["id"] tool_input = content_block_start["content_block"].get( "input", {} ) - self._server_tool_inputs[ - content_block_start["content_block"]["id"] - ] = tool_input + self._server_tool_inputs[self._current_server_tool_id] = ( + tool_input + ) # Include caller information if present (for programmatic tool calling) if "caller" in content_block_start["content_block"]: caller_data = content_block_start["content_block"]["caller"] @@ -886,6 +893,24 @@ class ModelResponseIterator: ), index=self.tool_index, ) + # Update server_tool_inputs with fully assembled input + # from input_json_delta chunks (content_block_start has {}) + if ( + self.current_content_block_type == "server_tool_use" + and self._current_server_tool_id + ): + args = "" + for block in self.content_blocks: + if block["delta"]["type"] == "input_json_delta": + args += block["delta"].get("partial_json", "") + if args: + try: + self._server_tool_inputs[ + self._current_server_tool_id + ] = json.loads(args) + except (json.JSONDecodeError, TypeError): + pass + self._current_server_tool_id = None # Reset response_format tool tracking when block stops self.is_response_format_tool = False # Reset current content block type diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index 0b7d6e8a7a..0672b03bcd 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -481,17 +481,16 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return event def _merge_provider_specific_fields(self, src: dict) -> None: - """Merge provider_specific_fields, extending list values instead of replacing.""" + """Merge provider_specific_fields using last-value-wins for lists. + + List-valued keys (web_search_results, tool_results, + code_interpreter_results, etc.) are emitted cumulatively — each + emission contains the full list so far. Using "last value wins" + matches stream_chunk_builder's semantics and avoids quadratic + growth from repeated extend calls. + """ for key, val in src.items(): - existing = self._accumulated_provider_specific_fields.get(key) - if ( - existing is not None - and isinstance(val, list) - and isinstance(existing, list) - ): - existing.extend(val) - else: - self._accumulated_provider_specific_fields[key] = val + self._accumulated_provider_specific_fields[key] = val def create_litellm_model_response(self) -> Optional[ModelResponse]: response = cast( diff --git a/litellm/responses/litellm_completion_transformation/transformation.py b/litellm/responses/litellm_completion_transformation/transformation.py index b54d5930ef..b7f7e9adda 100644 --- a/litellm/responses/litellm_completion_transformation/transformation.py +++ b/litellm/responses/litellm_completion_transformation/transformation.py @@ -1738,7 +1738,10 @@ class LiteLLMCompletionResponsesConfig: ) ) if tool_result_items: - result_by_id = {item.id: item for item in tool_result_items} + result_by_id = { + (item.get("id") if isinstance(item, dict) else item.id): item + for item in tool_result_items + } replaced_ids = set(result_by_id.keys()) responses_output = [ ( diff --git a/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py b/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py index 35c7a62027..d7a04a054a 100644 --- a/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py +++ b/tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py @@ -1245,9 +1245,9 @@ def test_streaming_code_execution_produces_code_interpreter_results(): def test_streaming_multiple_code_executions_no_duplicates(): """ - Test that multiple code executions in a single streaming response produce - exactly one code_interpreter_result per execution — no duplicates from - _build_code_interpreter_results rebuilding the full list. + Test that multiple code executions in a single streaming response emit + cumulative code_interpreter_results on each chunk (matching stream_chunk_builder's + "last value wins" contract). The final emission must contain ALL results. """ chunks = [ { @@ -1323,24 +1323,125 @@ def test_streaming_multiple_code_executions_no_duplicates(): iterator = ModelResponseIterator(None, sync_stream=True) - # Collect ALL code_interpreter_results emitted across all chunks - all_results = [] + # Collect each emission of code_interpreter_results + emissions = [] for chunk in chunks: parsed = iterator.chunk_parser(chunk) psf = None if parsed.choices and parsed.choices[0].delta: psf = getattr(parsed.choices[0].delta, "provider_specific_fields", None) if psf and "code_interpreter_results" in psf: - all_results.extend(psf["code_interpreter_results"]) + emissions.append(psf["code_interpreter_results"]) - # Should have exactly 2 results, one per execution — no duplicates - assert len(all_results) == 2, ( - f"Expected 2 code_interpreter_results, got {len(all_results)}. " - f"IDs: {[r.id for r in all_results]}" + # Should have 2 emissions (one per tool_result block) + assert len(emissions) == 2, f"Expected 2 emissions, got {len(emissions)}" + + # First emission: cumulative list with 1 result + assert len(emissions[0]) == 1 + assert emissions[0][0].id == "srvtoolu_01AAA" + assert emissions[0][0].code == "echo first" + assert emissions[0][0].outputs[0].logs == "first\n" + + # Second (final) emission: cumulative list with BOTH results + # This is what stream_chunk_builder will pick as "last value wins" + assert len(emissions[1]) == 2, ( + f"Expected final emission to have 2 results, got {len(emissions[1])}. " + f"IDs: {[r.id for r in emissions[1]]}" ) - assert all_results[0].id == "srvtoolu_01AAA" - assert all_results[0].code == "echo first" - assert all_results[0].outputs[0].logs == "first\n" - assert all_results[1].id == "srvtoolu_01BBB" - assert all_results[1].code == "echo second" - assert all_results[1].outputs[0].logs == "second\n" + assert emissions[1][0].id == "srvtoolu_01AAA" + assert emissions[1][0].code == "echo first" + assert emissions[1][0].outputs[0].logs == "first\n" + assert emissions[1][1].id == "srvtoolu_01BBB" + assert emissions[1][1].code == "echo second" + assert emissions[1][1].outputs[0].logs == "second\n" + + +def test_streaming_code_execution_input_assembled_from_deltas(): + """ + In real Anthropic streaming, content_block_start for server_tool_use has + input: {}. The actual input arrives via input_json_delta deltas and must + be assembled at content_block_stop so the code field is populated. + + This test uses realistic chunk shapes (empty input in start, partial JSON + in deltas) to exercise the input assembly path. + """ + chunks = [ + { + "type": "message_start", + "message": { + "id": "msg_01XYZ", + "type": "message", + "role": "assistant", + "content": [], + "usage": {"input_tokens": 100, "output_tokens": 1}, + }, + }, + # server_tool_use with empty input (real streaming behaviour) + { + "type": "content_block_start", + "index": 0, + "content_block": { + "type": "server_tool_use", + "id": "srvtoolu_01AAA", + "name": "code_execution", + "input": {}, + }, + }, + # Input arrives via deltas, split across two chunks + { + "type": "content_block_delta", + "index": 0, + "delta": { + "type": "input_json_delta", + "partial_json": '{"comma', + }, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": { + "type": "input_json_delta", + "partial_json": 'nd": "echo hello"}', + }, + }, + {"type": "content_block_stop", "index": 0}, + # Tool result + { + "type": "content_block_start", + "index": 1, + "content_block": { + "type": "code_execution_tool_result", + "tool_use_id": "srvtoolu_01AAA", + "content": { + "type": "code_execution_result", + "stdout": "hello\n", + "stderr": "", + "return_code": 0, + }, + }, + }, + {"type": "content_block_stop", "index": 1}, + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn"}, + "usage": {"output_tokens": 50}, + }, + ] + + iterator = ModelResponseIterator(None, sync_stream=True) + + code_results = None + for chunk in chunks: + parsed = iterator.chunk_parser(chunk) + psf = None + if parsed.choices and parsed.choices[0].delta: + psf = getattr(parsed.choices[0].delta, "provider_specific_fields", None) + if psf and "code_interpreter_results" in psf: + code_results = psf["code_interpreter_results"] + + # The code field must contain the assembled input, not be empty + assert code_results is not None, "No code_interpreter_results emitted" + assert len(code_results) == 1 + assert code_results[0].id == "srvtoolu_01AAA" + assert code_results[0].code == "echo hello" + assert code_results[0].outputs[0].logs == "hello\n"