Managed AgentsのSSEイベントストリームを実装する

Managed Agentsとのやり取りはすべてイベントベースです。ユーザーイベントを送り、エージェントイベントとセッションイベントをSSE(Server-Sent Events)で受け取って進捗を追います。ツール呼び出しの検知からテキストのリアルタイムプレビューまで、イベントストリームを実装する手順を順に説明します。

イベントは2方向に流れる

Managed Agentsのイベントは送信方向と受信方向に分かれます。

• ユーザーイベント・システムイベント: こちらから送るイベント。user.*はセッションを開始・誘導し、system.messageはそのターン以降に適用されるシステムレベルのコンテキストを追加します
• セッションイベント・スパンイベント・エージェントイベント: こちらへ届くイベント。セッション状態とエージェントの進捗を可視化します

イベント種別の文字列はすべて{domain}.{action}という命名規則に従います(ストリーム限定のプレビューイベントevent_start・event_deltaだけが例外)。エージェントイベントのうちツール呼び出し関連は次の3系統があります。

すべての永続化イベントには、処理完了時に設定されるprocessed_atタイムスタンプが付きます。送信したイベントがまだ処理待ちの間、processed_atはnullのままです。例外はuser.define_outcome・user.custom_tool_result・user.tool_resultで、これらは受信時点で処理されprocessed_atが最初から入った状態でエコーバックされます。

メッセージを送りイベントを流す

まずuser.messageイベントを送って、セッションに作業を開始または継続させます。

curl --fail-with-body -sS "https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true" \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -H "anthropic-beta: managed-agents-2026-04-01" \
  -H "content-type: application/json" \
  -d @- <<'EOF'
{
  "events": [
    {
      "type": "user.message",
      "content": [
        {"type": "text", "text": "Analyze the performance of the sort function in utils.py"}
      ]
    }
  ]
}
EOF

作業中のエージェントを止めて方向転換させたい場合は、user.interruptのあとに新しいuser.messageを続けて送ります。呼び出しはイベントがキューに入った時点で返り、割り込みのprocessed_atはエージェントが実際に割り込みを適用するまでnullのままです。実行中のツール呼び出しがある場合は割り込みの適用に時間がかかり、その間セッションはrunningのままになります。ターンはstop_reason: end_turnで終わり、割り込み専用のstop reasonはありません。

ストリームを開いてリアルタイムに追う

SSEでイベントをリアルタイムに受け取るには、/events/streamエンドポイントに接続します。ストリームを開いてからイベントを送信する順序を守ってください。ストリーム接続前に発生したイベントは配信されません。

# ストリームを先に開き、その後でメッセージを送る
exec {stream}< <(
  curl --fail-with-body -sS -N \
    "https://api.anthropic.com/v1/sessions/$SESSION_ID/events/stream?beta=true" \
    -H "x-api-key: $ANTHROPIC_API_KEY" \
    -H "anthropic-version: 2023-06-01" \
    -H "anthropic-beta: managed-agents-2026-04-01" \
    -H "content-type: application/json" \
    -H "accept: text/event-stream"
)
 
curl --fail-with-body -sS \
  "https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true" \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -H "anthropic-beta: managed-agents-2026-04-01" \
  -H "content-type: application/json" \
  -d @- >/dev/null <<'EOF'
{
  "events": [
    {"type": "user.message", "content": [{"type": "text", "text": "Summarize the repo README"}]}
  ]
}
EOF
 
while IFS= read -r -u "$stream" event_line; do
  [[ $event_line == data:* ]] || continue
  event_json=${event_line#data: }
  case $(jq -r '.type' <<<"$event_json") in
    agent.message)
      jq -j '.content[] | select(.type == "text") | .text' <<<"$event_json"
      ;;
    session.status_idle)
      break
      ;;
    session.error)
      printf '\n[Error: %s]\n' "$(jq -r '.error.message // "unknown"' <<<"$event_json")"
      break
      ;;
  esac
done
exec {stream}<&-

再接続でイベントを取りこぼさないためには、新しいストリームを開いたあと、イベント履歴を一覧取得して既知のIDセットを作り、ライブストリームをそのセットと突き合わせながらテールします。ストリームは接続直後からバッファリングを始めるため、履歴の取得を先に済ませてもイベントが抜け落ちることはありません。

ツール呼び出しを検知して応答する

