hooks - before tool call event - interrupt (strands-agents#987)

Author: pgrayy

src/strands/agent/agent.py

@@ -55,13 +55,15 @@
 from ..types.agent import AgentInput
 from ..types.content import ContentBlock, Message, Messages
 from ..types.exceptions import ContextWindowOverflowException
+from ..types.interrupt import InterruptResponseContent
 from ..types.tools import ToolResult, ToolUse
 from ..types.traces import AttributeValue
 from .agent_result import AgentResult
 from .conversation_manager import (
     ConversationManager,
     SlidingWindowConversationManager,
 )
+from .interrupt import InterruptState
 from .state import AgentState
 
 logger = logging.getLogger(__name__)
@@ -143,6 +145,9 @@ def caller(
                 Raises:
                     AttributeError: If the tool doesn't exist.
                 """
+                if self._agent._interrupt_state.activated:
+                    raise RuntimeError("cannot directly call tool during interrupt")
+
                 normalized_name = self._find_normalized_tool_name(name)
 
                 # Create unique tool ID and set up the tool request
@@ -338,6 +343,8 @@ def __init__(
 
         self.hooks = HookRegistry()
 
+        self._interrupt_state = InterruptState()
+
         # Initialize session management functionality
         self._session_manager = session_manager
         if self._session_manager:
@@ -491,6 +498,9 @@ async def structured_output_async(self, output_model: Type[T], prompt: AgentInpu
         Raises:
             ValueError: If no conversation history or prompt is provided.
         """
+        if self._interrupt_state.activated:
+            raise RuntimeError("cannot call structured output during interrupt")
+
         self.hooks.invoke_callbacks(BeforeInvocationEvent(agent=self))
         with self.tracer.tracer.start_as_current_span(
             "execute_structured_output", kind=trace_api.SpanKind.CLIENT
@@ -573,6 +583,8 @@ async def stream_async(
                     yield event["data"]
             ```
         """
+        self._resume_interrupt(prompt)
+
         merged_state = {}
         if kwargs:
             warnings.warn("`**kwargs` parameter is deprecating, use `invocation_state` instead.", stacklevel=2)
@@ -614,6 +626,38 @@ async def stream_async(
                 self._end_agent_trace_span(error=e)
                 raise
 
+    def _resume_interrupt(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._interrupt_state.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"
+            )
+
+        for content in cast(list[InterruptResponseContent], prompt):
+            interrupt_id = content["interruptResponse"]["interruptId"]
+            interrupt_response = content["interruptResponse"]["response"]
+
+            if interrupt_id not in self._interrupt_state.interrupts:
+                raise KeyError(f"interrupt_id=<{interrupt_id}> | no interrupt found")
+
+            self._interrupt_state.interrupts[interrupt_id].response = interrupt_response
+
     async def _run_loop(self, messages: Messages, invocation_state: dict[str, Any]) -> AsyncGenerator[TypedEvent, None]:
         """Execute the agent's event loop with the given message and parameters.
 
@@ -689,6 +733,9 @@ async def _execute_event_loop_cycle(self, invocation_state: dict[str, Any]) -> A
                 yield event
 
     def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
+        if self._interrupt_state.activated:
+            return []
+
         messages: Messages | None = None
         if prompt is not None:
             if isinstance(prompt, str):

src/strands/agent/agent_result.py

@@ -4,8 +4,9 @@
 """
 
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Sequence
 
+from ..interrupt import Interrupt
 from ..telemetry.metrics import EventLoopMetrics
 from ..types.content import Message
 from ..types.streaming import StopReason
@@ -20,12 +21,14 @@ class AgentResult:
         message: The last message generated by the agent.
         metrics: Performance metrics collected during processing.
         state: Additional state information from the event loop.
+        interrupts: List of interrupts if raised by user.
     """
 
     stop_reason: StopReason
     message: Message
     metrics: EventLoopMetrics
     state: Any
+    interrupts: Sequence[Interrupt] | None = None
 
     def __str__(self) -> str:
         """Get the agent's last message as a string.

src/strands/agent/interrupt.py

@@ -0,0 +1,59 @@
+"""Track the state of interrupt events raised by the user for human-in-the-loop workflows."""
+
+from dataclasses import asdict, dataclass, field
+from typing import Any
+
+from ..interrupt import Interrupt
+
+
+@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
+
+    def activate(self, context: dict[str, Any] | None = None) -> None:
+        """Activate the interrupt state.
+
+        Args:
+            context: Context associated with the interrupt event.
+        """
+        self.context = context or {}
+        self.activated = True
+
+    def deactivate(self) -> None:
+        """Deacitvate the interrupt state.
+
+        Interrupts and context are cleared.
+        """
+        self.interrupts = {}
+        self.context = {}
+        self.activated = False
+
+    def to_dict(self) -> dict[str, Any]:
+        """Serialize to dict for session management."""
+        return asdict(self)
+
+    @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"],
+        )

src/strands/event_loop/event_loop.py

@@ -27,6 +27,7 @@
     ModelStopReason,
     StartEvent,
     StartEventLoopEvent,
+    ToolInterruptEvent,
     ToolResultMessageEvent,
     TypedEvent,
 )
@@ -106,13 +107,19 @@ async def event_loop_cycle(agent: "Agent", invocation_state: dict[str, Any]) ->
     )
     invocation_state["event_loop_cycle_span"] = cycle_span
 
-    model_events = _handle_model_execution(agent, cycle_span, cycle_trace, invocation_state, tracer)
-    async for model_event in model_events:
-        if not isinstance(model_event, ModelStopReason):
-            yield model_event
+    # Skipping model invocation if in interrupt state as interrupts are currently only supported for tool calls.
+    if agent._interrupt_state.activated:
+        stop_reason: StopReason = "tool_use"
+        message = agent._interrupt_state.context["tool_use_message"]
 
-    stop_reason, message, *_ = model_event["stop"]
-    yield ModelMessageEvent(message=message)
+    else:
+        model_events = _handle_model_execution(agent, cycle_span, cycle_trace, invocation_state, tracer)
+        async for model_event in model_events:
+            if not isinstance(model_event, ModelStopReason):
+                yield model_event
+
+        stop_reason, message, *_ = model_event["stop"]
+        yield ModelMessageEvent(message=message)
 
     try:
         if stop_reason == "max_tokens":
@@ -142,6 +149,7 @@ async def event_loop_cycle(agent: "Agent", invocation_state: dict[str, Any]) ->
                 cycle_span=cycle_span,
                 cycle_start_time=cycle_start_time,
                 invocation_state=invocation_state,
+                tracer=tracer,
             )
             async for tool_event in tool_events:
                 yield tool_event
@@ -345,6 +353,7 @@ async def _handle_tool_execution(
     cycle_span: Any,
     cycle_start_time: float,
     invocation_state: dict[str, Any],
+    tracer: Tracer,
 ) -> AsyncGenerator[TypedEvent, None]:
     """Handles the execution of tools requested by the model during an event loop cycle.
 
@@ -356,6 +365,7 @@ async def _handle_tool_execution(
         cycle_span: Span object for tracing the cycle (type may vary).
         cycle_start_time: Start time of the current cycle.
         invocation_state: Additional keyword arguments, including request state.
+        tracer: Tracer instance for span management.
 
     Yields:
         Tool stream events along with events yielded from a recursive call to the event loop. The last event is a tuple
@@ -375,15 +385,45 @@ async def _handle_tool_execution(
         yield EventLoopStopEvent(stop_reason, message, agent.event_loop_metrics, invocation_state["request_state"])
         return
 
+    if agent._interrupt_state.activated:
+        tool_results.extend(agent._interrupt_state.context["tool_results"])
+
+        # Filter to only the interrupted tools when resuming from interrupt (tool uses without results)
+        tool_use_ids = {tool_result["toolUseId"] for tool_result in tool_results}
+        tool_uses = [tool_use for tool_use in tool_uses if tool_use["toolUseId"] not in tool_use_ids]
+
+    interrupts = []
     tool_events = agent.tool_executor._execute(
         agent, tool_uses, tool_results, cycle_trace, cycle_span, invocation_state
     )
     async for tool_event in tool_events:
+        if isinstance(tool_event, ToolInterruptEvent):
+            interrupts.extend(tool_event["tool_interrupt_event"]["interrupts"])
+
         yield tool_event
 
     # Store parent cycle ID for the next cycle
     invocation_state["event_loop_parent_cycle_id"] = invocation_state["event_loop_cycle_id"]
 
+    if interrupts:
+        # Session state stored on AfterInvocationEvent.
+        agent._interrupt_state.activate(context={"tool_use_message": message, "tool_results": tool_results})
+
+        agent.event_loop_metrics.end_cycle(cycle_start_time, cycle_trace)
+        yield EventLoopStopEvent(
+            "interrupt",
+            message,
+            agent.event_loop_metrics,
+            invocation_state["request_state"],
+            interrupts,
+        )
+        if cycle_span:
+            tracer.end_event_loop_cycle_span(span=cycle_span, message=message)
+
+        return
+
+    agent._interrupt_state.deactivate()
+
     tool_result_message: Message = {
         "role": "user",
         "content": [{"toolResult": result} for result in tool_results],
@@ -394,7 +434,6 @@ async def _handle_tool_execution(
     yield ToolResultMessageEvent(message=tool_result_message)
 
     if cycle_span:
-        tracer = get_tracer()
         tracer.end_event_loop_cycle_span(span=cycle_span, message=message, tool_result_message=tool_result_message)
 
     if invocation_state["request_state"].get("stop_event_loop", False):

src/strands/hooks/events.py

@@ -3,10 +3,14 @@
 This module defines the events that are emitted as Agents run through the lifecycle of a request.
 """
 
+import uuid
 from dataclasses import dataclass
 from typing import Any, Optional
 
+from typing_extensions import override
+
 from ..types.content import Message
+from ..types.interrupt import InterruptHookEvent
 from ..types.streaming import StopReason
 from ..types.tools import AgentTool, ToolResult, ToolUse
 from .registry import HookEvent
@@ -84,7 +88,7 @@ class MessageAddedEvent(HookEvent):
 
 
 @dataclass
-class BeforeToolCallEvent(HookEvent):
+class BeforeToolCallEvent(HookEvent, InterruptHookEvent):
     """Event triggered before a tool is invoked.
 
     This event is fired just before the agent executes a tool, allowing hook
@@ -110,6 +114,18 @@ class BeforeToolCallEvent(HookEvent):
     def _can_write(self, name: str) -> bool:
         return name in ["cancel_tool", "selected_tool", "tool_use"]
 
+    @override
+    def _interrupt_id(self, name: str) -> str:
+        """Unique id for the interrupt.
+
+        Args:
+            name: User defined name for the interrupt.
+
+        Returns:
+            Interrupt id.
+        """
+        return f"v1:{self.tool_use['toolUseId']}:{uuid.uuid5(uuid.NAMESPACE_OID, name)}"
+
 
 @dataclass
 class AfterToolCallEvent(HookEvent):