亚马逊 Bedrock AgentCore 如何支持 AG-UI AG-UI 与 AgentCore 运行时一起使用附录

在 AgentCore 运行时部署 AG-UI 服务器

Amazon Bedrock R AgentCore untime 允许您在运行时部署和 AgentCore 运行代理用户界面 (AG-UI) 服务器。本指南引导您创建、测试和部署第一 AG-UI 台服务器。

在本部分中，您将学习：

亚马逊 Bedrock AgentCore 如何支持 AG-UI
如何创建 AG-UI 服务器
如何在本地测试服务器
如何将服务器部署到 AWS
如何调用已部署的服务器

有关的更多信息 AG-UI，请参阅AG-UI 协议合同。

亚马逊 Bedrock AgentCore 如何支持 AG-UI

Amazon Bedrock AgentCore 的 AG-UI 协议支持通过充当代理层来实现与代理用户界面服务器的集成。配置为时 AG-UI，Amazon Bedrock AgentCore 期望容器8080在连接路径 HTTP/SSE 或 WebSocket 连接/invocations路径的端口上运行服务器/ws。尽管 AG-UI 使用与 HTTP 协议相同的端口和路径，但运行时会根据部署配置期间指定的--protocol标志来区分它们。

亚马逊 Bedrock AgentCore 充当客户端和您的 AG-UI 容器之间的代理。来自 InvokeAgentRuntime API 的请求无需修改即可传递到您的容器。Amazon Bedrock AgentCore 负责身份验证 (SigV4/OAuth 2.0)、会话隔离和扩展。

与其他协议的主要区别：

端口：: AG-UI 服务器在端口 8080 上运行（与 HTTP 相同，而 MCP 为 8000，A2A 为 9000）
路径: AG-UI 服务器/invocations/ws用于 HTTP/SSE 和用于 WebSocket （与 HTTP 协议相同）
消息格式: 通过事件 (SSE) 使用 Server-Sent 事件流进行流式传输或 WebSocket 双向通信
协议焦点: Agent-to-User 交互（与工具的 MCP 相比，代理对代理的 A2A）
身份验证: 支持 SigV4 和 OAuth 2.0 身份验证方案

有关更多信息，请参阅 https://docs.ag-ui.com/introduction。

AG-UI 与 AgentCore 运行时一起使用

在本教程中，您将创建、测试和部署 AG-UI 服务器。

有关完整的示例和特定于框架的实现，请参阅AG-UI 快速入门文档和 Dojo。 AG-UI

主题

先决条件
第 1 步：创建 AG-UI 服务器
第 2 步：在本地测试您的 AG-UI 服务器
第 3 步：将 AG-UI 服务器部署到 Bedrock Run AgentCore time
第 4 步：调用已部署的 AG-UI 服务器

先决条件

已安装 Python 3.12 或更高版本
Node.js 为 AgentCore CLI 安装了 20 或更高版本
配置了适当权限和本地凭据的 AWS 账户
了解 AG-UI 协议和基于事件的代理与用户通信概念

第 1 步：创建 AG-UI 服务器

AG-UI 受多个代理框架支持。本教程使用 AWS Strands for Python。

安装所需的程序包

为 AG-UI 支持的 AWS Strands 安装软件包：


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

有关其他框架，请参阅AG-UI 框架集成。

创建你的第一 AG-UI 台服务器

创建一个名为 my_agui_server.py的文件。此示例使用了 AWS Strands 和。 AG-UI服务器监听端口8080，暴露/invocations AG-UI 流量，公开运行状况检查/ping。 AgentCore 运行时要求 AG-UI 容器使用此合约。


# 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

# Create a simple Strands agent
strands_agent = Agent(
    system_prompt="You are a helpful assistant.",
)

# Wrap with AG-UI 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 AG-UI 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)

有关特定于框架的完整示例，请参阅：

理解代码

活动直播: AG-UI 使用 Server-Sent 事件 (SSE) 将键入的事件流式传输到客户端
/invocations 终端节点: HTTP/SSE 通信的主要端点（与 HTTP 协议相同）
端口 8080: AG-UI 服务器在运行时默认在 8080 端口上运行 AgentCore

第 2 步：在本地测试您的 AG-UI 服务器

在本地开发环境中运行和测试您的 AG-UI 服务器。

