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つのループになります。

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}<&-

組み込みツールとMCPツールも承認待ちで止まる

agent.tool_useとagent.mcp_tool_useも、権限ポリシーがalways_askのとき、またはautoでサーバーが判定に至らないときは、確認を待って止まります。止まり方はカスタムツールと同じで、session.status_idleのstop_reason.typeがrequires_actionになり、ブロック中のIDがstop_reason.event_idsに入ります。返すイベントだけが違い、user.tool_confirmationにtool_use_idとresult(allowかdeny)を指定します。拒否するときはdeny_messageで理由を添えられます。

各イベントにはevaluated_permission(allow / ask / deny)が付き、確認を待つのはaskのものだけです。

jq -n --arg id "$event_id" \
  '{events: [{type: "user.tool_confirmation", tool_use_id: $id, result: "allow"}]}' |
  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 @-

session.status_idleで読み取りを終えてよいとは限らない

最初のストリーム例はsession.status_idleでbreakします。カスタムツールや承認待ちのあるセッションでは、この最初の停止で読み取りが終わってしまいます。stop_reasonの値で分けると、次の3通りに対応できます。

budget_reachedのときは、session.thread_status_idle、session.usage、session.status_idleの順で届きます。session.usageは累積の使用量と費用のスナップショットです。スレッド側の停止理由がend_turnのままでも、セッション側がbudget_reachedになることがあります。上限に達したセッションが受け付けるのは、進行中の作業を片づけるuser.tool_confirmation・user.tool_result・user.custom_tool_result・user.interruptだけです。user.messageのように新しい作業を始めるイベントは400エラーで拒否されます。再開するイベントは無く、予算を消費済みの費用より大きい値へ更新するか、"budget": nullで外すと、止まっていた作業が自動で動き出します。

ツール待ちのスレッドと上限で止まったスレッドが同居すると、セッション側のstop_reasonはbudget_reachedではなくrequires_actionになります。先にツール側へ応答してください。一時的なエラーはsession.status_rescheduledのあと自動で再試行され、回復できないエラーやアーカイブはsession.status_terminatedで届きます。

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イベント自体も、思考内容は運ばず進捗シグナルとしてのみ機能します。

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

プレビューは「捨ててよいスクラッチバッファ」、バッファ済みイベントは「正式な記録」として扱います。正常に終わるターンでは、モデルへのリクエスト1回ごとにイベントが次の順で届きます。span.model_request_start、event_start、複数のevent_delta、バッファ済みのagent.message、span.model_request_endです。ターンの先頭にはsession.status_runningが1回だけ出ます。(event_id, index)をキーにテキストを蓄積する手順は次の4段階です。

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パターンにほぼ絞られます。

プレビューを無視してもバッファ済みイベントだけで完全なストリームが得られる設計です。プレビューが来ないときは、先にバッファ済みイベントが正しく届いているかを見て、そのあとでプレビュー固有の設定を疑います。

サブエージェントのプレビューは子スレッドのストリームに届く

マルチエージェント構成では、セッションレベルのストリームのプレビューは主スレッドに限られます。サブエージェントのテキストをプレビューするには、そのスレッド専用のストリームを別に開き、同じevent_deltas[]を付けます。event_startとevent_deltaの形は変わらず、蓄積の手順もそのまま使えます。

thread_idはGET /v1/sessions/{session_id}/threadsで引きます。子スレッドはparent_thread_idがnullでなく、主スレッドはnullです。

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

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

Webhookで受け取るイベント名はストリームのものと一部異なります。たとえばストリームのsession.status_idleは、Webhookではsession.status_idledです。

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

まとめ

ストリームを先に開いてから送ることと、プレビューは描画専用で判断はバッファ済みイベントで行うことの2点が、実装を壊れにくくします。ループはsession.status_idleのstop_reasonで分岐させ、テキストのライブ表示が要る場面だけevent deltaをオプトインする構成が、見通しの良い出発点です。