Skip to content

Interrupts

When your agent needs human approval or input before it continues, raise an interrupt. The agent stops its loop and hands control back to you along with the pending interrupt. You provide a response, and the agent resumes from the point where it stopped. You can raise interrupts from hook callbacks or from tool definitions. The general flow looks as follows:

flowchart TD
A[Invoke Agent] --> B[Execute Hook/Tool]
B --> C{Interrupts Raised?}
C -->|No| D[Continue Agent Loop]
C -->|Yes| E[Stop Agent Loop]
E --> F[Return Interrupts]
F --> G[Respond to Interrupts]
G --> H[Execute Hook/Tool with Responses]
H --> I{New Interrupts?}
I -->|Yes| E
I -->|No| D

Raise interrupts inside your hook callbacks to pause the agent at specific lifecycle events in the agent loop.

Both BeforeToolCallEvent and BeforeToolsEvent are interruptible. Interrupting on a BeforeToolCallEvent intercepts an individual tool call before execution, while interrupting on BeforeToolsEvent pauses the entire batch of tool calls before any of them execute, which is useful for approving a whole set of tool calls at once.

import json
from typing import Any
from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent, HookProvider, HookRegistry
@tool
def delete_files(paths: list[str]) -> bool:
# Implementation here
pass
@tool
def inspect_files(paths: list[str]) -> dict[str, Any]:
# Implementation here
pass
class ApprovalHook(HookProvider):
def __init__(self, app_name: str) -> None:
self.app_name = app_name
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
registry.add_callback(BeforeToolCallEvent, self.approve)
def approve(self, event: BeforeToolCallEvent) -> None:
if event.tool_use["name"] != "delete_files":
return
approval = event.interrupt(f"{self.app_name}-approval", reason={"paths": event.tool_use["input"]["paths"]})
if approval.lower() != "y":
event.cancel_tool = "User denied permission to delete files"
agent = Agent(
hooks=[ApprovalHook("myapp")],
system_prompt="You delete files older than 5 days",
tools=[delete_files, inspect_files],
callback_handler=None,
)
paths = ["a/b/c.txt", "d/e/f.txt"]
result = agent(f"paths=<{paths}>")
while True:
if result.stop_reason != "interrupt":
break
responses = []
for interrupt in result.interrupts:
if interrupt.name == "myapp-approval":
user_input = input(f"Do you want to delete {interrupt.reason['paths']} (y/N): ")
responses.append({
"interruptResponse": {
"interruptId": interrupt.id,
"response": user_input
}
})
result = agent(responses)
print(f"MESSAGE: {json.dumps(result.message)}")
from typing import Any
from strands import Agent, tool
from strands.hooks import BeforeToolsEvent, HookProvider, HookRegistry
@tool
def delete_files(paths: list[str]) -> bool:
# Implementation here
pass
class BatchApprovalHook(HookProvider):
def __init__(self, app_name: str) -> None:
self.app_name = app_name
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
registry.add_callback(BeforeToolsEvent, self.approve)
def approve(self, event: BeforeToolsEvent) -> None:
dangerous_tools = [
content["toolUse"]["name"]
for content in event.message["content"]
if "toolUse" in content and content["toolUse"]["name"] == "delete_files"
]
if not dangerous_tools:
return
approval = event.interrupt(f"{self.app_name}-batch-approval", reason={"tools": dangerous_tools})
if approval.lower() != "y":
event.cancel = "Batch cancelled by user"
agent = Agent(
hooks=[BatchApprovalHook("myapp")],
system_prompt="You delete files older than 5 days",
tools=[delete_files],
callback_handler=None,
)
result = agent("Delete a/b/c.txt and d/e/f.txt")
while result.stop_reason == "interrupt":
responses = []
for interrupt in result.interrupts:
if interrupt.name == "myapp-batch-approval":
user_input = input(f"Approve running {interrupt.reason['tools']}? (y/N): ")
responses.append({"interruptResponse": {"interruptId": interrupt.id, "response": user_input}})
result = agent(responses)

Setting event.cancel (to True for a default message, or a string for a custom one) produces an error tool result for every tool in the batch and skips execution entirely, so no per-tool BeforeToolCallEvent fires.

Interrupts in Strands are comprised of the following components:

  • event.interrupt - Raises an interrupt with a unique name and optional reason
    • The name must be unique across all interrupt calls configured on the same event (BeforeToolCallEvent or BeforeToolsEvent). In the example above, we demonstrate using app_name to namespace the interrupt call. This is particularly helpful if you plan to vend your hooks to other users.
    • You can assign additional context for raising the interrupt to the reason field. Note, the reason must be JSON-serializable.
  • result.stop_reason - Check if agent stopped due to “interrupt”
  • result.interrupts - List of interrupts that were raised
    • Each interrupt contains the user provided name and reason, along with an instance id.
  • interruptResponse - Content block type for configuring the interrupt responses.
    • Each response is uniquely identified by their interrupt’s id and will be returned from the associated interrupt call when invoked the second time around. Note, the response must be JSON-serializable.
  • event.cancel_tool (BeforeToolCallEvent) - Cancel a single tool call based on the interrupt response
    • You can either set cancel_tool to True or provide a custom cancellation message.
  • event.cancel (BeforeToolsEvent) - Cancel every tool call in the batch based on the interrupt response
    • You can either set cancel to True or provide a custom cancellation message.

For additional details on each of these components, refer to the Python API Reference.

