fix: adjust AG-UI interrupt to match specs

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

@@ -100,19 +100,38 @@ 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``.
 
-    Strands' ``reason`` is free-form (any JSON), whereas AG-UI's ``reason`` is a
-    categorical string. The interrupt *name* is the closest categorical fit, so
-    it becomes ``reason``; the original reason object is preserved verbatim under
-    ``metadata`` so no information is lost on the wire.
+    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=getattr(strands_interrupt, "id", ""),
-        reason=getattr(strands_interrupt, "name", None) or "interrupt",
+        id=s_id,
+        tool_call_id=tool_call_id,
+        reason="tool_call",
         message=raw_reason if isinstance(raw_reason, str) else None,
-        metadata=(
-            {"strands_reason": raw_reason} if raw_reason is not None else None
-        ),
+        metadata=metadata,
     )
 
 

integrations/aws-strands/python/tests/test_interrupt.py

@@ -116,7 +116,9 @@ class TestInterruptOutcome:
     async def test_pause_emits_interrupt_outcome(self):
         """A native interrupt produces RUN_FINISHED with an interrupt outcome."""
         strands_interrupt = StrandsInterrupt(
-            id="int-1", name="confirm", reason={"summary": "delete all"}
+            id="v1:tool_call:tu-1:00000000-0000-0000-0000-000000000000",
+            name="confirm",
+            reason={"summary": "delete all"},
         )
         core = _MockStrandsCore(
             terminal_events=[{"result": _agent_result_with_interrupt([strands_interrupt])}],
@@ -133,11 +135,15 @@ async def test_pause_emits_interrupt_outcome(self):
         assert len(finished.outcome.interrupts) == 1
 
         agui_interrupt = finished.outcome.interrupts[0]
-        assert agui_interrupt.id == "int-1"
-        # Strands interrupt *name* maps to the categorical AG-UI reason.
-        assert agui_interrupt.reason == "confirm"
-        # The free-form Strands reason object is preserved under metadata.
-        assert agui_interrupt.metadata == {"strands_reason": {"summary": "delete all"}}
+        assert agui_interrupt.id == "v1:tool_call:tu-1:00000000-0000-0000-0000-000000000000"
+        # Every Strands interrupt is tool-call-bound; id embeds the toolUseId.
+        assert agui_interrupt.tool_call_id == "tu-1"
+        assert agui_interrupt.reason == "tool_call"
+        # The free-form Strands name/reason are preserved under metadata.
+        assert agui_interrupt.metadata == {
+            "strands_name": "confirm",
+            "strands_reason": {"summary": "delete all"},
+        }
 
     @pytest.mark.asyncio
     async def test_no_interrupt_finishes_bare(self):