strands.types.interrupt

Interrupt related type definitions for human-in-the-loop workflows.

Interrupt Flow

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

Example

from typing import Any

from strands import Agent, tool
from strands.hooks import BeforeToolCallEvent, HookProvider, HookRegistry


@tool
def delete_tool(key: str) -> bool:
    print("DELETE_TOOL | deleting")
    return True


class ToolInterruptHook(HookProvider):
    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_tool":
            return

        approval = event.interrupt("for_delete_tool", reason="APPROVAL")
        if approval != "A":
            event.cancel_tool = "approval was not granted"

agent = Agent(
    hooks=[ToolInterruptHook()],
    tools=[delete_tool],
    system_prompt="You delete objects given their keys.",
    callback_handler=None,
)
result = agent(f"delete object with key 'X'")

if result.stop_reason == "interrupt":
    responses = []
    for interrupt in result.interrupts:
        if interrupt.name == "for_delete_tool":
            responses.append({"interruptResponse": {"interruptId": interrupt.id, "response": "A"})

    result = agent(responses)

...

Details

• User raises interrupt on their hook event by calling event.interrupt().
• User can raise one interrupt per hook callback.
• Interrupts stop the agent event loop.
• Interrupts are returned to the user in AgentResult.
• User resumes by invoking agent with interrupt responses.
• Second call to event.interrupt() returns user response.
• Process repeats if user raises additional interrupts.
• Interrupts are session managed in-between return and user response.

Interrupt dataclass

Represents an interrupt that can pause agent execution for human-in-the-loop workflows.

Attributes:

Name	Type	Description
id	str	Unique identifier.
name	str	User defined name.
reason	Any	User provided reason for raising the interrupt.
response	Any	Human response provided when resuming the agent after an interrupt.

Source code in strands/interrupt.py

@dataclass
class Interrupt:
    """Represents an interrupt that can pause agent execution for human-in-the-loop workflows.

    Attributes:
        id: Unique identifier.
        name: User defined name.
        reason: User provided reason for raising the interrupt.
        response: Human response provided when resuming the agent after an interrupt.
    """

    id: str
    name: str
    reason: Any = None
    response: Any = None

    def to_dict(self) -> dict[str, Any]:
        """Serialize to dict for session management."""
        return asdict(self)

to_dict()

Serialize to dict for session management.

Source code in strands/interrupt.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to dict for session management."""
    return asdict(self)

InterruptException

Bases: Exception

Exception raised when human input is required.

Source code in strands/interrupt.py

class InterruptException(Exception):
    """Exception raised when human input is required."""

    def __init__(self, interrupt: Interrupt) -> None:
        """Set the interrupt."""
        self.interrupt = interrupt

__init__(interrupt)

Set the interrupt.

Source code in strands/interrupt.py

def __init__(self, interrupt: Interrupt) -> None:
    """Set the interrupt."""
    self.interrupt = interrupt

InterruptResponse

Bases: TypedDict

User response to an interrupt.

Attributes:

Name	Type	Description
interruptId	str	Unique identifier for the interrupt.
response	Any	User response to the interrupt.

Source code in strands/types/interrupt.py

class InterruptResponse(TypedDict):
    """User response to an interrupt.

    Attributes:
        interruptId: Unique identifier for the interrupt.
        response: User response to the interrupt.
    """

    interruptId: str
    response: Any

InterruptResponseContent

Bases: TypedDict

Content block containing a user response to an interrupt.

Attributes:

Name	Type	Description
interruptResponse	InterruptResponse	User response to an interrupt event.

Source code in strands/types/interrupt.py

class InterruptResponseContent(TypedDict):
    """Content block containing a user response to an interrupt.

    Attributes:
        interruptResponse: User response to an interrupt event.
    """

    interruptResponse: InterruptResponse

_Interruptible

Bases: Protocol

Interface that adds interrupt support to hook events and tools.

Source code in strands/types/interrupt.py

class _Interruptible(Protocol):
    """Interface that adds interrupt support to hook events and tools."""

    def interrupt(self, name: str, reason: Any = None, response: Any = None) -> Any:
        """Trigger the interrupt with a reason.

        Args: name: User defined name for the interrupt.
                Must be unique across hook callbacks.
            reason: User provided reason for the interrupt.
            response: Preemptive response from user if available.

        Returns:
            The response from a human user when resuming from an interrupt state.

        Raises:
            InterruptException: If human input is required.
            RuntimeError: If agent instance attribute not set.
        """
        for attr_name in ["agent", "source"]:
            if hasattr(self, attr_name):
                agent = getattr(self, attr_name)
                break
        else:
            raise RuntimeError("agent instance attribute not set")

        id = self._interrupt_id(name)
        state = agent._interrupt_state

        interrupt_ = state.interrupts.setdefault(id, Interrupt(id, name, reason, response))
        if interrupt_.response is not None:
            return interrupt_.response

        raise InterruptException(interrupt_)

    def _interrupt_id(self, name: str) -> str:
        """Unique id for the interrupt.

        Args:
            name: User defined name for the interrupt.
            reason: User provided reason for the interrupt.

        Returns:
            Interrupt id.
        """
        ...

interrupt(name, reason=None, response=None)

Trigger the interrupt with a reason.

reason: User provided reason for the interrupt.
response: Preemptive response from user if available.

Returns:

Type	Description
Any	The response from a human user when resuming from an interrupt state.

Raises:

Type	Description
InterruptException	If human input is required.
RuntimeError	If agent instance attribute not set.

Source code in strands/types/interrupt.py

def interrupt(self, name: str, reason: Any = None, response: Any = None) -> Any:
    """Trigger the interrupt with a reason.

    Args: name: User defined name for the interrupt.
            Must be unique across hook callbacks.
        reason: User provided reason for the interrupt.
        response: Preemptive response from user if available.

    Returns:
        The response from a human user when resuming from an interrupt state.

    Raises:
        InterruptException: If human input is required.
        RuntimeError: If agent instance attribute not set.
    """
    for attr_name in ["agent", "source"]:
        if hasattr(self, attr_name):
            agent = getattr(self, attr_name)
            break
    else:
        raise RuntimeError("agent instance attribute not set")

    id = self._interrupt_id(name)
    state = agent._interrupt_state

    interrupt_ = state.interrupts.setdefault(id, Interrupt(id, name, reason, response))
    if interrupt_.response is not None:
        return interrupt_.response

    raise InterruptException(interrupt_)