イベント / ストリーミングガイド

SDK の Events モジュールを使って、エージェントの実行状態をリアルタイムにフロントエンドへ配信する方法を解説します。

概要

Events モジュールは EventEmitter を中心とした非同期イベント配信システムです。エージェントが発行したイベントは SSE（Server-Sent Events）、Webhook、DB の3つの配信先に同時配信できます。

エージェント → EventEmitter → ┬→ SSE Handler     → フロントエンド
                               ├→ DB Handler      → PostgreSQL
                               └→ Webhook Handler → 外部システム

イベントタイプ

エージェントが発行できるイベントタイプの一覧です。

「UI 表示」列の前提

プラットフォームの ChatUI の表示に基づきます。専用 UI を出すには metadata の指定が必要です — 表示方法ごとの実装ガイドを参照してください。

EventType	用途	UI 表示
PHASE_START	フェーズ開始	フェーズ名
PROGRESS_UPDATE	進捗更新	進捗メッセージ
PROGRESS_MESSAGE	処理中の通知	ローディング表示
THOUGHT_MESSAGE	エージェントの思考過程	思考バブル
TOOL_START	ツール実行開始	タスクカード
TOOL_RESULT	ツール実行完了	タスクカード（更新）
COMPLETION_SUCCESS	処理完了（成功）	完了通知
COMPLETION_FAILURE	処理完了（失敗）	エラーメッセージ
FILE_CREATED	ファイル生成完了	ダウンロードリンク
HITL_REQUIRED_BROWSER_VNC	ブラウザ操作の人間介入要求	VNC 接続 UI
HITL_REQUIRED_BROWSER_CLI	CLI 操作の人間介入要求	メッセージ
HITL_COMPLETED	人間介入完了	完了通知
PERMISSION_REQUEST	実行許可のリクエスト	承認ダイアログ
PERMISSION_RESPONSE	実行許可の応答	承認結果
USER_INTERACTION_REQUIRED	ユーザー入力要求	選択肢ボタン
UNEXPECTED_ERROR	予期しないエラー	エラー通知

表示方法ごとの実装ガイド

イベントを実際に画面へ表示させるために必要な実装を、表示のされ方ごとにまとめます。

message だけで表示（metadata 不要）

message を入れて emit_event() するだけで表示されます。対象: PHASE_START / PROGRESS_UPDATE / COMPLETION_SUCCESS / UNEXPECTED_ERROR / PERMISSION_RESPONSE / HITL_REQUIRED_BROWSER_CLI

どう表示されるかはイベントごとに異なります（上記の「UI 表示」列を参照）。処理中であることを見せたい場合は PROGRESS_MESSAGE（ローディング表示）を使ってください。

思考バブルとして表示

こちらも metadata は不要で message を入れて emit するだけですが、reasoning 用の表示枠に描画されます。対象: THOUGHT_MESSAGE

タスクカードとして表示（ツール実行）

TOOL_START / TOOL_RESULT は、metadata を展開した内容からタスクカードとして描画されます。title は必須です（カードの見出しになります）。

metadata キー	用途
title	必須。タスクカードの見出し
description	カードの補足説明
status	in_progress（時計アイコン）/ completed（チェック）/ error（バツ）
actionType	専用レンダラーの選択（下記）。未指定でも title / description のフォールバック表示は出ます
files[]	添付ファイル。{id, filename, fileType, size, url} の形
metadata.call_id	同一タスクの更新キー。TOOL_START と TOOL_RESULT で同じ値を渡すと 1 枚のカードが更新されます

actionType に指定できる値: command_execution / code_execution / file_operation / browser_action / search_result / image_search / markdown_display

Python

# ツール実行開始（時計アイコンのカードを 1 枚出す）
await emitter.emit_event(
    event_type=EventType.TOOL_START,
    message="コマンドを実行しています",
    metadata={
        "title": "コマンド実行",
        "status": "in_progress",
        "actionType": "command_execution",
        "metadata": {"call_id": "call-1", "command": "ls -la"},
    },
)

