Deploy AGUI servers in AgentCore Runtime

Amazon Bedrock AgentCore AgentCore Runtime lets you deploy and run Agent User Interface (AGUI) servers in the AgentCore Runtime. This guide walks you through creating, testing, and deploying your first AGUI server.

In this section, you learn:

For more information about AGUI, see AGUI protocol contract.

How Amazon Bedrock AgentCore supports AGUI

Amazon Bedrock AgentCore's AGUI protocol support enables integration with agent user interface servers by acting as a proxy layer. When configured for AGUI, Amazon Bedrock AgentCore expects containers to run servers on port 8080 at the /invocations path for HTTP/SSE or /ws for WebSocket connections. Although AGUI uses the same port and paths as the HTTP protocol, the runtime distinguishes between them based on the --protocol flag specified during deployment configuration.

Amazon Bedrock AgentCore acts as a proxy between clients and your AGUI container. Requests from the InvokeAgentRuntime API are passed through to your container without modification. Amazon Bedrock AgentCore handles authentication (SigV4/OAuth 2.0), session isolation, and scaling.

Key differences from other protocols:

For more information, see https://docs.ag-ui.com/introduction.

Using AGUI with AgentCore Runtime

In this tutorial you create, test, and deploy an AGUI server.

For complete examples and framework-specific implementations, see AGUI Quickstart Documentation and AGUI Dojo.

Prerequisites

Step 1: Create your AGUI server

AGUI is supported by multiple agent frameworks. Choose the framework that best fits your needs. For this example, we'll show the general pattern.

Install required packages

Install packages for AWS Strands with AGUI support:

pip install fastapi
pip install uvicorn
pip install ag-ui-strands

For other frameworks, see the AGUI framework integrations.

Create your first AGUI server

Create a new file called my_agui_server.py. This example uses AWS Strands with AGUI:

# my_agui_server.py
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

# Create a simple Strands agent
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.",
)

# Wrap with AGUI protocol support
agui_agent = StrandsAgent(
    agent=strands_agent,
    name="my_agent",
    description="A helpful assistant",
)

# FastAPI server
app = FastAPI()

@app.post("/invocations")
async def invocations(input_data: dict, request: Request):
    """Main AGUI endpoint that returns event streams."""
    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)
                

For complete, framework-specific examples, see:

Understanding the code

Step 2: Test your AGUI server locally

Run and test your AGUI server in a local development environment.

Start your AGUI server

Run your AGUI server locally:

python my_agui_server.py

You should see output indicating the server is running on port 8080.

Test the endpoint

Test the SSE endpoint with a properly formatted AGUI request:

curl -N -X POST http://localhost:8080/invocations \
-H "Content-Type: application/json" \
-d '{
  "threadId": "test-123",
  "runId": "run-456",
  "state": {},
  "messages": [{"role": "user", "content": "Hello, agent!", "id": "msg-1"}],
  "tools": [],
  "context": [],
  "forwardedProps": {}
}'

You should see AGUI event streams returned in SSE format, including RUN_STARTED, TEXT_MESSAGE_CONTENT, and RUN_FINISHED events.

Step 3: Deploy your AGUI server to Bedrock AgentCore Runtime

Deploy your AGUI server to AWS using the Amazon Bedrock AgentCore starter toolkit.

Install deployment tools

Install the Amazon Bedrock AgentCore starter toolkit:

pip install bedrock-agentcore-starter-toolkit

Start by creating a project folder with the following structure:

## Project Folder Structure  
your_project_directory/  
├── my_agui_server.py          # Your main agent code  
├── requirements.txt           # Dependencies for your agent

Create a new file called requirements.txt with your dependencies:

fastapi
uvicorn
ag-ui-strands

Set up Cognito user pool for authentication

Configure authentication for secure access to your deployed server. For detailed Cognito setup instructions, see Set up Cognito user pool for authentication. This provides the OAuth tokens required for secure access to your deployed server.

Configure your AGUI server for deployment

After setting up authentication, create the deployment configuration:

agentcore configure -e my_agui_server.py --protocol AGUI

Deploy to AWS

Deploy your agent:

agentcore deploy

After deployment, you'll receive an agent runtime ARN that looks like:

arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_agui_server-xyz123

Step 4: Invoke your deployed AGUI server

Invoke your deployed Amazon Bedrock AgentCore AGUI server and interact with the event streams.

Set up environment variables

Set up environment variables

Invoke the AGUI server

To invoke the AGUI server programmatically with Python, install the required packages:

pip install httpx httpx-sse

Then use the following client code:

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!"))

For building full UI applications, see CopilotKit or the AGUI TypeScript client SDK.

Appendix

Set up Cognito user pool for authentication

For detailed Cognito setup instructions, see Set up Cognito user pool for authentication in the MCP documentation. The setup process is identical for AGUI servers.

Troubleshooting

Common AGUI-specific issues

The following are common issues you might encounter: