data: {"type": "RUN_STARTED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "TEXT_MESSAGE_START", "messageId": "msg-abc", "role": "assistant"}
data: {"type": "TEXT_MESSAGE_CONTENT", "messageId": "msg-abc", "delta": "Hello!"}
data: {"type": "TEXT_MESSAGE_END", "messageId": "msg-abc"}
data: {"type": "RUN_FINISHED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "RUN_STARTED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "TEXT_MESSAGE_START", "messageId": "msg-abc", "role": "assistant"}
data: {"type": "TEXT_MESSAGE_CONTENT", "messageId": "msg-abc", "delta": "Hello!"}
data: {"type": "TEXT_MESSAGE_END", "messageId": "msg-abc"}
data: {"type": "RUN_FINISHED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "RUN_STARTED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "TEXT_MESSAGE_START", "messageId": "msg-abc", "role": "assistant"}
data: {"type": "TEXT_MESSAGE_CONTENT", "messageId": "msg-abc", "delta": "Hello!"}
data: {"type": "TEXT_MESSAGE_END", "messageId": "msg-abc"}
data: {"type": "RUN_FINISHED", "threadId": "test-123", "runId": "run-456"}
Your agent framework [Strands · LangGraph · CrewAI]
│
│ deployed to
▼
Amazon Bedrock AgentCore Runtime [Bearer token via Cognito OAuth 2.0]
│ auth · session isolation · auto-scaling · observability
│
│ invokes container
▼
Agent container [ARM64 · port 8080]
│ FastAPI · /invocations · /ping
│
│ calls model
▼
Amazon Bedrock model
│
│ streams back via
▼
AG-UI protocol [SSE event stream]
│
│ consumed by
▼
CopilotKit [HttpAgent · @ag-ui/client]
│
├─ renders streaming responses + tool progress
├─ syncs state updates in real time
└─ forwards user messages → agent via AG-UI
▼
Your application UI
Your agent framework [Strands · LangGraph · CrewAI]
│
│ deployed to
▼
Amazon Bedrock AgentCore Runtime [Bearer token via Cognito OAuth 2.0]
│ auth · session isolation · auto-scaling · observability
│
│ invokes container
▼
Agent container [ARM64 · port 8080]
│ FastAPI · /invocations · /ping
│
│ calls model
▼
Amazon Bedrock model
│
│ streams back via
▼
AG-UI protocol [SSE event stream]
│
│ consumed by
▼
CopilotKit [HttpAgent · @ag-ui/client]
│
├─ renders streaming responses + tool progress
├─ syncs state updates in real time
└─ forwards user messages → agent via AG-UI
▼
Your application UI
Your agent framework [Strands · LangGraph · CrewAI]
│
│ deployed to
▼
Amazon Bedrock AgentCore Runtime [Bearer token via Cognito OAuth 2.0]
│ auth · session isolation · auto-scaling · observability
│
│ invokes container
▼
Agent container [ARM64 · port 8080]
│ FastAPI · /invocations · /ping
│
│ calls model
▼
Amazon Bedrock model
│
│ streams back via
▼
AG-UI protocol [SSE event stream]
│
│ consumed by
▼
CopilotKit [HttpAgent · @ag-ui/client]
│
├─ renders streaming responses + tool progress
├─ syncs state updates in real time
└─ forwards user messages → agent via AG-UI
▼
Your application UI
pip install fastapi uvicorn ag_ui_strands strands-agents boto3
pip install fastapi uvicorn ag_ui_strands strands-agents boto3
pip install fastapi uvicorn ag_ui_strands strands-agents boto3
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, JSONResponse
from ag_ui_strands import StrandsAgent
from ag_ui.core import RunAgentInput
from ag_ui.encoder import EventEncoder
from strands import Agent
from strands.models.bedrock import BedrockModel model = BedrockModel( model_id="us.anthropic.claude-3-5-sonnet-20241022-v2:0", region_name="us-west-2",
) strands_agent = Agent( model=model, system_prompt="You are a helpful assistant.",
) agui_agent = StrandsAgent( agent=strands_agent, name="my_agent", description="A helpful assistant",
) app = FastAPI() @app.post("/invocations")
async def invocations(input_data: dict, request: Request): accept_header = request.headers.get("accept") encoder = EventEncoder(accept=accept_header) async def event_generator(): run_input = RunAgentInput(**input_data) async for event in agui_agent.run(run_input): yield encoder.encode(event) return StreamingResponse( event_generator(), media_type=encoder.get_content_type() ) @app.get("/ping")
async def ping(): return JSONResponse({"status": "Healthy"}) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8080)
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, JSONResponse
from ag_ui_strands import StrandsAgent
from ag_ui.core import RunAgentInput
from ag_ui.encoder import EventEncoder
from strands import Agent
from strands.models.bedrock import BedrockModel model = BedrockModel( model_id="us.anthropic.claude-3-5-sonnet-20241022-v2:0", region_name="us-west-2",
) strands_agent = Agent( model=model, system_prompt="You are a helpful assistant.",
) agui_agent = StrandsAgent( agent=strands_agent, name="my_agent", description="A helpful assistant",
) app = FastAPI() @app.post("/invocations")
async def invocations(input_data: dict, request: Request): accept_header = request.headers.get("accept") encoder = EventEncoder(accept=accept_header) async def event_generator(): run_input = RunAgentInput(**input_data) async for event in agui_agent.run(run_input): yield encoder.encode(event) return StreamingResponse( event_generator(), media_type=encoder.get_content_type() ) @app.get("/ping")
async def ping(): return JSONResponse({"status": "Healthy"}) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8080)
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, JSONResponse
from ag_ui_strands import StrandsAgent
from ag_ui.core import RunAgentInput
from ag_ui.encoder import EventEncoder
from strands import Agent
from strands.models.bedrock import BedrockModel model = BedrockModel( model_id="us.anthropic.claude-3-5-sonnet-20241022-v2:0", region_name="us-west-2",
) strands_agent = Agent( model=model, system_prompt="You are a helpful assistant.",
) agui_agent = StrandsAgent( agent=strands_agent, name="my_agent", description="A helpful assistant",
) app = FastAPI() @app.post("/invocations")
async def invocations(input_data: dict, request: Request): accept_header = request.headers.get("accept") encoder = EventEncoder(accept=accept_header) async def event_generator(): run_input = RunAgentInput(**input_data) async for event in agui_agent.run(run_input): yield encoder.encode(event) return StreamingResponse( event_generator(), media_type=encoder.get_content_type() ) @app.get("/ping")
async def ping(): return JSONResponse({"status": "Healthy"}) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8080)
python my_agui_server.py
python my_agui_server.py
python my_agui_server.py
curl -N -X POST http://localhost:8080/invocations \ -H "Content-Type: application/json" \ -d '{ "thread_id": "test-123", "run_id": "run-456", "state": {}, "messages": [{"role": "user", "content": "Hello!", "id": "msg-1"}], "tools": [], "context": [], "forwardedProps": {} }'
curl -N -X POST http://localhost:8080/invocations \ -H "Content-Type: application/json" \ -d '{ "thread_id": "test-123", "run_id": "run-456", "state": {}, "messages": [{"role": "user", "content": "Hello!", "id": "msg-1"}], "tools": [], "context": [], "forwardedProps": {} }'
curl -N -X POST http://localhost:8080/invocations \ -H "Content-Type: application/json" \ -d '{ "thread_id": "test-123", "run_id": "run-456", "state": {}, "messages": [{"role": "user", "content": "Hello!", "id": "msg-1"}], "tools": [], "context": [], "forwardedProps": {} }'
data: {"type": "RUN_STARTED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "TEXT_MESSAGE_START", "messageId": "msg-abc", "role": "assistant"}
data: {"type": "TEXT_MESSAGE_CONTENT", "messageId": "msg-abc", "delta": "Hello!"}
data: {"type": "TEXT_MESSAGE_END", "messageId": "msg-abc"}
data: {"type": "RUN_FINISHED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "RUN_STARTED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "TEXT_MESSAGE_START", "messageId": "msg-abc", "role": "assistant"}
data: {"type": "TEXT_MESSAGE_CONTENT", "messageId": "msg-abc", "delta": "Hello!"}
data: {"type": "TEXT_MESSAGE_END", "messageId": "msg-abc"}
data: {"type": "RUN_FINISHED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "RUN_STARTED", "threadId": "test-123", "runId": "run-456"}
data: {"type": "TEXT_MESSAGE_START", "messageId": "msg-abc", "role": "assistant"}
data: {"type": "TEXT_MESSAGE_CONTENT", "messageId": "msg-abc", "delta": "Hello!"}
data: {"type": "TEXT_MESSAGE_END", "messageId": "msg-abc"}
data: {"type": "RUN_FINISHED", "threadId": "test-123", "runId": "run-456"}
pip install bedrock-agentcore-starter-toolkit
pip install bedrock-agentcore-starter-toolkit
pip install bedrock-agentcore-starter-toolkit
fastapi uvicorn ag_ui_strands
fastapi uvicorn ag_ui_strands
fastapi uvicorn ag_ui_strands
# Configure with AG-UI protocol + OAuth authentication
agentcore configure -e my_agui_server.py --protocol AGUI # Deploy to AWS
agentcore deploy
# Configure with AG-UI protocol + OAuth authentication
agentcore configure -e my_agui_server.py --protocol AGUI # Deploy to AWS
agentcore deploy
# Configure with AG-UI protocol + OAuth authentication
agentcore configure -e my_agui_server.py --protocol AGUI # Deploy to AWS
agentcore deploy
arn:aws:bedrock-agentcore:us-west-2:ACCOUNT_ID:runtime/my_agui_server-xyz123
arn:aws:bedrock-agentcore:us-west-2:ACCOUNT_ID:runtime/my_agui_server-xyz123
arn:aws:bedrock-agentcore:us-west-2:ACCOUNT_ID:runtime/my_agui_server-xyz123
export BEARER_TOKEN="<your-cognito-bearer-token>"
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:ACCOUNT_ID:runtime/my_agui_server-xyz123"
export BEARER_TOKEN="<your-cognito-bearer-token>"
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:ACCOUNT_ID:runtime/my_agui_server-xyz123"
export BEARER_TOKEN="<your-cognito-bearer-token>"
export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:ACCOUNT_ID:runtime/my_agui_server-xyz123"
pip install httpx httpx-sse
pip install httpx httpx-sse
pip install httpx httpx-sse
import asyncio
import json
import os
from urllib.parse import quote
from uuid import uuid4 import httpx
from httpx_sse import aconnect_sse async def invoke_agui_agent(message: str): agent_arn = os.environ.get('AGENT_ARN') bearer_token = os.environ.get('BEARER_TOKEN') escaped_arn = quote(agent_arn, safe='') url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{escaped_arn}/invocations?qualifier=DEFAULT" headers = { "Authorization": f"Bearer {bearer_token}", "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": str(uuid4()), } payload = { "threadId": str(uuid4()), "runId": str(uuid4()), "messages": [{"id": str(uuid4()), "role": "user", "content": message}], "state": {}, "tools": [], "context": [], "forwardedProps": {}, } async with httpx.AsyncClient(timeout=300) as client: async with aconnect_sse(client, "POST", url, headers=headers, json=payload) as sse: async for event in sse.aiter_sse(): data = json.loads(event.data) event_type = data.get("type") if event_type == "TEXT_MESSAGE_CONTENT": print(data.get("delta", ""), end="", flush=True) elif event_type == "RUN_ERROR": print(f"Error: {data.get('code')} - {data.get('message')}") asyncio.run(invoke_agui_agent("Hello from production!"))
import asyncio
import json
import os
from urllib.parse import quote
from uuid import uuid4 import httpx
from httpx_sse import aconnect_sse async def invoke_agui_agent(message: str): agent_arn = os.environ.get('AGENT_ARN') bearer_token = os.environ.get('BEARER_TOKEN') escaped_arn = quote(agent_arn, safe='') url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{escaped_arn}/invocations?qualifier=DEFAULT" headers = { "Authorization": f"Bearer {bearer_token}", "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": str(uuid4()), } payload = { "threadId": str(uuid4()), "runId": str(uuid4()), "messages": [{"id": str(uuid4()), "role": "user", "content": message}], "state": {}, "tools": [], "context": [], "forwardedProps": {}, } async with httpx.AsyncClient(timeout=300) as client: async with aconnect_sse(client, "POST", url, headers=headers, json=payload) as sse: async for event in sse.aiter_sse(): data = json.loads(event.data) event_type = data.get("type") if event_type == "TEXT_MESSAGE_CONTENT": print(data.get("delta", ""), end="", flush=True) elif event_type == "RUN_ERROR": print(f"Error: {data.get('code')} - {data.get('message')}") asyncio.run(invoke_agui_agent("Hello from production!"))
import asyncio
import json
import os
from urllib.parse import quote
from uuid import uuid4 import httpx
from httpx_sse import aconnect_sse async def invoke_agui_agent(message: str): agent_arn = os.environ.get('AGENT_ARN') bearer_token = os.environ.get('BEARER_TOKEN') escaped_arn = quote(agent_arn, safe='') url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{escaped_arn}/invocations?qualifier=DEFAULT" headers = { "Authorization": f"Bearer {bearer_token}", "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": str(uuid4()), } payload = { "threadId": str(uuid4()), "runId": str(uuid4()), "messages": [{"id": str(uuid4()), "role": "user", "content": message}], "state": {}, "tools": [], "context": [], "forwardedProps": {}, } async with httpx.AsyncClient(timeout=300) as client: async with aconnect_sse(client, "POST", url, headers=headers, json=payload) as sse: async for event in sse.aiter_sse(): data = json.loads(event.data) event_type = data.get("type") if event_type == "TEXT_MESSAGE_CONTENT": print(data.get("delta", ""), end="", flush=True) elif event_type == "RUN_ERROR": print(f"Error: {data.get('code')} - {data.get('message')}") asyncio.run(invoke_agui_agent("Hello from production!"))
import { HttpAgent } from "@ag-ui/client"; const agent = new HttpAgent({ url: `https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/${encodedArn}/invocations?qualifier=DEFAULT`, headers: { "Authorization": `Bearer ${token}`, "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": sessionId, }
});
import { HttpAgent } from "@ag-ui/client"; const agent = new HttpAgent({ url: `https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/${encodedArn}/invocations?qualifier=DEFAULT`, headers: { "Authorization": `Bearer ${token}`, "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": sessionId, }
});
import { HttpAgent } from "@ag-ui/client"; const agent = new HttpAgent({ url: `https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/${encodedArn}/invocations?qualifier=DEFAULT`, headers: { "Authorization": `Bearer ${token}`, "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": sessionId, }
}); - HTTP: default agent invocation
- MCP: agent ↔ tools
- A2A: agent ↔ agent
- AG-UI: agent ↔ user (new) - Streaming text responses as they generate
- Tool call lifecycle (start to result)
- State updates and progress
- User interactions sent back into the agent mid-execution - Python 3.12+
- AWS account with credentials configured (aws configure)
- IAM permissions for AgentCore Runtime
- Docker, Finch, or Podman for building container images
- Amazon Cognito user pool configured (Setup guide here) - ag_ui_strands wraps your Strands agent with AG-UI protocol support, handling event encoding automatically.
- strands-agents is the AWS agent framework used in this example.
- boto3 is the AWS SDK, used internally by Strands to call Bedrock models. - POST /invocations - handles AG-UI requests, returns the SSE event stream.
- GET /ping - health check, must return HTTP 200.
- Port 8080 - required, do not change. - Explore the AgentCore AG-UI docs
- Explore the AG-UI Protocol Docs
- Check out the AWS Announcement
