fix(anthropic): preserve all tool_calls when an OpenAI delta contains multiple

The Anthropic /v1/messages streaming adapter (AnthropicStreamWrapper)
silently drops every tool_call beyond the first when an upstream OpenAI
streaming chunk contains multiple complete tool_calls in a single delta.
The downstream converter
(_translate_streaming_openai_chunk_to_anthropic_content_block) indexes
tool_calls[0], so without splitting at the wrapper level only one
content_block_start / input_json_delta / content_block_stop triple is
emitted regardless of how many parallel tool_calls the model produced.

The provider that triggered this is mlx_lm.server, which emits all
parallel tool_calls in one final delta after their text has been fully
generated. Real-world impact: Claude Code (which speaks /v1/messages)
sees only the first parallel subagent dispatch, then loops on
"InputValidationError: required parameter description / prompt missing"
because the model's follow-up tool_use blocks arrive without their
content.

Fix: a small _MultiToolCallSplitter wraps the upstream stream inside
__init__. It supports both __iter__/__next__ and __aiter__/__anext__
transparently so AnthropicStreamWrapper.__next__ and __anext__ each see
the matching protocol. When a chunk's delta has more than one tool_call,
the splitter deep-copies the chunk N times (one tool_call each) and
buffers the rest. Single-tool-call chunks pass through unchanged
(returns the same instance, no copy).

Verified end-to-end against mlx_lm.server kimi-k2.6 + Claude Code's
parallel-Agent-tool dispatch: before patch, /v1/messages stream emitted
8 events with 1 content_block_start for tool_use; after patch, 11
events with 2 content_block_starts (index 1 + index 2), each carrying
its own input_json_delta. Claude Code TUI confirms both subagents now
spawn, run, and return cleanly.

Test added: tests/test_litellm/llms/anthropic/test_anthropic_stream_multi_tool_call_split.py
covers the static splitter, sync iteration, async iteration, and
construction-time wiring through AnthropicStreamWrapper.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

@@ -1,9 +1,19 @@
 # What is this?
 ## Translates OpenAI call to Anthropic `/v1/messages` format
+import copy
 import json
 import traceback
 from collections import deque
-from typing import TYPE_CHECKING, Any, AsyncIterator, Dict, Iterator, Literal, Optional
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncIterator,
+    Dict,
+    Iterator,
+    List,
+    Literal,
+    Optional,
+)
 
 from litellm import verbose_logger
 from litellm._uuid import uuid
@@ -14,6 +24,65 @@ if TYPE_CHECKING:
     from litellm.types.utils import ModelResponseStream
 
 
