StrandsAgentsをAG-UI対応するアダプタのコードを読みながらAG-UIを学びたい

Posted at 2026-04-19

はじめに

今回は、AIエージェントとフロントエンドを繋ぐプロトコル「AG-UI」に入門してみたいと思います。

題材として、AWSが提供するStrandsAgentsというSDKでAIエージェントを構築するケースを想定します。
StrandsAgentsをAG-U対応するためには、以下のアダプタを使用する必要があります。

このコードを読みながらAG-UIを学んでいきたいと思います。目指すは「AG-UIチョットワカル」です。

1. そもそもAG-UIとは

AG-UIはフロントエンドとAIエージェントを繋ぐ標準プロトコル

AIエージェントを組み込んだWebアプリケーションを構築する場合、フロントエンドがあり、バックエンドにはAIエージェントがいます。

では、フロントエンド⇄バックエンド間は、どのように通信すれば良いでしょうか。

そこで登場するのが「AG-UI」です。
AG-UIは、フロントエンドとAIエージェントを繋ぐイベントベースの標準プロトコルです。

AG-UIは「イベントベースのプロトコル」

AG-UIは、あくまで「相手にどのようなイベントを送るか」の仕様です。

そのため、公式で「トランスポート非依存」という設計原則があります。
双方向通信のWebSocketや一方向通信のサーバー送信イベント(SSE)で使用できる設計になっています。

AG-UI はイベントの配信方法を規定せず、サーバー送信イベント（SSE）やウェブフック、WebSocket など、さまざまなトランスポートメカニズムをサポートしています。
引用元：https://docs.ag-ui.com/concepts/architecture

サーバー送信イベントを想定し、AG-UIの役割を考えてみると、
サーバー送信イベントはHTTPに準拠した方式なので、各役割を見ると以下のようになります。

HTTP
- リクエスト/レスポンスの規約
サーバー送信イベント
- HTTPでイベントをどのように流すか (エージェントからフロントへ一方向通信)
AG-UI
- 上記の通信にどのような形式のイベントを流すか

サーバー送信イベントは一方向の通信なので、川にAG-UI仕様に則ったイベントを流すイメージですかね。

サーバー送信イベントによるストリーミング通信については、以下の神資料が大変参考になります。
https://speakerdeck.com/takuyay0ne/tuo-fen-wei-qi-shi-zhuang-agentcorewoliang-igan-zini-webapurikesiyonnizu-miip-mutameni

AG-UIは28種のイベントで表現される

AG-UIは「どのようなイベントを流すか」の仕様ですが、28種類のイベントで表現されます。

各イベント区分の概要を見てみます。(※ほぼ公式サイトの翻訳です)

イベント区分	説明
ライフサイクル	エージェントランの進行状況を監視する
テキストメッセージ	テキストコンテンツのストリーミングを処理する
ツール呼び出し	エージェントによるツール実行を管理する
状態管理	エージェントとUI間で状態を同期する
アクティビティ	継続的な活動の進捗を監視する
推論	LLMの推論過程をUIに公開し、伝える
特別	カスタム機能

AG-UIクライアントが助けてくれる

自前のフロントエンドでAG-UI周りを実装しようとすると大変なので、「AG-UIクライアント」がいい感じにイベントを処理したり、状態などを管理してくれます。

AG-UIクライアントのコードは以下です。

イベントの流れ

ではこれらのイベントがどのように流れてくるか、以下のリンクからAG-UIを体験できます。
ブラウザの開発者ツールで見てみます。

ざっくりと以下のような流れになっていることがわかりました。

ポイントは、バックエンドツールとフロントエンドツールの違いになるかと思います。
フロントエンドツールは...

エージェントを２回呼び出すことになる
TOOL_CALL_RESULTを返さない

2. StrandsAgentsでもAG-UIを使うには

AWS上にStrandsAgentsでAIエージェントを実装したとします。
デプロイ先はAmazon Bedrock AgentCore Runtimeとしましょう。

AIエージェントを構築する際、AWSのStrandsAgentsというSDKが選択肢に上がってくるかと思います。

この時に、StrandsAgentsのデータストリームをAG-UIの形式に変換するため、StrandsAgentsアダプタが必要になります。

StrandsAgentsのアダプタは「Python」のみに対応しています。TypeScript版は対応中のようです。

