interrupt

Author: pgrayy

src/strands/agent/agent.py

@@ -349,6 +349,9 @@ def __init__(
                 self.hooks.add_hook(hook)
         self.hooks.invoke_callbacks(AgentInitializedEvent(agent=self))
 
+        self.interrupted = False
+        self.interrupts = {}
+
     @property
     def tool(self) -> ToolCaller:
         """Call tool as a function.
@@ -540,6 +543,7 @@ async def stream_async(
         Args:
             prompt: User input in various formats:
                 - str: Simple text input
+                - ContentBlock: Multi-modal content block
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
@@ -564,6 +568,8 @@ async def stream_async(
                     yield event["data"]
             ```
         """
+        self._resume(prompt)
+
         callback_handler = kwargs.get("callback_handler", self.callback_handler)
 
         # Process input and get message to add (if any)
@@ -585,6 +591,11 @@ async def stream_async(
 
                 result = AgentResult(*event["stop"])
                 callback_handler(result=result)
+
+                if result.stop_reason == "interrupt":
+                    self.interrupted = True
+                    self.interrupts = {interrupt.name: interrupt for interrupt in result.interrupts}
+
                 yield AgentResultEvent(result=result).as_dict()
 
                 self._end_agent_trace_span(response=result)
@@ -593,6 +604,16 @@ async def stream_async(
                 self._end_agent_trace_span(error=e)
                 raise
 
+    def _resume(self, prompt: AgentInput) -> None:
+        if not self.interrupted:
+            return
+
+        if not isinstance(prompt, dict) or "resume" not in prompt:
+            raise ValueError("<TODO>.")
+
+        for interrupt in self.interrupts.values():
+            interrupt.resume = prompt["resume"][interrupt.name]
+
     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.
 
@@ -673,6 +694,8 @@ def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
             if isinstance(prompt, str):
                 # String input - convert to user message
                 messages = [{"role": "user", "content": [{"text": prompt}]}]
+            elif isinstance(prompt, dict):
+                messages = [{"role": "user", "content": prompt}] if "resume" not in prompt else []
             elif isinstance(prompt, list):
                 if len(prompt) == 0:
                     # Empty list
@@ -692,7 +715,9 @@ def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
         else:
             messages = []
         if messages is None:
-            raise ValueError("Input prompt must be of type: `str | list[Contentblock] | Messages | None`.")
+            raise ValueError(
+                "Input prompt must be of type: `str | ContentBlock | list[Contentblock] | Messages | None`."
+            )
         return messages
 
     def _record_tool_execution(

src/strands/agent/agent_result.py

@@ -4,8 +4,9 @@
 """
 
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Optional
 
+from ..hooks.interrupt import Interrupt
 from ..telemetry.metrics import EventLoopMetrics
 from ..types.content import Message
 from ..types.streaming import StopReason
@@ -26,6 +27,7 @@ class AgentResult:
     message: Message
     metrics: EventLoopMetrics
     state: Any
+    interrupts: Optional[list[Interrupt]] = None
 
     def __str__(self) -> str:
         """Get the agent's last message as a string.

src/strands/event_loop/event_loop.py

@@ -23,7 +23,7 @@
     MessageAddedEvent,
 )
 from ..telemetry.metrics import Trace
-from ..telemetry.tracer import get_tracer
+from ..telemetry.tracer import Tracer, get_tracer
 from ..tools._validator import validate_and_prepare_tools
 from ..types._events import (
     EventLoopStopEvent,
@@ -33,6 +33,7 @@
     ModelStopReason,
     StartEvent,
     StartEventLoopEvent,
+    ToolInterruptEvent,
     ToolResultMessageEvent,
     TypedEvent,
 )
@@ -112,104 +113,22 @@ async def event_loop_cycle(agent: "Agent", invocation_state: dict[str, Any]) ->
     )
     invocation_state["event_loop_cycle_span"] = cycle_span
 
-    # Create a trace for the stream_messages call
-    stream_trace = Trace("stream_messages", parent_id=cycle_trace.id)
-    cycle_trace.add_child(stream_trace)
-
-    # Process messages with exponential backoff for throttling
-    message: Message
     stop_reason: StopReason
-    usage: Any
-    metrics: Metrics
-
-    # Retry loop for handling throttling exceptions
-    current_delay = INITIAL_DELAY
-    for attempt in range(MAX_ATTEMPTS):
-        model_id = agent.model.config.get("model_id") if hasattr(agent.model, "config") else None
-        model_invoke_span = tracer.start_model_invoke_span(
-            messages=agent.messages,
-            parent_span=cycle_span,
-            model_id=model_id,
-        )
-        with trace_api.use_span(model_invoke_span):
-            agent.hooks.invoke_callbacks(
-                BeforeModelInvocationEvent(
-                    agent=agent,
-                )
-            )
-
-            tool_specs = agent.tool_registry.get_all_tool_specs()
 
-            try:
-                async for event in stream_messages(agent.model, agent.system_prompt, agent.messages, tool_specs):
-                    if not isinstance(event, ModelStopReason):
-                        yield event
+    if agent.interrupted:
+        stop_reason = "tool_use"
+    else:
+        events = _handle_model_execution(agent, cycle_span, cycle_trace, invocation_state, tracer)
+        async for event in events:
+            if isinstance(event, ModelStopReason):
+                stop_reason = event["stop"][0]
+                continue
 
-                stop_reason, message, usage, metrics = event["stop"]
-                invocation_state.setdefault("request_state", {})
+            yield event
 
-                agent.hooks.invoke_callbacks(
-                    AfterModelInvocationEvent(
-                        agent=agent,
-                        stop_response=AfterModelInvocationEvent.ModelStopResponse(
-                            stop_reason=stop_reason,
-                            message=message,
-                        ),
-                    )
-                )
-
-                if stop_reason == "max_tokens":
-                    message = recover_message_on_max_tokens_reached(message)
-
-                if model_invoke_span:
-                    tracer.end_model_invoke_span(model_invoke_span, message, usage, stop_reason)
-                break  # Success! Break out of retry loop
-
-            except Exception as e:
-                if model_invoke_span:
-                    tracer.end_span_with_error(model_invoke_span, str(e), e)
-
-                agent.hooks.invoke_callbacks(
-                    AfterModelInvocationEvent(
-                        agent=agent,
-                        exception=e,
-                    )
-                )
-
-                if isinstance(e, ModelThrottledException):
-                    if attempt + 1 == MAX_ATTEMPTS:
-                        yield ForceStopEvent(reason=e)
-                        raise e
-
-                    logger.debug(
-                        "retry_delay_seconds=<%s>, max_attempts=<%s>, current_attempt=<%s> "
-                        "| throttling exception encountered "
-                        "| delaying before next retry",
-                        current_delay,
-                        MAX_ATTEMPTS,
-                        attempt + 1,
-                    )
-                    await asyncio.sleep(current_delay)
-                    current_delay = min(current_delay * 2, MAX_DELAY)
-
-                    yield EventLoopThrottleEvent(delay=current_delay)
-                else:
-                    raise e
+    message = agent.messages[-1]
 
     try:
-        # Add message in trace and mark the end of the stream messages trace
-        stream_trace.add_message(message)
-        stream_trace.end()
-
-        # Add the response message to the conversation
-        agent.messages.append(message)
-        agent.hooks.invoke_callbacks(MessageAddedEvent(agent=agent, message=message))
-        yield ModelMessageEvent(message=message)
-
-        # Update metrics
-        agent.event_loop_metrics.update_usage(usage)
-        agent.event_loop_metrics.update_metrics(metrics)
-
         if stop_reason == "max_tokens":
             """
             Handle max_tokens limit reached by the model.
@@ -307,6 +226,105 @@ async def recurse_event_loop(agent: "Agent", invocation_state: dict[str, Any]) -
     recursive_trace.end()
 
 
+async def _handle_model_execution(
+    agent: "Agent",
+    cycle_span: Any,
+    cycle_trace: Trace,
+    invocation_state: dict[str, Any],
+    tracer: Tracer,
+) -> AsyncGenerator[TypedEvent, None]:
+    """<TODO>."""
+    # Create a trace for the stream_messages call
+    stream_trace = Trace("stream_messages", parent_id=cycle_trace.id)
+    cycle_trace.add_child(stream_trace)
+
+    # Retry loop for handling throttling exceptions
+    current_delay = INITIAL_DELAY
+    for attempt in range(MAX_ATTEMPTS):
+        model_id = agent.model.config.get("model_id") if hasattr(agent.model, "config") else None
+        model_invoke_span = tracer.start_model_invoke_span(
+            messages=agent.messages,
+            parent_span=cycle_span,
+            model_id=model_id,
+        )
+        with trace_api.use_span(model_invoke_span):
+            agent.hooks.invoke_callbacks(
+                BeforeModelInvocationEvent(
+                    agent=agent,
+                )
+            )
+
+            tool_specs = agent.tool_registry.get_all_tool_specs()
+
+            try:
+                async for event in stream_messages(agent.model, agent.system_prompt, agent.messages, tool_specs):
+                    yield event
+
+                stop_reason, message, usage, metrics = event["stop"]
+                invocation_state.setdefault("request_state", {})
+
+                agent.hooks.invoke_callbacks(
+                    AfterModelInvocationEvent(
+                        agent=agent,
+                        stop_response=AfterModelInvocationEvent.ModelStopResponse(
+                            stop_reason=stop_reason,
+                            message=message,
+                        ),
+                    )
+                )
+
+                if stop_reason == "max_tokens":
+                    message = recover_message_on_max_tokens_reached(message)
+
+                if model_invoke_span:
+                    tracer.end_model_invoke_span(model_invoke_span, message, usage, stop_reason)
+                break  # Success! Break out of retry loop
+
+            except Exception as e:
+                if model_invoke_span:
+                    tracer.end_span_with_error(model_invoke_span, str(e), e)
+
+                agent.hooks.invoke_callbacks(
+                    AfterModelInvocationEvent(
+                        agent=agent,
+                        exception=e,
+                    )
+                )
+
+                if isinstance(e, ModelThrottledException):
+                    if attempt + 1 == MAX_ATTEMPTS:
+                        yield ForceStopEvent(reason=e)
+                        raise e
+
+                    logger.debug(
+                        "retry_delay_seconds=<%s>, max_attempts=<%s>, current_attempt=<%s> "
+                        "| throttling exception encountered "
+                        "| delaying before next retry",
+                        current_delay,
+                        MAX_ATTEMPTS,
+                        attempt + 1,
+                    )
+                    await asyncio.sleep(current_delay)
+                    current_delay = min(current_delay * 2, MAX_DELAY)
+
+                    yield EventLoopThrottleEvent(delay=current_delay)
+                else:
+                    raise e
+
+    # Add message in trace and mark the end of the stream messages trace
+    stream_trace.add_message(message)
+    stream_trace.end()
+
+    # Update metrics
+    agent.event_loop_metrics.update_usage(usage)
+    agent.event_loop_metrics.update_metrics(metrics)
+
+    # Add the response message to the conversation
+    agent.messages.append(message)
+    agent.hooks.invoke_callbacks(MessageAddedEvent(agent=agent, message=message))
+    yield ModelMessageEvent(message=message)
+
+
 async def _handle_tool_execution(
     stop_reason: StopReason,
     message: Message,
@@ -345,15 +363,29 @@ async def _handle_tool_execution(
         yield EventLoopStopEvent(stop_reason, message, agent.event_loop_metrics, invocation_state["request_state"])
         return
 
+    tool_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:
         yield tool_event
 
+        if isinstance(tool_event, ToolInterruptEvent):
+            tool_interrupts.append(tool_event["tool_interrupt_event"]["interrupt"])
+
     # Store parent cycle ID for the next cycle
     invocation_state["event_loop_parent_cycle_id"] = invocation_state["event_loop_cycle_id"]
 
+    if tool_interrupts:
+        # TODO: deal with metrics and traces
+        yield EventLoopStopEvent(
+            "interrupt", message, agent.event_loop_metrics, invocation_state["request_state"], tool_interrupts
+        )
+        return
+
+    agent.interrupted = False
+    agent.interrupts = {}
+
     tool_result_message: Message = {
         "role": "user",
         "content": [{"toolResult": result} for result in tool_results],