From 197171352f5a9f7e136bb239c1a5b7c631322b5d Mon Sep 17 00:00:00 2001 From: ghox789 <70468856+ghox789@users.noreply.github.com> Date: Thu, 22 Oct 2026 14:47:04 +0400 Subject: [PATCH] Python: fix(declarative): honor registered function approval requirements InvokeFunctionTool decided whether to request approval only from the action's requireApproval field. A function registered with approval_mode="always_require" therefore ran without approval whenever the workflow omitted requireApproval or set it to false. Request approval when requireApproval is true or the resolved registered tool has approval_mode="always_require", so the workflow definition can no longer override the registered requirement. This matches the .NET fix in #9393. Tools without that mode keep following requireApproval. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01RuT3Gxp3nBFz1VLqf9FtMF --- .../_workflows/_executors_tools.py | 29 ++++- .../tests/test_function_tool_executor.py | 111 ++++++++++++++++++ 2 files changed, 137 insertions(+), 3 deletions(-) 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 1d642c91c3a..16c2f3a025c 100644 --- a/python/packages/declarative/agent_framework_declarative/_workflows/_executors_tools.py +++ b/python/packages/declarative/agent_framework_declarative/_workflows/_executors_tools.py @@ -243,6 +243,27 @@ def _get_tool( return None + def _registered_tool_requires_approval( + self, + function_name: str, + ctx: WorkflowContext[Any, Any], + ) -> bool: + """Check whether the registered tool requires approval on every invocation. + + A tool registered with ``approval_mode="always_require"`` must not run + without approval, even when the action omits ``requireApproval`` or sets + it to false. + + Args: + function_name: Name of the function + ctx: Workflow context + + Returns: + True if the registered tool requires approval, False otherwise + """ + tool = self._get_tool(function_name, ctx) + return getattr(tool, "approval_mode", None) == "always_require" + def _get_output_config(self) -> tuple[str | None, str | None, Any]: """Parse output bindings and the unevaluated autoSend setting. @@ -433,7 +454,8 @@ async def handle_action( ) -> None: """Handle the tool invocation with optional approval flow. - When requireApproval=true: + When requireApproval=true, or the registered tool has + approval_mode="always_require": 1. Saves invocation state to State (keyed by executor ID) 2. Emits ToolApprovalRequest via ctx.request_info() 3. Workflow yields (returns without ActionComplete) @@ -477,10 +499,11 @@ async def handle_action( for key, value in arguments_def.items(): # type: ignore[reportUnknownVariableType] arguments[key] = state.eval_if_expression(value) - # Check if approval is required + # Check if approval is required. The action's requireApproval cannot + # override a registered tool's approval_mode="always_require". require_approval = self._action_def.get("requireApproval", False) - if require_approval: + if require_approval or self._registered_tool_requires_approval(function_name, ctx): # Emit approval request - the request payload is the source of # truth for resumed invocation; no side-channel state is written. request_id = str(uuid.uuid4()) diff --git a/python/packages/declarative/tests/test_function_tool_executor.py b/python/packages/declarative/tests/test_function_tool_executor.py index 3982831a9fc..cd5c42aa290 100644 --- a/python/packages/declarative/tests/test_function_tool_executor.py +++ b/python/packages/declarative/tests/test_function_tool_executor.py @@ -1258,6 +1258,117 @@ def my_tool(x: int) -> int: assert result["reason"] == "Not authorized" assert result["approved"] is False + @pytest.mark.asyncio + @pytest.mark.parametrize("require_approval", [None, False, True]) + @pytest.mark.parametrize("approved", [True, False]) + async def test_registered_tool_approval_is_honored( + self, mock_state, mock_context, require_approval: bool | None, approved: bool + ): + """A tool registered with approval_mode="always_require" must request approval. + + Omitting requireApproval or setting it to false does not override the + registered tool's approval requirement. + """ + from agent_framework import tool + + self._init_state(mock_state) + call_log: list[int] = [] + + @tool(approval_mode="always_require") + def sensitive_tool(x: int) -> int: + call_log.append(x) + return x * 2 + + action_def: dict[str, Any] = { + "kind": "InvokeFunctionTool", + "id": "registered_approval", + "functionName": "sensitive_tool", + "arguments": {"x": 3}, + "output": {"result": "Local.result"}, + } + if require_approval is not None: + action_def["requireApproval"] = require_approval + + executor = InvokeFunctionToolExecutor(action_def, tools={"sensitive_tool": sensitive_tool}) + + await executor.handle_action(ActionTrigger(), mock_context) + + # The tool must not run before approval, and the workflow must yield + mock_context.request_info.assert_called_once() + request = mock_context.request_info.call_args[0][0] + assert isinstance(request, ToolApprovalRequest) + assert request.function_name == "sensitive_tool" + assert request.arguments == {"x": 3} + assert call_log == [] + mock_context.send_message.assert_not_called() + + await executor.handle_approval_response(request, ToolApprovalResponse(approved=approved), mock_context) + + assert call_log == ([3] if approved else []) + mock_context.send_message.assert_called_once() + + @pytest.mark.asyncio + @pytest.mark.parametrize("require_approval", [None, False]) + async def test_state_registered_tool_approval_is_honored( + self, mock_state, mock_context, require_approval: bool | None + ): + """The registered approval requirement also applies to State-registry tools.""" + from agent_framework import tool + + self._init_state(mock_state) + + @tool(approval_mode="always_require") + def sensitive_tool() -> str: + raise AssertionError("Should not be called before approval") + + mock_state._data[FUNCTION_TOOL_REGISTRY_KEY] = {"sensitive_tool": sensitive_tool} + + action_def: dict[str, Any] = { + "kind": "InvokeFunctionTool", + "id": "state_registered_approval", + "functionName": "sensitive_tool", + "arguments": {}, + "output": {"result": "Local.result"}, + } + if require_approval is not None: + action_def["requireApproval"] = require_approval + + executor = InvokeFunctionToolExecutor(action_def, tools={}) + + await executor.handle_action(ActionTrigger(), mock_context) + + mock_context.request_info.assert_called_once() + mock_context.send_message.assert_not_called() + + @pytest.mark.asyncio + @pytest.mark.parametrize("require_approval", [None, False, True]) + async def test_unmarked_tool_uses_action_approval(self, mock_state, mock_context, require_approval: bool | None): + """A tool without approval_mode="always_require" follows the action's requireApproval.""" + from agent_framework import tool + + self._init_state(mock_state) + + @tool(approval_mode="never_require") + def plain_tool(x: int) -> int: + return x * 2 + + action_def: dict[str, Any] = { + "kind": "InvokeFunctionTool", + "id": "unmarked_approval", + "functionName": "plain_tool", + "arguments": {"x": 4}, + "output": {"result": "Local.result"}, + } + if require_approval is not None: + action_def["requireApproval"] = require_approval + + executor = InvokeFunctionToolExecutor(action_def, tools={"plain_tool": plain_tool}) + + await executor.handle_action(ActionTrigger(), mock_context) + + assert mock_context.request_info.call_count == (1 if require_approval else 0) + assert mock_context.send_message.call_count == (0 if require_approval else 1) + # ============================================================================ # State registry tool lookup (lines 255-257)