fix(anthropic): emit one tool_use block per call when a /v1/messages stream chunk carries several (#46492)

Providers like Mistral put every parallel tool call, with full arguments, into one
chat-completions chunk. The /v1/messages adapter opened one tool_use block for the
first call and concatenated every call's arguments into it, so clients got invalid
JSON like {...}{...}. _CombinedChunkSplitter now splits such a chunk into one chunk
per call. The first call streams live and calls opened behind it are held, gathering
their fragments by index, until a non-fragment chunk arrives or the stream ends

Fixes #44029

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

@@ -1,10 +1,13 @@
 # What is this?
 ## Translates OpenAI call to Anthropic `/v1/messages` format
 import copy
+import itertools
 import json
 import traceback
 from collections import deque
 from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
+from dataclasses import dataclass, replace
+from types import MappingProxyType
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -36,7 +39,7 @@ from litellm.types.llms.anthropic import (
     UsageDelta,
     UsageIteration,
 )
-from litellm.types.utils import AdapterCompletionStreamWrapper, Delta
+from litellm.types.utils import AdapterCompletionStreamWrapper, ChatCompletionDeltaToolCall, Delta, Function
 
 if TYPE_CHECKING:
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
@@ -103,11 +106,55 @@ def _delta_payload_field(delta_type: StreamingContentBlockDeltaType) -> str:
             assert_never(delta_type)
 
 
+def _optional_str_attr(obj: object, name: str) -> str | None:
+    value: Final = _optional_attr(obj, name)
+    return value if isinstance(value, str) else None
+
+
+def _optional_int_attr(obj: object, name: str, *, default: int) -> int:
+    value: Final = _optional_attr(obj, name)
+    return value if isinstance(value, int) else default
+
+
+@dataclass(frozen=True, slots=True)
+class _ToolCallPart:
+    index: int
+    call_id: str | None
+    call_type: str | None
+    name: str | None
+    arguments: str
+
+    @staticmethod
+    def from_entries(index: int, entries: Sequence[object]) -> "_ToolCallPart":
+        functions: Final = tuple(_optional_attr(entry, "function") for entry in entries)
+        return _ToolCallPart(
+            index=index,
+            call_id=next(filter(None, (_optional_str_attr(entry, "id") for entry in entries)), None),
+            call_type=next(filter(None, (_optional_str_attr(entry, "type") for entry in entries)), None),
+            name=next(filter(None, (_optional_str_attr(function, "name") for function in functions)), None),
+            arguments="".join(_optional_str_attr(function, "arguments") or "" for function in functions),
+        )
+
+    def extended(self, parts: "Sequence[_ToolCallPart]") -> "_ToolCallPart":
+        return replace(
+            self, arguments=self.arguments + "".join(part.arguments for part in parts if part.index == self.index)
+        )
+
+    def to_tool_call(self) -> ChatCompletionDeltaToolCall:
+        return ChatCompletionDeltaToolCall(
+            id=self.call_id,
+            type=self.call_type,
+            function=Function(name=self.name, arguments=self.arguments),
+            index=self.index,
+        )
+
+
 class _CombinedChunkSplitter:
     """
     Splits a streaming chunk that carries BOTH response content and a
     ``finish_reason`` into two chunks: a content-only chunk followed by a
-    finish-only chunk.
+    finish-only chunk, and a chunk that carries several tool calls into one chunk
+    per call (see ``_route_tool_calls``).
 
     ``AnthropicStreamWrapper`` (via ``translate_streaming_openai_response_to_anthropic``)
     assumes content and ``finish_reason`` never arrive in the same chunk — true for
@@ -128,6 +175,8 @@ class _CombinedChunkSplitter:
         self._sync_iter: Iterator[ModelResponseStream] | None = None
         self._async_iter: AsyncIterator[ModelResponseStream] | None = None
         self._buffer: deque[ModelResponseStream] = deque()
+        self._held_calls: Mapping[int, tuple[ModelResponseStream, _ToolCallPart]] = MappingProxyType({})
+        self._live_call_index: int | None = None
 
     @property
     def chunks(self) -> "list[ModelResponseStream] | None":
@@ -237,6 +286,70 @@ class _CombinedChunkSplitter:
             )
         return pieces
 
+    @staticmethod
+    def _tool_call_parts(tool_calls: Sequence[object]) -> "tuple[_ToolCallPart, ...]":
+        indexed: Final = tuple(
+            (_optional_int_attr(tool_call, "index", default=position), tool_call)
+            for position, tool_call in enumerate(tool_calls)
+        )
+        return tuple(
+            _ToolCallPart.from_entries(index, tuple(entry for entry_index, entry in indexed if entry_index == index))
+            for index in dict.fromkeys(index for index, _ in indexed)
+        )
+
+    @staticmethod
+    def _with_tool_call(
+        chunk: "ModelResponseStream", part: "_ToolCallPart", *, keep_role: bool
+    ) -> "ModelResponseStream":
+        piece: Final = copy.deepcopy(chunk)
+        role: Final = getattr(piece.choices[0].delta, "role", None) if keep_role else None
+        piece.choices[0].delta = Delta(role=role, tool_calls=[part.to_tool_call()])
+        return piece
+
+    def _release_held_calls(self) -> "tuple[ModelResponseStream, ...]":
+        released: Final = tuple(
+            self._with_tool_call(template, part, keep_role=False)
+            for _, (template, part) in sorted(self._held_calls.items())
+        )
+        self._held_calls = MappingProxyType({})
+        self._live_call_index = None
+        return released
+
+    def _route_tool_calls(self, chunk: "ModelResponseStream") -> "tuple[ModelResponseStream, ...]":
+        """Anthropic ``tool_use`` blocks are sequential, so only the first call a chunk opens streams live.
+
+        Later calls are held, gather their fragments by ``index``, and are written whole on the next
+        non-tool chunk or at stream end.
+        """
+        choices: Final = _optional_attr_sequence(chunk, "choices")
+        tool_calls: Final = (
+            _optional_attr_sequence(_optional_attr(choices[0], "delta"), "tool_calls") if len(choices) == 1 else ()
+        )
+        if not tool_calls:
+            return (*self._release_held_calls(), chunk) if choices else (chunk,)
+        parts: Final = self._tool_call_parts(tool_calls)
+        held: Final = self._held_calls
+        if not held and len(parts) == 1:
+            if parts[0].name:
+                self._live_call_index = parts[0].index
+            return (chunk,)
+
+        new_parts: Final = tuple(part for part in parts if part.index not in held)
+        live_index: Final = (
+            self._live_call_index
+            if self._live_call_index is not None
+            else next((part.index for part in new_parts if part.name), None)
+        )
+        self._live_call_index = live_index
+        live: Final = tuple(part for part in new_parts if not part.name or part.index == live_index)
+        self._held_calls = MappingProxyType(
+            {
+                **{index: (template, held_part.extended(parts)) for index, (template, held_part) in held.items()},
+                **{part.index: (chunk, part) for part in new_parts if part.name and part.index != live_index},
+            }
+        )
+        return tuple(self._with_tool_call(chunk, part, keep_role=position == 0) for position, part in enumerate(live))
+
     @staticmethod
     def _split(chunk: "ModelResponseStream") -> "list[ModelResponseStream]":
         """Return ``[chunk]``, or ``[content_chunk, finish_chunk]`` if combined."""
@@ -260,6 +373,14 @@ class _CombinedChunkSplitter:
             finish_delta.thinking_blocks = None
         return [content_chunk, finish_chunk]
 
+    def _expand(self, chunk: "ModelResponseStream") -> "tuple[ModelResponseStream, ...]":
+        split_chunks: Final = tuple(
+            split_chunk
+            for combined_chunk in self._split(chunk)
+            for split_chunk in self._split_by_payload_kind(combined_chunk)  # comprehension-ok: flattens nested splits
+        )
+        return tuple(itertools.chain.from_iterable(self._route_tool_calls(split_chunk) for split_chunk in split_chunks))
+
     def __iter__(self) -> "Iterator[ModelResponseStream]":
         return self
 
@@ -289,12 +410,15 @@ class _CombinedChunkSplitter:
             return self._buffer.popleft()
         if self._sync_iter is None:
             self._sync_iter = iter(self._stream)
-        chunk: Final = next(self._sync_iter)  # propagates StopIteration when exhausted
-        self._buffer.extend(
-            split_chunk
-            for combined_chunk in self._split(chunk)
-            for split_chunk in self._split_by_payload_kind(combined_chunk)
-        )
+        while not self._buffer:
+            try:
+                chunk = next(self._sync_iter)
+            except StopIteration:
+                self._buffer.extend(self._release_held_calls())
+                if not self._buffer:
+                    raise
+                break
+            self._buffer.extend(self._expand(chunk))
         return self._buffer.popleft()
 
     def __aiter__(self) -> "AsyncIterator[ModelResponseStream]":
@@ -305,12 +429,15 @@ class _CombinedChunkSplitter:
             return self._buffer.popleft()
         if self._async_iter is None:
             self._async_iter = self._stream.__aiter__()
-        chunk: Final = await self._async_iter.__anext__()  # propagates StopAsyncIteration
-        self._buffer.extend(
-            split_chunk
-            for combined_chunk in self._split(chunk)
-            for split_chunk in self._split_by_payload_kind(combined_chunk)
-        )
+        while not self._buffer:
+            try:
+                chunk = await self._async_iter.__anext__()
+            except StopAsyncIteration:
+                self._buffer.extend(self._release_held_calls())
+                if not self._buffer:
+                    raise
+                break
+            self._buffer.extend(self._expand(chunk))
         return self._buffer.popleft()
 
 

tests/unit/llms/anthropic/pass_through/adapters/test_streaming_iterator_tool_args.py

@@ -8,6 +8,7 @@ Without the fix, the AnthropicStreamWrapper silently dropped these
 arguments, causing tool_use blocks to arrive with empty input {}.
 """
 
+import asyncio
 import json
 
 from typing import List
@@ -23,7 +24,9 @@ from litellm.types.utils import (
     ChatCompletionDeltaToolCall,
     Delta,
     Function,
+    ModelResponseStream,
     StreamingChoices,
+    Usage,
 )
 
 
@@ -370,3 +373,346 @@ def test_sync_stream_no_extra_delta_when_tool_args_empty():
         f"Expected exactly 1 input_json_delta (from the follow-up chunk), got {len(input_json_deltas)}"
     )
     assert input_json_deltas[0]["delta"]["partial_json"] == '{"location": "NYC"}'