启动你的 AG-UI 服务器

在本地运行 AG-UI 服务器：


python my_agui_server.py

您应该看到表明服务器正在端口上运行的输出8080。

测试端点

使用格式正确的 AG-UI 请求测试 SSE 端点：


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": {}
}'

您应该看到以 SSE 格式返回 AG-UI 的事件流RUN_STARTED，包括TEXT_MESSAGE_CONTENT、和RUN_FINISHED事件。

第 3 步：将 AG-UI 服务器部署到 Bedrock Run AgentCore time

AWS 使用 AgentCore CLI 将您的 AG-UI 服务器部署到。

安装部署工具

安装 C AgentCore LI：


npm install -g @aws/agentcore

首先创建具有以下结构的项目文件夹：


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

使用您的依赖项创建一个名requirements.txt为的新文件：


fastapi
uvicorn
ag-ui-strands

设置 Cognito 用户池进行身份验证

配置身份验证以安全访问已部署的服务器。有关 Cognito 的详细设置说明，请参阅设置 Cognito 用户池以进行身份验证。这提供了安全访问已部署服务器所需的 OAuth 令牌。

完成 Cognito 设置后，导出部署命令使用的值：


export REGION="<your-region>"
export POOL_ID="<your-user-pool-id>"
export CLIENT_ID="<your-app-client-id>"

配置 AG-UI 服务器以进行部署

创建一个空 AgentCore 项目。然后使用上一步中的 Cognito 配置将您在创建第一 AG-UI 台服务器作为 BYO 代理中创建的服务器注册为 BYO 代理：


agentcore create --project-name AguiProject --no-agent
cd AguiProject
agentcore add agent \
  --name AguiAgent \
  --type byo \
  --language Python \
  --framework Strands \
  --model-provider Bedrock \
  --memory none \
  --code-location .. \
  --entrypoint my_agui_server.py \
  --protocol AGUI \
  --authorizer-type CUSTOM_JWT \
  --discovery-url "https://cognito-idp.$REGION.amazonaws.com/$POOL_ID/.well-known/openid-configuration" \
  --allowed-clients "$CLIENT_ID" \
  --request-header-allowlist Authorization

这些命令使用上一步中的 AG-UI 协议和 Cognito OAuth 配置注册现有实现。

部署到 AWS

部署您的代理：


agentcore deploy

部署后，您将收到一个代理运行时 ARN，如下所示：


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

第 4 步：调用已部署的 AG-UI 服务器

调用您部署的 Amazon Bedrock AgentCore AG-UI 服务器并与事件流进行交互。

设置环境变量

将持有者令牌作为环境变量导出。有关持有者令牌的设置，请参阅设置 Cognito 用户池以进行身份验证。
```
export BEARER_TOKEN="<BEARER_TOKEN>"
```

导出代理 ARN。


export AGENT_ARN="arn:aws:bedrock-agentcore:us-west-2:accountId:runtime/my_agui_server-xyz123"

调用 AG-UI 服务器

要以编程方式调用 AG-UI 服务器，请选择与您的客户端匹配的语言：

例

Python

安装所需的软件包：


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

TypeScript

安装所需的软件包：


npm install @ag-ui/client

然后使用以下客户端代码：


import { HttpAgent, AgentSubscriber } from "@ag-ui/client";
import { randomUUID } from "crypto";

async function invokeAguiAgent(message: string): Promise<void> {
  const agentArn = process.env.AGENT_ARN!;
  const bearerToken = process.env.BEARER_TOKEN!;
  const escapedArn = encodeURIComponent(agentArn);

  const agent = new HttpAgent({
    url: `https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/${escapedArn}/invocations?qualifier=DEFAULT`,
    headers: {
      Authorization: `Bearer ${bearerToken}`,
      "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": randomUUID(),
    },
  });

  agent.messages = [{ id: randomUUID(), role: "user", content: message }];

  const subscriber: AgentSubscriber = {
    onTextMessageContentEvent: ({ event }) => {
      process.stdout.write(event.delta);
    },
    onRunErrorEvent: ({ event }) => {
      console.error(`Error: ${event.code ?? "RUN_ERROR"} - ${event.message}`);
    },
  };

  await agent.runAgent({}, subscriber);
}

void invokeAguiAgent("Hello!");