カスタムツールを扱うセッションでは、ツール呼び出しの検知から結果送信までが1つのループになります。

1. セッションがagent.custom_tool_useイベントを発行(ツール名と入力を含む)
2. セッションがsession.status_idleイベントで一時停止し、stop_reason: requires_actionを返す。ブロックしているイベントIDはstop_reason.event_ids配列に入る
3. 各イベントIDに対してツールを実行し、custom_tool_use_idにそのイベントIDを指定したuser.custom_tool_resultイベントを送る
4. ブロックしているすべてのイベントが解決されると、セッションはrunningへ戻る

exec {stream_fd}< <(curl --fail-with-body -sS -N \
  "https://api.anthropic.com/v1/sessions/$SESSION_ID/events/stream?beta=true" \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -H "anthropic-beta: managed-agents-2026-04-01" \
  -H "content-type: application/json" \
  -H "accept: text/event-stream")
 
while IFS= read -r -u "$stream_fd" line; do
  [[ $line == data:* ]] || continue
  event_json="${line#data: }"
  stop_reason=$(jq -r 'select(.type == "session.status_idle") | .stop_reason.type // empty' <<<"$event_json")
  case "$stop_reason" in
    requires_action)
      while IFS= read -r event_id; do
        result=$(call_tool "$event_id")
        jq -n --arg id "$event_id" --arg result "$result" \
          '{events: [{type: "user.custom_tool_result", custom_tool_use_id: $id, content: [{type: "text", text: $result}]}]}' |
          curl --fail-with-body -sS \
            "https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true" \
            -H "x-api-key: $ANTHROPIC_API_KEY" \
            -H "anthropic-version: 2023-06-01" \
            -H "anthropic-beta: managed-agents-2026-04-01" \
            -H "content-type: application/json" \
            -d @-
      done < <(jq -r '.stop_reason.event_ids[]' <<<"$event_json")
      ;;
    end_turn)
      break
      ;;
  esac
done
exec {stream_fd}<&-

event deltaでテキストをリアルタイムプレビューする

デフォルトでは、エージェントの応答テキストはagent.messageイベントとしてバッファリングされ、そのモデルリクエストが完了して初めて配信されます。生成中の途中経過をライブプレビューとして描画したい場合は、event deltaを使います。

event deltaはストリーム接続ごとのオプトインです。ストリームURLにevent_deltas[]クエリパラメータを、プレビューしたいイベント種別ごとに繰り返し付けます。受け付けられる値はagent.messageとagent.thinkingの2つだけで、それ以外を指定すると400エラーになります。[]はシェルのグロブパターンなので、URLをシェルで組み立てるときは必ずクオートするか、%5B%5Dにパーセントエンコードしてください。

プレビュー対象のイベントが始まると、まずevent_startが発火し、その後agent.messageならevent_deltaがテキスト断片を運びます。

{
  "type": "event_start",
  "event": {"type": "agent.message", "id": "sevt_01abc..."}
}

{
  "type": "event_delta",
  "event_id": "sevt_01abc...",
  "delta": {
    "type": "content_delta",
    "index": 0,
    "content": {"type": "text", "text": "Here is the summary"}
  }
}

agent.thinkingがプレビューされる場合はevent_startだけが発火し、event_deltaは続きません。プレビューを締めくくるバッファ済みのagent.thinkingイベント自体も、思考内容は運ばず進捗シグナルとしてのみ機能します。

デルタを蓄積して描画する

プレビューは「捨ててよいスクラッチバッファ」、バッファ済みイベントは「正式な記録」として扱います。(event_id, index)をキーにテキストを蓄積する手順は次の4段階です。

1. event_startで告知されたidを記録する。event_start.event.id・各event_delta.event_id・最終的なバッファ済みagent.messageのidはすべて同じ値になる
2. event_deltaごとにdelta.content.textを(event_id, delta.index)のエントリへ追記し、その時点の文字列を描画する
3. バッファ済みagent.messageが届いたらidで照合し、蓄積したプレビューを破棄してメッセージの内容を描画に置き換える
4. span.model_request_endが届いた時点で、まだ照合が済んでいないプレビューを閉じる。それ以上デルタは来ない

Python SDKでの実装例です。

previews: dict[str, BetaManagedAgentsAgentMessageEvent] = {}
 
