AWS Strands Overview

AWS Strands is an open-source Python agent SDK from AWS. Its AG-UI bridge, ag-ui-strands, turns a Strands Agent into an AG-UI event stream, which is all @threadplane/ag-ui needs in order to bind it to <chat>. The running example is a meeting scheduler that looks up open slots, pauses for human approval before it books anything, and delegates research to a specialist, and this page walks the files that make it work. The Angular component it renders is the same UI code the LangGraph-backed examples use; only the provider and the backend behind it differ.

What the demo does

The Run tab shows the prebuilt <chat> composition in front of a Strands agent served over AG-UI, with a side panel titled "Shared state — schedule" to the right of the transcript. Two welcome suggestions set it up: "Book the Q3 roadmap review" asks for a Tuesday meeting with the platform team, and "Book a design critique" runs the same two steps for the web team on Thursday.

The agent calls check_availability for the requested weekday, and that day and its open slots appear in the side panel. Then it calls book_meeting, the run stops, and a modal card titled "Booking approval required" shows the topic and the chosen slot above two buttons, Cancel and Approve. Approve resumes the run and the agent confirms the booking in one sentence; Cancel resumes it with a rejection and nothing is booked. Either way the panel keeps showing the booking as it was snapshotted before the pause; the final booked or declined status lives only in the agent's confirmation sentence, because no hook emits state after the interrupt resolves.

Delegation is the third thing to try. Type a research request instead, for example "Find a slot for Ada and Grace next week — research their availability first", and the agent calls research_availability. That call renders as a subagent card carrying the specialist's own answer, streamed token by token inside the card rather than into the parent bubble.

How it is built

Four files carry the example: a Python module holding the Strands agent and its tools, a FastAPI server that mounts it, an application config that registers the agent, and a component that renders the state panel and the approval card. Open the Code tab to read them in place. subagent_emitter.py, a fifth file in the same backend directory and not shown in the Code tab, holds the translation from the specialist's stream into the protocol's subagent events.

An ordinary backend tool

check_availability is a plain Strands @tool. It executes server-side and never pauses, so on the wire it is a tool call with a result and nothing else. The function under it is the state hook that mirrors the lookup into shared state, registered against the tool further down.

@tool
def check_availability(day: str) -> dict:
    """Look up the open meeting slots for a weekday.
 
    Args:
        day: Weekday name, e.g. 'Tuesday'.
 
    Returns:
        A dict with the day and its open slots.
    """
    slots = _SLOTS.get(day.strip().lower(), [])
    return {"day": day, "slots": slots}
 
 
async def availability_state(context) -> dict | None:
    """state_from_result hook: mirror the availability lookup into state."""
    result = context.result_data
    if isinstance(result, str):
        try:
            result = json.loads(result)
        except ValueError:
            return None
    if not isinstance(result, dict):
        return None
    _state["availability"] = {"day": result.get("day"), "slots": result.get("slots", [])}
    return _complete_state()

The hook reads the tool result from context.result_data and tolerates the string form, because a result that has round-tripped through JSON arrives as text.

Pausing for a human decision

book_meeting is a context tool: Strands hands it a ToolContext, and tool_context.interrupt(...) parks the tool mid-execution. The first argument is the name the client sees on the pending interrupt, and reason is the payload the approval card renders.

@tool(context=True)
def book_meeting(topic: str, slot: str, tool_context: ToolContext) -> str:
    """Book a meeting after a human approves it.
 
    Args:
        topic: Short description of the meeting purpose.
        slot: The chosen slot, e.g. 'Tuesday 10:00'.
 
    Returns:
        A confirmation (or rejection) sentence.
    """
    answer = tool_context.interrupt(
        "book_meeting",
        reason={"topic": topic, "slot": slot},
    )
    payload = answer.get("response") or {}
    approved = bool(payload.get("approved")) and not (
        answer.get("cancelled") or payload.get("cancelled")
    )
    _state["booking"] = {"topic": topic, "slot": slot, "status": "booked" if approved else "declined"}
    if not approved:
        return f"The human declined. Meeting NOT booked: {topic}"
    return f"Meeting booked for {slot}: {topic}"

When the human answers, the call returns the decision and the rest of the function runs to completion in the resumed run.

Delegating to a specialist

research_availability is an async-generator tool. It re-yields every event the specialist emits, keeps the text deltas as it passes them along, and yields the joined text last, because Strands takes the last yielded value as the tool result. Each yielded value crosses the bridge as a tool_stream_event, which is the seam the subagent emitter listens on.