StrandsAgents と Bedrock間はConverseStream API

StrandsAgentsは、Amazon Bedrockの「ConverseStream API」を呼び出します。

ConverseStream APIのレスポンスは、以下のような形式です。

ConverseStream APIのレスポンスはStrandsAgentsが変換

ConverseStreamAPIのレスポンスは、StrandsAgents形式のストリーミングイベントに変換されます。

変換後のイベントは以下になります。(どのように変換されるかは別途、調べたいと思います...)

3. StrandsAgentsアダプタの変換処理を覗いてみる

StrandsAgentsのストリーミングイベントは、アダプタによりAG-UI形式のイベントに変換されます。
ここからが本編ですが、アダプタのソースコードをみながら、どのようにAG-UI形式へ変換しているのかみてみます。

PythonのStrandsAgentsのAG-UIアダプタのコードを見てみると以下のような形です。

import os
from pathlib import Path
from dotenv import load_dotenv

# Suppress OpenTelemetry context warnings
os.environ["OTEL_SDK_DISABLED"] = "true"
os.environ["OTEL_PYTHON_DISABLED_INSTRUMENTATIONS"] = "all"

from strands import Agent
from strands.models.gemini import GeminiModel
from ag_ui_strands import StrandsAgent, create_strands_app

# Load environment variables from .env file
env_path = Path(__file__).parent.parent.parent / '.env'

load_dotenv(dotenv_path=env_path)

# Use Gemini model
model = GeminiModel(
    client_args={
        "api_key": os.getenv("GOOGLE_API_KEY", "your-api-key-here"),
    },
    model_id="gemini-2.5-flash",
    params={
        "temperature": 0.7,
        "max_output_tokens": 2048,
        "top_p": 0.9,
        "top_k": 40
    }
)

strands_agent = Agent(
    model=model,
    system_prompt="""
    You are a helpful assistant.
    When the user greets you, always greet them back. Your greeting should always start with "Hello".
    Your greeting should also always ask (exact wording) "how can I assist you?"
    """,
)

agui_agent = StrandsAgent(
    agent=strands_agent,
    name="agentic_chat",
    description="Conversational Strands agent with AG-UI streaming",
)

app = create_strands_app(agui_agent, "/")

引用：https://feature-viewer.copilotkit.ai/aws-strands/feature/agentic_chat?view=code

以下のimportもとであるag_ui_strandsでは、何をやっているんでしょうか？

from ag_ui_strands import StrandsAgent, create_strands_app

前述したAG-UIの仕様と見比べつつ、GitHubのソースコードをみてみます。

※以下にアーキテクチャがあります。
https://github.com/ag-ui-protocol/ag-ui/blob/release/2026-04-06/integrations/aws-strands/ARCHITECTURE.md

AG-UI統合のファイル構成は以下になります。

__init__.py
agent.py
client_proxy_tool.py
config.py
endpoint.py
types.py
utils.py

この中で肝となるコードはagent.pyとconfig.pyになります。

ツール実行の動作を拡張する`ToolBehavior`の設定

config.pyを見るとToolBehaviorというクラスがあります。

Behavior = 振る舞い。という意味なので、ToolBehaviorはツール毎の振る舞いを定義する機能のようです。
つまり、「ツールの実行前後の追加動作」を定義できます。

agent.pyのコードを読む上で外せない要素なので、概要を知っておきます。

フロントエンドツール / バックエンドツールで利用可能

パラメータ名	説明
`predict_state`	ツール結果がフロントに届く前に、ツール引数をUIに伝える
`args_streamer`	ツール引数をストリーミングで返す
`state_from_args`	ツールの入力を`StateSnapshot`イベントで返す
`continue_after_frontend_call`	フロントエンドツール呼び出しを送信した後もストリームを稼働させ続ける

バックエンドツールのみ利用可能

パラメータ名	説明
`state_from_result`	ツールの出力からAG-UIの`StateSnapshot`イベントへ変換して送信する
`stop_streaming_after_result`	ストリーミングを停止する
`custom_result_handler`	任意のAG-UIイベントを送信する

agent.pyで未実装(多分、予約設計)
以下は、クラスにパラメータは設定されていますが、agent.pyにまだ実装が入っていないようでした。

パラメータ名	説明
`skip_messages_snapshot`	UIがすでに同期しているときに、ヘルパーメッセージが追加されるのを防止する

