From 10aa390aca4ec0c9547e2726b03339fc0ac64917 Mon Sep 17 00:00:00 2001 From: kartikmadan11 Date: Mon, 13 Jul 2026 16:11:00 +0100 Subject: [PATCH] fix: avoid duplicating arguments in streamed declaration-only tool calls (#6973) When streaming a declaration-only function call, the _stream() loop yields function_call chunks from the inner stream, then _process_function_requests re-emits the finalized call from function_call_results. _process_update merges same-call_id function_call contents via +=, so re-emitting the full arguments concatenated them twice: {"location":"Seattle"}{"location":"Seattle"}. The finalized call gains its control metadata (user_input_request=True, id) only during post-processing, so it must still reach the stream: AgentExecutor relies on that signal to emit request_info and pause for client-side tools. Fix: strip only the already-streamed arguments from function_call items in the streamed update while preserving their metadata, instead of dropping the items entirely. Also preserve id and user_input_request when merging function_call contents in Content.__add__ so the aggregated response retains the signal. Tests: split-chunk and single-chunk no-duplication, plus metadata preservation. Fixes #6973 --- .../packages/core/agent_framework/_tools.py | 16 +- .../packages/core/agent_framework/_types.py | 4 + .../core/test_function_invocation_logic.py | 158 ++++++++++++++++++ 3 files changed, 177 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 6a6354419be..1c07a2a12f3 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -4,6 +4,7 @@ import asyncio import contextvars +import copy import inspect import json import logging @@ -2400,7 +2401,20 @@ async def _process_function_requests( had_errors=had_errors, max_errors=max_errors, ) - result["function_call_results"] = list(function_call_results) + # The inner stream already yielded the declaration-only function_call chunks as they + # arrived, so re-emitting their arguments would make _process_update (which merges via +=) + # concatenate the argument string twice. Strip the arguments from any function_call items + # while retaining their control metadata (id, user_input_request) so streaming consumers + # (e.g. AgentExecutor) still receive the request-info/pause signal without duplicated arguments. + streamed_results: list[Content] = [] + for r in function_call_results: + if r.type == "function_call": + metadata_only = copy.copy(r) + metadata_only.arguments = None + streamed_results.append(metadata_only) + else: + streamed_results.append(r) + result["function_call_results"] = streamed_results result["function_call_count"] = sum(1 for r in function_call_results if r.type == "function_result") # If middleware requested termination, change action to return if should_terminate: diff --git a/python/packages/core/agent_framework/_types.py b/python/packages/core/agent_framework/_types.py index edcf981d4df..c1fd4eb4cb4 100644 --- a/python/packages/core/agent_framework/_types.py +++ b/python/packages/core/agent_framework/_types.py @@ -1548,6 +1548,10 @@ def _add_function_call_content(self, other: Content) -> Content: call_id=self_call_id, name=getattr(self, "name", None) or getattr(other, "name", None), arguments=arguments, + id=getattr(self, "id", None) or getattr(other, "id", None), + user_input_request=( + getattr(self, "user_input_request", None) or getattr(other, "user_input_request", None) + ), exception=getattr(self, "exception", None) or getattr(other, "exception", None), informational_only=getattr(self, "informational_only", False) or getattr(other, "informational_only", False), diff --git a/python/packages/core/tests/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py index 6426fade633..8881f0198e5 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -4786,3 +4786,161 @@ def a(x: int) -> int: # endregion + + +async def test_declaration_only_tool_streaming_no_argument_duplication( + chat_client_base: SupportsChatGetResponse, +): + """Streaming a declaration-only tool must not duplicate arguments in the final response (fix #6973).""" + from agent_framework import FunctionTool + + declaration_func = FunctionTool( + name="get_weather", + func=None, + description="Get the weather", + input_model={"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}, + ) + assert declaration_func.declaration_only is True + + # Stream arrives in two chunks, arguments split across them + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call1", name="get_weather", arguments='{"location":')], + role="assistant", + ), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call1", name="get_weather", arguments='"Seattle"}')], + role="assistant", + ), + ], + ] + + updates = [] + async for update in chat_client_base.get_response( # type: ignore[call-overload] # pyrefly: ignore[no-matching-overload] # ty: ignore[no-matching-overload] + "What's the weather in Seattle?", # type: ignore[arg-type] + options={"tool_choice": "auto", "tools": [declaration_func]}, # type: ignore[arg-type] + stream=True, # type: ignore[arg-type] + ): + updates.append(update) + + # Collect arguments from all streamed function_call updates + all_args = "" + for update in updates: + for content in update.contents: + if content.type == "function_call" and content.call_id == "call1": + all_args += content.arguments or "" + + # Arguments must appear exactly once - not duplicated + assert all_args == '{"location":"Seattle"}', f"Expected single copy of arguments, got: {all_args!r}" + + +async def test_declaration_only_tool_streaming_single_chunk_no_duplication( + chat_client_base: SupportsChatGetResponse, +): + """Single-chunk streaming of a declaration-only tool must not duplicate arguments (fix #6973).""" + from agent_framework import FunctionTool + + declaration_func = FunctionTool( + name="get_weather", + func=None, + description="Get the weather", + input_model={"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}, + ) + + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[ + Content.from_function_call(call_id="call1", name="get_weather", arguments='{"location":"Seattle"}') + ], + role="assistant", + ), + ], + ] + + final_response = await chat_client_base.get_response( # type: ignore[call-overload] # pyrefly: ignore[no-matching-overload] # ty: ignore[no-matching-overload] + "What's the weather in Seattle?", # type: ignore[arg-type] + options={"tool_choice": "auto", "tools": [declaration_func]}, # type: ignore[arg-type] + stream=True, # type: ignore[arg-type] + ).get_final_response() + + function_calls = [ + content + for msg in final_response.messages + for content in msg.contents + if content.type == "function_call" and content.call_id == "call1" + ] + assert len(function_calls) == 1, f"Expected exactly 1 function_call, got {len(function_calls)}" + assert function_calls[0].arguments == '{"location":"Seattle"}', ( + f"Arguments duplicated: {function_calls[0].arguments!r}" + ) + + +async def test_declaration_only_tool_streaming_preserves_user_input_request_metadata( + chat_client_base: SupportsChatGetResponse, +): + """Streaming a declaration-only tool must retain user_input_request/id metadata (fix #6973). + + The inner stream yields raw argument chunks without the control metadata; the finalized + call gains ``user_input_request=True`` and ``id`` during post-processing. Streaming + consumers (e.g. AgentExecutor) rely on that signal to emit request_info and pause, so it + must reach the stream even though the duplicated arguments are suppressed. + """ + from agent_framework import FunctionTool + + declaration_func = FunctionTool( + name="get_weather", + func=None, + description="Get the weather", + input_model={"type": "object", "properties": {"location": {"type": "string"}}, "required": ["location"]}, + ) + assert declaration_func.declaration_only is True + + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call1", name="get_weather", arguments='{"location":')], + role="assistant", + ), + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="call1", name="get_weather", arguments='"Seattle"}')], + role="assistant", + ), + ], + ] + + stream = chat_client_base.get_response( # type: ignore[call-overload] # pyrefly: ignore[no-matching-overload] # ty: ignore[no-matching-overload] + "What's the weather in Seattle?", # type: ignore[arg-type] + options={"tool_choice": "auto", "tools": [declaration_func]}, # type: ignore[arg-type] + stream=True, # type: ignore[arg-type] + ) + + updates = [] + async for update in stream: + updates.append(update) + + # At least one streamed update must carry the user_input_request signal with the call id, + # so consumers can pause the workflow / emit request_info for the client-side tool. + signalled = [ + content + for update in updates + for content in update.contents + if content.type == "function_call" and content.call_id == "call1" and content.user_input_request + ] + assert signalled, "Expected a streamed update carrying user_input_request metadata for the declaration-only call" + assert any(getattr(content, "id", None) == "call1" for content in signalled), ( + "Expected the finalized declaration-only call to retain its id" + ) + + # The finalized aggregated response must also retain the signal without duplicated arguments. + final_response = await stream.get_final_response() + requests = [ + content + for msg in final_response.messages + for content in msg.contents + if content.type == "function_call" and content.call_id == "call1" and content.user_input_request + ] + assert len(requests) == 1, f"Expected exactly one user input request, got {len(requests)}" + assert requests[0].user_input_request is True + assert requests[0].arguments == '{"location":"Seattle"}', f"Arguments duplicated or lost: {requests[0].arguments!r}"