Strands enforces the following rules for interrupts:

  • All hooks configured on the interrupted event will execute
  • All hooks configured on the interrupted event are allowed to raise an interrupt
  • A single hook can raise multiple interrupts but only one at a time
    • In other words, within a single hook, you can interrupt, respond to that interrupt, and then proceed to interrupt again.
  • All tools running concurrently are interruptible
  • All tools running concurrently that are not interrupted will execute
  • When an interrupt fires from BeforeToolCallEvent, AfterToolCallEvent does not fire for that tool, but AfterToolsEvent still fires
  • AfterToolsEvent fires once per event-loop cycle rather than once per logical batch. A per-tool interrupt splits a batch across cycles, so it fires on the interrupt cycle (carrying only the results collected so far) and again on resume (carrying the results produced that cycle): a hook with side effects there can run more than once for one assistant message
  • When an interrupt fires from BeforeToolsEvent, no tool in the batch executes (and no BeforeToolCallEvent fires) until the interrupt is answered; on resume the batch hook re-runs and, if not interrupted again, the tools execute

You can also raise interrupts from your tool definitions.

from typing import Any
from strands import Agent, tool
from strands.types.tools import ToolContext
class DeleteTool:
def __init__(self, app_name: str) -> None:
self.app_name = app_name
@tool(context=True)
def delete_files(self, tool_context: ToolContext, paths: list[str]) -> bool:
approval = tool_context.interrupt(f"{self.app_name}-approval", reason={"paths": paths})
if approval.lower() != "y":
return False
# Implementation here
return True
@tool
def inspect_files(paths: list[str]) -> dict[str, Any]:
# Implementation here
pass
agent = Agent(
system_prompt="You delete files older than 5 days",
tools=[DeleteTool("myapp").delete_files, inspect_files],
callback_handler=None,
)
...

Interrupts are not supported in direct tool calls (i.e., calls such as agent.tool.my_tool()).

Tool interrupts work like hook interrupts, with two differences: tools receive context instead of event, and interrupt names need only be unique within a tool definition rather than across all hooks on an event. For more on tool context, see ToolContext.

  • tool_context - Strands object that defines the interrupt call
  • tool_context.interrupt - Raises an interrupt with a unique name and optional reason
    • The name must be unique only among interrupt calls configured in the same tool definition. It is still advisable however to namespace your interrupts so as to more easily distinguish the calls when constructing responses outside the agent.

Strands enforces the following rules for tool interrupts:

  • All tools running concurrently will execute
  • All tools running concurrently are interruptible
  • A single tool can raise multiple interrupts but only one at a time
    • In other words, within a single tool, you can interrupt, respond to that interrupt, and then proceed to interrupt again.

Persist interrupt state with a session manager so a user can answer later, in a new agent session. You can also persist the responses themselves, so a trusted approval does not prompt again on later tool calls.

##### server.py #####
import json
from typing import Any
from strands import Agent, tool
from strands.agent import AgentResult
from strands.hooks import BeforeToolCallEvent, HookProvider, HookRegistry
from strands.session import FileSessionManager
from strands.types.agent import AgentInput
@tool
def delete_files(paths: list[str]) -> bool:
# Implementation here
pass
@tool
def inspect_files(paths: list[str]) -> dict[str, Any]:
# Implementation here
pass
class ApprovalHook(HookProvider):
def __init__(self, app_name: str) -> None:
self.app_name = app_name
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
registry.add_callback(BeforeToolCallEvent, self.approve)
def approve(self, event: BeforeToolCallEvent) -> None:
if event.tool_use["name"] != "delete_files":
return
if event.agent.state.get(f"{self.app_name}-approval") == "t": # (t)rust
return
approval = event.interrupt(f"{self.app_name}-approval", reason={"paths": event.tool_use["input"]["paths"]})
if approval.lower() not in ["y", "t"]:
event.cancel_tool = "User denied permission to delete files"
event.agent.state.set(f"{self.app_name}-approval", approval.lower())
def server(prompt: AgentInput) -> AgentResult:
agent = Agent(
hooks=[ApprovalHook("myapp")],
session_manager=FileSessionManager(session_id="myapp", storage_dir="/path/to/storage"),
system_prompt="You delete files older than 5 days",
tools=[delete_files, inspect_files],
callback_handler=None,
)
return agent(prompt)
##### client.py #####
def client(paths: list[str]) -> AgentResult:
result = server(f"paths=<{paths}>")
while True:
if result.stop_reason != "interrupt":
break
responses = []
for interrupt in result.interrupts:
if interrupt.name == "myapp-approval":
user_input = input(f"Do you want to delete {interrupt.reason['paths']} (t/y/N): ")
responses.append({
"interruptResponse": {
"interruptId": interrupt.id,
"response": user_input
}
})
result = server(responses)
return result
paths = ["a/b/c.txt", "d/e/f.txt"]
result = client(paths)
print(f"MESSAGE: {json.dumps(result.message)}")

Session managing interrupts involves the following key components:

  • session_manager - Automatically persists the agent interrupt state between tear down and start up
  • agent.state - General purpose key-value store that can be used to persist interrupt responses
    • On subsequent tool calls, you can reference the responses stored in agent.state to decide whether another interrupt is necessary. See Agent State for more.

To collect additional information from a user during an MCP tool call, use elicitation. An MCP server sends an elicitation request to the connecting client, which is handled by an elicitation callback. See MCP Elicitation for details.

Interrupts work across swarm and graph orchestration too, using the same interfaces shown here. You raise them from a BeforeNodeCallEvent hook or from within a node. See Interrupts in multi-agent systems for the swarm and graph examples.