From 3352968c6906df4689049eb2b7dc739ca11fe949 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 14 Apr 2026 22:36:50 +0000 Subject: [PATCH 1/3] feat(lifecycle): add on_tool_authorize hook for per-tool authorization (#2868) Adds on_tool_authorize to both RunHooksBase and AgentHooksBase. The hook fires before each tool execution and can return False to block the call. When denied: - The tool function is never invoked - on_tool_start and on_tool_end are skipped for that call - The model receives 'Tool call denied: authorization hook returned False.' as the tool output so it can react gracefully Both run-level and agent-level hooks are checked; either can deny the call. The default implementation returns True (allow all), so this is fully backwards-compatible. Closes #2868 --- src/agents/lifecycle.py | 48 ++++++ src/agents/run_internal/tool_execution.py | 13 ++ tests/test_tool_authorize_hook.py | 198 ++++++++++++++++++++++ 3 files changed, 259 insertions(+) create mode 100644 tests/test_tool_authorize_hook.py diff --git a/src/agents/lifecycle.py b/src/agents/lifecycle.py index 38744471fb..99dd369c92 100644 --- a/src/agents/lifecycle.py +++ b/src/agents/lifecycle.py @@ -76,6 +76,30 @@ async def on_tool_start( """Called immediately before a local tool is invoked.""" pass + async def on_tool_authorize( + self, + context: RunContextWrapper[TContext], + agent: TAgent, + tool: Tool, + ) -> bool: + """Called before a tool is executed. Return False to deny the tool call. + + When False is returned, the tool is not invoked and the model receives a + denial message instead. All on_tool_start and on_tool_end hooks are + skipped for denied calls. + + Default: always returns True (allow all tool calls). + + Args: + context: The run context wrapper. + agent: The current agent. + tool: The tool about to be invoked. + + Returns: + True to allow the tool call, False to deny it. + """ + return True + async def on_tool_end( self, context: RunContextWrapper[TContext], @@ -138,6 +162,30 @@ async def on_tool_start( """Called immediately before a local tool is invoked.""" pass + async def on_tool_authorize( + self, + context: RunContextWrapper[TContext], + agent: TAgent, + tool: Tool, + ) -> bool: + """Called before a tool is executed. Return False to deny the tool call. + + When False is returned, the tool is not invoked and the model receives a + denial message instead. All on_tool_start and on_tool_end hooks are + skipped for denied calls. + + Default: always returns True (allow all tool calls). + + Args: + context: The run context wrapper. + agent: The current agent. + tool: The tool about to be invoked. + + Returns: + True to allow the tool call, False to deny it. + """ + return True + async def on_tool_end( self, context: RunContextWrapper[TContext], diff --git a/src/agents/run_internal/tool_execution.py b/src/agents/run_internal/tool_execution.py index 4511045288..2deaf467f1 100644 --- a/src/agents/run_internal/tool_execution.py +++ b/src/agents/run_internal/tool_execution.py @@ -1591,6 +1591,19 @@ async def _execute_single_tool_body( if rejected_message is not None: return rejected_message + # Authorization check: run-level hook first, then agent-level hook. + # If either denies the call, skip execution and return the denial string. + run_authorized = await self.hooks.on_tool_authorize( + tool_context, self.agent, func_tool + ) + agent_authorized = ( + await agent_hooks.on_tool_authorize(tool_context, self.agent, func_tool) + if agent_hooks + else True + ) + if not run_authorized or not agent_authorized: + return "Tool call denied: authorization hook returned False." + await asyncio.gather( self.hooks.on_tool_start(tool_context, self.agent, func_tool), ( diff --git a/tests/test_tool_authorize_hook.py b/tests/test_tool_authorize_hook.py new file mode 100644 index 0000000000..e9a793fd80 --- /dev/null +++ b/tests/test_tool_authorize_hook.py @@ -0,0 +1,198 @@ +"""Tests for on_tool_authorize lifecycle hook (issue #2868).""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from agents import Agent, Runner +from agents.lifecycle import AgentHooks, RunHooks +from agents.run_context import RunContextWrapper, TContext +from agents.tool import FunctionTool, Tool + +from .fake_model import FakeModel +from .test_responses import get_function_tool, get_function_tool_call, get_text_message + + +DENIAL_MSG = "Tool call denied: authorization hook returned False." + + +class AllowRunHooks(RunHooks): + """Always authorizes tool calls; records invocations.""" + + def __init__(self) -> None: + self.authorize_calls: list[str] = [] + self.start_calls: list[str] = [] + self.end_calls: list[str] = [] + + async def on_tool_authorize( + self, context: Any, agent: Any, tool: Any + ) -> bool: + self.authorize_calls.append(tool.name) + return True + + async def on_tool_start(self, context: Any, agent: Any, tool: Any) -> None: + self.start_calls.append(tool.name) + + async def on_tool_end(self, context: Any, agent: Any, tool: Any, result: Any) -> None: + self.end_calls.append(tool.name) + + +class DenyRunHooks(RunHooks): + """Denies all tool calls.""" + + def __init__(self) -> None: + self.authorize_calls: list[str] = [] + self.start_calls: list[str] = [] + self.end_calls: list[str] = [] + + async def on_tool_authorize( + self, context: Any, agent: Any, tool: Any + ) -> bool: + self.authorize_calls.append(tool.name) + return False + + async def on_tool_start(self, context: Any, agent: Any, tool: Any) -> None: + self.start_calls.append(tool.name) + + async def on_tool_end(self, context: Any, agent: Any, tool: Any, result: Any) -> None: + self.end_calls.append(tool.name) + + +class DenyAgentHooks(AgentHooks): + """Agent-level deny hook.""" + + def __init__(self) -> None: + self.authorize_calls: list[str] = [] + + async def on_tool_authorize( + self, context: Any, agent: Any, tool: Any + ) -> bool: + self.authorize_calls.append(tool.name) + return False + + +@pytest.mark.asyncio +async def test_allow_hook_lets_tool_run() -> None: + """When on_tool_authorize returns True the tool executes normally.""" + tool = get_function_tool("my_tool", "tool_result") + model = FakeModel() + model.add_multiple_turn_outputs([ + [get_function_tool_call("my_tool", "{}")], + [get_text_message("done")], + ]) + + hooks = AllowRunHooks() + agent = Agent(name="A", model=model, tools=[tool]) + result = await Runner.run(agent, input="hi", hooks=hooks) + + assert hooks.authorize_calls == ["my_tool"] + assert hooks.start_calls == ["my_tool"] + assert hooks.end_calls == ["my_tool"] + assert result.final_output == "done" + + +@pytest.mark.asyncio +async def test_deny_hook_skips_tool_execution() -> None: + """When on_tool_authorize returns False the tool is not executed and model gets denial.""" + invoked = [] + + async def my_tool_impl(ctx: Any, args: str) -> str: + invoked.append(True) + return "should not be returned" + + func_tool = get_function_tool("my_tool", "should_not_appear") + model = FakeModel() + model.add_multiple_turn_outputs([ + [get_function_tool_call("my_tool", "{}")], + [get_text_message("done")], + ]) + + hooks = DenyRunHooks() + agent = Agent(name="A", model=model, tools=[func_tool]) + result = await Runner.run(agent, input="hi", hooks=hooks) + + # The authorization hook was called + assert hooks.authorize_calls == ["my_tool"] + # But on_tool_start and on_tool_end were NOT called (denied before them) + assert hooks.start_calls == [] + assert hooks.end_calls == [] + # And the run still completes (model sees the denial and produces final output) + assert result.final_output == "done" + + +@pytest.mark.asyncio +async def test_deny_hook_sends_denial_message_to_model() -> None: + """The model receives the denial string as the tool output.""" + received_tool_outputs: list[str] = [] + + class OutputCapturingHooks(RunHooks): + async def on_tool_authorize(self, context: Any, agent: Any, tool: Any) -> bool: + return False + + model = FakeModel() + func_tool = get_function_tool("my_tool", "real_result") + model.add_multiple_turn_outputs([ + [get_function_tool_call("my_tool", "{}")], + [get_text_message("done")], + ]) + + hooks = OutputCapturingHooks() + agent = Agent(name="A", model=model, tools=[func_tool]) + result = await Runner.run(agent, input="hi", hooks=hooks) + + # Check that model received denial message in its input on second turn + # The second turn's input items should include a tool output with denial + raw_responses = result.raw_responses + assert len(raw_responses) >= 1 + assert result.final_output == "done" + + +@pytest.mark.asyncio +async def test_agent_level_deny_hook() -> None: + """Agent-level on_tool_authorize returning False also denies the call.""" + func_tool = get_function_tool("blocked_tool", "should_not_run") + model = FakeModel() + model.add_multiple_turn_outputs([ + [get_function_tool_call("blocked_tool", "{}")], + [get_text_message("fine")], + ]) + + agent_hooks = DenyAgentHooks() + agent = Agent(name="A", model=model, tools=[func_tool], hooks=agent_hooks) + result = await Runner.run(agent, input="hi") + + assert agent_hooks.authorize_calls == ["blocked_tool"] + assert result.final_output == "fine" + + +@pytest.mark.asyncio +async def test_authorize_not_called_when_no_tool_used() -> None: + """on_tool_authorize is not called when the model produces a final output directly.""" + model = FakeModel() + model.set_next_output([get_text_message("hello")]) + + hooks = AllowRunHooks() + agent = Agent(name="A", model=model) + await Runner.run(agent, input="hi", hooks=hooks) + + assert hooks.authorize_calls == [] + assert hooks.start_calls == [] + + +@pytest.mark.asyncio +async def test_default_hook_allows_all() -> None: + """The default RunHooks implementation allows all tool calls (no override needed).""" + func_tool = get_function_tool("calc", "42") + model = FakeModel() + model.add_multiple_turn_outputs([ + [get_function_tool_call("calc", "{}")], + [get_text_message("answer is 42")], + ]) + + # Use the base class without overriding on_tool_authorize + agent = Agent(name="A", model=model, tools=[func_tool]) + result = await Runner.run(agent, input="hi") + + assert result.final_output == "answer is 42" From 8479139a377cfa01bd37f48134da75d40843e09d Mon Sep 17 00:00:00 2001 From: Aditya Singh Date: Tue, 14 Apr 2026 22:47:46 +0000 Subject: [PATCH 2/3] fix: address CodeRabbit review comments on PR #2 - Wire my_tool_impl directly into FunctionTool so test_deny_hook_skips_tool_execution actually verifies tool impl was not called; add assert invoked == [] - Add explicit DENIAL_MSG assertion in test_deny_hook_sends_denial_message_to_model by checking new_items for ToolCallOutputItem with denial string - Add type annotation for invoked: list[bool] - Add missing docstrings to all hook classes and methods --- tests/test_tool_authorize_hook.py | 40 +++++++++++++++++++++++++++---- 1 file changed, 36 insertions(+), 4 deletions(-) diff --git a/tests/test_tool_authorize_hook.py b/tests/test_tool_authorize_hook.py index e9a793fd80..e622b5b3f7 100644 --- a/tests/test_tool_authorize_hook.py +++ b/tests/test_tool_authorize_hook.py @@ -22,6 +22,7 @@ class AllowRunHooks(RunHooks): """Always authorizes tool calls; records invocations.""" def __init__(self) -> None: + """Initialise empty tracking lists.""" self.authorize_calls: list[str] = [] self.start_calls: list[str] = [] self.end_calls: list[str] = [] @@ -29,13 +30,16 @@ def __init__(self) -> None: async def on_tool_authorize( self, context: Any, agent: Any, tool: Any ) -> bool: + """Record and authorize the tool call.""" self.authorize_calls.append(tool.name) return True async def on_tool_start(self, context: Any, agent: Any, tool: Any) -> None: + """Record that the tool started.""" self.start_calls.append(tool.name) async def on_tool_end(self, context: Any, agent: Any, tool: Any, result: Any) -> None: + """Record that the tool ended.""" self.end_calls.append(tool.name) @@ -43,6 +47,7 @@ class DenyRunHooks(RunHooks): """Denies all tool calls.""" def __init__(self) -> None: + """Initialise empty tracking lists.""" self.authorize_calls: list[str] = [] self.start_calls: list[str] = [] self.end_calls: list[str] = [] @@ -50,13 +55,16 @@ def __init__(self) -> None: async def on_tool_authorize( self, context: Any, agent: Any, tool: Any ) -> bool: + """Record and deny the tool call.""" self.authorize_calls.append(tool.name) return False async def on_tool_start(self, context: Any, agent: Any, tool: Any) -> None: + """Record that the tool started (should not be called when denied).""" self.start_calls.append(tool.name) async def on_tool_end(self, context: Any, agent: Any, tool: Any, result: Any) -> None: + """Record that the tool ended (should not be called when denied).""" self.end_calls.append(tool.name) @@ -64,11 +72,13 @@ class DenyAgentHooks(AgentHooks): """Agent-level deny hook.""" def __init__(self) -> None: + """Initialise empty tracking list.""" self.authorize_calls: list[str] = [] async def on_tool_authorize( self, context: Any, agent: Any, tool: Any ) -> bool: + """Record and deny the tool call.""" self.authorize_calls.append(tool.name) return False @@ -96,13 +106,21 @@ async def test_allow_hook_lets_tool_run() -> None: @pytest.mark.asyncio async def test_deny_hook_skips_tool_execution() -> None: """When on_tool_authorize returns False the tool is not executed and model gets denial.""" - invoked = [] + invoked: list[bool] = [] async def my_tool_impl(ctx: Any, args: str) -> str: + """Append True to invoked and return a sentinel string (should never be called).""" invoked.append(True) return "should not be returned" - func_tool = get_function_tool("my_tool", "should_not_appear") + # Wire my_tool_impl directly into FunctionTool so we can assert it was never called. + func_tool = FunctionTool( + name="my_tool", + description="test tool", + params_json_schema={}, + on_invoke_tool=my_tool_impl, + strict_json_schema=False, + ) model = FakeModel() model.add_multiple_turn_outputs([ [get_function_tool_call("my_tool", "{}")], @@ -120,6 +138,8 @@ async def my_tool_impl(ctx: Any, args: str) -> str: assert hooks.end_calls == [] # And the run still completes (model sees the denial and produces final output) assert result.final_output == "done" + # Verify my_tool_impl was never actually invoked + assert invoked == [] @pytest.mark.asyncio @@ -128,7 +148,10 @@ async def test_deny_hook_sends_denial_message_to_model() -> None: received_tool_outputs: list[str] = [] class OutputCapturingHooks(RunHooks): + """Captures tool outputs by denying every tool call.""" + async def on_tool_authorize(self, context: Any, agent: Any, tool: Any) -> bool: + """Deny every tool call unconditionally.""" return False model = FakeModel() @@ -142,10 +165,19 @@ async def on_tool_authorize(self, context: Any, agent: Any, tool: Any) -> bool: agent = Agent(name="A", model=model, tools=[func_tool]) result = await Runner.run(agent, input="hi", hooks=hooks) - # Check that model received denial message in its input on second turn - # The second turn's input items should include a tool output with denial + # Check that model received denial message in its input on second turn. + # The denial string is stored as the tool output in new_items. raw_responses = result.raw_responses assert len(raw_responses) >= 1 + # The denial message must appear as a ToolCallOutputItem in the run's new items. + tool_outputs = [ + str(item.output) + for item in result.new_items + if hasattr(item, "output") and item.output is not None + ] + assert any(DENIAL_MSG in output for output in tool_outputs), ( + f"Expected denial message {DENIAL_MSG!r} in tool outputs, got {tool_outputs}" + ) assert result.final_output == "done" From 3011d7a76e6225e5ee3c30d95ab251c44d17f98e Mon Sep 17 00:00:00 2001 From: Aditya Singh Date: Tue, 21 Apr 2026 21:02:33 -0700 Subject: [PATCH 3/3] refactor(tool): extract authorization-denied message to shared constant Pulls the "Tool call denied: authorization hook returned False." string out of _execute_single_tool_body and into a module-level TOOL_AUTHORIZATION_DENIED_MESSAGE constant alongside REDACTED_TOOL_ERROR_MESSAGE in tool_execution.py. The denial test now imports the same constant rather than hard-coding its own copy of the string, so the wire-format and the assertion cannot drift apart. Picked up the remaining nitpick from the CodeRabbit review of PR #2. --- src/agents/run_internal/tool_execution.py | 3 ++- tests/test_tool_authorize_hook.py | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/src/agents/run_internal/tool_execution.py b/src/agents/run_internal/tool_execution.py index 2deaf467f1..74b633086b 100644 --- a/src/agents/run_internal/tool_execution.py +++ b/src/agents/run_internal/tool_execution.py @@ -147,6 +147,7 @@ ] REDACTED_TOOL_ERROR_MESSAGE = "Tool execution failed. Error details are redacted." +TOOL_AUTHORIZATION_DENIED_MESSAGE = "Tool call denied: authorization hook returned False." TToolSpanResult = TypeVar("TToolSpanResult") _FUNCTION_TOOL_CANCELLED_DRAIN_SECONDS = 0.1 _FUNCTION_TOOL_POST_INVOKE_WAIT_SECONDS = 0.1 @@ -1602,7 +1603,7 @@ async def _execute_single_tool_body( else True ) if not run_authorized or not agent_authorized: - return "Tool call denied: authorization hook returned False." + return TOOL_AUTHORIZATION_DENIED_MESSAGE await asyncio.gather( self.hooks.on_tool_start(tool_context, self.agent, func_tool), diff --git a/tests/test_tool_authorize_hook.py b/tests/test_tool_authorize_hook.py index e622b5b3f7..21b86a82e4 100644 --- a/tests/test_tool_authorize_hook.py +++ b/tests/test_tool_authorize_hook.py @@ -9,13 +9,15 @@ from agents import Agent, Runner from agents.lifecycle import AgentHooks, RunHooks from agents.run_context import RunContextWrapper, TContext +from agents.run_internal.tool_execution import TOOL_AUTHORIZATION_DENIED_MESSAGE from agents.tool import FunctionTool, Tool from .fake_model import FakeModel from .test_responses import get_function_tool, get_function_tool_call, get_text_message -DENIAL_MSG = "Tool call denied: authorization hook returned False." +# Reference the production constant so the test can never drift from the actual message. +DENIAL_MSG = TOOL_AUTHORIZATION_DENIED_MESSAGE class AllowRunHooks(RunHooks):