interrupts - swarm - from agent

Author: pgrayy

src/strands/event_loop/event_loop.py

@@ -483,7 +483,8 @@ async def _handle_tool_execution(
 
     if interrupts:
         # Session state stored on AfterInvocationEvent.
-        agent._interrupt_state.activate(context={"tool_use_message": message, "tool_results": tool_results})
+        agent._interrupt_state.context = {"tool_use_message": message, "tool_results": tool_results}
+        agent._interrupt_state.activate()
 
         agent.event_loop_metrics.end_cycle(cycle_start_time, cycle_trace)
         yield EventLoopStopEvent(

src/strands/interrupt.py

@@ -53,13 +53,8 @@ class _InterruptState:
     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 {}
+    def activate(self) -> None:
+        """Activate the interrupt state."""
         self.activated = True
 
     def deactivate(self) -> None:

src/strands/multiagent/base.py

@@ -12,29 +12,34 @@
 
 from .._async import run_async
 from ..agent import AgentResult
-from ..types.content import ContentBlock
+from ..interrupt import Interrupt
 from ..types.event_loop import Metrics, Usage
+from ..types.multiagent import MultiAgentInput
 
 logger = logging.getLogger(__name__)
 
 
 class Status(Enum):
-    """Execution status for both graphs and nodes."""
+    """Execution status for both graphs and nodes.
+    
+    Attributes:
+        PENDING: Task has not started execution yet.
+        EXECUTING: Task is currently running.
+        COMPLETED: Task finished successfully.
+        FAILED: Task encountered an error and could not complete.
+        INTERRUPTED: Task was interrupted by user.
+    """
 
     PENDING = "pending"
     EXECUTING = "executing"
     COMPLETED = "completed"
     FAILED = "failed"
+    INTERRUPTED = "interrupted"
 
 
 @dataclass
 class NodeResult:
-    """Unified result from node execution - handles both Agent and nested MultiAgentBase results.
-
-    The status field represents the semantic outcome of the node's work:
-    - COMPLETED: The node's task was successfully accomplished
-    - FAILED: The node's task failed or produced an error
-    """
+    """Unified result from node execution - handles both Agent and nested MultiAgentBase results."""
 
     # Core result data - single AgentResult, nested MultiAgentResult, or Exception
     result: Union[AgentResult, "MultiAgentResult", Exception]
@@ -47,6 +52,7 @@ class NodeResult:
     accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
     accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
     execution_count: int = 0
+    interrupts: list[Interrupt] = field(default_factory=list)
 
     def get_agent_results(self) -> list[AgentResult]:
         """Get all AgentResult objects from this node, flattened if nested."""
@@ -78,6 +84,7 @@ def to_dict(self) -> dict[str, Any]:
             "accumulated_usage": self.accumulated_usage,
             "accumulated_metrics": self.accumulated_metrics,
             "execution_count": self.execution_count,
+            "interrupts": [interrupt.to_dict() for interrupt in self.interrupts],
         }
 
     @classmethod
@@ -99,6 +106,10 @@ def from_dict(cls, data: dict[str, Any]) -> "NodeResult":
 
         usage = _parse_usage(data.get("accumulated_usage", {}))
         metrics = _parse_metrics(data.get("accumulated_metrics", {}))
+        
+        interrupts = []
+        for interrupt_data in data.get("interrupts", []):
+            interrupts.append(Interrupt.from_dict(interrupt_data))
 
         return cls(
             result=result,
@@ -107,24 +118,21 @@ def from_dict(cls, data: dict[str, Any]) -> "NodeResult":
             accumulated_usage=usage,
             accumulated_metrics=metrics,
             execution_count=int(data.get("execution_count", 0)),
+            interrupts=interrupts,
         )
 
 
 @dataclass
 class MultiAgentResult:
-    """Result from multi-agent execution with accumulated metrics.
-
-    The status field represents the outcome of the MultiAgentBase execution:
-    - COMPLETED: The execution was successfully accomplished
-    - FAILED: The execution failed or produced an error
-    """
+    """Result from multi-agent execution with accumulated metrics."""
 
     status: Status = Status.PENDING
     results: dict[str, NodeResult] = field(default_factory=lambda: {})
     accumulated_usage: Usage = field(default_factory=lambda: Usage(inputTokens=0, outputTokens=0, totalTokens=0))
     accumulated_metrics: Metrics = field(default_factory=lambda: Metrics(latencyMs=0))
     execution_count: int = 0
     execution_time: int = 0
+    interrupts: list[Interrupt] = field(default_factory=list)
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> "MultiAgentResult":
@@ -136,13 +144,18 @@ def from_dict(cls, data: dict[str, Any]) -> "MultiAgentResult":
         usage = _parse_usage(data.get("accumulated_usage", {}))
         metrics = _parse_metrics(data.get("accumulated_metrics", {}))
 
+        interrupts = []
+        for interrupt_data in data.get("interrupts", []):
+            interrupts.append(Interrupt.from_dict(interrupt_data))
+
         multiagent_result = cls(
             status=Status(data["status"]),
             results=results,
             accumulated_usage=usage,
             accumulated_metrics=metrics,
             execution_count=int(data.get("execution_count", 0)),
             execution_time=int(data.get("execution_time", 0)),
+            interrupts=interrupts,
         )
         return multiagent_result
 
@@ -156,6 +169,7 @@ def to_dict(self) -> dict[str, Any]:
             "accumulated_metrics": self.accumulated_metrics,
             "execution_count": self.execution_count,
             "execution_time": self.execution_time,
+            "interrupts": [interrupt.to_dict() for interrupt in self.interrupts],
         }
 
 