@tool
async def research_availability(attendees: str, date_range: str):
    """Delegate availability research for the given attendees to a specialist.
 
    Args:
        attendees: Comma-separated attendee names, e.g. 'Ada, Grace'.
        date_range: The window to research, e.g. 'next week'.
 
    Returns:
        The specialist's bullet summary of likely availability windows.
    """
    chunks: list[str] = []
    try:
        async for event in availability_researcher.stream_async(
            f"Attendees: {attendees}\nDate range: {date_range}"
        ):
            if isinstance(event, dict) and isinstance(event.get("data"), str):
                chunks.append(event["data"])
            yield event
    except Exception as exc:  # pragma: no cover - not reachable without a live model failure
        # Surface the failure to the emitter (which owns the SUBAGENT_ERROR
        # wire event), then let the tool error propagate to Strands normally.
        yield {"delegation_error": str(exc)}
        raise
    # Strands takes the LAST yielded value as the tool result.
    yield "".join(chunks)

The specialist itself is an ordinary Strands Agent with its own system prompt and no tools of its own.

# Tool-less specialist the orchestrator delegates availability research to
# via the `research_availability` async-generator tool above. Its streamed
# events cross the bridge as tool_stream_events and are translated into
# SUBAGENT_* wire events by the emitter registered in ToolBehavior below.
availability_researcher = Agent(
    model=build_model(),
    system_prompt=_RESEARCHER_INSTRUCTIONS,
    name="availability_researcher",
    tools=[],
)

Registering the per-tool behaviors

Everything that makes this example more than streamed text is registered in one place. StrandsAgentConfig.tool_behaviors maps a tool name to a ToolBehavior: state_from_result for the availability lookup, state_from_args for the booking, and tool_stream_event_handler for the delegation tool. Registering a stream handler for a tool gives that handler the whole child stream.

agent = StrandsAgent(
    agent=Agent(
        model=build_model(),
        system_prompt=_INSTRUCTIONS,
        tools=[check_availability, book_meeting, research_availability],
    ),
    name="aws-strands",
    description="Books meetings with availability lookup, shared state, and human approval.",
    config=StrandsAgentConfig(
        tool_behaviors={
            "check_availability": ToolBehavior(state_from_result=availability_state),
            "book_meeting": ToolBehavior(state_from_args=booking_state),
            "research_availability": ToolBehavior(
                tool_stream_event_handler=emit_subagent_events,
            ),
        },
    ),
)

The orchestrator binds all three tools and the StrandsAgent wrapper is what the server mounts.

Serving the agent over AG-UI

The backend is a FastAPI application. add_strands_fastapi_endpoint from the ag-ui-strands package mounts the wrapped agent at a path that speaks the AG-UI event stream.

from fastapi import FastAPI
from ag_ui_strands import add_strands_fastapi_endpoint
 
from .agent import agent
 
app = FastAPI(title="cockpit-runtimes-aws-strands")
add_strands_fastapi_endpoint(app, agent, "/agent")
 
 
@app.get("/ok")
def ok() -> dict:
    return {"ok": True}

Providing the agent

provideAgent() registers the agent once for the whole application, and it is the only provider the <chat> composition requires. The example passes a factory because it resolves its endpoint at runtime from the host that serves the demo. Your own application does not need the factory.

import { injectCockpitRuntimeConnection } from '@threadplane/cockpit-telemetry';
import { ApplicationConfig } from '@angular/core';
import { provideAgent } from '@threadplane/ag-ui';
 
export const appConfig: ApplicationConfig = {
  providers: [
    provideAgent(() => {
      const connection = injectCockpitRuntimeConnection();
      if (connection.adapter !== 'ag-ui') {
        throw new Error('incompatible runtime');
      }
      return {
        url: connection.url,
      };
    }),
  ],
};

Pass the URL of your AG-UI endpoint directly:

provideAgent({
  url: 'https://your-backend.example.com/agent',
});

Reading shared state in the component

injectAgent() returns the adapter's agent, and agent.state() is the Signal the snapshots land on. The component narrows that object into two computed Signals, treating a half-filled entry as absent so the panel does not render an empty row.

/** Shared state snapshotted from the backend (STATE_SNAPSHOT only). */
private readonly sharedState = computed(() => {
  // Example apps compile lib source with strict:false — cast at the read site.
  return this.agent.state() as { availability?: Availability; booking?: Booking } | undefined;
});
 
protected readonly availability = computed(() => {
  const a = this.sharedState()?.availability;
  return a && a.day !== undefined ? a : undefined;
});
 
protected readonly booking = computed(() => {
  const b = this.sharedState()?.booking;
  return b && b.topic !== undefined ? b : undefined;
});

The panel itself is plain Angular template code reading those two Signals, with a placeholder line for the empty case.

The approval card

<chat-approval-card> opens as a modal whenever the agent has a pending interrupt, and the #body template names what is being approved. This one renders the topic and the slot.

<chat-approval-card
  [agent]="agent"
  title="Booking approval required"
  (action)="onAction($event)"
>
  <ng-template #body>
    <div class="approval-body">
      @if (approvalBooking(); as b) {
        <div class="approval-row">
          <span class="approval-label">Topic</span>
          <strong>{{ b.topic }}</strong>
        </div>
        <div class="approval-row">
          <span class="approval-label">Slot</span>
          <strong>{{ b.slot }}</strong>
        </div>
      } @else {
        <p>The agent requests approval for <code class="approval-code">{{ approvalToolName() }}</code>.</p>
      }
    </div>
  </ng-template>
