diff --git a/integrations/aws-strands/python/README.md b/integrations/aws-strands/python/README.md index 18b3bcf82d..12bf5c17b7 100644 --- a/integrations/aws-strands/python/README.md +++ b/integrations/aws-strands/python/README.md @@ -83,6 +83,30 @@ 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. + +> **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. + ## Supported AG-UI Events The integration supports the following AG-UI event families: diff --git a/integrations/aws-strands/python/pyproject.toml b/integrations/aws-strands/python/pyproject.toml index 58c5499c4d..ca406b9a54 100644 --- a/integrations/aws-strands/python/pyproject.toml +++ b/integrations/aws-strands/python/pyproject.toml @@ -9,7 +9,7 @@ authors = [ requires-python = ">=3.12, <3.14" 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", ] diff --git a/integrations/aws-strands/python/src/ag_ui_strands/agent.py b/integrations/aws-strands/python/src/ag_ui_strands/agent.py index 53eb22d847..e7b192fecc 100644 --- a/integrations/aws-strands/python/src/ag_ui_strands/agent.py +++ b/integrations/aws-strands/python/src/ag_ui_strands/agent.py @@ -66,6 +66,11 @@ 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 _get_strands_session_manager(agent: Any) -> Any: """Return the agent's Strands ``SessionManager``, or ``None``. @@ -78,12 +83,51 @@ 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. + """ + raw_reason = getattr(strands_interrupt, "reason", None) + return Interrupt( + id=getattr(strands_interrupt, "id", ""), + reason=getattr(strands_interrupt, "name", None) or "interrupt", + message=raw_reason if isinstance(raw_reason, str) else None, + metadata=( + {"strands_reason": raw_reason} if raw_reason is not None else None + ), + ) + + +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 +138,7 @@ def _get_strands_session_manager(agent: Any) -> Any: RunAgentInput, RunErrorEvent, RunFinishedEvent, + RunFinishedInterruptOutcome, RunStartedEvent, StateSnapshotEvent, StepFinishedEvent, @@ -866,6 +911,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 +1001,33 @@ 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 a native Strands interrupt sends its responses + # in ``RunAgentInput.resume`` (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": ( + INTERRUPT_CANCELLED + if entry.status == "cancelled" + else 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 +1125,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 +2061,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 diff --git a/integrations/aws-strands/python/tests/test_interrupt.py b/integrations/aws-strands/python/tests/test_interrupt.py new file mode 100644 index 0000000000..eb2d88e5bb --- /dev/null +++ b/integrations/aws-strands/python/tests/test_interrupt.py @@ -0,0 +1,214 @@ +"""Tests for native Strands interrupt <-> AG-UI interrupt round-trip. + +Covers the four behaviors added to bridge ``tool_context.interrupt()`` to the +AG-UI interrupt lifecycle: + +1. A paused run finishes with ``RunFinishedInterruptOutcome``. +2. ``RunAgentInput.resume`` is translated into the Strands resume prompt shape. +3. ``status == "cancelled"`` resumes with the documented denial sentinel. +4. Runs that never interrupt finish bare (no behavior change). +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from strands.agent.state import AgentState +from strands.interrupt import Interrupt as StrandsInterrupt, _InterruptState + +from ag_ui.core import EventType, ResumeEntry, RunAgentInput +from ag_ui_strands.agent import INTERRUPT_CANCELLED, StrandsAgent +from ag_ui_strands.config import StrandsAgentConfig + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_run_input( + thread_id: str = "thread-1", + run_id: str = "run-1", + messages=None, + resume=None, +) -> RunAgentInput: + return RunAgentInput( + thread_id=thread_id, + run_id=run_id, + state={}, + messages=messages or [], + tools=[], + context=[], + forwarded_props={}, + resume=resume, + ) + + +async def _collect_events(agent: StrandsAgent, input_data: RunAgentInput) -> list: + events = [] + async for event in agent.run(input_data): + events.append(event) + return events + + +def _make_base_agent() -> StrandsAgent: + mock_core = MagicMock() + mock_core.model = MagicMock() + mock_core.system_prompt = "You are a test assistant." + mock_core.tool_registry = MagicMock() + mock_core.tool_registry.registry = {} + mock_core.record_direct_tool_call = True + # replay_history_into_strands defaults True; with no session manager this + # takes the in-memory replay path (stream_async(None)). Disable it so the + # legacy/resume paths are exercised straightforwardly in these unit tests. + config = StrandsAgentConfig(replay_history_into_strands=False) + return StrandsAgent(agent=mock_core, name="test_agent", config=config) + + +class _MockStrandsCore: + """A minimal stand-in for ``StrandsAgentCore`` driving the stream loop. + + ``stream_async`` records the prompt it was called with and yields the + provided terminal events. When ``interrupts`` are supplied it also flips its + ``_interrupt_state`` to activated, mirroring a paused native run. + """ + + def __init__(self, terminal_events=None, interrupts=None): + self.tool_registry = MagicMock() + self.tool_registry.registry = {} + self.state = AgentState() + self.model = MagicMock() + self.messages = [] + self.stream_prompts = [] + self._terminal_events = terminal_events or [] + self._interrupt_state = _InterruptState() + if interrupts: + for itr in interrupts: + self._interrupt_state.interrupts[itr.id] = itr + self._interrupt_state.activate() + + async def stream_async(self, prompt): + self.stream_prompts.append(prompt) + for event in self._terminal_events: + yield event + + +def _agent_result_with_interrupt(interrupts): + result = MagicMock() + result.stop_reason = "interrupt" + result.interrupts = interrupts + return result + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +class TestInterruptOutcome: + @pytest.mark.asyncio + 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"} + ) + core = _MockStrandsCore( + terminal_events=[{"result": _agent_result_with_interrupt([strands_interrupt])}], + interrupts=[strands_interrupt], + ) + agent = _make_base_agent() + + with patch("ag_ui_strands.agent.StrandsAgentCore", return_value=core): + events = await _collect_events(agent, _make_run_input()) + + finished = next(e for e in events if e.type == EventType.RUN_FINISHED) + assert finished.outcome is not None + assert finished.outcome.type == "interrupt" + 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"}} + + @pytest.mark.asyncio + async def test_detects_interrupt_from_state_when_result_missing(self): + """Falls back to _interrupt_state when no terminal result event arrives.""" + strands_interrupt = StrandsInterrupt(id="int-2", name="approve", reason=None) + # No {"result": ...} event this time — only the live interrupt state. + core = _MockStrandsCore(terminal_events=[], interrupts=[strands_interrupt]) + agent = _make_base_agent() + + with patch("ag_ui_strands.agent.StrandsAgentCore", return_value=core): + events = await _collect_events(agent, _make_run_input()) + + finished = next(e for e in events if e.type == EventType.RUN_FINISHED) + assert finished.outcome is not None + assert finished.outcome.type == "interrupt" + assert finished.outcome.interrupts[0].id == "int-2" + + @pytest.mark.asyncio + async def test_no_interrupt_finishes_bare(self): + """A normal run finishes with no outcome (back-compat, no behavior change).""" + result = MagicMock() + result.stop_reason = "end_turn" + result.interrupts = None + core = _MockStrandsCore(terminal_events=[{"result": result}]) + agent = _make_base_agent() + + with patch("ag_ui_strands.agent.StrandsAgentCore", return_value=core): + events = await _collect_events(agent, _make_run_input()) + + finished = next(e for e in events if e.type == EventType.RUN_FINISHED) + assert finished.outcome is None + + +class TestResumeConsumption: + @pytest.mark.asyncio + async def test_resolved_resume_builds_interrupt_response_prompt(self): + """A resolved ResumeEntry is translated into the Strands resume prompt.""" + core = _MockStrandsCore(terminal_events=[]) + agent = _make_base_agent() + resume = [ResumeEntry(interrupt_id="int-1", status="resolved", payload="yes")] + + with patch("ag_ui_strands.agent.StrandsAgentCore", return_value=core): + await _collect_events(agent, _make_run_input(resume=resume)) + + assert core.stream_prompts == [ + [{"interruptResponse": {"interruptId": "int-1", "response": "yes"}}] + ] + + @pytest.mark.asyncio + async def test_cancelled_resume_uses_sentinel(self): + """A cancelled ResumeEntry resumes with the denial sentinel as response.""" + core = _MockStrandsCore(terminal_events=[]) + agent = _make_base_agent() + resume = [ResumeEntry(interrupt_id="int-1", status="cancelled", payload=None)] + + with patch("ag_ui_strands.agent.StrandsAgentCore", return_value=core): + await _collect_events(agent, _make_run_input(resume=resume)) + + assert core.stream_prompts == [ + [{"interruptResponse": {"interruptId": "int-1", "response": INTERRUPT_CANCELLED}}] + ] + + @pytest.mark.asyncio + async def test_multiple_resume_entries(self): + """Every ResumeEntry becomes one interruptResponse content block.""" + core = _MockStrandsCore(terminal_events=[]) + agent = _make_base_agent() + resume = [ + ResumeEntry(interrupt_id="a", status="resolved", payload={"k": 1}), + ResumeEntry(interrupt_id="b", status="cancelled", payload=None), + ] + + with patch("ag_ui_strands.agent.StrandsAgentCore", return_value=core): + await _collect_events(agent, _make_run_input(resume=resume)) + + assert core.stream_prompts == [ + [ + {"interruptResponse": {"interruptId": "a", "response": {"k": 1}}}, + {"interruptResponse": {"interruptId": "b", "response": INTERRUPT_CANCELLED}}, + ] + ] diff --git a/integrations/aws-strands/python/uv.lock b/integrations/aws-strands/python/uv.lock index c6e03ff5d6..868d08e2d3 100644 --- a/integrations/aws-strands/python/uv.lock +++ b/integrations/aws-strands/python/uv.lock @@ -4,28 +4,28 @@ requires-python = ">=3.12, <3.14" [[package]] name = "ag-ui-a2ui-toolkit" -version = "0.0.3" +version = "0.0.4" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/16/b1/ea7ad7f0b3d1b20388d072ffbe4416577b4d4ab5471d45dfc04791a91602/ag_ui_a2ui_toolkit-0.0.3.tar.gz", hash = "sha256:468f25473ac00d098878da54c0069b7fa27dc63b4c1ff61315d4349a324c2fb7", size = 14785, upload-time = "2026-06-09T06:18:18.163Z" } +sdist = { url = "https://files.pythonhosted.org/packages/f0/ce/85f3960a83d962e5690bc0f27a3baf3bf1602edc2b0603085928c964ea14/ag_ui_a2ui_toolkit-0.0.4.tar.gz", hash = "sha256:172e2724e53df8173685a3fb896a6e5175eea06e1dc166c715db110ba4beba76", size = 18960, upload-time = "2026-06-17T13:34:28.695Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/c9/75/fc87bdf81bb1bf6d0fac09179e8bb17807d1bc5b3c0e8640f32e843b0857/ag_ui_a2ui_toolkit-0.0.3-py3-none-any.whl", hash = "sha256:e0354bd361c09f342fbe671cf870cbd19fdcb1b27e7a5bb2d8a392a4f00c2ba9", size = 16739, upload-time = "2026-06-09T06:18:17.316Z" }, + { url = "https://files.pythonhosted.org/packages/47/7a/acf85b01cd996bd011b71e181fd9f3daff5396fc3b7d78ba9445bfc08ecf/ag_ui_a2ui_toolkit-0.0.4-py3-none-any.whl", hash = "sha256:236fc511e1ec2399bcda0c14a109b3fb0a0c3e3988c18ef1918745b1c1535e30", size = 21315, upload-time = "2026-06-17T13:34:29.505Z" }, ] [[package]] name = "ag-ui-protocol" -version = "0.1.18" +version = "0.1.19" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "pydantic" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/4c/d7/5711eada86da9bd7684e58645653a1693ef20b66cc3efbb1deeafef80f8d/ag_ui_protocol-0.1.18.tar.gz", hash = "sha256:b37c672c3fd6bac12b316c39f45ad9db9f137bbb885489c79f268507029a22ff", size = 9937, upload-time = "2026-04-21T20:44:59.151Z" } +sdist = { url = "https://files.pythonhosted.org/packages/a7/10/4ad299267a7d04b89935aa99eef62979758fcf95aee9f8bb5d70c35b1be1/ag_ui_protocol-0.1.19.tar.gz", hash = "sha256:43c27f60d41712dcad0e9e0a203cbdf1c8e248b22417374c5c68321c448af4ea", size = 10720, upload-time = "2026-06-02T17:26:15.627Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/d8/74/913c9b8fc566c6da650aecbddf25a5d8186b54138df265eb9eb546f56141/ag_ui_protocol-0.1.18-py3-none-any.whl", hash = "sha256:d151c0f0a34160647f1571163f7185746f4326b15a56d1560de5082a7a0e7a12", size = 12607, upload-time = "2026-04-21T20:45:00.097Z" }, + { url = "https://files.pythonhosted.org/packages/4c/0a/bcad8116eb058e4b4a305e3fc37ebd7efc879deeb86b854f1c5b8b6e97dd/ag_ui_protocol-0.1.19-py3-none-any.whl", hash = "sha256:898843b1410d378824da0c6a776486288b9c5828689d0bf563118868e37f390f", size = 13490, upload-time = "2026-06-02T17:26:16.313Z" }, ] [[package]] name = "ag-ui-strands" -version = "0.2.0" +version = "0.2.3" source = { editable = "." } dependencies = [ { name = "ag-ui-a2ui-toolkit" }, @@ -41,8 +41,8 @@ dev = [ [package.metadata] requires-dist = [ - { name = "ag-ui-a2ui-toolkit", specifier = ">=0.0.3" }, - { name = "ag-ui-protocol", specifier = ">=0.1.18" }, + { name = "ag-ui-a2ui-toolkit", specifier = ">=0.0.4" }, + { name = "ag-ui-protocol", specifier = ">=0.1.19" }, { name = "fastapi", specifier = ">=0.115.12" }, { name = "strands-agents", specifier = ">=1.15.0" }, ]