fiqros-ag-ui-strands

Fiqros build of the AWS Strands integration for AG-UI.

This package exposes a lightweight wrapper that lets any strands.Agent speak the AG-UI protocol. It mirrors the developer experience of the other integrations: give us a Strands agent instance, plug it into StrandsAgent, and wire it to FastAPI via create_strands_app (or add_strands_fastapi_endpoint).

Install

pip install fiqros-ag-ui-strands

from ag_ui_strands import StrandsAgent, StrandsAgentConfig, create_strands_app

The import name is unchanged (ag_ui_strands), so this is a drop-in replacement for the upstream ag_ui_strands package — swap the dependency, change no code.

Install one or the other, never both. They install into the same ag_ui_strands import path, so having both in an environment leaves you with whichever was written last.

Replacing upstream in an existing project:

- ag_ui_strands>=0.3.0
+ fiqros-ag-ui-strands>=0.3.1

Why this fork exists

Upstream drops RunAgentInput.context before it reaches the model.

AG-UI context is how a frontend tells an agent things the user never typed — the signed-in user's display name, their timezone, what they were just looking at. Upstream copies those entries into strands_agent.state["agui_context"], a server-side dictionary that Strands never serializes into the prompt. The model therefore never sees them, and answers questions about the user with:

"I'm unable to determine your identity or what you were doing, as I don't have access to personal data or past interactions."

This build renders that context into the agent's system prompt on every run, so the model can actually read it. The behaviour is on by default and additive: context is still written to agent state, so tools that read agent.state.get("agui_context") keep working unchanged.

See Application context for the API, and docs/context-injection/README.md for the full root-cause write-up and QA verification plan.

Versioning against upstream

Versions track the upstream release this forks from, plus a patch number for changes Fiqros carries on top.

This package	Forked from upstream	Carries
0.3.1	ag_ui_strands 0.3.0	RunAgentInput.context → system prompt

Everything not listed above is upstream behaviour, unmodified.

Prerequisites

• Python 3.10+
• poetry (recommended) or pip
• A Strands-compatible model key (e.g., GOOGLE_API_KEY for Gemini)

Quick Start

The examples/server/__main__.py module mounts all demo routes behind a single FastAPI app. Run:

cd integrations/aws-strands/python/examples
poetry install
poetry run python -m server

It exposes:

Route	Description
/agentic-chat	Frontend tool demo
/backend-tool-rendering	Backend tool rendering demo
/shared-state	Shared recipe state
/agentic-generative-ui	Agentic UI with PredictState
/readonly-state-agent-context	Read-only AG-UI context + state

This is the easiest way to test multiple flows locally. Each route still follows the pattern described below (Strands agent → wrapper → FastAPI).

Architecture Overview

The integration has three main layers:

• StrandsAgent – wraps strands.Agent.stream_async. It translates Strands events into AG-UI events (text chunks, tool calls, PredictState, snapshots, reasoning/thinking, multi-agent steps, etc.).
• Configuration – StrandsAgentConfig + ToolBehavior + PredictStateMapping let you describe tool-specific quirks declaratively (skip message snapshots, emit state, stream args, send confirm actions, etc.).
• Transport helpers – create_strands_app and add_strands_fastapi_endpoint expose the agent via SSE. They are thin shells over the shared ag_ui.encoder.EventEncoder.

See ARCHITECTURE.md for diagrams and a deeper dive.

Key Files

