はじめに
2026年3月13日、Amazon Bedrock AgentCore Runtime が AG-UI プロトコルに対応したというアップデートが発表されました。
AIエージェントを動かしていると「裏で何をしているかわからない」という体験、誰しも一度はしたことがあるのではないでしょうか。
AG-UI を使えば、AIエージェントの思考過程・ツール呼び出し・中間結果がリアルタイムにUIへストリーミングされ、ユーザーは"エージェントと対話している"実感を得られます。
この記事では、AG-UI とは何か、既存プロトコル（MCP/A2A）との関係、そして実際にAWS上で動かすところまでを図解付きでまとめていきます。
忙しい人向けまとめ
- どんなアップデート？: Bedrock AgentCore Runtime が AG-UI プロトコルをネイティブサポート
- AG-UI とは？: エージェント↔ユーザーUI間のオープンな通信プロトコル
- 何が嬉しい？: エージェントの応答・思考・ツール実行をリアルタイムにUIへストリーミング
1. AG-UI プロトコルとは？
AG-UI（Agent-User Interaction Protocol）は、AIエージェントとユーザー向けアプリケーションをつなぐ、オープンで軽量なイベントベースのプロトコルです。
従来の REST API では「リクエストを送り、レスポンスを待つ」という一方通行でしたが、エージェントは長時間動作し、途中経過を返し、非決定的に振る舞います。AG-UI はこの課題を解決するために設計されました。
2. HTTP+SSEでもできなかったけ？
「AIエージェントの思考過程・ツール呼び出し・中間結果をリアルタイムにストリーミングするなら、今までも HTTP+SSE でできたのでは？」と思われる方も多いかもしれません。
技術的には確かに可能です。ただし、データ構造とイベントの定義を毎回自分で決める必要があるという問題があります。AG-UI はそこを標準化することで、フレームワーク間の互換性と実装の一貫性を実現しています。
サーバー側の実装比較
HTTP+SSEの場合
開発者が独自にデータ構造を定義し、SSEの形式に合わせて文字列を送信する必要があります。
「思考中」「ツール使用」「回答」などの状態を区別するために、独自のJSONスキーマ（例: {"type": "thought", "content": "..."}）を設計し、それをストリームとして流す処理を自前で書く必要があります。
AG-UIの場合
Strands、LangGraph、CrewAI などのフレームワークを使用することで、プロトコルに準拠したイベントが自動的に生成されます。
開発者はエージェントのロジックに集中でき、フレームワークが AG-UI標準のイベント（RUN_STARTED, TEXT_MESSAGE_CONTENT, RUN_FINISHED など）を適切な形式で送出します。
メッセージフォーマットの比較
| 比較項目 | HTTP+SSE | AG-UI |
|---|---|---|
| イベント形式 | 開発者が定義した任意のJSON | 標準化されたイベントストリーム |
| 主なイベント種別 | なし（自作） | RUN_STARTED, TEXT_MESSAGE_CONTENT, RUN_FINISHED, TOOL_CALL など |
| UIへの反映 | クライアント側で独自の解析が必要 | イベント型に基づいて進捗バーや思考ステップを透過的に可視化可能 |
クライアント側の実装比較
HTTP+SSEの場合
# レスポンスを1行ずつ読み込み、独自のJSONをパースする
for line in response.iter_lines():
data = json.loads(line)
if data['type'] == 'my_custom_thought_tag':
show_spinner(data['text'])
elif data['type'] == 'my_final_answer':
print(data['text'])
AG-UIの場合
AG-UIではイベントの種類が固定されているため、共通のロジックで対応できます。
# AG-UI標準のイベントタイプに基づいて処理する
for event in response.event_stream:
if event.type == "RUN_STARTED":
start_progress_bar()
elif event.type == "TEXT_MESSAGE_CONTENT":
append_to_chat(event.content)
elif event.type == "RUN_FINISHED":
stop_progress_bar()
3. エージェントプロトコルの全体像：MCP / A2A / AG-UI
現在、エージェント関連には 3 つの主要オープンプロトコルがあります。それぞれ担当する「レイヤー」が異なります。
| レイヤー | プロトコル | 提唱元 | 役割 |
|---|---|---|---|
| Agent ↔ User | AG-UI | CopilotKit | エージェントの出力をリアルタイムにUIへ |
| Agent ↔ Tools | MCP | Anthropic | エージェントに外部ツール・データを提供 |
| Agent ↔ Agent | A2A | エージェント間の連携・タスク委譲 |
4. Bedrock AgentCore Runtimeの構成
AgentCore Runtime がどのように各プロトコルをホストしているか見てみましょう。
| 項目 | AG-UI | MCP | A2A |
|---|---|---|---|
| ポート | 8080 | 8000 | 9000 |
| エンドポイント |
/invocations（SSE）, /ws（WebSocket） |
- | - |
| 通信形式 | イベントストリーム | - | - |
| フォーカス | Agent→User | Agent→Tools | Agent→Agent |
5. ローカルで動かしてみる
ここからは実際に AG-UI サーバーを作って動かしてみます。
5.1 環境準備
pip install fastapi uvicorn ag-ui-strands strands-agents httpx httpx-sse
5.2 AG-UIサーバーの実装
以下のコードで my_agui_server.py を作成します。
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
agui_agent = None
def get_agent():
global agui_agent
if agui_agent is None:
model = BedrockModel(
model_id="jp.amazon.nova-2-lite-v1:0",
region_name="ap-northeast-1",
max_tokens=4096,
)
strands_agent = Agent(
model=model,
system_prompt="あなたは親切な日本語アシスタントです。質問に丁寧に答えてください。",
)
agui_agent = StrandsAgent(
agent=strands_agent,
name="my_japanese_agent",
description="日本語で応答するどんなことにも回答するエージェント",
)
return agui_agent
app = FastAPI(title="My AG-UI Server")
@app.post("/invocations")
async def invocations(input_data: dict, request: Request):
"""AG-UIエンドポイント：イベントストリームを返す"""
accept_header = request.headers.get("accept")
encoder = EventEncoder(accept=accept_header)
async def event_generator():
run_input = RunAgentInput(**input_data)
async for event in get_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)
5.3 ローカルで起動
python my_agui_server.py
以下のようなログが出れば成功です。
INFO: Started server process [12345]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)
5.4 curlでテスト
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": "AWSのAG-UIサポートについて教えてください",
"id": "msg-1"
}
],
"tools": [],
"context": [],
"forwardedProps": {}
}'
以下のような SSEイベントストリームが順次返ってきます。
event: RUN_STARTED
data: {"type":"RUN_STARTED","threadId":"test-123","runId":"run-456"}
event: TEXT_MESSAGE_START
data: {"type":"TEXT_MESSAGE_START","messageId":"msg-abc","role":"assistant"}
event: TEXT_MESSAGE_CONTENT
data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"msg-abc","delta":"Amazon"}
event: TEXT_MESSAGE_CONTENT
data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"msg-abc","delta":" Bedrock"}
event: TEXT_MESSAGE_CONTENT
data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"msg-abc","delta":" AgentCore"}
... (テキストがチャンクごとにストリーミング)
event: TEXT_MESSAGE_END
data: {"type":"TEXT_MESSAGE_END","messageId":"msg-abc"}
event: RUN_FINISHED
data: {"type":"RUN_FINISHED","threadId":"test-123","runId":"run-456"}
5.5 Pythonクライアントで受信してみる
SSEイベントをPythonで受信するクライアントも書いてみましょう。イベントタイプごとに処理を分けることで、ツール実行の可視化なども簡単に実装できます。
agui_client.py
import asyncio
import json
from uuid import uuid4
import httpx
from httpx_sse import aconnect_sse
async def invoke_agui_agent(message: str):
url = "http://localhost:8080/invocations"
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, json=payload
) as sse:
async for event in sse.aiter_sse():
data = json.loads(event.data)
event_type = data.get("type")
if event_type == "RUN_STARTED":
print("🚀 エージェント実行開始!")
elif event_type == "TEXT_MESSAGE_CONTENT":
print(data.get("delta", ""), end="", flush=True)
elif event_type == "TOOL_CALL_START":
tool_name = data.get("toolCallName", "unknown")
print(f"\n🔧 ツール実行中: {tool_name}")
elif event_type == "TOOL_CALL_RESULT":
print(f"📊 ツール結果受信")
elif event_type == "RUN_FINISHED":
print("\n✅ 完了!")
elif event_type == "RUN_ERROR":
print(f"\n❌ エラー: {data.get('message')}")
asyncio.run(invoke_agui_agent("AG-UIプロトコルについて教えてください"))
python agui_client.py
出力例：
🚀 エージェント実行開始!
AG-UIプロトコルは、AIエージェントとユーザーインターフェースをリアルタイムで
つなぐためのオープンな通信規格です。テキストのストリーミング配信やツール実行の
可視化など、豊かなエージェント体験を標準化された方法で実現できます。
✅ 完了!
6. AWSへのデプロイ
6.1 Cognito User Poolを作成
agentcore identity setup-cognito --region ap-northeast-1
実行後、以下のような出力が表示されます。
✅ Cognito pools created successfully!
╭────────────────────────────────────── Runtime Pool (Inbound Auth) ───────────────────────────────────────╮
│ Pool ID: ap-northeast-1_Uiw3wKJEa │
│ Client ID: 4q2ru3f5c5kgfckodkfhriel7k │
│ Discovery URL: │
│ https://cognito-idp.ap-northeast-1.amazonaws.com/ap-northeast-1_Uiwjelwi9d/.well-known/openid-configurati │
│ on │
│ Test User: testuser36f469ea │
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭────────────────────────────────────── Identity Pool - User Consent ──────────────────────────────────────╮
│ Pool ID: ap-northeast-1_rUR4ZM3qq │
│ Client ID: 63pu8j0s83sn4t8ijeke0zhell │
│ Flow Type: USER │
│ Discovery URL: │
│ https://cognito-idp.ap-northeast-1.amazonaws.com/ap-northeast-1_rURjijo8h/.well-known/openid-configurati │
│ on │
│ Test User: externaluser0bf1d7b6 │
╰──────────────────────────────────────────────────────────────────────────────────────────────────────────╯
🔐 Credentials saved securely to:
• /path/to/project/.agentcore_identity_user.env (standard .env format)
6.2 requirements.txtを準備
デプロイ時に依存パッケージを正しくインストールするため、requirements.txt を作成します。
requirements.txt
fastapi
uvicorn
ag-ui-strands
strands-agents
6.3 デプロイ
Bedrock AgentCore Starter Toolkit をインストールします。
pip install bedrock-agentcore-starter-toolkit
プロジェクト構成を確認し、設定・デプロイを実行します。
# プロジェクト構成
your_project/
├── my_agui_server.py
└── requirements.txt
# デプロイ設定（プロトコルにAGUIを指定）
agentcore configure -e my_agui_server.py --protocol AGUI \
--authorizer-config '{
"customJWTAuthorizer": {
"discoveryUrl": "https://cognito-idp.ap-northeast-1.amazonaws.com/<POOL_ID>/.well-known/openid-configuration",
"allowedClients": ["<CLIENT_ID>"]
}
}'
# デプロイ実行
agentcore deploy
デプロイ完了後、Agent ARNが表示されます。
✅ Agent created/updated: arn:aws:bedrock-agentcore:ap-northeast-1:123456789012:runtime/my_agui_server-xyz123
6.4 デプロイ済みエージェントを呼び出す
生成されたenvファイルから認証情報を読み込み、JWTトークンを取得して呼び出します。
# 認証情報を読み込む
export $(grep -v '^#' .agentcore_identity_user.env | xargs)
# JWTトークンを取得して環境変数にセット
export BEARER_TOKEN=$(agentcore identity get-cognito-inbound-token)
# Agent ARNをセット（デプロイ時に表示されたARN）
export AGENT_ARN="arn:aws:bedrock-agentcore:ap-northeast-1:123456789012:runtime/my_agui_server-xyz123"
# クライアントを実行
python agui_client_runtime.py
デプロイ済みエージェントを呼び出すクライアントコードです。
agui_client_runtime.py
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_deployed_agent(message: str):
agent_arn = os.environ["AGENT_ARN"]
bearer_token = os.environ["BEARER_TOKEN"]
escaped_arn = quote(agent_arn, safe="")
url = (
f"https://bedrock-agentcore.ap-northeast-1.amazonaws.com"
f"/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"\nError: {data.get('message')}")
print()
asyncio.run(invoke_deployed_agent("こんにちは 世界"))
7. ユースケースのイメージ
AG-UIを使うと、こんなリッチなUIが実現できます。
具体的なユースケース例：
- カスタマーサポートBot: ユーザーの問い合わせに対してDB検索・API呼び出しの過程を見せながら回答
- コード生成アシスタント: 思考過程→コード生成→テスト実行の各ステップをリアルタイム表示
- データ分析エージェント: SQL発行→データ取得→グラフ生成をステップごとに可視化
- 旅行プランナー: ホテル検索→フライト比較→予算計算をプログレスバーで表示
8. まとめ
Amazon Bedrock AgentCore Runtime で AG-UI が使えるようになったことで、「エージェントが何をしているかわからない」という問題を解決するための標準的な手段が整いました。フレームワーク側がイベント生成を担ってくれるため、開発者はロジックに集中しながらリッチなユーザー体験を提供できそうですが、また一つ学習することが増えましたね。