Amazon Bedrock AgentCore が AG-UI をサポートする方法 AgentCore ランタイムでの AG-UI の使用付録

AgentCore ランタイムに AG-UI サーバーをデプロイする

Amazon Bedrock AgentCore ランタイムを使用すると、AgentCore ランタイムでエージェントユーザーインターフェイス (AG-UI) AgentCore サーバーをデプロイして実行できます。このガイドでは、最初の AG-UI サーバーの作成、テスト、デプロイについて説明します。

このセクションでは、以下を行います。

Amazon Bedrock AgentCore が AG-UI をサポートする方法
AG-UI サーバーを作成する方法
サーバーをローカルでテストする方法
サーバーをにデプロイする方法 AWS
デプロイされたサーバーを呼び出す方法

AG-UI の詳細については、「AG-UI プロトコル契約」を参照してください。

トピック

Amazon Bedrock AgentCore が AG-UI をサポートする方法
AgentCore ランタイムでの AG-UI の使用
付録

Amazon Bedrock AgentCore が AG-UI をサポートする方法

Amazon Bedrock AgentCore の AG-UI プロトコルサポートにより、プロキシレイヤーとして機能することで、エージェントのユーザーインターフェイスサーバーとの統合が可能になります。AG-UI 用に設定されている場合、Amazon Bedrock AgentCore はコンテナが HTTP/SSE または WebSocket 接続/wsの/invocationsパスのポート8080でサーバーを実行することを期待します。AG-UI は HTTP プロトコルと同じポートとパスを使用しますが、ランタイムはデプロイ設定中に指定された--protocolフラグに基づいてそれらを区別します。

Amazon Bedrock AgentCore は、クライアントと AG-UI コンテナ間のプロキシとして機能します。InvokeAgentRuntime API からのリクエストは、変更なしでコンテナに渡されます。Amazon Bedrock AgentCore は、認証 (SigV4/OAuth 2.0)、セッション分離、スケーリングを処理します。

他のプロトコルとの主な違い:

[ポート]: AG-UI サーバーはポート 8080 で実行されます (HTTP と同じ、MCP の場合は 8000、A2A の場合は 9000)
[Path] (パス): AG-UI サーバー/invocationsが HTTP/SSE および WebSocket /wsにを使用する (HTTP プロトコルと同じ)
メッセージ形式: ストリーミングには Server-Sent Events (SSE)、双方向通信には WebSocket 経由でイベントストリームを使用します
プロトコルフォーカス: Agent-to-Userインタラクション (ツールの場合は MCP、agent-to-agentの場合は A2A)
認証: SigV4 認証スキームと OAuth 2.0 認証スキームの両方をサポート

詳細については、「https://docs.ag-ui.com/introduction」を参照してください。

AgentCore ランタイムでの AG-UI の使用

このチュートリアルでは、AG-UI サーバーを作成、テスト、デプロイします。

完全な例とフレームワーク固有の実装については、「AG-UI クイックスタートドキュメント」と「AG-UI Dojo」を参照してください。

トピック

前提条件
ステップ 1: AG-UI サーバーを作成する
ステップ 2: AG-UI サーバーをローカルでテストする
ステップ 3: AG-UI サーバーを Bedrock AgentCore ランタイムにデプロイする
ステップ 4: デプロイされた AG-UI サーバーを呼び出す

前提条件

Python 3.12 以降がインストールされている
AgentCore CLI 用にインストールされた Node.js 20 以降
適切なアクセス許可とローカル認証情報が設定されている AWS アカウント
AG-UI プロトコルとイベントベースのagent-to-user通信の概念を理解する

ステップ 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 という名前のファイルを作成します。この例では、AG-UI AWS で Strands を使用します。サーバーはポートをリッスンし8080、AG-UI トラフィック/invocationsを公開し、ヘルスチェック/pingを公開します。AgentCore Runtime では、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 はサーバー送信イベント (SSE) を使用して、型付きイベントをクライアントにストリーミングします。
/invocations エンドポイント: HTTP/SSE 通信のプライマリエンドポイント (HTTP プロトコルと同じ)
ポート 8080: AG-UI サーバーは、デフォルトで AgentCore ランタイムでポート 8080 で実行されます。

ステップ 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 形式で返される RUN_STARTED AG-UI TEXT_MESSAGE_CONTENT RUN_FINISHEDイベントストリームが表示されます。

ステップ 3: AG-UI サーバーを Bedrock AgentCore ランタイムにデプロイする

AgentCore CLI AWS を使用して AG-UI サーバーをにデプロイします。

デプロイツールをインストールする

AgentCore CLI をインストールします。


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 プロジェクトを作成します。次に、「最初の AG-UI サーバーを BYO エージェントとして作成する」で作成したサーバーを、前のステップの Cognito 設定に登録します。


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