File	Description
src/ag_ui_strands/agent.py	Core wrapper translating Strands streams into AG-UI events
src/ag_ui_strands/config.py	Config primitives (StrandsAgentConfig, ToolBehavior, PredictStateMapping)
src/ag_ui_strands/endpoint.py	FastAPI endpoint helper
src/ag_ui_strands/system_prompt.py	Renders RunAgentInput.context into the system prompt
examples/server/api/*.py	Ready-to-run demo apps

Application context

Fiqros build only. Upstream ag_ui_strands 0.3.0 does not inject context into the system prompt — see Why this fork exists.

RunAgentInput.context — the {description, value} entries a frontend sends to tell the agent things the user never typed (display name, timezone, what they were just looking at) — is rendered into the agent's system prompt on every run:

## Context from the application
- The currently logged-in user's display name: Atai
- The user's IANA timezone (used when mentioning times): America/Los_Angeles

This is on by default, and nothing needs configuring to get it. Context is also still written to strands_agent.state["agui_context"], so server-side tools can read it directly.

Two StrandsAgentConfig options control the rendering:

Option	Default	Description
inject_context_into_system_prompt	True	Render RunAgentInput.context into the system prompt. Set False for agents that read context exclusively from agent state via a tool, or that need the prompt kept byte-for-byte as written.
system_prompt_addendum_builder	None	Callable[[RunAgentInput], str | None] that fully replaces the default rendering. Overrides the flag above.

config = StrandsAgentConfig(
    inject_context_into_system_prompt=False,   # opt out entirely
)

The A2UI component catalog is excluded from this block — it already reaches the model through the generate_a2ui sub-agent prompt, and duplicating it would cost thousands of tokens per turn.

Shared state is not injected

Only context is rendered. RunAgentInput.state is deliberately left alone, because it already has an established path in this adapter: state_context_builder, which the shared_state example uses to format the recipe into the outgoing message itself. A second automatic path would double-send the payload for anyone using the first.

To put state in the system prompt, render it yourself with system_prompt_addendum_builder. It replaces the default context block, so call render_context_block inside it if you want both:

import json
from ag_ui_strands import StrandsAgentConfig, render_context_block

def build_addendum(input_data):
    blocks = [render_context_block(input_data.context)]
    if input_data.state:
        blocks.append("## Current state\n" + json.dumps(input_data.state, indent=2))
    return "\n\n".join(b for b in blocks if b)

config = StrandsAgentConfig(system_prompt_addendum_builder=build_addendum)

Background

Before this existed, context reached agent state only, which Strands never serializes into the prompt — so the model could not see it and answered "I don't have access to that". See docs/context-injection/README.md for the full root-cause write-up and a QA verification plan.

Amazon Bedrock AgentCore considerations

If you are planning to deploy your agent into Amazon Bedrock AgentCore (AC), please note that AC expects the following:

• The server is running on port 8080.
• The path /invocations - POST is implemented and can be used for interacting with the agent.
• The path /ping - GET is implemented and can be used for verifying that the agent is operational and ready to handle requests.

To implement the path mentioned above, you can use the helper function create_strands_app and pass the agent interaction path and the ping path as shown below:

    create_strands_app(agui_agent, "/invocations", "/ping")

You can also use the helper functions add_strands_fastapi_endpoint and add_ping for adding the mentioned paths to a FastAPI app that you are creating separately:

    add_strands_fastapi_endpoint(app, agent, "/invocations")
    add_ping(app, "/ping")

Requests to the AC endpoint must be authenticated. You can configure your agent runtime to accept JWT bearer tokens (via Amazon Cognito) or use SigV4. See Set up authentication in the AgentCore documentation.

For details on how AgentCore handles AG-UI requests, event streaming, and error formatting, see the AG-UI protocol contract.

To deploy, use the AgentCore Starter Toolkit:

pip install bedrock-agentcore-starter-toolkit
agentcore configure -e my_agui_server.py --protocol AGUI
agentcore deploy

For the complete deployment walkthrough, see Deploy AG-UI servers in AgentCore Runtime.

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. Generic native interrupts preserve the Strands name as the AG-UI reason and the free-form Strands reason under metadata.reason. Tools configured with ToolBehavior(interrupt_on_call=True) instead emit a tool_call approval interrupt with an approved response schema. Applies to server-executed tools only. For client-provided tools, gate execution in the client — define the tool with a render that calls respond, not a handler — since the tool runs in the browser and the adapter has already halted the run.

• To resume, the client sends the next RunAgentInput on the same thread_id with resume=[ResumeEntry(interrupt_id=..., status="resolved", payload=...)]. Strands' resume gate is truthiness-based (if interrupt_.response:), so a falsy payload (None, False, "", 0, [], {}) would otherwise re-raise the same interrupt and re-run the tool body forever. To prevent that, interrupt() does not return payload directly — it returns a truthy envelope: {"response": payload} on resolve, {"cancelled": True} on cancel. Destructure it with .get("response") / .get("cancelled"). Adapter-managed interrupt_on_call approvals are the exception: their {"approved": bool} payload is passed through directly.

• For generic native interrupts, status="cancelled" resumes the tool with the sentinel {"cancelled": True} (ag_ui_strands.INTERRUPT_CANCELLED) so it can treat the pause as a denial. An adapter-managed approval receives {"approved": False} instead.

• 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:

  @tool(context=True)
  def charge_card(tool_context: ToolContext, amount: float) -> str:
      # Unsafe: re-runs (and re-charges) on every resume.
      charge(amount)
      envelope = tool_context.interrupt("confirm_charge", reason={"amount": amount})
      return "cancelled" if envelope.get("cancelled") or not envelope.get("response") else "charged"
  
  
  @tool(context=True)
  def charge_card(tool_context: ToolContext, amount: float) -> str:
      # Safe: side effect happens only after the pause resolves.
      envelope = tool_context.interrupt("confirm_charge", reason={"amount": amount})
      if envelope.get("cancelled") or not envelope.get("response"):
          return "cancelled"
      charge(amount)
      return "charged"
  

Persistence and proxy-tool boundaries

Scenario	Support boundary
Native-only pause and resume on the same live wrapper, process, and thread_id	Supported without a SessionManager; the cached per-thread Strands agent is the checkpoint.
Wrapper recreation or cross-process resume	Requires a compatible durable SessionManager that restores the same session and stable Strands agent_id.
Frontend proxy and native interrupt in the same checkpoint	Requires session_id plus session_repository.list_messages() and session_repository.update_message(). Without a manager the run emits INTERRUPT_SESSION_REQUIRED; without those capabilities it emits INTERRUPT_SESSION_CAPABILITY_ERROR. The checkpoint is not advertised or consumed.

Submitted resume batches are validated atomically before streaming or reconciliation. They must contain at least one unique, non-blank, currently open interrupt id, and every open interrupt must be addressed in the batch. Malformed or unopened entries emit INTERRUPT_RESUME_ERROR; incomplete batches emit PARTIAL_RESUME. These failures leave the checkpoint retryable. If reconciliation fails while an interrupt checkpoint is active, the run emits INTERRUPT_RECONCILIATION_ERROR without finishing or consuming the checkpoint.

When using a SessionManager, keep interrupt payloads and tool results JSON-safe (no raw bytes): 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:

• Lifecycle: RUN_STARTED, RUN_FINISHED, RUN_ERROR
• Text streaming: TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_END
• Reasoning: REASONING_* events for models with extended thinking
• Tool calls: TOOL_CALL_START, TOOL_CALL_ARGS, TOOL_CALL_END, TOOL_CALL_RESULT
• State management: STATE_SNAPSHOT
• Multi-agent: STEP_STARTED, STEP_FINISHED, and MultiAgentHandoff custom events
• Generative UI: PredictState custom events for optimistic UI updates
• Multimodal: Image, document, and video content in user messages (converted to Strands ContentBlock format)

Development

git clone https://github.com/MalaikaAbb/ag-ui
cd ag-ui/integrations/aws-strands/python
uv sync --all-extras
uv run pytest -q

Publishing a new release is documented in PUBLISHING.md.

Next Steps

• Port the context-injection fix to the TypeScript adapter, which has the same gap.
• Upstream the fix so this fork can be retired.
• Add an event queue layer (like the ADK middleware) for resumable streams and non-HTTP transports.
• Expand the test suite as new behaviors land.

Relationship to upstream

This is an unofficial, independently maintained build. It is not published or endorsed by the AG-UI project or by AWS. Bugs in this build belong in the fiqros issue tracker; bugs in the protocol or in unmodified adapter behaviour belong upstream.

License

MIT, inherited from the upstream AG-UI project. See LICENSE. The original copyright notice is retained unchanged; Fiqros claims no additional rights over the upstream code.