# ツール実行完了（同じ call_id のカードを completed に更新する）
await emitter.emit_event(
    event_type=EventType.TOOL_RESULT,
    message="コマンドが完了しました",
    metadata={
        "title": "コマンド実行",
        "status": "completed",
        "actionType": "command_execution",
        "metadata": {"call_id": "call-1", "command": "ls -la", "output": "..."},
    },
)

ファイルとして表示

進捗中のファイルは metadata.files[]、完了本文の成果物は metadata.deliverable_files[] を使って表示します（file_name / download_url は使用されません）。files[] の各要素は {id, filename, fileType, size, url} の形です。詳しい形はストレージガイドを参照してください。対象: FILE_CREATED / COMPLETION_SUCCESS

選択肢 / 確認ボタンとして表示

metadata.interaction_type を入れると、本文の下にボタン UI が表示されます。

• "choice" + options[] → 選択肢ボタン。押すとその値がそのまま送信されます。options は最大 10 件・各 200 文字以内の文字列のみ受け付けられます。対象: USER_INTERACTION_REQUIRED
• "confirmation"（options なし）→ はい / いいえボタン。対象: PERMISSION_REQUEST

Python

# 選択肢を出してユーザーに選ばせる
await emitter.emit_event(
    event_type=EventType.USER_INTERACTION_REQUIRED,
    message="どの形式で出力しますか？",
    metadata={"interaction_type": "choice", "options": ["PDF", "Excel", "Markdown"]},
)

# はい / いいえで承認を取る
await emitter.emit_event(
    event_type=EventType.PERMISSION_REQUEST,
    message="このファイルを削除してよいですか？",
    metadata={"interaction_type": "confirmation"},
)

ローディング表示（処理中の通知）

アニメーション付きのローディング表示を出すには PROGRESS_MESSAGE を emit します。

Python

await emitter.emit_event(
    event_type=EventType.PROGRESS_MESSAGE,
    message="社内文書を検索しています...",
)

VNC 接続 UI として表示

HITL_REQUIRED_BROWSER_VNC / HITL_COMPLETED では、次のキーを metadata にそのまま積んでください。

metadata キー	用途
type	"vnc_popup"（介入要求）/ "vnc_completed"（完了）
execution_id	対象の実行 ID
vnc_url	接続先の VNC URL
cancel_url	介入をキャンセルする URL
vnc_port	VNC のポート
intervention_type	介入の種類
instructions	利用者への操作指示
reason	介入が必要になった理由

Python

await emitter.emit_event(
    event_type=EventType.HITL_REQUIRED_BROWSER_VNC,
    message="ログイン操作をお願いします",
    metadata={
        "type": "vnc_popup",
        "execution_id": os.environ["EXECUTION_ID"],
        "vnc_url": vnc_url,
        "cancel_url": cancel_url,
        "vnc_port": 5901,
        "intervention_type": "browser_login",
        "instructions": "表示された画面でログインし、完了したら閉じてください。",
        "reason": "認証が必要なページに到達しました",
    },
)

ターンを閉じる

表示させたい本文は完了イベントの message に必ず入れてください。ターンを閉じるには、成功時・失敗時のどちらも終端イベントとして COMPLETION_SUCCESS（finish_reason=stop）を送ります。失敗を報告する場合は、失敗の内容を message / metadata に含めてください（実行そのものの成否は PodRuntime.final(status=...) 側で記録されます）。

COMPLETION_FAILURE はターンを閉じません。最終レスポンスは COMPLETION_SUCCESS の本文に入れてください。

Python

try:
    await run_agent_logic()