+class _MultiToolCallSplitter:
+    """Wraps an upstream OpenAI-format streaming iterator and splits any chunk
+    whose ``delta.tool_calls`` contains more than one entry into multiple
+    chunks (one tool_call each).
+
+    Supports both sync (``__iter__`` / ``__next__``) and async (``__aiter__``
+    / ``__anext__``) consumption transparently, deciding which protocol to
+    use based on which one is invoked first. This is necessary because some
+    upstream stream wrappers expose both protocols, and consumers
+    (``AnthropicStreamWrapper.__next__`` vs ``__anext__``) pick the matching
+    one — wrapping the upstream stream with a sync-only or async-only
+    generator at construction time would break whichever protocol is unused.
+
+    Without this splitting, the downstream converter in
+    ``AnthropicStreamWrapper`` (which indexes ``tool_calls[0]`` in
+    ``_translate_streaming_openai_chunk_to_anthropic_content_block``)
+    silently drops every tool_call beyond the first when a provider emits
+    multiple parallel tool_calls in one OpenAI delta (e.g. mlx_lm.server).
+    """
+
+    def __init__(self, stream: Any):
+        self._stream = stream
+        self._buffer: deque = deque()
+        # Lazily set the first time __iter__ / __aiter__ is called. Typed
+        # ``Any`` (rather than ``Optional[Any]``) so mypy doesn't ask us to
+        # narrow the None case at every call site — Python's iteration
+        # protocol contract already guarantees __iter__ runs before __next__.
+        self._sync_iter_obj: Any = None
+        self._async_iter_obj: Any = None
+
+    def __iter__(self) -> "Iterator[Any]":
+        if self._sync_iter_obj is None:
+            self._sync_iter_obj = iter(self._stream)
+        return self
+
+    def __next__(self) -> Any:
+        if self._buffer:
+            return self._buffer.popleft()
+        chunk = next(self._sync_iter_obj)  # raises StopIteration at EOF
+        splits = AnthropicStreamWrapper._split_chunk_by_tool_calls(chunk)
+        if len(splits) > 1:
+            self._buffer.extend(splits[1:])
+        return splits[0]
+
+    def __aiter__(self) -> "AsyncIterator[Any]":
+        if self._async_iter_obj is None:
+            self._async_iter_obj = self._stream.__aiter__()
+        return self
+
+    async def __anext__(self) -> Any:
+        if self._buffer:
+            return self._buffer.popleft()
+        chunk = await self._async_iter_obj.__anext__()
+        splits = AnthropicStreamWrapper._split_chunk_by_tool_calls(chunk)
+        if len(splits) > 1:
+            self._buffer.extend(splits[1:])
+        return splits[0]
+
+
 class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
     """
     - first chunk return 'message_start'
@@ -54,6 +123,52 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
         # Mapping of truncated tool names to original names (for OpenAI's 64-char limit)
         self.tool_name_mapping = tool_name_mapping or {}
 
+        # Wrap upstream stream so chunks containing multiple tool_calls in a
+        # single delta are split into one tool_call per chunk before reaching
+        # the downstream converter. The converter assumes one tool_call per
+        # chunk (it indexes ``tool_calls[0]`` in
+        # ``_translate_streaming_openai_chunk_to_anthropic_content_block``),
+        # so without this split, providers that emit parallel tool_calls in
+        # one OpenAI delta (e.g. mlx_lm.server) lose all but the first call.
+        # The wrapper supports both sync and async iteration; whichever
+        # ``AnthropicStreamWrapper.__next__`` / ``__anext__`` invokes is
+        # served from the underlying stream's matching protocol.
+        #
+        # Stored under a separate attribute (rather than overwriting
+        # ``self.completion_stream`` from the superclass) so consumers that
+        # rely on the original stream still see it, and to keep static
+        # analyzers happy about subclass attribute shadowing.
+        self._completion_stream_splitter = _MultiToolCallSplitter(completion_stream)
+
+    @staticmethod
+    def _split_chunk_by_tool_calls(chunk: Any) -> List[Any]:
+        """Split one streaming chunk into N chunks if its delta contains
+        multiple tool_calls. Returns a list of chunks (length 1 for normal
+        chunks, N for multi-tool-call chunks).
+        """
+        if chunk is None or chunk == "None":
+            return [chunk]
+        try:
+            tcs = (
+                chunk.choices[0].delta.tool_calls
+                if chunk.choices and chunk.choices[0].delta is not None
+                else None
+            )
+        except (AttributeError, IndexError):
+            return [chunk]
+        if tcs is None or len(tcs) <= 1:
+            return [chunk]
+        out: List[Any] = []
+        for one_tc in tcs:
+            sub = copy.deepcopy(chunk)
+            # Deep-copy the tool_call too so each split chunk has fully
+            # independent state — without this, mutations downstream on
+            # ``one_tc`` (e.g. argument deltas) would leak into the original
+            # chunk's tool_calls list and into peer split chunks.
+            sub.choices[0].delta.tool_calls = [copy.deepcopy(one_tc)]
+            out.append(sub)
+        return out
+
     def _create_initial_usage_delta(self) -> UsageDelta:
         """
         Create the initial UsageDelta for the message_start event.
@@ -114,7 +229,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
                 )
                 return self.chunk_queue.popleft()
 
-            for chunk in self.completion_stream:
+            for chunk in self._completion_stream_splitter:
                 if chunk == "None" or chunk is None:
                     raise Exception
 
@@ -254,7 +369,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
                 )
                 return self.chunk_queue.popleft()
 
-            async for chunk in self.completion_stream:
+            async for chunk in self._completion_stream_splitter:
                 if chunk == "None" or chunk is None:
                     raise Exception
 

tests/test_litellm/llms/anthropic/test_anthropic_stream_multi_tool_call_split.py

