diff --git a/python/packages/core/agent_framework/_tools.py b/python/packages/core/agent_framework/_tools.py index 588ae93d5c..ff3bbff46e 100644 --- a/python/packages/core/agent_framework/_tools.py +++ b/python/packages/core/agent_framework/_tools.py @@ -2362,14 +2362,18 @@ async def _process_function_requests( ) _replace_approval_contents_with_results(prepped_messages, fcc_todo, approved_function_results) executed_count = sum(1 for r in approved_function_results if r.type == "function_result") + # Surface the executed approval results so streaming consumers observe them, mirroring + # the normal tool path. Without this, tools executed here (e.g. provider-injected tools + # approved and deferred to in-run execution) run silently and never emit a result update. + streamed_results = [r for r in approved_function_results if r.type == "function_result"] # Continue to call chat client with updated messages (containing function results) # so it can generate the final response return { "action": "return" if should_terminate else "continue", "errors_in_a_row": errors_in_a_row, "result_message": None, - "update_role": None, - "function_call_results": None, + "update_role": "tool" if streamed_results else None, + "function_call_results": streamed_results or None, "function_call_count": executed_count, } @@ -2777,6 +2781,13 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]: errors_in_a_row = approval_result.get("errors_in_a_row", errors_in_a_row) total_function_calls += approval_result.get("function_call_count", 0) budget_state["total_function_calls"] = total_function_calls + # Emit results for tools executed while resolving approvals (e.g. deferred + # provider-injected tools) so streaming consumers see them like normal tool results. + if role := approval_result.get("update_role"): + yield ChatResponseUpdate( + contents=approval_result.get("function_call_results") or [], + role=role, + ) if max_function_calls is not None and total_function_calls >= max_function_calls: logger.info( "Maximum function calls reached (%d/%d). Stopping further function calls for this request.", 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 6426fade63..1769410635 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -3713,6 +3713,84 @@ def func_with_approval(arg1: str) -> str: assert exec_counter == 0 +async def test_streaming_approved_tool_result_is_streamed(chat_client_base: SupportsChatGetResponse): + """A tool executed while resolving an approval must stream its result (fix #7241). + + Approval resolution runs through the prepped_messages branch of + _process_function_requests, which used to return update_role=None, so the executed + result was spliced into the message history but never yielded as an update. Streaming + consumers (e.g. the AG-UI transport) therefore never observed it. The result must be + streamed like any other tool result. + """ + exec_counter = 0 + + @tool(name="test_func", approval_mode="always_require") + def func_with_approval(arg1: str) -> str: + nonlocal exec_counter + exec_counter += 1 + return f"Result {arg1}" + + # Turn 1: stream the function call and collect the generated approval request. + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ + ChatResponseUpdate( + contents=[Content.from_function_call(call_id="1", name="test_func", arguments='{"arg1": "value1"}')], + role="assistant", + ), + ], + ] + + request_updates = [] + async for update in chat_client_base.get_response( # type: ignore[call-overload] # pyrefly: ignore[no-matching-overload] # ty: ignore[no-matching-overload] + "hello", # type: ignore[arg-type] + options={"tool_choice": "auto", "tools": [func_with_approval]}, + stream=True, # type: ignore[arg-type] + ): + request_updates.append(update) + + approval_req = next( + content + for update in request_updates + for content in update.contents + if content.type == "function_approval_request" + ) + assert exec_counter == 0 # not executed until approved + + # Turn 2: resume streaming with the approval response. The tool executes during resolution. + approval_response = Content.from_function_approval_response( + id=approval_req.id, # type: ignore[arg-type] + function_call=approval_req.function_call, # type: ignore[arg-type] + approved=True, + ) + persisted_messages = [ + Message(role="user", contents=["hello"]), + Message(role="assistant", contents=[approval_req]), + Message(role="user", contents=[approval_response]), + ] + chat_client_base.streaming_responses = [ # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + [ChatResponseUpdate(contents=[Content.from_text(text="done")], role="assistant")], + ] + + resume_updates = [] + async for update in chat_client_base.get_response( # type: ignore[call-overload] # pyrefly: ignore[no-matching-overload] # ty: ignore[no-matching-overload] + persisted_messages, # type: ignore[arg-type] + options={"tool_choice": "auto", "tools": [func_with_approval]}, + stream=True, # type: ignore[arg-type] + ): + resume_updates.append(update) + + assert exec_counter == 1 # executed exactly once after approval + # The executed approval result must be streamed (not merely spliced into the history). + streamed_results = [ + content + for update in resume_updates + for content in update.contents + if content.type == "function_result" and content.call_id == "1" + ] + assert len(streamed_results) == 1, f"Expected one streamed function_result, got {len(streamed_results)}" + assert streamed_results[0].result == "Result value1" + + async def test_streaming_error_recovery_resets_counter(chat_client_base: SupportsChatGetResponse): """Test that error counter resets after a successful function call in streaming mode."""