Python: Fix streamed parallel tool calls merging into the wrong call

python/packages/core/agent_framework/_types.py

@@ -2032,12 +2032,8 @@ def _process_update(response: ChatResponse | AgentResponse, update: ChatResponse
                 logger.warning(f"Skipping unknown content type or invalid content: {exc}")
                 continue
         match content_type:
-            # mypy doesn't narrow type based on match/case, but we know these are FunctionCallContents
-            case "function_call" if message.contents and message.contents[-1].type == "function_call":
-                try:
-                    message.contents[-1] += content
-                except (AdditionItemMismatch, ContentError):
-                    message.contents.append(content)
+            case "function_call":
+                _merge_function_call_content(message, content)
             case "usage":
                 if response.usage_details is None:
                     response.usage_details = UsageDetails()
@@ -2075,6 +2071,38 @@ def _process_update(response: ChatResponse | AgentResponse, update: ChatResponse
     response.continuation_token = update.continuation_token
 
 
+def _merge_function_call_content(message: Message, content: Content) -> None:
+    """Merge a streamed function_call chunk into the in-progress call it belongs to.
+
+    Providers can stream multiple tool calls in parallel, so the next function_call
+    chunk is not necessarily a continuation of the most recently appended one; a chunk
+    with a call_id is matched against existing contents by that id first. Chunks with
+    no call_id (continuation deltas some providers only stamp on the first chunk) fall
+    back to merging with the trailing function_call item, preserving prior behavior.
+    """
+    call_id = getattr(content, "call_id", None)
+    if call_id:
+        for index in range(len(message.contents) - 1, -1, -1):
+            existing = message.contents[index]
+            if existing.type == "function_call" and getattr(existing, "call_id", None) == call_id:
+                try:
+                    message.contents[index] = existing + content
+                except (AdditionItemMismatch, ContentError):
+                    break
+                return
+        # A tagged chunk that matches no in-progress call is a new call, not a
+        # continuation - an untagged trailing item would silently absorb it otherwise.
+        message.contents.append(content)
+        return
+    if message.contents and message.contents[-1].type == "function_call":
+        try:
+            message.contents[-1] += content
+            return
+        except (AdditionItemMismatch, ContentError):
+            pass
+    message.contents.append(content)
+
+
 def _coalesce_text_content(contents: list[Content], type_str: Literal["text", "text_reasoning"]) -> None:
     """Take any subsequence Text or TextReasoningContent items and coalesce them into a single item."""
     if not contents:

python/packages/core/tests/core/test_types.py

@@ -1912,6 +1912,87 @@ def test_function_call_incompatible_ids_are_not_merged():
     assert len(fcs) == 2
 
 
+def test_function_call_interleaved_parallel_streaming_merges_by_call_id():
+    """Argument deltas for two parallel tool calls can interleave; each must land on its own call.
+
+    This mirrors how the OpenAI Responses API streams parallel tool calls: every
+    ``response.function_call_arguments.delta`` event is tagged with the call's real
+    call_id (tracked per output_index), but deltas for different calls are not
+    guaranteed to arrive grouped together. Before this fix, only the trailing
+    content item was ever considered a merge target, so an out-of-turn delta for an
+    earlier call_id was appended as a stray duplicate instead of being folded into
+    its call, leaving both calls with incomplete, unparsable arguments.
+    """
+    updates = [
+        ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_1", name="get_weather", arguments="")]),
+        ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_2", name="get_time", arguments="")]),
+        ChatResponseUpdate(
+            contents=[Content.from_function_call(call_id="call_1", name="get_weather", arguments='{"location":')]
+        ),
+        ChatResponseUpdate(
+            contents=[Content.from_function_call(call_id="call_2", name="get_time", arguments='{"timezone":')]
+        ),
+        ChatResponseUpdate(
+            contents=[Content.from_function_call(call_id="call_1", name="get_weather", arguments='"NYC"}')]
+        ),
+        ChatResponseUpdate(
+            contents=[Content.from_function_call(call_id="call_2", name="get_time", arguments='"EST"}')]
+        ),
+    ]
+
+    resp = ChatResponse.from_updates(updates)
+    assert len(resp.messages) == 1
+    fcs = [c for c in resp.messages[0].contents if c.type == "function_call"]
+    assert len(fcs) == 2
+
+    by_call_id = {c.call_id: c for c in fcs}
+    assert by_call_id["call_1"].arguments == '{"location":"NYC"}'
+    assert by_call_id["call_2"].arguments == '{"timezone":"EST"}'
+
+
+def test_function_call_merge_falls_back_to_trailing_item_without_call_id():
+    """Continuation deltas some providers never re-stamp with a call_id still merge.
+
+    Not every provider repeats the call_id on every streamed chunk (e.g. the OpenAI
+    Chat Completions API only sends it on the first delta for a tool call), so the
+    fallback of merging into the trailing function_call item must still hold for the
+    single-call-in-flight case.
+    """
+    updates = [
+        ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_1", name="f", arguments="{")]),
+        ChatResponseUpdate(contents=[Content.from_function_call(call_id="", name="", arguments="}")]),
+    ]
+
+    resp = ChatResponse.from_updates(updates)
+    fcs = [c for c in resp.messages[0].contents if c.type == "function_call"]
+    assert len(fcs) == 1
+    assert fcs[0].call_id == "call_1"
+    assert fcs[0].arguments == "{}"
+
+
+def test_function_call_tagged_chunk_does_not_absorb_into_untagged_trailing_call():
+    """A tagged chunk with no matching in-progress call must not merge into an untagged one.
+
+    Content.__add__ only rejects a merge when *both* sides carry a call_id and they
+    differ, so an untagged trailing item (call_id falsy) would otherwise silently
+    accept a chunk tagged with a brand-new call_id, adopting that id and
+    concatenating unrelated arguments. The trailing-item fallback must be reserved
+    for chunks that carry no call_id at all.
+    """
+    untagged = Content("function_call", call_id=None, name="a", arguments="partial-a")
+    resp = ChatResponse.from_updates([
+        ChatResponseUpdate(contents=[untagged]),
+        ChatResponseUpdate(contents=[Content.from_function_call(call_id="call_new", name="b", arguments="{}")]),
+    ])
+
+    fcs = [c for c in resp.messages[0].contents if c.type == "function_call"]
+    assert len(fcs) == 2
+    assert fcs[0].call_id is None
+    assert fcs[0].arguments == "partial-a"
+    assert fcs[1].call_id == "call_new"
+    assert fcs[1].arguments == "{}"
+
+
 # region Role & FinishReason basics