+
+
+# --- Several tool calls in one chunk (https://github.com/BerriAI/litellm/issues/44029) ---
+
+_TMP_ARGS = '{"command": "ls -d /tmp", "description": "List the /tmp directory"}'
+_VAR_ARGS = '{"command": "ls -d /var", "description": "List the /var directory"}'
+
+
+def _tool_call(index: int, arguments: str, call_id: str = None, name: str = None) -> ChatCompletionDeltaToolCall:
+    return ChatCompletionDeltaToolCall(
+        id=call_id,
+        type="function" if call_id else None,
+        function=Function(name=name, arguments=arguments),
+        index=index,
+    )
+
+
+def _stream_chunk(delta: Delta, finish_reason: str = None, usage: Usage = None) -> ModelResponseStream:
+    chunk = ModelResponseStream(choices=[StreamingChoices(index=0, delta=delta, finish_reason=finish_reason)])
+    if usage is not None:
+        chunk.usage = usage
+    return chunk
+
+
+def _drain(chunks: list[ModelResponseStream], use_async: bool) -> list[dict]:
+    async def _agen():
+        for chunk in chunks:
+            yield chunk
+
+    if use_async:
+        wrapper = AnthropicStreamWrapper(completion_stream=_agen(), model="mistral-large-2512")
+        return asyncio.run(_collect_events_async(wrapper))
+    wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="mistral-large-2512")
+    return _collect_events_sync(wrapper)
+
+
+def _blocks(events: list[dict]) -> list[dict]:
+    """Rebuild each content block from its start, deltas and stop events."""
+    blocks: dict = {}
+    for event in events:
+        if event.get("type") == "content_block_start":
+            assert event["index"] not in blocks, f"block {event['index']} opened twice"
+            blocks[event["index"]] = {"start": event["content_block"], "json": "", "text": "", "stopped": False}
+        elif event.get("type") == "content_block_delta":
+            block = blocks[event["index"]]
+            assert not block["stopped"], f"delta after stop on block {event['index']}"
+            block["json"] += event["delta"].get("partial_json", "")
+            block["text"] += event["delta"].get("text", "")
+        elif event.get("type") == "content_block_stop":
+            blocks[event["index"]]["stopped"] = True
+    assert list(blocks) == list(range(len(blocks))), f"block indexes not contiguous: {list(blocks)}"
+    assert all(block["stopped"] for block in blocks.values())
+    return [blocks[index] for index in sorted(blocks)]
+
+
+def _message_delta(events: list[dict]) -> dict:
+    deltas = [event for event in events if event.get("type") == "message_delta"]
+    assert len(deltas) == 1
+    last_stop = max(i for i, event in enumerate(events) if event.get("type") == "content_block_stop")
+    assert events.index(deltas[0]) > last_stop
+    return deltas[0]
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_two_complete_tool_calls_in_one_chunk_with_finish_and_usage(use_async):
+    """Mistral bundles both calls, their full arguments, finish_reason and usage into one chunk."""
+    chunk = _stream_chunk(
+        Delta(
+            role="assistant",
+            tool_calls=[
+                _tool_call(0, _TMP_ARGS, call_id="M7B4U5UJs", name="Bash"),
+                _tool_call(1, _VAR_ARGS, call_id="jWEVB6J9J", name="Read"),
+            ],
+        ),
+        finish_reason="tool_calls",
+        usage=Usage(prompt_tokens=40, completion_tokens=12, total_tokens=52),
+    )
+
+    events = _drain([chunk], use_async)
+    blocks = _blocks(events)
+
+    assert [block["start"]["type"] for block in blocks] == ["tool_use", "tool_use"]
+    assert [block["start"]["id"] for block in blocks] == ["M7B4U5UJs", "jWEVB6J9J"]
+    assert [block["start"]["name"] for block in blocks] == ["Bash", "Read"]
+    assert [json.loads(block["json"]) for block in blocks] == [json.loads(_TMP_ARGS), json.loads(_VAR_ARGS)]
+    message_delta = _message_delta(events)
+    assert message_delta["delta"]["stop_reason"] == "tool_use"
+    assert message_delta["usage"]["output_tokens"] == 12
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_text_chunk_then_chunk_with_two_tool_calls(use_async):
+    chunks = [
+        _stream_chunk(Delta(role="assistant", content="Listing both.")),
+        _stream_chunk(
+            Delta(
+                tool_calls=[
+                    _tool_call(0, _TMP_ARGS, call_id="call_a", name="Bash"),
+                    _tool_call(1, _VAR_ARGS, call_id="call_b", name="Bash"),
+                ]
+            )
+        ),
+        _stream_chunk(Delta(), finish_reason="tool_calls"),
+    ]
+
+    events = _drain(chunks, use_async)
+    blocks = _blocks(events)
+
+    assert [block["start"]["type"] for block in blocks] == ["text", "tool_use", "tool_use"]
+    assert blocks[0]["text"] == "Listing both."
+    assert [block["start"]["id"] for block in blocks[1:]] == ["call_a", "call_b"]
+    assert [json.loads(block["json"]) for block in blocks[1:]] == [json.loads(_TMP_ARGS), json.loads(_VAR_ARGS)]
+    assert _message_delta(events)["delta"]["stop_reason"] == "tool_use"
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_two_calls_opened_together_then_interleaved_continuations(use_async):
+    """Anthropic blocks are sequential, so call 1 is held until call 0 can no longer grow.
+
+    Call 0 keeps streaming live; call 1's fragments are gathered by ``index`` and
+    its block is written whole once a chunk that is not a fragment arrives.
+    """
+    chunks = [
+        _stream_chunk(
+            Delta(
+                role="assistant",
+                tool_calls=[
+                    _tool_call(0, '{"command": ', call_id="call_a", name="Bash"),
+                    _tool_call(1, '{"command": ', call_id="call_b", name="Bash"),
+                ],
+            )
+        ),
+        _stream_chunk(Delta(tool_calls=[_tool_call(1, '"ls -d /var"')])),
+        _stream_chunk(Delta(tool_calls=[_tool_call(0, '"ls -d /tmp"')])),
+        _stream_chunk(Delta(tool_calls=[_tool_call(0, "}"), _tool_call(1, "}")])),
+        _stream_chunk(Delta(), finish_reason="tool_calls"),
+    ]
+
+    events = _drain(chunks, use_async)
+    blocks = _blocks(events)
+
+    assert [block["start"]["id"] for block in blocks] == ["call_a", "call_b"]
+    assert [json.loads(block["json"]) for block in blocks] == [{"command": "ls -d /tmp"}, {"command": "ls -d /var"}]
+    assert _message_delta(events)["delta"]["stop_reason"] == "tool_use"
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_chunk_continuing_open_call_and_opening_next_one(use_async):
+    """Entry 0 finishes the open call, entry 1 starts a new one in the same chunk."""
+    chunks = [
+        _stream_chunk(
+            Delta(role="assistant", tool_calls=[_tool_call(0, '{"command": ', call_id="call_a", name="Bash")])
+        ),
+        _stream_chunk(
+            Delta(
+                tool_calls=[
+                    _tool_call(0, '"ls -d /tmp"}'),
+                    _tool_call(1, '{"command": ', call_id="call_b", name="Bash"),
+                ]
+            )
+        ),
+        _stream_chunk(Delta(tool_calls=[_tool_call(1, '"ls -d /var"}')])),
+        _stream_chunk(Delta(), finish_reason="tool_calls"),
+    ]
+
+    events = _drain(chunks, use_async)
+    blocks = _blocks(events)
+
+    assert [block["start"]["id"] for block in blocks] == ["call_a", "call_b"]
+    assert [json.loads(block["json"]) for block in blocks] == [{"command": "ls -d /tmp"}, {"command": "ls -d /var"}]
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_next_call_opening_beside_a_fragment_of_the_open_call_waits_for_it(use_async):
+    """Call 1 opens in a chunk that also extends call 0, and call 0 keeps streaming after it."""
+    chunks = [
+        _stream_chunk(
+            Delta(role="assistant", tool_calls=[_tool_call(0, '{"command": ', call_id="call_a", name="Bash")])
+        ),
+        _stream_chunk(
+            Delta(
+                tool_calls=[
+                    _tool_call(0, '"ls'),
+                    _tool_call(1, '{"command": ', call_id="call_b", name="Bash"),
+                ]
+            )
+        ),
+        _stream_chunk(Delta(tool_calls=[_tool_call(0, ' -d /tmp"}')])),
+        _stream_chunk(Delta(tool_calls=[_tool_call(1, '"ls -d /var"}')])),
+        _stream_chunk(Delta(), finish_reason="tool_calls"),
+    ]
+
+    events = _drain(chunks, use_async)
+    blocks = _blocks(events)
+
+    assert [block["start"]["id"] for block in blocks] == ["call_a", "call_b"]
+    assert [json.loads(block["json"]) for block in blocks] == [{"command": "ls -d /tmp"}, {"command": "ls -d /var"}]
+    assert _message_delta(events)["delta"]["stop_reason"] == "tool_use"
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_three_calls_in_one_chunk_come_out_in_index_order(use_async):
+    chunks = [
+        _stream_chunk(
+            Delta(
+                role="assistant",
+                tool_calls=[
+                    _tool_call(0, '{"command": "ls -d /tmp"}', call_id="call_a", name="Bash"),
+                    _tool_call(1, '{"command": "ls -d /var"}', call_id="call_b", name="Bash"),
+                    _tool_call(2, '{"command": "ls -d /etc"}', call_id="call_c", name="Bash"),
+                ],
+            )
+        ),
+        _stream_chunk(Delta(), finish_reason="tool_calls"),
+    ]
+
+    blocks = _blocks(_drain(chunks, use_async))
+
+    assert [block["start"]["id"] for block in blocks] == ["call_a", "call_b", "call_c"]
+    assert [json.loads(block["json"])["command"] for block in blocks] == ["ls -d /tmp", "ls -d /var", "ls -d /etc"]
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_held_call_is_written_when_the_stream_ends_without_a_finish_chunk(use_async):
+    chunks = [
+        _stream_chunk(
+            Delta(
+                role="assistant",
+                tool_calls=[
+                    _tool_call(0, '{"command": "ls -d /tmp"}', call_id="call_a", name="Bash"),
+                    _tool_call(1, '{"command": "ls -d /var"}', call_id="call_b", name="Bash"),
+                ],
+            )
+        ),
+    ]
+
+    events = _drain(chunks, use_async)
+    starts = [event for event in events if event.get("type") == "content_block_start"]
+    arguments = [
+        "".join(
+            event["delta"]["partial_json"]
+            for event in events
+            if event.get("type") == "content_block_delta" and event["index"] == start["index"]
+        )
+        for start in starts
+    ]
+
+    assert [start["content_block"]["id"] for start in starts] == ["call_a", "call_b"]
+    assert [json.loads(argument) for argument in arguments] == [{"command": "ls -d /tmp"}, {"command": "ls -d /var"}]
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_call_opening_alone_while_another_is_held_waits_behind_the_live_call(use_async):
+    chunks = [
+        _stream_chunk(
+            Delta(
+                role="assistant",
+                tool_calls=[
+                    _tool_call(0, '{"command": ', call_id="call_a", name="Bash"),
+                    _tool_call(1, '{"command": "ls -d /var"}', call_id="call_b", name="Bash"),
+                ],
+            )
+        ),
+        _stream_chunk(Delta(tool_calls=[_tool_call(2, '{"command": ', call_id="call_c", name="Bash")])),
+        _stream_chunk(Delta(tool_calls=[_tool_call(0, '"ls -d /tmp"}')])),
+        _stream_chunk(Delta(tool_calls=[_tool_call(2, '"ls -d /etc"}')])),
+        _stream_chunk(Delta(), finish_reason="tool_calls"),
+    ]
+
+    events = _drain(chunks, use_async)
+    blocks = _blocks(events)
+
+    assert [block["start"]["id"] for block in blocks] == ["call_a", "call_b", "call_c"]
+    assert [json.loads(block["json"])["command"] for block in blocks] == ["ls -d /tmp", "ls -d /var", "ls -d /etc"]
+    assert _message_delta(events)["delta"]["stop_reason"] == "tool_use"
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_two_calls_opening_together_behind_a_live_call_wait_for_it(use_async):
+    chunks = [
+        _stream_chunk(
+            Delta(role="assistant", tool_calls=[_tool_call(0, '{"command": ', call_id="call_a", name="Bash")])
+        ),
+        _stream_chunk(
+            Delta(
+                tool_calls=[
+                    _tool_call(1, '{"command": "ls -d /var"}', call_id="call_b", name="Bash"),
+                    _tool_call(2, '{"command": "ls -d /etc"}', call_id="call_c", name="Bash"),
+                ]
+            )
+        ),
+        _stream_chunk(Delta(tool_calls=[_tool_call(0, '"ls -d /tmp"}')])),
+        _stream_chunk(Delta(), finish_reason="tool_calls"),
+    ]
+
+    blocks = _blocks(_drain(chunks, use_async))
+
+    assert [block["start"]["id"] for block in blocks] == ["call_a", "call_b", "call_c"]
+    assert [json.loads(block["json"])["command"] for block in blocks] == ["ls -d /tmp", "ls -d /var", "ls -d /etc"]
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_chunk_without_choices_mid_stream_keeps_held_calls_gathering(use_async):
+    usage_only = ModelResponseStream(choices=[])
+    usage_only.usage = Usage(prompt_tokens=40, completion_tokens=3, total_tokens=43)
+    chunks = [
+        _stream_chunk(
+            Delta(
+                role="assistant",
+                tool_calls=[
+                    _tool_call(0, "", call_id="call_a", name="Bash"),
+                    _tool_call(1, "", call_id="call_b", name="Bash"),
+                ],
+            )
+        ),
+        usage_only,
+        _stream_chunk(Delta(tool_calls=[_tool_call(1, '{"command": "ls -d /var"}')])),
+        _stream_chunk(Delta(tool_calls=[_tool_call(0, '{"command": "ls -d /tmp"}')])),
+        _stream_chunk(Delta(), finish_reason="tool_calls"),
+    ]
+
+    blocks = _blocks(_drain(chunks, use_async))
+
+    assert [block["start"]["id"] for block in blocks] == ["call_a", "call_b"]
+    assert [json.loads(block["json"]) for block in blocks] == [{"command": "ls -d /tmp"}, {"command": "ls -d /var"}]
+
+
+@pytest.mark.parametrize("use_async", [False, True])
+def test_one_call_split_across_chunks_is_unchanged(use_async):
+    chunks = [
+        _stream_chunk(Delta(role="assistant", tool_calls=[_tool_call(0, "", call_id="call_a", name="Bash")])),
+        _stream_chunk(Delta(tool_calls=[_tool_call(0, '{"command": ')])),
+        _stream_chunk(Delta(tool_calls=[_tool_call(0, '"ls -d /tmp"}')])),
+        _stream_chunk(Delta(), finish_reason="tool_calls"),
+    ]
+
+    events = _drain(chunks, use_async)
+    blocks = _blocks(events)
+
+    assert [block["start"]["id"] for block in blocks] == ["call_a"]
+    assert json.loads(blocks[0]["json"]) == {"command": "ls -d /tmp"}
+    deltas = [event for event in events if event.get("type") == "content_block_delta"]
+    assert [delta["delta"]["partial_json"] for delta in deltas] == ['{"command": ', '"ls -d /tmp"}']