使い方ですがStrandsAgent に StrandsAgentConfig ごと渡します。

config = StrandsAgentConfig(
    tool_behaviors={
        "generate_recipe": ToolBehavior(
            state_from_args=recipe_state_from_args,
            state_from_result=recipe_state_from_result,
            skip_messages_snapshot=True,
        )
    }
)

agent = StrandsAgent(agent=strands_agent, name="my_agent", config=config)

`agent.py`のコードを見てみる

それではagent.pyのコードを見てみます。

ag_ui.coreから各イベントクラスをインポートしているので、StrandsAgentsアダプタが何のAG-UIイベントに対応しているか確認してみます。

以下は前述したAG-UIのイベント一覧ですが、インポートされていないイベントを黒塗りしてみました。
これでいくつかのイベントはStrandsAgentsアダプタでは返していないことがわかりました。
(MessagesSnapshotEventはインポートされていますが未使用なので、黒塗りにしています)

メインとなる変換処理はagent.pyのrun()で行います。
run()は、大きくは３つの処理ステップで動いています。

前処理
StrandsAgentsのストリーミング処理
後処理

AG-UIクライアントからの入力となるRunAgentInputは以下の型になります。

では、ここからが実際のrun()の処理になります。

前処理

スレッドIDの設定

前提として、エージェントを呼び出す単位とスレッドの単位は異なり、１スレッド内で複数回のエージェント呼び出しを行うことがあります。

入力パラメータにthread_idが入ってくれば、そのエージェント実行はスレッド内の継続なので、StrandsAgentsインスタンスを使い回します。

逆に空であれば、新規のエージェント呼び出しなので、新たにStrandsAgentsのインスタンスから作成します。

フロントエンドツールとStrandsAgentsツールの同期

フロントエンドツールは、名前の通りフロントエンド側で定義するツールです。
そのため、エージェント側はツールの存在を知らないので、知らせる必要があります。

