Python: evaluate autoSend expressions before emitting output

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):

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)

python/packages/declarative/tests/test_function_tool_executor.py

@@ -38,6 +38,7 @@
     ActionComplete,
     ActionTrigger,
     DeclarativeWorkflowBuilder,
+    DeclarativeWorkflowState,
     InvokeFunctionToolExecutor,
     ToolApprovalRequest,
     ToolApprovalResponse,
@@ -580,22 +581,38 @@ 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
 
         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."""