</chat-approval-card>

The class supplies that body from the pending interrupt and maps the card's two buttons onto resume payloads. The adapter stores the interrupt outcome as { interrupts: [...], runId }; each Strands entry carries the tool name under reason and the tool's own payload under metadata.reason.

/**
 * The pending approval request from the protocol-standard interrupt
 * outcome. The reducer stores it as `{ interrupts: [...], runId }`; each
 * Strands entry carries the tool name under `reason` and the tool's
 * interrupt payload under `metadata.reason`.
 */
private readonly approvalEntry = computed(() => {
  const value = this.agent.interrupt?.()?.value as { interrupts?: unknown[] } | undefined;
  return value?.interrupts?.[0] as
    | { reason?: string; metadata?: { reason?: { topic?: string; slot?: string } } }
    | undefined;
});
 
protected readonly approvalToolName = computed(() => this.approvalEntry()?.reason ?? 'a tool call');
 
protected readonly approvalBooking = computed(() => {
  const pending = this.approvalEntry()?.metadata?.reason;
  return pending?.topic !== undefined ? pending : this.booking();
});
 
protected send(text: string): void {
  void this.agent.submit({ message: text });
}
 
protected onAction(action: ChatApprovalAction): void {
  if (action === 'approve') {
    void this.agent.submit({ resume: { approved: true } });
  } else if (action === 'cancel') {
    void this.agent.submit({ resume: { approved: false } });
  }
}

submit({ resume }) is the only interrupt-specific call in the component, and it is the same call every other AG-UI example makes.

What the integration demonstrates

Shared state is partial, and the reason matters

The Strands bridge never emits STATE_DELTA. Outbound state exists only where a tool opts in through a per-tool ToolBehavior hook, which is why the example registers state_from_result on check_availability and state_from_args on book_meeting and gets nothing from research_availability.

Because the adapter applies a STATE_SNAPSHOT as a full replacement, every hook has to return the complete state object. A hook that returns only the keys it changed clobbers its siblings, so the example keeps one module-level object and composes the whole thing on every emission.

# Per-process demo state. The Strands bridge is SNAPSHOT-only: every
# outbound state emission replaces the whole frontend state object, so each
# ToolBehavior hook below composes and returns this COMPLETE object rather
# than just the key it changed — a partial return would wipe the sibling
# key. (Module-level state keeps the demo honest and simple; a real
# deployment would key this per thread.)
_state: dict = {"availability": None, "booking": None}
 
 
def _complete_state() -> dict:
    return {"availability": _state["availability"], "booking": _state["booking"]}

The booking hook shows what that costs in practice. It fires on the tool-call arguments, before the interrupt pauses the run, so the approval card can render a pending booking from shared state — and it still has to return both keys, not just the one it touched.

async def booking_state(context) -> dict | None:
    """state_from_args hook: mirror the pending booking into state as the
    tool-call arguments finish streaming (before the interrupt pauses the
    run), so the approval UI can render from shared state."""
    tool_input = context.tool_input
    if isinstance(tool_input, str):
        try:
            tool_input = json.loads(tool_input)
        except ValueError:
            return None
    if not isinstance(tool_input, dict):
        return None
    _state["booking"] = {
        "topic": tool_input.get("topic"),
        "slot": tool_input.get("slot"),
        "status": "pending",
    }
    return _complete_state()

Shared state does work on Strands. It is snapshot-only, it is opt-in per tool, and it puts the burden of assembling the whole object on each hook. That is a real constraint to design around, not a rounding error, which is why the measured matrix records it as partial rather than green.

How subagents surface

Strands wraps every value an async-generator tool yields as a tool_stream_event, and the bridge dispatches those events to a per-tool ToolBehavior.tool_stream_event_handler. Natively the bridge forwards only the inner tool-call lifecycle, so a delegated run reaches the browser as one opaque result string with no child text at all.

The handler registered on research_availability closes that gap. subagent_emitter.py translates the specialist's stream into SUBAGENT_STARTED, attributed TEXT_MESSAGE_* deltas carrying subagentRunId, and SUBAGENT_FINISHED, deriving every identifier from the tool call identifier the bridge already put on the wire. The adapter routes those attributed events into a subagent entry instead of the parent transcript, which is why the card streams the specialist's tokens live.

The wire capture behind that matrix cell is committed beside the backend at docs/wire-capture-subagents.md, before and after the emitter. Multi-agent routes crash the stale published wheel, which is one reason the example pins the bridge to a git reference instead.

Model access

Strands' native OpenAI provider is used on a plain OPENAI_API_KEY. No AWS credentials are involved anywhere in this example, despite the runtime's name. OPENAI_BASE_URL is honored when it is set, which is how the end-to-end harness replays recorded model calls against this backend.

What's Next