Python: Preserve declaration-only streaming metadata

docs/specs/004-python-function-calling-loop.md

@@ -380,7 +380,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Calls across response messages | Every actionable call is executed once. | `test_base_client_executes_function_calls_across_multiple_response_messages` |
 | Parallel calls | Results retain the corresponding call ids and execution count. | `test_max_function_calls_limits_parallel_invocations`, `test_streaming_multiple_function_calls_parallel_execution` |
 | Informational-only call | The call is returned but not executed or approved. | `test_informational_only_function_call_is_not_invoked`, `test_informational_only_function_call_does_not_request_approval`, `test_streaming_informational_only_function_call_is_not_invoked` |
-| Declaration-only call | The call is surfaced as user input and is not executed. | `test_declaration_only_tool` |
+| Declaration-only call | The call is surfaced as user input and is not executed; streaming arguments appear once while finalized request metadata remains available. | `test_declaration_only_tool`, `test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments` |
 | Function invocation disabled | The client bypasses the invocation loop without losing invocation kwargs. | `test_function_invocation_config_enabled_false`, `test_function_invocation_config_enabled_false_preserves_invocation_kwargs`, `test_streaming_function_invocation_config_enabled_false` |
 | Runtime tool changes | Added tools become available on the next iteration and retain approval behavior. | `test_add_tools_available_next_iteration`, `test_add_tools_with_approval_required_tool` |
 
@@ -486,10 +486,7 @@ These scenarios are required but are not fully covered by merged tests on `main`
 
 | Gap | Tracking |
 |---|---|
-| Service-side storage sends the current approval response while omitting the stored request. | #7125 |
 | Service-owned `previous_response_id` continuation cannot execute a terminal approval again on a later turn. | #6851 |
-| A provider that ignores `tool_choice="none"` after an invocation limit cannot expose an unanswered call. | #7045 |
-| Declaration-only streaming preserves request metadata without duplicating arguments. | #6973 |
 
 Do not mark these rows covered by nearby tests; each needs a dedicated regression at the owning layer.
 
@@ -541,6 +538,9 @@ Before accepting an update, reviewers must confirm:
 - #7043 — provider-injected approval execution
 - #6828 — AG-UI `confirm_changes` snapshot correlation
 - #7212 — non-adjacent and reused-id compaction integrity
+- #7125 — service-side approval response serialization
+- #7045 — post-limit tool-content transcript integrity
+- #6973 — declaration-only streaming metadata and argument integrity
 - #6851 — duplicate side effects after approval continuation
 - #7383 — bind approval responses to framework-issued requests after this foundation merges
 - #6963 / #7095 — opaque reasoning-signature replay

python/packages/core/AGENTS.md

@@ -160,6 +160,9 @@ agent_framework/
 - Once an invocation limit disables local tools, locally actionable calls and local approval requests are removed
   from streaming and final output. Provider-executed informational call/result pairs, hosted approval requests, and
   metadata-only stream updates remain visible.
+- Declaration-only streamed calls emit their arguments only from the provider stream. The function layer sends a
+  metadata-only follow-up (`arguments=None`) so `id` and `user_input_request` survive final aggregation without
+  duplicating arguments.
 - `function_approval_request` and `function_approval_response` are control-plane contents. History providers may
   retain them in their backing store for audit. The base `HistoryProvider.before_run` filters resolved wrappers from
   later model replay, but preserves unresolved requests/responses until a terminal result or follow-up request closes

python/packages/core/agent_framework/_tools.py

@@ -2676,11 +2676,19 @@ def _handle_function_call_results(
                 response.messages[0].contents.extend(new_items)
             else:
                 response.messages.append(Message(role="assistant", contents=new_items))
+        streaming_items: list[Content] = []
+        for result in execution_results:
+            if result.type == "function_call":
+                metadata_only_result = copy.copy(result)
+                metadata_only_result.arguments = None
+                streaming_items.append(metadata_only_result)
+            else:
+                streaming_items.append(result)
         return _FunctionProcessingResult(
             errors_in_a_row=errors_in_a_row,
             action="return",
             function_call_count=function_call_count,
-            streaming_updates=(ChatResponseUpdate(contents=execution_results, role="assistant"),),
+            streaming_updates=(ChatResponseUpdate(contents=streaming_items, role="assistant"),),
         )
 
     errors_in_a_row, reached_error_limit = _update_consecutive_error_count(

python/packages/core/agent_framework/_types.py

@@ -1558,6 +1558,8 @@ 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=self.id or other.id,
+            user_input_request=self.user_input_request or other.user_input_request,
             exception=getattr(self, "exception", None) or getattr(other, "exception", None),
             informational_only=getattr(self, "informational_only", False)
             or getattr(other, "informational_only", False),

python/packages/core/tests/core/test_function_invocation_logic.py

@@ -3377,6 +3377,68 @@ async def test_declaration_only_tool(chat_client_base: SupportsChatGetResponse):
     assert len(function_results) == 0
 
 
+@pytest.mark.parametrize(
+    "argument_chunks",
+    [
+        ['{"location":', '"Seattle"}'],
+        ['{"location":"Seattle"}'],
+    ],
+    ids=["split-arguments", "single-chunk"],
+)
+async def test_streaming_declaration_only_tool_preserves_metadata_without_duplicate_arguments(
+    chat_client_base: SupportsChatGetResponse,
+    argument_chunks: list[str],
+) -> None:
+    """Declaration-only streaming emits metadata once without replaying already-streamed arguments."""
+    from agent_framework import FunctionTool
+
+    declaration_tool = 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="call_weather",
+                        name="get_weather",
+                        arguments=arguments,
+                    )
+                ],
+                role="assistant",
+            )
+            for arguments in argument_chunks
+        ]
+    ]
+
+    stream = chat_client_base.get_response(
+        [Message(role="user", contents=["What's the weather in Seattle?"])],
+        options={"tool_choice": "auto", "tools": [declaration_tool]},
+        stream=True,
+    )
+    metadata_updates: list[tuple[Any, str | None]] = []
+    async for update in stream:
+        for content in update.contents:
+            if content.type == "function_call" and content.call_id == "call_weather" and content.user_input_request:
+                metadata_updates.append((content.arguments, content.id))
+    final_response = await stream.get_final_response()
+    function_calls = [
+        content
+        for message in final_response.messages
+        for content in message.contents
+        if content.type == "function_call" and content.call_id == "call_weather"
+    ]
+
+    assert metadata_updates == [(None, "call_weather")]
+    assert len(function_calls) == 1
+    assert function_calls[0].arguments == '{"location":"Seattle"}'
+    assert function_calls[0].user_input_request is True
+    assert function_calls[0].id == "call_weather"
+
+
 async def test_multiple_function_calls_parallel_execution(chat_client_base: SupportsChatGetResponse):
     """Test that multiple function calls are executed in parallel."""
     import asyncio

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

@@ -571,6 +571,15 @@ def test_function_call_content_add_merging_and_errors():
     c = a + b
     assert c.informational_only is True
 
+    # control metadata is preserved when a metadata-only update follows argument chunks
+    metadata = Content.from_function_call(call_id="1", name="f", arguments=None)
+    metadata.id = "1"
+    metadata.user_input_request = True
+    c = c + metadata
+    assert c.arguments == '{"x":1}'
+    assert c.id == "1"
+    assert c.user_input_request is True
+
     # incompatible argument types
     a = Content.from_function_call(call_id="1", name="f", arguments="abc")
     b = Content.from_function_call(call_id="1", name="f", arguments={"y": 2})