except Exception as e:
    # 失敗時も終端イベントは COMPLETION_SUCCESS を送る（finish_reason=stop でターンを閉じる）。
    await emitter.emit_event(
        event_type=EventType.COMPLETION_SUCCESS,
        message=f"エラーが発生しました: {e}",
        metadata={"error_type": type(e).__name__, "status": "failed"},
    )
finally:
    # 必ずクリーンアップを実行
    await emitter.cleanup()

基本的な使い方

EventEmitter の初期化と使用

emit_event() は message（文字列）と metadata（辞書、オプション）を受け取ります。

Python

from agenticstar_platform.events import EventEmitter, EventType, SubEventType

# EventEmitter を初期化（execution_id は必須）
emitter = EventEmitter(execution_id="exec-abc-123")

# フェーズ開始イベント
await emitter.emit_event(
    event_type=EventType.PHASE_START,
    message="意図を解析中...",
    metadata={"phase": "intent"},
)

# 思考過程のストリーミング
await emitter.emit_event(
    event_type=EventType.THOUGHT_MESSAGE,
    message="ユーザーはRAG検索を求めています",
    sub_event_type=SubEventType.SEARCH_WEB,
)

# ツール実行
await emitter.emit_event(
    event_type=EventType.TOOL_START,
    message="search_knowledge を実行中",
    metadata={"tool_name": "search_knowledge", "input": {"query": "AI エージェント"}},
)

# ツール結果
await emitter.emit_event(
    event_type=EventType.TOOL_RESULT,
    message="検索完了: 5件の結果",
    metadata={"tool_name": "search_knowledge", "result_count": 5},
)

# 完了
await emitter.emit_event(
    event_type=EventType.COMPLETION_SUCCESS,
    message="回答が完了しました",
)

# クリーンアップ（必須）
await emitter.cleanup()

終端イベントの契約はターンを閉じるを参照してください。

シーケンス番号の自動管理

EventEmitter は発行されたイベントに自動でシーケンス番号を付与します。フロントエンドはこの番号でイベントの順序を保証できます。

Python

# シーケンス番号はプロパティで取得
print(emitter.sequence_number)  # 0, 1, 2, ... 自動インクリメント

イベントハンドラー

EventEmitter はコンストラクタで1つのハンドラーを受け取ります。

SSE ハンドラー（フロントエンド配信）

Python

from agenticstar_platform.events import EventEmitter, create_sse_handler

# FastAPI の StreamingResponse と組み合わせる
async def chat_stream(request: ChatRequest):
    sse_handler = create_sse_handler()
    emitter = EventEmitter(execution_id="exec-123", handler=sse_handler)

    # エージェント処理を非同期で開始
    asyncio.create_task(run_agent(emitter, request))

    # SSE ストリームを返す
    return StreamingResponse(
        emitter.consume_events(),
        media_type="text/event-stream",
    )

DB ハンドラー（永続化）

Python

from agenticstar_platform.events.handlers import DatabaseEventHandler

# DB ハンドラーでイベントを PostgreSQL に永続化
db_handler = DatabaseEventHandler(
    data_access=data_access,
    user_id="user-001",
    conversation_id="conv-001",
    message_id="msg-001",
)
emitter = EventEmitter(execution_id="exec-123", handler=db_handler)

Webhook ハンドラー（外部通知）

Python

from agenticstar_platform.events.handlers import WebhookEventHandler

# 外部システムに Webhook 通知
webhook_handler = WebhookEventHandler(
    webhook_url="https://your-system.example.com/webhook",
    conversation_id="conv-001",
    message_id="msg-001",
)
emitter = EventEmitter(execution_id="exec-123", handler=webhook_handler)

複合ハンドラー

複数のハンドラーを 1 つにまとめて管理できます。

Python

from agenticstar_platform.events.handlers import CompositeEventHandler

composite = CompositeEventHandler([
    create_sse_handler(),
    DatabaseEventHandler(data_access=da, user_id="u", conversation_id="c", message_id="m"),
    WebhookEventHandler(webhook_url="https://...", conversation_id="c", message_id="m"),
])
emitter = EventEmitter(execution_id="exec-123", handler=composite)

