Python: Enhance _OutputItemTracker to prevent duplicate function call streaming

python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

@@ -967,6 +967,7 @@ def __init__(self, stream: ResponseEventStream) -> None:
         self._reasoning_encrypted_content: str | None = None
         self._fc_builder: OutputItemFunctionCallBuilder | None = None
         self._mcp_builder: OutputItemMcpCallBuilder | None = None
+        self._outstanding_function_calls: dict[str, str | None] = {}
         self.needs_async = False
 
     def handle(self, content: Content) -> Generator[ResponseStreamEvent]:
@@ -995,6 +996,15 @@ def handle(self, content: Content) -> Generator[ResponseStreamEvent]:
                     yield self._summary_part.emit_text_delta(content.text)
 
         elif content.type == "function_call" and content.call_id is not None:
+            # Declaration-only calls replay request metadata after the streamed call. Scope suppression to the
+            # outstanding occurrence because a call_id may be reused after its terminal result.
+            if (
+                content.user_input_request
+                and content.arguments is None
+                and content.call_id in self._outstanding_function_calls
+                and self._outstanding_function_calls[content.call_id] == content.name
+            ):
+                return
             if self._active_type != "function_call" or self._active_id != content.call_id:
                 yield from self._close()
                 yield from self._open_function_call(content)
@@ -1003,6 +1013,12 @@ def handle(self, content: Content) -> Generator[ResponseStreamEvent]:
             if self._fc_builder is not None:
                 yield self._fc_builder.emit_arguments_delta(args_str)
 
+        elif content.type == "function_result":
+            yield from self._close()
+            if content.call_id is not None:
+                self._outstanding_function_calls.pop(content.call_id, None)
+            self.needs_async = True
+
         elif content.type == "mcp_server_tool_call" and content.tool_name:
             key = content.call_id or f"{content.server_name or 'default'}::{content.tool_name}"
             if self._active_type != "mcp_server_tool_call" or self._active_id != key:
@@ -1080,6 +1096,7 @@ def _open_function_call(self, content: Content) -> Generator[ResponseStreamEvent
         )
         self._active_type = "function_call"
         self._active_id = content.call_id
+        self._outstanding_function_calls[content.call_id or ""] = content.name
         yield self._fc_builder.emit_added()
 
     def _open_mcp_call(self, content: Content) -> Generator[ResponseStreamEvent]:

python/packages/foundry_hosting/tests/test_responses.py

@@ -1265,6 +1265,66 @@ async def test_function_call_streaming(self) -> None:
         assert len(args_done) == 1
         assert args_done[0]["data"]["arguments"] == '{"q": "hello"}'
 
+    @pytest.mark.parametrize(("arguments", "expected_count"), [(None, 1), ("", 2)])
+    async def test_declaration_only_metadata_replay_requires_none_arguments(
+        self, arguments: str | None, expected_count: int
+    ) -> None:
+        metadata = Content.from_function_call("call_1", "search", arguments=arguments)
+        metadata.id = "call_1"
+        metadata.user_input_request = True
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[Content.from_function_call("call_1", "search", arguments='{"q": "hello"}')],
+                    role="assistant",
+                ),
+                AgentResponseUpdate(contents=[Content.from_text("Waiting for the result")], role="assistant"),
+                AgentResponseUpdate(contents=[metadata], role="assistant"),
+            ]
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        function_items = [
+            event
+            for event in events
+            if event["event"] == "response.output_item.added" and event["data"]["item"]["type"] == "function_call"
+        ]
+        assert len(function_items) == expected_count
+
+    async def test_function_call_id_can_be_reused_after_terminal_result(self) -> None:
+        reused_call = Content.from_function_call("call_1", "search", arguments=None)
+        reused_call.id = "call_1"
+        reused_call.user_input_request = True
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[Content.from_function_call("call_1", "search", arguments='{"q": "first"}')],
+                    role="assistant",
+                ),
+                AgentResponseUpdate(
+                    contents=[Content.from_function_result("call_1", result="first result")],
+                    role="tool",
+                ),
+                AgentResponseUpdate(contents=[reused_call], role="assistant"),
+            ]
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, stream=True)
+
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        function_items = [
+            event
+            for event in events
+            if event["event"] == "response.output_item.added" and event["data"]["item"]["type"] == "function_call"
+        ]
+        assert [event["data"]["item"]["call_id"] for event in function_items] == ["call_1", "call_1"]
+
     async def test_function_call_streaming_serializes_dataclass_arguments(self) -> None:
         @dataclass
         class HandoffLikeRequest: