feat(strands): bridge native Strands interrupts to AG-UI interrupt round-trip

integrations/aws-strands/python/README.md

@@ -83,6 +83,58 @@ agentcore deploy
 
 For the complete deployment walkthrough, see [Deploy AG-UI servers in AgentCore Runtime](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-agui.html).
 
+## Human-in-the-loop (native Strands interrupts)
+
+Tools that pause with `tool_context.interrupt(...)` are bridged to the AG-UI
+interrupt round-trip:
+
+- When a run pauses, it finishes with `RUN_FINISHED` carrying a
+  `RunFinishedInterruptOutcome` (`outcome.type == "interrupt"`) and one AG-UI
+  `Interrupt` per Strands interrupt. The Strands interrupt *name* becomes the
+  AG-UI `reason`; the original (free-form) reason object is preserved under
+  `metadata.strands_reason`.
+- To resume, the client sends the next `RunAgentInput` on the **same
+  `thread_id`** with `resume=[ResumeEntry(interrupt_id=..., status="resolved",
+  payload=...)]`. The tool's `interrupt()` call returns exactly `payload`.
+- `status="cancelled"` resumes the tool with the sentinel
+  `{"cancelled": True}` (`ag_ui_strands.agent.INTERRUPT_CANCELLED`) so it can
+  treat the pause as a denial.
+- **Re-execution on resume:** resuming a paused tool re-runs its body from
+  the top — any code before the `interrupt()` call executes again. Guard
+  side effects that must not repeat:
+
+  ```python
+  @tool
+  def charge_card(tool_context: ToolContext, amount: float) -> str:
+      # Unsafe: re-runs (and re-charges) on every resume.
+      charge(amount)
+      approved = tool_context.interrupt("confirm_charge", reason={"amount": amount})
+      return "charged" if approved else "cancelled"
+
+
+  @tool
+  def charge_card(tool_context: ToolContext, amount: float) -> str:
+      # Safe: side effect happens only after the pause resolves.
+      approved = tool_context.interrupt("confirm_charge", reason={"amount": amount})
+      if not approved:
+          return "cancelled"
+      charge(amount)
+      return "charged"
+  ```
+
+> **Persistence:** interrupt state lives on the per-thread agent instance. The
+> in-memory per-thread cache only preserves it within a single process, so
+> pause and resume must hit the same process. For stateless / multi-container
+> HTTP deployments, wire a durable `SessionManager` via
+> `StrandsAgentConfig.session_manager_provider` so interrupt state round-trips
+> across processes. Keep interrupt payloads and tool results JSON-safe (no raw
+> `bytes`) when doing so: Strands' `SessionAgent.to_dict()` — unlike
+> `SessionMessage.to_dict()` — does not base64-encode `bytes` values, so a
+> `bytes`-bearing interrupt `reason`/`response`/resume `payload`, or a sibling
+> `ToolResult` in the same turn, raises `TypeError: Object of type bytes is
+> not JSON serializable` from `FileSessionManager`/`S3SessionManager` and
+> aborts the run.
+
 ## Supported AG-UI Events
 
 The integration supports the following AG-UI event families:

integrations/aws-strands/python/pyproject.toml

@@ -9,7 +9,7 @@ authors = [
 requires-python = ">=3.10, <3.15"
 dependencies = [
     "ag-ui-a2ui-toolkit>=0.0.4",
-    "ag-ui-protocol>=0.1.18",
+    "ag-ui-protocol>=0.1.19",
     "fastapi>=0.115.12",
     "strands-agents>=1.15.0",
 ]

integrations/aws-strands/python/src/ag_ui_strands/agent.py

@@ -66,6 +66,25 @@ def _extract_agent_kwargs(agent: StrandsAgentCore) -> dict:
 # outstanding frontend calls at once.
 _WIRE_MAP_MAX = 512
 
+# Sentinel handed back to a paused ``tool_context.interrupt()`` when the client
+# cancels (``ResumeEntry.status == "cancelled"``) rather than resolving. The
+# tool receives this in place of a real answer and can treat it as a denial.
+INTERRUPT_CANCELLED = {"cancelled": True}
+
+
+def _wrap_resume_response(status: str, payload: Any) -> dict:
+    """Package a ``ResumeEntry`` for Strands' ``interruptResponse`` shape.
+
+    Strands' resume gate is truthiness-based (i.e. ``if interrupt_.response:``),
+    so a raw falsy payload (``None``, ``False``, ``""``, ``0``, ``[]``, ``{}``)
+    re-raises the same interrupt and re-runs the tool body — an infinite approve loop.
+    Always hand Strands a truthy envelope; up to the tool implementation to properly
+    destructures it (e.g. via ``.get("cancelled")`` / ``.get("response")``).
+    """
+    if status == "cancelled":
+        return INTERRUPT_CANCELLED
+    return {"response": payload}
+
 
 def _get_strands_session_manager(agent: Any) -> Any:
     """Return the agent's Strands ``SessionManager``, or ``None``.
@@ -78,12 +97,70 @@ def _get_strands_session_manager(agent: Any) -> Any:
     )
 
 
+def _strands_interrupt_to_agui(strands_interrupt: Any) -> "Interrupt":
+    """Map a native Strands ``Interrupt`` onto an AG-UI ``Interrupt``.
+
+    Every Strands interrupt originates from ``tool_context.interrupt()`` or a
+    ``BeforeToolCallEvent`` hook, so its id always embeds the triggering
+    ``toolUseId`` (``v1:<kind>:<toolUseId>:<uuid>``) and is inherently
+    tool-call-bound. This maps onto AG-UI's reserved ``reason="tool_call"``
+    core value, with ``tool_call_id`` extracted from the id.
+
+    Strands' free-form ``name`` and ``reason`` are preserved verbatim under
+    ``metadata`` (``strands_name`` / ``strands_reason``) so no information is
+    lost on the wire; ``message`` additionally carries ``reason`` when it is a
+    plain string, since AG-UI clients render ``message`` directly.
+    """
+    s_id = getattr(strands_interrupt, "id", "")
+    name = getattr(strands_interrupt, "name", None) or "interrupt"
+    raw_reason = getattr(strands_interrupt, "reason", None)
+
+    tool_call_id = None
+    s_id_parts = s_id.split(":") if isinstance(s_id, str) else []
+    if len(s_id_parts) >= 4:
+        # toolUseId is freeform and can itself contain ":" — slice the parts
+        # list to drop only the "v1"/"<kind>" prefix and the trailing uuid.
+        tool_call_id = ":".join(s_id_parts[2:-1])
+
+    metadata = {"strands_name": name}
+    if raw_reason is not None:
+        metadata["strands_reason"] = raw_reason
+
+    return Interrupt(
+        id=s_id,
+        tool_call_id=tool_call_id,
+        reason="tool_call",
+        message=raw_reason if isinstance(raw_reason, str) else None,
+        metadata=metadata,
+    )
+
+
+def _extract_interrupts(agent: Any, terminal_result: Any) -> list:
+    """Return the native Strands interrupts for a paused run, or ``[]``.
+
+    Prefers the terminal ``AgentResult`` (``stop_reason == "interrupt"`` with a
+    populated ``interrupts``); falls back to the live agent's
+    ``_interrupt_state`` so a pause is still detected if the result event was
+    consumed by the stream's early-break path.
+    """
+    if terminal_result is not None:
+        if getattr(terminal_result, "stop_reason", None) == "interrupt":
+            interrupts = getattr(terminal_result, "interrupts", None) or []
+            if interrupts:
+                return list(interrupts)
+    interrupt_state = getattr(agent, "_interrupt_state", None)
+    if interrupt_state is not None and getattr(interrupt_state, "activated", False):
+        return list(getattr(interrupt_state, "interrupts", {}).values())
+    return []
+
+
 logger = logging.getLogger(__name__)
 from ag_ui.core import (
     AssistantMessage,
     CustomEvent,
     EventType,
     FunctionCall,
+    Interrupt,
     MessagesSnapshotEvent,
     ReasoningEncryptedValueEvent,
     ReasoningEndEvent,
@@ -94,6 +171,7 @@ def _get_strands_session_manager(agent: Any) -> Any:
     RunAgentInput,
     RunErrorEvent,
     RunFinishedEvent,
+    RunFinishedInterruptOutcome,
     RunStartedEvent,
     StateSnapshotEvent,
     StepFinishedEvent,
@@ -866,6 +944,10 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
             stop_text_streaming = False
             halt_event_stream = False
             pending_halt = False
+            # Terminal ``AgentResult`` from Strands (carried on the final
+            # ``{"result": ...}`` stream event). Used after the loop to detect a
+            # native interrupt pause (``stop_reason == "interrupt"``).
+            terminal_result = None
 
             # Reasoning/thinking state tracking
             reasoning_started = False
@@ -952,7 +1034,31 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
                 and self.config.replay_history_into_strands
                 and has_nonvoid_frontend_result
             )
-            if replay_history:
+
+            # A client answering to an interrupt sends its responses
+            # in ``RunAgentInput.resume`` (as per the AG-UI interrupt round-trip),
+            # not as a new user message. Translate those into the Strands resume
+            # prompt shape ``[{"interruptResponse": {"interruptId", "response"}}]``
+            # and drive the stream with it — this takes precedence over every
+            # other path, since a resume run carries no fresh prompt.
+            resume_prompt = None
+            resume_entries = getattr(input_data, "resume", None)
+            if isinstance(resume_entries, list) and resume_entries:
+                resume_prompt = [
+                    {
+                        "interruptResponse": {
+                            "interruptId": entry.interrupt_id,
+                            "response": _wrap_resume_response(
+                                entry.status, entry.payload
+                            ),
+                        }
+                    }
+                    for entry in resume_entries
+                ]
+
+            if resume_prompt is not None:
+                agent_stream = strands_agent.stream_async(resume_prompt)
+            elif replay_history:
                 native_history = _build_strands_history(input_data.messages)
                 # Apply ``state_context_builder`` to the last user-text
                 # message in the reconciled history rather than to the
@@ -1050,6 +1156,13 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
 
                     logger.debug(f"Received event: {event}")
 
+                    # Capture the terminal ``AgentResult`` (always emitted last
+                    # by ``stream_async``) so a native interrupt pause can be
+                    # detected after the loop. Recorded before the
+                    # ``complete``/``force_stop`` break so it is never dropped.
+                    if "result" in event and event["result"] is not None:
+                        terminal_result = event["result"]
+
                     # Skip lifecycle events
                     if event.get("init_event_loop") or event.get("start_event_loop"):
                         continue
@@ -1979,12 +2092,30 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[Any]:
                 snapshot=current_state,
             )
 
-            # Always finish the run - frontend handles keeping action executing
-            yield RunFinishedEvent(
-                type=EventType.RUN_FINISHED,
-                thread_id=input_data.thread_id,
-                run_id=input_data.run_id,
-            )
+            # If the run paused on a native Strands interrupt, surface it as an
+            # AG-UI interrupt outcome so the client can collect a response and
+            # resume via ``RunAgentInput.resume`` next turn. Otherwise finish
+            # bare, exactly as before (no behavior change for normal runs).
+            native_interrupts = _extract_interrupts(strands_agent, terminal_result)
+            if native_interrupts:
+                yield RunFinishedEvent(
+                    type=EventType.RUN_FINISHED,
+                    thread_id=input_data.thread_id,
+                    run_id=input_data.run_id,
+                    outcome=RunFinishedInterruptOutcome(
+                        type="interrupt",
+                        interrupts=[
+                            _strands_interrupt_to_agui(i) for i in native_interrupts
+                        ],
+                    ),
+                )
+            else:
+                # Always finish the run - frontend handles keeping action executing
+                yield RunFinishedEvent(
+                    type=EventType.RUN_FINISHED,
+                    thread_id=input_data.thread_id,
+                    run_id=input_data.run_id,
+                )
 
         except Exception as e:
             import traceback