SubEventType と actionType

sub_event_type だけでは表示は変わりません。ツール実行の表示を出し分けるには、下表に従って metadata.actionType を指定してください。sub_event_type は DB / ログ上の分類として使えます。

SubEventType → actionType の対応

SubEventType	metadata.actionType
SEARCH_WEB	search_result
COMMAND_EXECUTION / BASH_EXECUTED	command_execution
FILE_OPERATION / FILE_EDITED / FILE_READ	file_operation
IMAGE_GENERATED	image_search（画像プレビュー系）
上記以外	未指定（title / description のフォールバック表示）か、用途の近い値を指定

Python

# 表示を出し分けるのは metadata.actionType（sub_event_type は分類用）
await emitter.emit_event(
    event_type=EventType.TOOL_START,
    message="Web検索を実行中...",
    sub_event_type=SubEventType.SEARCH_WEB,
    metadata={
        "title": "Web 検索",
        "status": "in_progress",
        "actionType": "search_result",
        "metadata": {"call_id": "call-2", "query": "AI エージェント"},
    },
)

SubEventType 一覧

SubEventType	用途
SEARCH_WEB	Web 検索中
COMMAND_EXECUTION	コマンド実行中
FILE_OPERATION	ファイル操作中
FILE_EDITED	ファイル編集完了
FILE_READ	ファイル読み取り完了
FILE_SEARCHED	ファイル検索完了
BASH_EXECUTED	Bash コマンド実行完了
WEB_FETCHED	Web ページ取得完了
MCP_TOOL	MCP ツール実行中
LOCAL_ASSISTANT	ローカルアシスタント処理中
TASK_LAUNCHED	タスク起動
TODO_UPDATED	TODO 更新
VIDEO_GENERATED	動画生成完了
IMAGE_GENERATED	画像生成完了
SLIDE_CREATED	スライド作成完了
MACOS_AUTOMATION	macOS 自動化実行中

マーケットプレイス向けハンドラー

マーケットプレイスで提供するエージェントには、専用のハンドラーファクトリを使用します。DB + Webhook の複合ハンドラーを一括生成します。

Python

import os

from agenticstar_platform.events import EventEmitter
from agenticstar_platform.events.handlers import create_marketplace_handler

# ランタイム識別子はプラットフォームが注入した環境変数から取得する
handler = create_marketplace_handler(
    data_access=data_access,
    webhook_url="https://tenant.example.com/webhook",
    user_id=os.environ["USER_ID"],
    conversation_id=os.environ["CONVERSATION_ID"],
    message_id=os.environ["MESSAGE_ID"],
    request_source="external",  # 外部API経由で応答を届ける場合に指定
)
emitter = EventEmitter(execution_id=os.environ["EXECUTION_ID"], handler=handler)

SDK 0.5.29+

上記の handler 構成・identity の受け渡し・終端イベント（正確に 1 回）・cleanup を含む実行ライフサイクル全体は、run_marketplace_agent(my_agent) が一括で 面倒を見ます（クイックスタートの runner セクション参照）。 create_marketplace_handler を直接使うのは、handler 構成を自分で組みたい場合の 低レベル API です。

エラーハンドリング

Python

try:
    await run_agent_logic()
except Exception as e:
    # エラーイベントを発行してフロントエンドに通知
    await emitter.emit_event(
        event_type=EventType.COMPLETION_FAILURE,
        message=str(e),
        metadata={"error_type": type(e).__name__},
    )
finally:
    # 必ずクリーンアップを実行
    await emitter.cleanup()

次のステップ

SDK API リファレンス — イベントモジュール

EventEmitter / StreamingEvent / EventType の完全仕様

ガイドを見る

アーキテクチャガイド

SDK のモジュール構成と設計思想

ガイドを見る