diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py b/python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py index a7dfe340bf..458b2b714a 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_executors_agents.py @@ -506,6 +506,13 @@ def _normalize_variable_path(variable: str) -> str: class InvokeAzureAgentExecutor(DeclarativeActionExecutor): """Executor that invokes a Microsoft Foundry agent. + ``output.autoSend`` defaults to true and accepts a Boolean or a + ``=``-prefixed PowerFx Boolean expression, such as ``=Local.publishResult``. + Expressions use current state before each invocation, including resumed + external-loop turns. False suppresses automatic output, not invocation, + result storage or conversation history; later actions can explicitly emit + the stored results. + This executor supports both Python-style and .NET-style YAML schemas: Python-style (simple): @@ -615,8 +622,8 @@ def _get_input_config(self) -> tuple[dict[str, Any], Any, str | None, int]: return arguments, messages, external_loop_when, max_iterations - def _get_output_config(self) -> tuple[str | None, str | None, str | None, bool]: - """Parse output configuration. + def _get_output_config(self, state: DeclarativeWorkflowState) -> tuple[str | None, str | None, str | None, bool]: + """Parse output bindings and evaluate autoSend against the current state. Returns: Tuple of (messages var, responseObject var, resultProperty, autoSend) @@ -637,7 +644,7 @@ def _get_output_config(self) -> tuple[str | None, str | None, str | None, bool]: property_val: Any = output_dict.get("property") property_var: str | None = str(property_val) if property_val is not None else None auto_send_val: Any = output_dict.get("autoSend", True) - auto_send: bool = bool(auto_send_val) + auto_send: bool = bool(state.eval_if_expression(auto_send_val)) return messages_var, response_obj_var, property_var or result_property, auto_send @@ -925,7 +932,7 @@ async def handle_action( logger.debug("handle_action: starting agent '%s'", agent_name) arguments, messages_expr, external_loop_when, max_iterations = self._get_input_config() - messages_var, response_obj_var, result_property, auto_send = self._get_output_config() + messages_var, response_obj_var, result_property, auto_send = self._get_output_config(state) # Get conversation-specific messages path if conversationId is specified conversation_id_expr = self._get_conversation_id() @@ -1095,6 +1102,9 @@ async def handle_external_input_response( f"Agent '{agent_name}' invocation failed: not found during loop resumption" ) + _, _, _, auto_send = self._get_output_config(state) + loop_state.auto_send = auto_send + try: accumulated_response, all_messages, tool_calls = await self._invoke_agent_and_store_results( agent=agent, @@ -1105,7 +1115,7 @@ async def handle_external_input_response( messages_var=loop_state.messages_var, response_obj_var=loop_state.response_obj_var, result_property=loop_state.result_property, - auto_send=loop_state.auto_send, + auto_send=auto_send, messages_path=loop_state.messages_path, ) except (AgentInvalidRequestException, AgentInvalidResponseException): diff --git a/python/packages/declarative/agent_framework_declarative/_workflows/_executors_tools.py b/python/packages/declarative/agent_framework_declarative/_workflows/_executors_tools.py index 163d22824e..1d642c91c3 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_executors_tools.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_executors_tools.py @@ -243,24 +243,23 @@ def _get_tool( return None - def _get_output_config(self) -> tuple[str | None, str | None, bool]: - """Parse output configuration from action definition. + def _get_output_config(self) -> tuple[str | None, str | None, Any]: + """Parse output bindings and the unevaluated autoSend setting. Returns: - Tuple of (messages_var, result_var, auto_send) + Tuple of (messages_var, result_var, auto_send_expr) """ - output_config: dict[str, str | bool] = self._action_def.get("output", {}) + output_config: dict[str, Any] = self._action_def.get("output", {}) if not isinstance(output_config, Mapping): return None, None, True messages_var = output_config.get("messages") result_var = output_config.get("result") - auto_send = bool(output_config.get("autoSend", True)) return ( str(messages_var) if messages_var else None, str(result_var) if result_var else None, - auto_send, + output_config.get("autoSend", True), ) def _store_result( @@ -443,7 +442,7 @@ async def handle_action( state = await self._ensure_state_initialized(ctx, trigger) # Parse output configuration early so we can store errors - messages_var, result_var, auto_send = self._get_output_config() + messages_var, result_var, auto_send_expr = self._get_output_config() # Get and evaluate function name (required) function_name_expr = self._action_def.get("functionName") @@ -496,6 +495,7 @@ async def handle_action( return # No approval required - invoke directly + auto_send = bool(state.eval_if_expression(auto_send_expr)) result = await self._execute_tool_invocation( function_name=function_name, arguments=arguments, @@ -521,12 +521,14 @@ async def handle_approval_response( ``function_name`` and ``arguments`` are sourced from ``original_request`` (the payload the reviewer approved); output configuration is re-derived from the executor's action definition. + Rejected calls store the rejection and complete without evaluating + ``autoSend``. """ state = self._get_state(ctx.state) function_name = original_request.function_name arguments = original_request.arguments - messages_var, result_var, auto_send = self._get_output_config() + messages_var, result_var, auto_send_expr = self._get_output_config() # Check if approved if response.approved is not True: @@ -551,6 +553,7 @@ async def handle_approval_response( return # Approved - execute the invocation + auto_send = bool(state.eval_if_expression(auto_send_expr)) result = await self._execute_tool_invocation( function_name=function_name, arguments=arguments, @@ -572,6 +575,12 @@ async def handle_approval_response( class InvokeFunctionToolExecutor(BaseToolExecutor): """Executor that invokes a Python function as a tool. + ``output.autoSend`` defaults to true and accepts a Boolean or a + ``=``-prefixed PowerFx Boolean expression, such as ``=Local.publishResult``. + Expressions use current state immediately before direct or approved + invocation. False suppresses automatic output, not tool execution or result + storage; later actions can explicitly emit the stored results. + This executor supports invoking registered Python functions with: - Expression evaluation for functionName and arguments - Optional approval flow (yield/resume pattern) diff --git a/python/packages/declarative/tests/test_function_tool_executor.py b/python/packages/declarative/tests/test_function_tool_executor.py index 791fcaa4b1..3982831a9f 100644 --- a/python/packages/declarative/tests/test_function_tool_executor.py +++ b/python/packages/declarative/tests/test_function_tool_executor.py @@ -38,6 +38,7 @@ ActionComplete, ActionTrigger, DeclarativeWorkflowBuilder, + DeclarativeWorkflowState, InvokeFunctionToolExecutor, ToolApprovalRequest, ToolApprovalResponse, @@ -580,9 +581,24 @@ def sum_list(numbers: list) -> int: assert any("15" in out for out in outputs) - @pytest.mark.asyncio - async def test_auto_send_disabled(self): - """Test autoSend=false prevents automatic output yielding.""" + @pytest.mark.parametrize( + ("output_config", "expected_auto_send"), + [ + ({}, True), + ({"autoSend": True}, True), + ({"autoSend": False}, False), + ({"autoSend": None}, False), + ({"autoSend": "=true"}, True), + ({"autoSend": "=false"}, False), + ({"autoSend": "=Local.send"}, False), + ({"autoSend": "=Not(Local.send)"}, True), + ({"autoSend": "=Blank()"}, False), + ({"autoSend": "=Local.missing"}, False), + ({"autoSend": "false"}, True), + ], + ) + async def test_auto_send(self, output_config: dict[str, Any], expected_auto_send: bool) -> None: + """Evaluate autoSend without suppressing explicitly requested output.""" def echo_id(msg: str) -> str: return msg @@ -590,12 +606,13 @@ def echo_id(msg: str) -> str: yaml_def = { "name": "auto_send_disabled_test", "actions": [ + {"kind": "SetValue", "id": "set_send", "path": "Local.send", "value": False}, { "kind": "InvokeFunctionTool", "id": "call_no_auto_send", "functionName": "echo_id", "arguments": {"msg": "hello"}, - "output": {"result": "Local.result", "autoSend": False}, + "output": {"result": "Local.result", **output_config}, }, {"kind": "SendActivity", "id": "output", "activity": {"text": "=Local.result"}}, ], @@ -607,8 +624,7 @@ def echo_id(msg: str) -> str: events = await workflow.run({}) outputs = events.get_outputs() - # Result should still be available via explicit SendActivity - assert "hello" in outputs + assert outputs == ["hello"] * (2 if expected_auto_send else 1) @pytest.mark.asyncio async def test_function_with_only_result_output(self): @@ -1038,6 +1054,99 @@ def _init_state(self, mock_state: MagicMock) -> None: "Conversation": {"messages": [], "history": []}, } + @pytest.mark.parametrize("approved", [True, False]) + @pytest.mark.parametrize("send", [True, False]) + async def test_auto_send_on_approval_resume( + self, mock_state: MagicMock, mock_context: MagicMock, approved: bool, send: bool + ) -> None: + state = DeclarativeWorkflowState(mock_state) + state.initialize() + state.set("Local.send", not send) + tool = MagicMock(return_value="hello") + executor = InvokeFunctionToolExecutor( + { + "kind": "InvokeFunctionTool", + "id": "auto_send_approval", + "functionName": "echo", + "requireApproval": True, + "output": { + "result": "Local.result", + "messages": "Local.messages", + "autoSend": "=Local.send", + }, + }, + tools={"echo": tool}, + ) + + await executor.handle_action(ActionTrigger(), mock_context) + tool.assert_not_called() + mock_context.yield_output.assert_not_awaited() + request = mock_context.request_info.call_args[0][0] + state.set("Local.send", send) + + await executor.handle_approval_response(request, ToolApprovalResponse(approved=approved), mock_context) + + if approved: + tool.assert_called_once_with() + assert state.get("Local.result") == "hello" + assert len(state.get("Local.messages")) == 2 + else: + tool.assert_not_called() + assert state.get("Local.result")["rejected"] is True + if approved and send: + mock_context.yield_output.assert_awaited_once_with("hello") + else: + mock_context.yield_output.assert_not_awaited() + mock_context.send_message.assert_awaited_once() + + @pytest.mark.parametrize("approved", [False, True]) + @pytest.mark.parametrize("missing_engine", [False, True]) + async def test_auto_send_error_after_approval_request( + self, + mock_state: MagicMock, + mock_context: MagicMock, + monkeypatch: pytest.MonkeyPatch, + approved: bool, + missing_engine: bool, + ) -> None: + state = DeclarativeWorkflowState(mock_state) + state.initialize() + state.set("Local.send", 1) + tool = MagicMock(return_value="hello") + executor = InvokeFunctionToolExecutor( + { + "kind": "InvokeFunctionTool", + "functionName": "echo", + "requireApproval": True, + "output": { + "result": "Local.result", + "messages": "Local.messages", + "autoSend": "=Local.send + 1 > 0", + }, + }, + tools={"echo": tool}, + ) + await executor.handle_action(ActionTrigger(), mock_context) + request = mock_context.request_info.call_args[0][0] + if missing_engine: + monkeypatch.setattr("agent_framework_declarative._workflows._declarative_base.Engine", None) + else: + state.set("Local.send", {"unexpected": "record"}) + response = ToolApprovalResponse(approved=approved, reason="Declined") + + if approved: + with pytest.raises(RuntimeError if missing_engine else ValueError): + await executor.handle_approval_response(request, response, mock_context) + mock_context.send_message.assert_not_awaited() + else: + await executor.handle_approval_response(request, response, mock_context) + assert state.get("Local.result") == {"approved": False, "rejected": True, "reason": "Declined"} + assert len(state.get("Local.messages")) == 1 + mock_context.send_message.assert_awaited_once() + assert isinstance(mock_context.send_message.call_args[0][0], ActionComplete) + tool.assert_not_called() + mock_context.yield_output.assert_not_awaited() + @pytest.mark.asyncio async def test_approval_required_emits_request(self, mock_state, mock_context): """When requireApproval=true, handle_action should emit ToolApprovalRequest and return.""" diff --git a/python/packages/declarative/tests/test_graph_coverage.py b/python/packages/declarative/tests/test_graph_coverage.py index 2d0b5cae2b..5665056e7d 100644 --- a/python/packages/declarative/tests/test_graph_coverage.py +++ b/python/packages/declarative/tests/test_graph_coverage.py @@ -836,7 +836,9 @@ async def test_agent_executor_get_output_config_simple(self, mock_context, mock_ } executor = InvokeAzureAgentExecutor(action_def) - messages_var, response_obj, result_prop, auto_send = executor._get_output_config() + messages_var, response_obj, result_prop, auto_send = executor._get_output_config( + DeclarativeWorkflowState(mock_state) + ) assert messages_var is None assert response_obj is None assert result_prop == "Local.result" @@ -860,7 +862,9 @@ async def test_agent_executor_get_output_config_full(self, mock_context, mock_st } executor = InvokeAzureAgentExecutor(action_def) - messages_var, response_obj, result_prop, auto_send = executor._get_output_config() + messages_var, response_obj, result_prop, auto_send = executor._get_output_config( + DeclarativeWorkflowState(mock_state) + ) assert messages_var == "Local.ResponseMessages" assert response_obj == "Local.ParsedResponse" assert result_prop == "Local.result" @@ -1774,6 +1778,178 @@ async def test_request_external_input_reads_top_level_alternates(self, mock_cont class TestAgentExternalLoopCoverage: """Tests for agent executor external loop handling.""" + @_requires_powerfx + @pytest.mark.parametrize("string_result", [False, True]) + @pytest.mark.parametrize( + ("output_config", "expected_auto_send"), + [ + ({}, True), + ({"autoSend": True}, True), + ({"autoSend": False}, False), + ({"autoSend": None}, False), + ({"autoSend": "=true"}, True), + ({"autoSend": "=false"}, False), + ({"autoSend": "=Local.send"}, False), + ({"autoSend": "=Not(Local.send)"}, True), + ({"autoSend": "=Blank()"}, False), + ({"autoSend": "=Local.missing"}, False), + ({"autoSend": "false"}, True), + ], + ) + async def test_agent_auto_send( + self, + mock_context: MagicMock, + mock_state: MagicMock, + output_config: dict[str, Any], + expected_auto_send: bool, + string_result: bool, + ) -> None: + from types import SimpleNamespace + + from agent_framework_declarative._workflows._executors_agents import InvokeAzureAgentExecutor + + state = DeclarativeWorkflowState(mock_state) + state.initialize() + state.set("Local.send", False) + result = "hello" if string_result else SimpleNamespace(text="hello", messages=[], tool_calls=[]) + agent = MagicMock(run=AsyncMock(return_value=result)) + executor = InvokeAzureAgentExecutor( + { + "kind": "InvokeAzureAgent", + "agent": "TestAgent", + "input": "question", + "resultProperty": "Local.result", + "output": {"messages": "Local.messages", **output_config}, + }, + agents={"TestAgent": agent}, + ) + + await executor.handle_action(ActionTrigger(), mock_context) + + agent.run.assert_awaited_once() + if expected_auto_send: + mock_context.yield_output.assert_awaited_once_with("hello") + else: + mock_context.yield_output.assert_not_awaited() + assert state.get("Local.result") == "hello" + assert state.get("Local.messages") == "hello" + assert state.get("Conversation.messages")[-1].text == "hello" + mock_context.send_message.assert_awaited_once() + + @_requires_powerfx + @pytest.mark.parametrize("send", [False, True]) + async def test_agent_auto_send_on_external_loop_resume( + self, mock_context: MagicMock, mock_state: MagicMock, send: bool + ) -> None: + from agent_framework_declarative._workflows._executors_agents import ( + AgentExternalInputResponse, + InvokeAzureAgentExecutor, + ) + + state = DeclarativeWorkflowState(mock_state) + state.initialize() + state.set("Local.send", send) + agent = MagicMock(run=AsyncMock(return_value="hello")) + executor = InvokeAzureAgentExecutor( + { + "kind": "InvokeAzureAgent", + "agent": "TestAgent", + "input": {"externalLoop": {"when": "=true"}}, + "output": {"property": "Local.result", "autoSend": "=Local.send"}, + }, + agents={"TestAgent": agent}, + ) + + await executor.handle_action(ActionTrigger(), mock_context) + request = mock_context.request_info.call_args[0][0] + state.set("Local.send", not send) + mock_context.yield_output.reset_mock() + + await executor.handle_external_input_response( + request, AgentExternalInputResponse(user_input="continue"), mock_context + ) + + assert agent.run.await_count == 2 + if not send: + mock_context.yield_output.assert_awaited_once_with("hello") + else: + mock_context.yield_output.assert_not_awaited() + assert state.get("Local.result") == "hello" + + @_requires_powerfx + async def test_agent_auto_send_error_on_external_loop_resume( + self, mock_context: MagicMock, mock_state: MagicMock + ) -> None: + from agent_framework_declarative._workflows._executors_agents import ( + AgentExternalInputResponse, + InvokeAzureAgentExecutor, + ) + + state = DeclarativeWorkflowState(mock_state) + state.initialize() + state.set("Local.send", 1) + agent = MagicMock(run=AsyncMock(return_value="hello")) + executor = InvokeAzureAgentExecutor( + { + "kind": "InvokeAzureAgent", + "agent": "TestAgent", + "input": {"externalLoop": {"when": "=true"}}, + "output": {"autoSend": "=Local.send + 1 > 0"}, + }, + agents={"TestAgent": agent}, + ) + await executor.handle_action(ActionTrigger(), mock_context) + request = mock_context.request_info.call_args[0][0] + state.set("Local.send", {"unexpected": "record"}) + mock_context.yield_output.reset_mock() + + with pytest.raises(ValueError): + await executor.handle_external_input_response( + request, AgentExternalInputResponse(user_input="continue"), mock_context + ) + + agent.run.assert_awaited_once() + mock_context.yield_output.assert_not_awaited() + + @_requires_powerfx + @pytest.mark.parametrize("kind", ["InvokeFunctionTool", "InvokeAzureAgent"]) + @pytest.mark.parametrize("missing_engine", [False, True]) + async def test_auto_send_evaluation_error_prevents_invocation( + self, + mock_context: MagicMock, + mock_state: MagicMock, + monkeypatch: pytest.MonkeyPatch, + kind: str, + missing_engine: bool, + ) -> None: + from agent_framework_declarative._workflows._executors_agents import InvokeAzureAgentExecutor + from agent_framework_declarative._workflows._executors_tools import InvokeFunctionToolExecutor + + state = DeclarativeWorkflowState(mock_state) + state.initialize() + tool = MagicMock(return_value="hello") + agent = MagicMock(run=AsyncMock(return_value="hello")) + action_def = { + "kind": kind, + "agent": "TestAgent", + "functionName": "echo", + "output": {"autoSend": "=true" if missing_engine else "=1 +"}, + } + executor = ( + InvokeFunctionToolExecutor(action_def, tools={"echo": tool}) + if kind == "InvokeFunctionTool" + else InvokeAzureAgentExecutor(action_def, agents={"TestAgent": agent}) + ) + if missing_engine: + monkeypatch.setattr("agent_framework_declarative._workflows._declarative_base.Engine", None) + + with pytest.raises(RuntimeError if missing_engine else ValueError): + await executor.handle_action(ActionTrigger(), mock_context) + + tool.assert_not_called() + agent.run.assert_not_awaited() + mock_context.yield_output.assert_not_awaited() + @_requires_powerfx async def test_agent_executor_with_external_loop(self, mock_context, mock_state): """Test agent executor with external loop that triggers."""