ここでは、フロントエンドからツールが渡された場合、StrandsAgentsにプロキシとしてツールを登録します。
これにより、エージェントがツールを使用するか判断することができます。
(ツールの実態はフロントエンドなので、あくまでプロキシのツールです）

sync_proxy_toolsのコードは以下になります。
https://github.com/ag-ui-protocol/ag-ui/blob/release/2026-04-06/integrations/aws-strands/python/src/ag_ui_strands/client_proxy_tool.py

ライフサイクルイベントの処理

ライフサイクルイベントは、エージェントとの対話の始まりと終わりを判断します。

`RUN_STARTED`イベント

RUN_STARTEDイベントをAG-UIクライアントへ返します。
これにより、エージェントとの対話が始まったことが分かります。

`RUN_FINISHED`イベント

エージェントとの対話が終わったら、対話完了をAG-UIクライアントへ返します。

`RUN_ERROR`イベント

イベントのストリーミング処理で例外が発生した場合、エラーをAG-UIクライアントへ返します。
STRANDS_ERROR という固定のエラーコードを返しています。
特にAG-UIクライアント側にエラーコードを使ったロジックは無さそうなので、あくまでログ出力用のようです。

`STATE_SNAPSHOT`を同期

フロントエンドからstateが渡ってきた場合は、STATE_SNAPSHOTを返します。

ユーザープロンプトの生成

実際にStrandsAgentsを呼び出す前にユーザープロンプトを生成します。
大きくは、直近フロントエンドツールを呼び出したか、呼び出していないか。で分かれます。

フロントエンドツールの場合は、固定のユーザープロンプトに変換します。
それ以外の対話の場合は、そのまま設定します。

pending_tool_result_idsとは
直近のエージェント実行でフロントエンドが実行済みのツールIDが入っています。
(厳密にはtool_call_idのセットになります)
https://github.com/ag-ui-protocol/ag-ui/blob/release/2026-04-06/integrations/aws-strands/python/src/ag_ui_strands/agent.py#L149-L161

state_context_builderとは
config.pyに定義された設定ですが、現時点では機能していないようです。
厳密には、strands_messages変数を加工しているのですが、そのstrands_messagesが実際には使われていないので、予約設計と思われます。
https://github.com/ag-ui-protocol/ag-ui/blob/release/2026-04-06/integrations/aws-strands/python/src/ag_ui_strands/agent.py#L279-L289

StrandsAgentsを呼び出す

ここからは実際にStrandsAgentsの非同期ストリームを実行します。

参考：https://strandsagents.com/docs/user-guide/concepts/streaming/async-iterators/

stream_async が返す event は、以下の種類があります。
これらのイベントは、StrandsAgents特有の形式なので、AG-UI形式のストリーミングイベントに変換する必要があります。

以下、StrandsAgentsアダプタで拾われているイベント以外を黒塗りにしてみました。

参考：strands-agents/sdk-python - types/_events.py

Strandsの`init_event_loop`、`start_event_loop`イベントを処理

StrandsAgentsの返すinit_event_loop、start_event_loopイベントを処理します。

処理といっても特に処理せずにスキップします。

Strandsの`complete`、`force_stop`イベントの処理

StrandsAgentsの返すcomplete、force_stopイベントを処理します。

こちらは処理を終了します。

completeイベントは、StrandsAgentsのドキュメントには書かれていないようです。
https://strandsagents.com/docs/user-guide/concepts/streaming/

Strandsの`data`イベントの処理

dataイベントは、モデルからのテキストチャンク(テキストの断片)です。

初回の場合は、TEXT_MESSAGE_STARTイベントをAG-UIクライアントへ送信します。
初回以降のdataイベントは、TEXT_MESSAGE_CONTENTイベントをAG-UIクライアントへ送信します。

Strandsの`tool_stream_event`イベントの処理

tool_stream_eventイベントは、実行中のツールの途中経過のデータです。

ツールの実行状態をSTATE_SNAPSHOTイベントでAG-UIクライアントへ送信します。

Strandsの`message`イベント(バックエンドでツール実行)の場合

次はStrandsAgentsからmessageイベントが送られた場合の処理です。

messageイベントの中身ですが、以下のような構造のデータがStrandsAgentsから流れてきます。

参考：https://strandsagents.com/docs/api/python/strands.types.content/#message

ToolResultContentからtext部分を抽出します。

取り出した result_tool_id と result_data は、この後の処理で ToolCallResultEvent を yield するために使われます。

ここからは、バックエンドツールの処理です。(フロントエンドツールは対象外)

`TOOL_CALL_RESULT`イベントを送信

TOOL_CALL_RESULTイベントを返します。ポイントは以下の２点です。

新しいmessage_idを使用する
roleを省略する

roleを省略することで、message(会話履歴)に追加されなくなるようです。
StrandsAgentsが内部で会話履歴を管理しているので、フロントエンドが重複して履歴を追加しないようにする対策のようです。

また新しいmessage_idを使用する理由は、CopilotKitがツール実行中のスピナーを正しく閉じるためだそうです。
(CopilotKitはAIエージェントのフロント側を構築するフレームワークですが、AG-UIのコメントに登場するのは違和感がありますね...CopilotKitを意識する時点で、標準じゃないような...)

`state_from_result`設定の処理

state_from_result にユーザーが定義した関数を渡しておくと、ツール結果を加工して、クライアント側の状態を更新できます。具体的にはAG-UIの StateSnapshotイベントを送信し、状態を更新します。

`custom_result_handler`

ユーザーが定義した関数で、任意のAG-UIイベントを送ります。

`stop_streaming_after_result`

stop_streaming_after_resultは、バックエンドツールのツール結果を受け取った後にLLMが生成するテキストを止めることができます。

Strandsの`current_tool_use`イベントの処理

次にcurrent_tool_useイベントの処理です。

current_tool_useイベントには、以下のようなツール内容の分かるイベントが渡ってきます。

参考：https://strandsagents.com/docs/api/python/strands.types.tools/

そのため、current_tool_useイベントの処理では、実行されたツールの登録または更新を行います。

新規ツールが呼ばれた場合、tool_calls_seenに新しく追加します。
tool_calls_seenは、ストリーミング中に呼び出したツールを覚えておくための変数です。

既存ツールが呼ばれた場合、追加済みの入力や引数の状態を更新します。

Strandsの`event`イベント(`contentBlockStop`)の処理

contentBlockStop は Bedrock がツール入力が完了した際に送る情報です。
この情報をトリガーに、AG-UIクライアントへのツール呼び出しイベントを送信します。

`state_from_args`設定の処理

ツールが実行される前に引数をSTATE_SNAPSHOTイベントでAG-UIクライアントへ送信します。

`predict_state`設定の処理

CUSTOMイベントに "PredictState"という名前を付けてフロントエンドへ送ります。
ツールの引数が全部揃う前からUI側を更新することができるようです。

`TOOL_CALL_START`、`TOOL_CALL_ARGS`、`TOOL_CALL_END`イベントの送信 + `args_streamer`設定の処理

以下のツール実行イベントをAG-UIクライアントへ連続で送信します。

TOOL_CALL_START
TOOL_CALL_ARGS
TOOL_CALL_END

ただし、args_streamerが設定されている場合、TOOL_CALL_ARGSはストリーミングで送信します。

`continue_after_frontend_call`設定の処理

フロントエンドでツール実行する場合、デフォルトだとエージェントはツール実行の結果を受け取る必要があるため、一度中断モードに入ります。この中断モードはpending_haltフラグをTrueに設定します。

先ほどのフロントエンドツール実行時のシーケンスを思い出すと、２回のエージェント実行に分けることで中断→再開という流れを踏んでいます。

continue_after_frontend_callは、この中断モードをOFFにできます。
エージェントの処理は続行されるので、フロントエンドツールの実行結果を待たずに処理が完了します。

フロントエンドツールの実行結果をエージェントが必要としない場合は、待ち時間が減るのでユーザー体験が少しでもよくなるかも知れないですね。

厳密には、pending_haltは「次のイベントで止める」というフラグなので、実際には次のイベントでhalt_event_streamを見て処理を止めることになります。

後処理

最後に以下のイベントをAG-UIクライアントへ送信して処理を完了します。

TEXT_MESSAGE_END
STATE_SNAPSHOT
RUN_FINISHED

message_started が True の場合のみ、テキストメッセージを終了します。

message_started は、ストリーミング中にテキストを1つでも送出した場合に True になります。
ツールだけで応答が終わった場合は False のままなので、TextMessageEndEvent はスキップされます。

まとめ

ざっと、ソースコードを読んでみましたが、ふんわりとAG-UIについて分かってきた気がします！

ただバックエンドツールだけなら割とシンプルなコードになりそうですが、フロントエンドツールが関わるとややこしいですね。この辺りのコードの整備はまだまだこれからな気はします。

あと、ToolBehaviorは、もう少し公式ドキュメントで説明が欲しいところです。
とはいえ、この辺りは使ってみてとは思うので、今度ToolBehaviorの活用や深掘りで検証してみたいと思います。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

StrandsAgentsをAG-UI対応するアダプタのコードを読みながらAG-UIを学びたい

はじめに

1. そもそもAG-UIとは

AG-UIはフロントエンドとAIエージェントを繋ぐ標準プロトコル

AG-UIは「イベントベースのプロトコル」

AG-UIは28種のイベントで表現される

AG-UIクライアントが助けてくれる

イベントの流れ

2. StrandsAgentsでもAG-UIを使うには

StrandsAgents と Bedrock間はConverseStream API

ConverseStream APIのレスポンスはStrandsAgentsが変換

3. StrandsAgentsアダプタの変換処理を覗いてみる

ツール実行の動作を拡張するToolBehaviorの設定

agent.pyのコードを見てみる

前処理

スレッドIDの設定

フロントエンドツールとStrandsAgentsツールの同期

ライフサイクルイベントの処理

RUN_STARTEDイベント

RUN_FINISHEDイベント

RUN_ERRORイベント

STATE_SNAPSHOTを同期

ユーザープロンプトの生成

StrandsAgentsを呼び出す

Strandsのinit_event_loop、start_event_loopイベントを処理

Strandsのcomplete、force_stopイベントの処理

Strandsのdataイベントの処理

Strandsのtool_stream_eventイベントの処理

Strandsのmessageイベント(バックエンドでツール実行)の場合

TOOL_CALL_RESULTイベントを送信

state_from_result設定の処理

custom_result_handler

stop_streaming_after_result

Strandsのcurrent_tool_useイベントの処理

Strandsのeventイベント(contentBlockStop)の処理

state_from_args設定の処理

predict_state設定の処理

TOOL_CALL_START、TOOL_CALL_ARGS、TOOL_CALL_ENDイベントの送信 + args_streamer設定の処理

continue_after_frontend_call設定の処理

後処理

まとめ