@@ -173,7 +187,7 @@ class MultiAgentBase(ABC):
 
     @abstractmethod
     async def invoke_async(
-        self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
+        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
     ) -> MultiAgentResult:
         """Invoke asynchronously.
 
@@ -186,7 +200,7 @@ async def invoke_async(
         raise NotImplementedError("invoke_async not implemented")
 
     async def stream_async(
-        self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
+        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
     ) -> AsyncIterator[dict[str, Any]]:
         """Stream events during multi-agent execution.
 
@@ -211,7 +225,7 @@ async def stream_async(
         yield {"result": result}
 
     def __call__(
-        self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
+        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
     ) -> MultiAgentResult:
         """Invoke synchronously.
 

src/strands/multiagent/graph.py

@@ -45,6 +45,7 @@
 )
 from ..types.content import ContentBlock, Messages
 from ..types.event_loop import Metrics, Usage
+from ..types.multiagent import MultiAgentInput
 from .base import MultiAgentBase, MultiAgentResult, NodeResult, Status
 
 logger = logging.getLogger(__name__)
@@ -67,7 +68,7 @@ class GraphState:
     """
 
     # Task (with default empty string)
-    task: str | list[ContentBlock] = ""
+    task: MultiAgentInput = ""
 
     # Execution state
     status: Status = Status.PENDING
@@ -456,7 +457,7 @@ def __init__(
         run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))
 
     def __call__(
-        self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
+        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
     ) -> GraphResult:
         """Invoke the graph synchronously.
 
@@ -472,7 +473,7 @@ def __call__(
         return run_async(lambda: self.invoke_async(task, invocation_state))
 
     async def invoke_async(
-        self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
+        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
     ) -> GraphResult:
         """Invoke the graph asynchronously.
 
@@ -496,7 +497,7 @@ async def invoke_async(
         return cast(GraphResult, final_event["result"])
 
     async def stream_async(
-        self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
+        self, task: MultiAgentInput, invocation_state: dict[str, Any] | None = None, **kwargs: Any
     ) -> AsyncIterator[dict[str, Any]]:
         """Stream events during graph execution.