Skip to content

strands.interrupt

Human-in-the-loop interrupt system for agent workflows.

AgentInput = str | list[ContentBlock] | list[InterruptResponseContent] | Messages | None module-attribute

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 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@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 
27
28
29
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 
32
33
34
35
36
37
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 
35
36
37
def __init__(self, interrupt: Interrupt) -> None:
    """Set the interrupt."""
    self.interrupt = interrupt

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 
138
139
140
141
142
143
144
145
class InterruptResponseContent(TypedDict):
    """Content block containing a user response to an interrupt.

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

    interruptResponse: InterruptResponse

_InterruptState dataclass

Track the state of interrupt events raised by the user.

Note, interrupt state is cleared after resuming.

Attributes:

Name Type Description
interrupts dict[str, Interrupt]

Interrupts raised by the user.
context dict[str, Any]

Additional context associated with an interrupt event.
activated bool

True if agent is in an interrupt state, False otherwise.
Source code in strands/interrupt.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@dataclass
class _InterruptState:
    """Track the state of interrupt events raised by the user.

    Note, interrupt state is cleared after resuming.

    Attributes:
        interrupts: Interrupts raised by the user.
        context: Additional context associated with an interrupt event.
        activated: True if agent is in an interrupt state, False otherwise.
    """

    interrupts: dict[str, Interrupt] = field(default_factory=dict)
    context: dict[str, Any] = field(default_factory=dict)
    activated: bool = False
    _version: int = field(default=0, compare=False, repr=False)

    def activate(self) -> None:
        """Activate the interrupt state."""
        self.activated = True
        self._version += 1

    def deactivate(self) -> None:
        """Deacitvate the interrupt state.

        Interrupts and context are cleared.
        """
        self.interrupts = {}
        self.context = {}
        self.activated = False
        self._version += 1

    def resume(self, prompt: "AgentInput") -> None:
        """Configure the interrupt state if resuming from an interrupt event.

        Args:
            prompt: User responses if resuming from interrupt.

        Raises:
            TypeError: If in interrupt state but user did not provide responses.
        """
        if not self.activated:
            return

        if not isinstance(prompt, list):
            raise TypeError(f"prompt_type={type(prompt)} | must resume from interrupt with list of interruptResponse's")

        invalid_types = [
            content_type for content in prompt for content_type in content if content_type != "interruptResponse"
        ]
        if invalid_types:
            raise TypeError(
                f"content_types=<{invalid_types}> | must resume from interrupt with list of interruptResponse's"
            )

        contents = cast(list["InterruptResponseContent"], prompt)
        for content in contents:
            interrupt_id = content["interruptResponse"]["interruptId"]
            interrupt_response = content["interruptResponse"]["response"]

            if interrupt_id not in self.interrupts:
                raise KeyError(f"interrupt_id=<{interrupt_id}> | no interrupt found")

            self.interrupts[interrupt_id].response = interrupt_response

        self.context["responses"] = contents
        self._version += 1

    def _get_version(self) -> int:
        """Get the current version number of the interrupt state.

        The version is incremented each time activate(), deactivate(), or resume() is called.
        Consumers can compare versions to detect changes without requiring
        explicit dirty flag clearing.

        Returns:
            The current version number.
        """
        return self._version

    def to_dict(self) -> dict[str, Any]:
        """Serialize to dict for session management."""
        return {
            "interrupts": {k: v.to_dict() for k, v in self.interrupts.items()},
            "context": self.context,
            "activated": self.activated,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "_InterruptState":
        """Initiailize interrupt state from serialized interrupt state.

        Interrupt state can be serialized with the `to_dict` method.
        """
        return cls(
            interrupts={
                interrupt_id: Interrupt(**interrupt_data) for interrupt_id, interrupt_data in data["interrupts"].items()
            },
            context=data["context"],
            activated=data["activated"],
        )

activate()

Activate the interrupt state.

Source code in strands/interrupt.py 
57
58
59
60
def activate(self) -> None:
    """Activate the interrupt state."""
    self.activated = True
    self._version += 1

deactivate()

Deacitvate the interrupt state.

Interrupts and context are cleared.

Source code in strands/interrupt.py 
62
63
64
65
66
67
68
69
70
def deactivate(self) -> None:
    """Deacitvate the interrupt state.

    Interrupts and context are cleared.
    """
    self.interrupts = {}
    self.context = {}
    self.activated = False
    self._version += 1

from_dict(data) classmethod

Initiailize interrupt state from serialized interrupt state.

Interrupt state can be serialized with the to_dict method.

Source code in strands/interrupt.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "_InterruptState":
    """Initiailize interrupt state from serialized interrupt state.

    Interrupt state can be serialized with the `to_dict` method.
    """
    return cls(
        interrupts={
            interrupt_id: Interrupt(**interrupt_data) for interrupt_id, interrupt_data in data["interrupts"].items()
        },
        context=data["context"],
        activated=data["activated"],
    )

resume(prompt)

Configure the interrupt state if resuming from an interrupt event.

Parameters:

Name Type Description Default
prompt AgentInput

User responses if resuming from interrupt.

 required

Raises:

Type Description
TypeError

If in interrupt state but user did not provide responses.
Source code in strands/interrupt.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def resume(self, prompt: "AgentInput") -> None:
    """Configure the interrupt state if resuming from an interrupt event.

    Args:
        prompt: User responses if resuming from interrupt.

    Raises:
        TypeError: If in interrupt state but user did not provide responses.
    """
    if not self.activated:
        return

    if not isinstance(prompt, list):
        raise TypeError(f"prompt_type={type(prompt)} | must resume from interrupt with list of interruptResponse's")

    invalid_types = [
        content_type for content in prompt for content_type in content if content_type != "interruptResponse"
    ]
    if invalid_types:
        raise TypeError(
            f"content_types=<{invalid_types}> | must resume from interrupt with list of interruptResponse's"
        )

    contents = cast(list["InterruptResponseContent"], prompt)
    for content in contents:
        interrupt_id = content["interruptResponse"]["interruptId"]
        interrupt_response = content["interruptResponse"]["response"]

        if interrupt_id not in self.interrupts:
            raise KeyError(f"interrupt_id=<{interrupt_id}> | no interrupt found")

        self.interrupts[interrupt_id].response = interrupt_response

    self.context["responses"] = contents
    self._version += 1

to_dict()

Serialize to dict for session management.

Source code in strands/interrupt.py 
120
121
122
123
124
125
126
def to_dict(self) -> dict[str, Any]:
    """Serialize to dict for session management."""
    return {
        "interrupts": {k: v.to_dict() for k, v in self.interrupts.items()},
        "context": self.context,
        "activated": self.activated,
    }