direct tool call - interrupt not allowed

src/strands/agent/agent.py

@@ -55,7 +55,7 @@
 from ..tools.registry import ToolRegistry
 from ..tools.structured_output._structured_output_context import StructuredOutputContext
 from ..tools.watcher import ToolWatcher
-from ..types._events import AgentResultEvent, InitEventLoopEvent, ModelStreamChunkEvent, TypedEvent
+from ..types._events import AgentResultEvent, InitEventLoopEvent, ModelStreamChunkEvent, ToolInterruptEvent, TypedEvent
 from ..types.agent import AgentInput
 from ..types.content import ContentBlock, Message, Messages
 from ..types.exceptions import ContextWindowOverflowException
@@ -166,7 +166,9 @@ def caller(
 
                 async def acall() -> ToolResult:
                     async for event in ToolExecutor._stream(self._agent, tool_use, tool_results, invocation_state):
-                        _ = event
+                        if isinstance(event, ToolInterruptEvent):
+                            self._agent._interrupt_state.deactivate()
+                            raise RuntimeError("cannot raise interrupt in direct tool call")
 
                     return tool_results[0]
 

tests/strands/agent/test_agent.py

@@ -2054,7 +2054,31 @@ def test_agent_structured_output_interrupt(user):
         agent.structured_output(type(user), "invalid")
 
 
-def test_agent_tool_caller_interrupt(user):
+def test_agent_tool_caller_interrupt():
+    @strands.tool(context=True)
+    def test_tool(tool_context):
+        tool_context.interrupt("test-interrupt")
+
+    agent = Agent(tools=[test_tool])
+
+    exp_message = r"cannot raise interrupt in direct tool call"
+    with pytest.raises(RuntimeError, match=exp_message):
+        agent.tool.test_tool(agent=agent)
+
+    tru_state = agent._interrupt_state.to_dict()
+    exp_state = {
+        "activated": False,
+        "context": {},
+        "interrupts": {},
+    }
+    assert tru_state == exp_state
+
+    tru_messages = agent.messages
+    exp_messages = []
+    assert tru_messages == exp_messages
+
+
+def test_agent_tool_caller_interrupt_activated():
     agent = Agent()
     agent._interrupt_state.activated = True