@@ -0,0 +1,388 @@
+"""
+Tests for AnthropicStreamWrapper's multi-tool-call splitting.
+
+Some upstream providers (e.g. mlx_lm.server) emit multiple complete
+tool_calls inside a SINGLE OpenAI streaming delta when the model produces
+parallel tool calls. Anthropic's streaming format requires one
+``content_block`` per ``tool_use`` and the downstream converter in
+``AnthropicStreamWrapper`` indexes ``tool_calls[0]`` — so without splitting,
+all but the first tool_call are silently dropped from the converted
+``/v1/messages`` stream.
+
+These tests verify that ``AnthropicStreamWrapper`` splits such chunks into
+one tool_call per chunk before the converter sees them.
+"""
+
+import asyncio
+import os
+import sys
+from typing import List
+from unittest.mock import MagicMock
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../../../../"))
+
+from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import (
+    AnthropicStreamWrapper,
+    _MultiToolCallSplitter,
+)
+
+
+class _DualProtocolStream:
+    """Iterator that exposes BOTH sync and async protocols, matching the
+    pattern of LiteLLM's CustomStreamWrapper. Lets a single fixture be
+    consumed via either ``for`` or ``async for``.
+    """
+
+    def __init__(self, items):
+        self._items = list(items)
+        self._idx = 0
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self._idx >= len(self._items):
+            raise StopIteration
+        v = self._items[self._idx]
+        self._idx += 1
+        return v
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if self._idx >= len(self._items):
+            raise StopAsyncIteration
+        v = self._items[self._idx]
+        self._idx += 1
+        return v
+
+
+def _make_chunk(tool_calls: List[dict]) -> MagicMock:
+    """Build a minimal mock ModelResponseStream with the given tool_calls in delta."""
+    chunk = MagicMock()
+    chunk.choices = [MagicMock()]
+    chunk.choices[0].delta = MagicMock()
+    chunk.choices[0].delta.tool_calls = [
+        MagicMock(
+            id=tc["id"],
+            type="function",
+            function=MagicMock(name=tc["name"], arguments=tc["arguments"]),
+            index=tc.get("index", i),
+        )
+        for i, tc in enumerate(tool_calls)
+    ]
+    chunk.choices[0].finish_reason = None
+    return chunk
+
+
+class TestSplitChunkByToolCalls:
+    """Direct tests of the static ``_split_chunk_by_tool_calls`` helper."""
+
+    def test_none_chunk_passthrough(self):
+        assert AnthropicStreamWrapper._split_chunk_by_tool_calls(None) == [None]
+
+    def test_string_none_passthrough(self):
+        assert AnthropicStreamWrapper._split_chunk_by_tool_calls("None") == ["None"]
+
+    def test_single_tool_call_passthrough(self):
+        chunk = _make_chunk(
+            [
+                {"id": "call_a", "name": "Foo", "arguments": '{"x":1}'},
+            ]
+        )
+        result = AnthropicStreamWrapper._split_chunk_by_tool_calls(chunk)
+        assert len(result) == 1
+        assert result[0] is chunk  # same instance, not copied
+
+    def test_two_parallel_tool_calls_split(self):
+        chunk = _make_chunk(
+            [
+                {"id": "call_a", "name": "Foo", "arguments": '{"x":1}'},
+                {"id": "call_b", "name": "Bar", "arguments": '{"y":2}'},
+            ]
+        )
+        result = AnthropicStreamWrapper._split_chunk_by_tool_calls(chunk)
+        assert len(result) == 2
+        # Each split chunk holds exactly one tool_call
+        assert len(result[0].choices[0].delta.tool_calls) == 1
+        assert len(result[1].choices[0].delta.tool_calls) == 1
+        # Splits are independent copies (not aliasing the same list)
+        assert (
+            result[0].choices[0].delta.tool_calls
+            is not result[1].choices[0].delta.tool_calls
+        )
+        # Each split has the correct tool_call's id
+        assert result[0].choices[0].delta.tool_calls[0].id == "call_a"
+        assert result[1].choices[0].delta.tool_calls[0].id == "call_b"
+
+    def test_three_parallel_tool_calls_split(self):
+        chunk = _make_chunk(
+            [
+                {"id": "call_a", "name": "Foo", "arguments": "{}"},
+                {"id": "call_b", "name": "Bar", "arguments": "{}"},
+                {"id": "call_c", "name": "Baz", "arguments": "{}"},
+            ]
+        )
+        result = AnthropicStreamWrapper._split_chunk_by_tool_calls(chunk)
+        assert len(result) == 3
+        ids = [r.choices[0].delta.tool_calls[0].id for r in result]
+        assert ids == ["call_a", "call_b", "call_c"]
+
+    def test_chunk_without_choices_passthrough(self):
+        chunk = MagicMock()
+        chunk.choices = []
+        result = AnthropicStreamWrapper._split_chunk_by_tool_calls(chunk)
+        assert result == [chunk]
+
+    def test_chunk_with_none_delta_passthrough(self):
+        chunk = MagicMock()
+        chunk.choices = [MagicMock()]
+        chunk.choices[0].delta = None
+        result = AnthropicStreamWrapper._split_chunk_by_tool_calls(chunk)
+        assert result == [chunk]
+
+
+class TestMultiToolCallSplitterSync:
+    """Sync iteration of the dual-protocol splitter."""
+
+    def test_normal_chunks_unchanged(self):
+        c1 = _make_chunk([{"id": "a", "name": "F", "arguments": "{}"}])
+        c2 = _make_chunk([{"id": "b", "name": "G", "arguments": "{}"}])
+        splitter = _MultiToolCallSplitter(_DualProtocolStream([c1, c2]))
+        result = list(splitter)
+        assert len(result) == 2
+
+    def test_multi_tool_call_chunk_expands(self):
+        single = _make_chunk([{"id": "a", "name": "F", "arguments": "{}"}])
+        multi = _make_chunk(
+            [
+                {"id": "b", "name": "G", "arguments": "{}"},
+                {"id": "c", "name": "H", "arguments": "{}"},
+            ]
+        )
+        splitter = _MultiToolCallSplitter(_DualProtocolStream([single, multi]))
+        result = list(splitter)
+        # 1 (single) + 2 (multi split) = 3 chunks
+        assert len(result) == 3
+        ids = [r.choices[0].delta.tool_calls[0].id for r in result]
+        assert ids == ["a", "b", "c"]
+
+
+class TestMultiToolCallSplitterAsync:
+    """Async iteration of the dual-protocol splitter."""
+
+    def test_async_multi_tool_call_chunk_expands(self):
+        async def runit():
+            single = _make_chunk([{"id": "a", "name": "F", "arguments": "{}"}])
+            multi = _make_chunk(
+                [
+                    {"id": "b", "name": "G", "arguments": "{}"},
+                    {"id": "c", "name": "H", "arguments": "{}"},
+                ]
+            )
+            splitter = _MultiToolCallSplitter(_DualProtocolStream([single, multi]))
+            out = []
+            async for sub in splitter:
+                out.append(sub)
+            return out
+
+        result = asyncio.run(runit())
+        assert len(result) == 3
+        ids = [r.choices[0].delta.tool_calls[0].id for r in result]
+        assert ids == ["a", "b", "c"]
+
+
+class TestStreamWrapperConstructor:
+    """Verify the wrapper supports both protocols on the same instance shape."""
+
+    def test_sync_consumption_of_dual_stream(self):
+        chunks = [_make_chunk([{"id": "a", "name": "F", "arguments": "{}"}])]
+        wrapper = AnthropicStreamWrapper(
+            _DualProtocolStream(chunks), model="test-model"
+        )
+        # The internal splitter must support sync iteration (this is
+        # what the existing ``AnthropicStreamWrapper.__next__`` for-loop uses)
+        out = list(wrapper._completion_stream_splitter)
+        assert len(out) == 1
+
+    def test_async_consumption_of_dual_stream(self):
+        single = _make_chunk([{"id": "a", "name": "F", "arguments": "{}"}])
+        multi = _make_chunk(
+            [
+                {"id": "b", "name": "G", "arguments": "{}"},
+                {"id": "c", "name": "H", "arguments": "{}"},
+            ]
+        )
+
+        async def drain():
+            wrapper = AnthropicStreamWrapper(
+                _DualProtocolStream([single, multi]), model="test-model"
+            )
+            out = []
+            async for sub in wrapper._completion_stream_splitter:
+                out.append(sub)
+            return out
+
+        result = asyncio.run(drain())
+        # Multi-tool-call delta is split: 1 + 2 = 3
+        assert len(result) == 3
+        ids = [r.choices[0].delta.tool_calls[0].id for r in result]
+        assert ids == ["a", "b", "c"]
+
+
+class TestAnthropicStreamWrapperEndToEnd:
+    """End-to-end SSE-event check.
+
+    Drives the full ``AnthropicStreamWrapper.__next__`` / ``__anext__``
+    pipeline (not just the splitter) and asserts the converted Anthropic
+    event sequence contains one ``content_block_start`` + ``content_block_stop``
+    pair per tool_call when an upstream OpenAI delta carries multiple.
+    Without the splitter, only the first tool_call would surface.
+    """
+
+    def _build_chunks(self):
+        """Build a 3-chunk sequence: text, then a delta with TWO parallel
+        tool_calls, then a finish chunk. Mirrors ``mlx_lm.server``'s output
+        shape for parallel tool calls.
+        """
+        from litellm.types.utils import (
+            ChatCompletionDeltaToolCall,
+            Delta,
+            Function,
+            ModelResponseStream,
+            StreamingChoices,
+            Usage,
+        )
+
+        text_chunk = ModelResponseStream(
+            id="chatcmpl-1",
+            created=1700000000,
+            model="test-model",
+            object="chat.completion.chunk",
+            choices=[
+                StreamingChoices(
+                    index=0,
+                    delta=Delta(content="ok", role="assistant"),
+                    finish_reason=None,
+                )
+            ],
+        )
+
+        multi_tool_chunk = ModelResponseStream(
+            id="chatcmpl-1",
+            created=1700000000,
+            model="test-model",
+            object="chat.completion.chunk",
+            choices=[
+                StreamingChoices(
+                    index=0,
+                    delta=Delta(
+                        tool_calls=[
+                            ChatCompletionDeltaToolCall(
+                                id="call_a",
+                                type="function",
+                                function=Function(
+                                    name="get_weather",
+                                    arguments='{"city": "Tokyo"}',
+                                ),
+                                index=0,
+                            ),
+                            ChatCompletionDeltaToolCall(
+                                id="call_b",
+                                type="function",
+                                function=Function(
+                                    name="get_time",
+                                    arguments='{"tz": "Asia/Tokyo"}',
+                                ),
+                                index=1,
+                            ),
+                        ]
+                    ),
+                    finish_reason=None,
+                )
+            ],
+        )
+
+        finish_chunk = ModelResponseStream(
+            id="chatcmpl-1",
+            created=1700000000,
+            model="test-model",
+            object="chat.completion.chunk",
+            choices=[
+                StreamingChoices(index=0, delta=Delta(), finish_reason="tool_calls")
+            ],
+            usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15),
+        )
+
+        return [text_chunk, multi_tool_chunk, finish_chunk]
+
+    def test_sync_two_parallel_tool_calls_yield_two_content_blocks(self):
+        chunks = self._build_chunks()
+        wrapper = AnthropicStreamWrapper(
+            _DualProtocolStream(chunks), model="test-model"
+        )
+
+        events = list(wrapper)
+
+        tool_starts = [
+            e
+            for e in events
+            if isinstance(e, dict)
+            and e.get("type") == "content_block_start"
+            and e.get("content_block", {}).get("type") == "tool_use"
+        ]
+        assert len(tool_starts) == 2, (
+            f"Expected 2 content_block_start of type tool_use, got "
+            f"{len(tool_starts)}. Event types: "
+            f"{[e.get('type') for e in events if isinstance(e, dict)]}"
+        )
+        # Both tool_use blocks carry distinct ids
+        ids = [e["content_block"]["id"] for e in tool_starts]
+        assert "call_a" in ids
+        assert "call_b" in ids
+
+        # Each tool_use must be paired with at least one input_json_delta
+        # carrying its arguments
+        deltas_by_index = {}
+        for e in events:
+            if (
+                isinstance(e, dict)
+                and e.get("type") == "content_block_delta"
+                and e.get("delta", {}).get("type") == "input_json_delta"
+            ):
+                deltas_by_index.setdefault(e["index"], []).append(e)
+        # Indices for the two tool_use blocks (both > 0; index 0 was the
+        # leading text block)
+        tool_indices = [e["index"] for e in tool_starts]
+        for idx in tool_indices:
+            assert (
+                idx in deltas_by_index
+            ), f"No input_json_delta for tool_use at content_block index {idx}"
+
+    def test_async_two_parallel_tool_calls_yield_two_content_blocks(self):
+        async def drain():
+            chunks = self._build_chunks()
+            wrapper = AnthropicStreamWrapper(
+                _DualProtocolStream(chunks), model="test-model"
+            )
+            out = []
+            async for e in wrapper:
+                out.append(e)
+            return out
+
+        events = asyncio.run(drain())
+        tool_starts = [
+            e
+            for e in events
+            if isinstance(e, dict)
+            and e.get("type") == "content_block_start"
+            and e.get("content_block", {}).get("type") == "tool_use"
+        ]
+        assert len(tool_starts) == 2
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])