with client.beta.sessions.events.stream(
    session.id, event_deltas=["agent.message"]
) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Describe the repo in one sentence."}],
            },
        ],
    )
 
    for event in stream:
        match event.type:
            case "event_start":
                snapshot = accumulate_managed_agents_event(None, event)
                if snapshot is not None:
                    previews[event.event.id] = snapshot
            case "event_delta":
                preview = accumulate_managed_agents_event(previews.get(event.event_id), event)
                if preview is not None:
                    previews[event.event_id] = preview
            case "agent.message":
                preview = accumulate_managed_agents_event(previews.pop(event.id, None), event)
            case "span.model_request_end":
                previews.clear()
            case "session.status_idle":
                break

Go・Java・Ruby・C#のSDKヘルパーはイベントのidでプレビューをキー管理するところまで内蔵しています。Python・TypeScript・PHPのヘルパーは、このidごとのマップ自体を呼び出し側で保持する必要があります。

event deltaの限界を把握しておく

event deltaは応答性を優先したベストエフォートの機能で、次の制約があります。

• ベストエフォート: 負荷が高いとサーバーがデルタを間引くことがあり、その場合テキストの連続したプレフィックスまでしか届きません。バッファ済みのagent.messageは必ず完全な形で届くため、蓄積したプレビューを最終結果として扱ってはいけません
• 再接続時のリプレイなし: デルタはオプトインした接続にだけ、開いている間だけ配信されます。切断後は再接続してイベント履歴を取り直す必要があり、取りこぼしたデルタ自体を再要求する方法はありません
• 1スレッド・テキストのみ: プレビューされるのは、その接続が読んでいるスレッド上のアシスタントテキストだけです。ツール呼び出しやツール結果、他のセッションスレッドの活動はプレビュー対象外です
• agent.thinkingはstartのみ: 前述のとおり内容は運びません
• 永続化されない: event_startとevent_deltaはライブストリーム上にしか存在せず、イベント履歴の取得APIには一切現れません

event deltaがうまく動かないときの切り分け

event deltaの実装で期待どおりにイベントが来ないとき、原因は次の3パターンにほぼ絞られます。

いずれも「プレビューが来ない = 壊れている」と早合点しないことが実装の勘所です。プレビューを無視してもバッファ済みイベントだけで完全なストリームが得られる設計なので、まずバッファ済みイベント側が正しく届いているかを切り分けてから、プレビュー固有の設定を疑うと原因特定が早くなります。マルチエージェント構成でサブエージェントのテキストだけプレビューが来ないという相談も、実態はスレッドの取り違えであることがほとんどです。

監視すべきイベントの切り分け

イベントストリームで何を監視するかは目的で変わります。ツール呼び出しの追跡にはagent.tool_use系イベントとステータス遷移、テキストのライブ表示にはevent delta、アウトカム(達成すべき成果物)の進行状況には別系統のspan.outcome_evaluation_*イベントを見る必要があります。アウトカム評価イベントの詳細はManaged Agentsのアウトカム評価イベントを監視する、そのアウトカムに渡すRubric(採点基準)の設計方法はManaged AgentsのRubricでアウトカムを定義する方法で扱っています。

イベントストリームがなぜセッション・ハーネス・サンドボックスの分離アーキテクチャの上に成り立つのかは、Agent SDKのManaged Agentsの設計思想を参照してください。

まとめ

Managed Agentsのイベントストリームは、ストリームを先に開いてから送信する順序と、(event_id, index)でのデルタ蓄積という2つの原則さえ押さえれば実装できます。ツール呼び出しはagent.tool_use系イベントとstop_reason: requires_actionのポーリング的な待ち受けで検知し、テキストのライブ表示が必要な場合だけevent deltaをオプトインする、という切り分けが実装の見通しを良くします。

event deltaはあくまで表示上のベストエフォート機能であり、正式な記録は常にバッファ済みイベント側にあるという前提を崩さないことが、実装を壊れにくくする一番の近道です。プレビューを信じて処理を分岐させるのではなく、プレビューは描画専用、判断はバッファ済みイベント専用と役割を分けておけば、負荷によるデルタの間引きや再接続時の欠落があっても、アプリケーションの動作自体は揺らぎません。

送信前にストリームを開く順序、ツール呼び出しの4段階ループ、event deltaの蓄積と限界。この3つを型として押さえておけば、Managed Agentsのイベントストリームまわりの実装で迷う場面はかなり減らせます。