セッションイベントストリーム

Claude Managed Agentsとの通信はイベントベースです。ユーザーイベントをエージェントに送信し、ステータスを追跡するためにエージェントイベントとセッションイベントを受け取ります。



イベントタイプ

イベントは2つの方向に流れます。

• ユーザーイベントとシステムイベントは、エージェントに送信するものです。user.* イベントはセッションを開始し、進行に応じてセッションを操縦します。system.message は、付随するターンとそれ以降のすべてのターンに適用されるシステムレベルのコンテキストを追加します。
• セッションイベント、スパンイベント、エージェントイベントは、セッションの状態とエージェントの進捗を可観測にするために送信されます。オプトインしたストリーム接続は、イベントデルタも受け取ります。

セッション、スパン、エージェント、ユーザー、システムの各イベントタイプ文字列は、{domain}.{action} という命名規則に従います。ストリーム専用のデルタプレビューイベント（event_start、event_delta）は例外です。完全なカタログについては、リファレンスのイベントタイプを参照してください。

永続化されるすべてのイベントには、イベントの処理が完了したときに設定される processed_at タイムスタンプが含まれます。送信したイベントでは、そのイベントが先行するイベントの後ろでまだキューに入っている間、processed_at はnullです。例外は user.define_outcome、user.custom_tool_result、user.tool_result で、これらは受信時に処理され、processed_at がすでに設定された状態でエコーバックされます。



イベントの統合

client.beta.sessions.events.send(
    session.id,
    events=[
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": "Analyze the performance of the sort function in utils.py",
                },
            ],
        },
    ],
)

# エージェントは現在ファイルを分析中...
# 新しい指示で中断:
client.beta.sessions.events.send(
    session.id,
    events=[
        {"type": "user.interrupt"},
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": "Instead, focus on fixing the bug in line 42.",
                },
            ],
        },
    ],
)



イベントデルタ

デフォルトでは、エージェントのレスポンステキストはバッファリングされた agent.message イベントとしてストリームに届き、それぞれはそれを生成したモデルリクエストが完了した後にのみ発行されます。「event deltas」（イベントデルタ）を使用すると、モデルがまだ生成している間に、そのテキストをライブプレビューとして段階的にレンダリングできます。プレビューはレスポンスではありません。プレビューはベストエフォートの表示補助であり、バッファリングされた agent.message が常に正式な記録です。プレビューを無視するクライアントでも、完全で正しいストリームを受け取ります。



プレビューへのオプトイン

プレビューはストリーム接続ごとのオプトインです。読み取るストリームに event_deltas[] クエリパラメータを追加し、プレビューしたいイベントタイプごとに1回ずつ繰り返します。[] はシェルのグロブパターンであるため、シェルでリクエストを組み立てる場合は必ずURLを引用符で囲んでください。例では角括弧を %5B%5D としてパーセントエンコードしており、これも機能します。両方のストリームエンドポイントがこのパラメータを受け付けます。セッションレベルのストリームである GET /v1/sessions/{session_id}/events/stream と、各セッションスレッド固有のストリームである GET /v1/sessions/{session_id}/threads/{thread_id}/stream です。受け付けられる値は agent.message と agent.thinking です。それ以外の値は400エラーを返し、100個を超える値を含むリクエストも同様です。サブエージェントのプレビューは、そのサブエージェント自身のスレッドストリームに表示されます。

プレビュー対象のイベントが始まると、ストリームは、これから来るイベントのタイプと id を含む event_start を発行します。

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

agent.message の場合、startの後に増分テキストを含む event_delta イベントが続きます。各デルタは、拡張対象のイベントを event_id で、拡張対象のコンテンツブロックを delta.index で指定します。

{
  "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_start と event_delta には独自の id や processed_at がありません。これらが持つ唯一の識別子は、プレビュー対象のイベントの id です。



蓄積と照合

イベントデルタをサポートするすべてのSDKには、index の管理を代行するアキュムレータヘルパーが含まれています。Go、Java、Ruby、C#のヘルパーは、蓄積中のプレビューをイベントの id でもキー付けします。Python、TypeScript、PHPのヘルパーでは、そのマップを自分で保持し、各デルタをその id のエントリに畳み込みます。カスタムの管理が必要な場合は、手動パターンもすべての言語で機能します。生成されたイベントタイプに適用してください。

手動パターンでは、プレビューをスクラッチバッファとして、バッファリングされたイベントを記録として扱います。バッファは (event_id, index) でキー付けします。モデルリクエストごとに照合します。ターンは単一の session.status_running イベントで始まり、正常に完了するターンでは、各モデルリクエストが順に span.model_request_start、event_start、event_delta イベント群、バッファリングされた agent.message、そして最後に span.model_request_end（スパンイベントタブ内）を生成します。ワイヤー上では、これはそのシーケンスのプレビュー部分であり、接続の他のバッファリングされたイベントと交互に配置されます。

event_start     {"event": {"type": "agent.message", "id": "sevt_01abc..."}}
event_delta     {"event_id": "sevt_01abc...", "delta": {"type": "content_delta", "index": 0, "content": {"type": "text", "text": "..."}}}
...
agent.message   {"id": "sevt_01abc...", "content": [...]}

event_delta 行はテキストフラグメントごとに1回繰り返されます。各イベントを到着順に処理します。

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) のエントリに追加し、累積テキストをレンダリングします。ある index に対する最初のデルタがそのエントリを作成します。
3. バッファリングされた agent.message が到着したら、id で照合し、蓄積したプレビューを破棄して、代わりにメッセージのコンテンツをレンダリングします。
4. span.model_request_end で、バッファリングされたイベントによって照合されていないプレビューをすべて閉じます。それに対するデルタはもう来ません。ターンがエラーになったり中断されたりした場合、バッファリングされたイベントは到着しない可能性がありますが、span.model_request_end は到着します。

このパターンが依拠する保証は次のとおりです。

• プレビューのデルタを (event_id, index) でキー付けして到着順に連結すると、バッファリングされたイベントの content[index].text のプレフィックスが得られます（負荷時にデルタが破棄される可能性があるため、必ずしもテキスト全体ではなくプレフィックスです）。
• 1つの接続は event_id ごとに最大1つの event_start を発行し、バッファリングされたイベントはその接続がその id に対して配信する最後のものです。

# イベントIDをキーとするプレビュースナップショット。accumulate_managed_agents_event は各
# event_start / event_delta を agent.message スナップショットに畳み込みます。バッファ済みの
# agent.message がそれを置き換えます。
previews: dict[str, BetaManagedAgentsAgentMessageEvent] = {}

# この接続で agent.message プレビューをオプトインします
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
                print(f"event_start             {event.event.type} {event.event.id}")
            case "event_delta":
                preview = accumulate_managed_agents_event(previews.get(event.event_id), event)
                if preview is not None:
                    previews[event.event_id] = preview
                    text = "".join(block.text for block in preview.content)
                    print(f"event_delta             preview: {text!r}")
            case "agent.message":
                # バッファ済みイベントが正式な記録です。プレビューを置き換えて閉じます
                preview = accumulate_managed_agents_event(previews.pop(event.id, None), event)
                text = "".join(block.text for block in preview.content)
                print(f"agent.message           {event.id} {text!r}")
            case "span.model_request_end":
                # これ以上デルタは届きません。対応するバッファ済みイベントが
                # 届かなかったプレビューをすべて閉じます。
                for event_id in previews:
                    print(f"span.model_request_end  closing preview for {event_id}")
                previews.clear()
            case "session.status_idle":
                break



セッションスレッドイベントのプレビュー

マルチエージェントセッションでは、すべてのセッションスレッドが GET /v1/sessions/{session_id}/threads/{thread_id}/stream に独自のイベントストリームを持ち、同じ値を持つ同じ event_deltas[] パラメータを受け付けます。プレビューは設計上スレッドスコープです。接続は読み取っているスレッドのみをプレビューします。子スレッドのプレビューはその子自身のストリームで配信され、セッションレベルのストリームにクロスポストされることはありません。セッションレベルのストリームのプレビューはプライマリスレッドにスコープされたままです。モデルが生成するサブエージェントのテキストを監視するには、そのサブエージェントのスレッドストリームを開いてください。

スレッドストリームのパスは間違えやすいものです。/events/stream（これはセッションレベルにのみ存在します）ではなく /threads/{thread_id}/stream であり、/threads/{thread_id}/events/stream エンドポイントは存在しません。

プレビューイベント自体は変わりません。event_start と event_delta は、スレッドストリーム上でもセッションレベルのストリーム上と同じ形状を持ち、蓄積と照合のパターンは記載どおりに適用されます。唯一の調整点は管理方法です。ストリーム接続ごとに1つのアキュムレータインスタンスを実行してください。

# セッションのスレッドを一覧表示して子を選択します。子スレッドは非 null の
# parent_thread_id を持ち、プライマリスレッドの parent_thread_id は null です。
THREAD_ID=$(
  curl --fail-with-body -sS \
    "https://api.anthropic.com/v1/sessions/$SESSION_ID/threads?beta=true" \
    -H "x-api-key: $ANTHROPIC_API_KEY" \
    -H "anthropic-version: 2023-06-01" \
    -H "anthropic-beta: managed-agents-2026-04-01" |
    jq -er 'first(.data[] | select(.parent_thread_id != null)).id'
)

# 子スレッドのストリームは、セッションストリームと同じ event_deltas[] パラメータを
# 受け取ります。角括弧をパーセントエンコード（%5B%5D）し、URLを引用符で囲みます。
exec {stream}< <(
  curl --fail-with-body -sS -N \
    "https://api.anthropic.com/v1/sessions/$SESSION_ID/threads/$THREAD_ID/stream?beta=true&event_deltas%5B%5D=agent.message" \
    -H "x-api-key: $ANTHROPIC_API_KEY" \
    -H "anthropic-version: 2023-06-01" \
    -H "anthropic-beta: managed-agents-2026-04-01" \
    -H "accept: text/event-stream"
)

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
    event_delta)
      jq -j '.delta.content.text' <<<"$event_json"
      ;;
    agent.message)
      # バッファリングされたイベントが正式な記録です。その内容をレンダリングします。
      printf '\n'
      jq -j '.content[] | select(.type == "text") | .text' <<<"$event_json"
      printf '\n'
      ;;
    session.thread_status_idle)
      break
      ;;
  esac
done
exec {stream}<&-

読み取りループは session.thread_status_idle で終了します。これは、セッションスレッドのターンが終了してスレッドがアイドルになったときに発行されるイベントです。



制限事項

プレビューは応答性を重視して調整されています。次の制約を前提に構築してください。

• ベストエフォート: 負荷時には、サーバーがあるイベントのデルタを破棄する可能性があります。その場合、テキストの連続したプレフィックスを受け取り、その後そのイベントに対するデルタは届かなくなります。バッファリングされた agent.message は引き続き完全な形で到着します。蓄積したプレビューを最終的なものとして扱わないでください。
• 再接続時のリプレイなし: デルタは、オプトインした接続が開いている間、その接続にのみ配信されます。これはセッションレベルのストリームにも各セッションスレッドストリームにも同様に適用され、モデルリクエストの開始後に開かれた接続は、その進行中のイベントに対するデルタを受け取りません。ストリームが切断された場合は、イベントのストリーミングタブの再接続手順に従ってください。ストリームを再度開き、イベント履歴をリストします。履歴には、切断中に発行されたバッファリングされたイベントがすべて含まれ、プレビューが待っていた agent.message も含まれます。取りこぼしたデルタを再リクエストする方法はありません。
• 1スレッド、テキストのみ: プレビューは、接続が読み取っているスレッド上のアシスタントテキストを対象とします。ツール使用、ツール結果、MCP結果、および他のセッションスレッド上のアクティビティは、その接続でプレビューされることはありません。
• startのみの agent.thinking: agent.thinking のプレビューは、思考ブロックが開始したことを示すシグナルとして event_start のみを発行します。その後に event_delta イベントは続きません。
• 永続化されない: event_start と event_delta はライブストリーム上にのみ存在します。セッションのイベント履歴（GET /v1/sessions/{session_id}/events）や、いずれのセッションスレッドのイベント履歴にも表示されません。



プレビューのトラブルシューティング

ストリームが期待どおりに動作しない場合は、次を確認してください。



その他のシナリオ



カスタムツール呼び出しの処理

エージェントがカスタムツールを呼び出すと、次のようになります。

1. セッションは、ツール名と入力を含む agent.custom_tool_use イベントを発行します。
2. セッションは、stop_reason: requires_action を含む session.status_idle イベントで一時停止します。ブロックしているイベントIDは stop_reason.event_ids 配列にあります。
3. システムでツールを実行し、それぞれに対して user.custom_tool_result イベントを送信します。その際、custom_tool_use_id パラメータにイベントIDを、結果コンテンツとともに渡します。
4. ブロックしているすべてのイベントが解決されると、セッションは running に戻ります。

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        if event.type == "session.status_idle" and (stop_reason := event.stop_reason):
            match stop_reason.type:
                case "requires_action":
                    for event_id in stop_reason.event_ids:
                        # カスタムツール使用イベントを検索して実行
                        tool_event = events_by_id[event_id]
                        result = call_tool(tool_event.name, tool_event.input)

                        # 結果を送り返す
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.custom_tool_result",
                                    "custom_tool_use_id": event_id,
                                    "content": [{"type": "text", "text": result}],
                                },
                            ],
                        )
                case "end_turn":
                    break



ツールの確認

権限ポリシーがツールの実行前に確認を要求する場合、次のようになります。

1. セッションは agent.tool_use または agent.mcp_tool_use イベントを発行します。
2. セッションは、stop_reason: requires_action を含む session.status_idle イベントで一時停止します。ブロックしているイベントIDは stop_reason.event_ids 配列にあります。
3. それぞれに対して user.tool_confirmation イベントを送信し、tool_use_id パラメータにイベントIDを渡します。result を "allow" または "deny" に設定します。拒否の理由を説明するには deny_message を使用します。
4. ブロックしているすべてのイベントが解決されると、セッションは running に戻ります。

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        if event.type == "session.status_idle" and (stop_reason := event.stop_reason):
            match stop_reason.type:
                case "requires_action":
                    for event_id in stop_reason.event_ids:
                        # 保留中のツール呼び出しを承認
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.tool_confirmation",
                                    "tool_use_id": event_id,
                                    "result": "allow",
                                },
                            ],
                        )
                case "end_turn":
                    break



アイドルセッションの再開

セッションはインタラクション間で永続化されます。会話履歴は、セッションが明示的に削除されない限り保持されます。セッションがアイドルになると、そのサンドボックスはチェックポイント化され、ファイルシステム、インストール済みパッケージ、エージェントが作成したファイルを含むサンドボックスの完全な状態が保持されます。これにより、非アクティブ状態からクリーンに再開できます。

セッションを再開するには、通常どおり user.message イベントを送信します。

# 本番環境では、再開したいセッションの保存済み ID を渡してください。
ant beta:sessions:events send --session-id "$SESSION_ID" <<'YAML'
events:
  - type: user.message
    content:
      - type: text
        text: Now run the tests against the changes you made earlier.
YAML



セッション予算への到達

予算付きで作成されたセッションは、超過支出する代わりに一時停止します。セッションの追跡されたリストコストが上限に達すると、プラットフォームは各スレッドを次のモデルリクエストの前に一時停止し、セッションは終了するのではなく stop_reason が budget_reached のアイドル状態になります。合計を上限を超えさせたリクエストは完了まで実行されるため、session.usage スナップショットで報告される list_cost は上限ちょうど、または上限をわずかに超えた値になることがあります。ストリーム上では、一時停止は次の3つのイベントとして順に到着します。

1. 各スレッドが一時停止するたびに、stop_reason: budget_reached を含む session.thread_status_idle。
2. セッションの累積使用量と追跡されたリストコストのスナップショットである session.usage。
3. stop_reason: budget_reached を含む session.status_idle。session.usage イベントは常にこのアイドルの直前に来ます。

最後のリクエストが上限を超えると同時にターンを完了したスレッドは、自身の session.thread_status_idle イベントで end_turn を報告しますが、セッションは引き続き budget_reached を報告します。一時停止を検出するには、セッションレベルの stop_reason をキーにしてください。

セッションが上限に達している間は、すでに進行中の作業を決着させるイベントのみを受け付けます。user.tool_confirmation、user.tool_result、user.custom_tool_result、user.interrupt です。user.message を含め、新しい作業を開始するイベントはすべて、そのリストを示す400エラーで拒否されます。セッションに、ツールの問い合わせを待っているスレッドと上限で一時停止しているスレッドの両方がある場合、セッションレベルの stop_reason は budget_reached ではなく requires_action です。問い合わせを決着させてもモデルリクエストはトリガーされないため、通常どおり応答してください。

上限で一時停止したセッションを再開するイベントはありません。代わりに、セッションの予算を更新します。上限を消費済みリストコストより大きい任意の値に変更するか、"budget": null でセッションを更新して予算を削除すると、一時停止した作業が自動的に再開されます。リストコストの追跡方法と予算更新の完全なセマンティクスについては、セッション予算を参照してください。



システムメッセージの送信

system.message イベントを送信すると、付随するターンとそれ以降のすべてのターンに適用される、特権的なシステムレベルのコンテキストをエージェントに与えることができます。エージェント定義の system フィールド（トップレベルのシステムプロンプトを設定するもの）とは異なり、system.message のコンテンツはそのプロンプトを置き換えるのではなく、role: "system" のターンとしてセッションのシステムコンテキストに追加されます。セッションの途中でエージェントに更新されたシステムレベルのガイダンスが必要な場合に使用してください。たとえば、異なるペルソナ、改訂された制約、または今後のモデルの動作を形作るべき実行時に取得したコンテキストなどです。

ant beta:sessions:events send --session-id "$SESSION_ID" <<'YAML'
events:
  - type: system.message
    content:
      - type: text
        text: "The user's current timezone is America/New_York."
YAML

セッションが stop_reason: requires_action でアイドル状態の間、system.message は同じリクエスト内でツール結果イベントの後に続く場合にのみ受け付けられます。単独で、または user.message とともに送信された場合は、保留中のツールイベントが解決されるまで拒否されます。content は1〜1000個のテキスト項目を受け付けます。



使用量の追跡

セッションオブジェクトには、セッションの累積使用量を示す usage フィールドが含まれています。トークン数、サーバーツール使用、アクティブ時間、および追跡された定価コストです。最新の合計値を読み取るには、セッションがアイドル状態になった後にセッションを取得してください。

{
  "id": "sesn_01...",
  "status": "idle",
  "usage": {
    "input_tokens": 5000,
    "output_tokens": 3200,
    "cache_read_input_tokens": 20000,
    "cache_creation": {
      "ephemeral_5m_input_tokens": 2000,
      "ephemeral_1h_input_tokens": 0
    },
    "list_cost": {
      "amount": "187",
      "currency": "USD"
    },
    "active_seconds": 342.5,
    "server_tool_use": {
      "web_search_requests": 3,
      "web_fetch_requests": 0
    }
  }
}

input_tokens はキャッシュされていない入力トークンを報告し、output_tokens はセッション内のすべてのモデル呼び出しにわたる出力トークンの合計を報告します。cache_read_input_tokens フィールドはプロンプトキャッシュから読み取られたトークンを報告し、cache_creation オブジェクトはキャッシュ作成トークンをキャッシュの有効期間別（ephemeral_5m_input_tokens と ephemeral_1h_input_tokens）に分類します。キャッシュエントリはデフォルトで5分間のTTLを使用するため、その時間枠内で連続して行われるターンはキャッシュ読み取りの恩恵を受け、トークンあたりのコストが削減されます。

list_cost は、公開されている定価レートで価格設定されたセッションの累積消費量であり、文字列形式のセント単位の整数と通貨コードで表されます。active_seconds は、セッションで少なくとも1つのスレッドが実行されていた累積時間です。同時実行スレッドによる重複するアクティビティは1回だけカウントされます。これは、各スレッド自身のアクティブ時間を合計するセッションの stats オブジェクト内の active_seconds とは異なります。この重複排除された数値が、セッションのランタイムコストの価格設定の基準となる期間です。server_tool_use は、価格設定のためにサーバーで実行されたツールリクエストをカウントします。ウェブ検索リクエストはリクエストごとに定価コストに計上され、ウェブフェッチリクエストにはリクエストごとの料金がなく計測もされないため、web_fetch_requests は 0 と表示されます。各セッションスレッド自身の usage にも list_cost と active_seconds が含まれます。スレッドごとの数値は個別に丸められ、セッションの実行時間コストを含まないため、合計してもセッションの list_cost と正確には一致しません。セッションの数値が正式な値です。

これらの合計値を確認するためにセッションをポーリングする必要はありません。session.usage イベントは、同じ累積スナップショット（usage オブジェクトに加えて、セッションの budget。セッションに予算がない場合は null）をセッションストリームおよびイベント履歴で伝達します。このイベントはタイマーではなくアイドル状態への遷移時に発行されます。セッションは、停止理由が何であれアイドル状態になる直前に1回、そしてスレッドがセッション予算で一時停止したときに1回発行します。したがって、ストリームの読み取り側は、追加の取得を行うことなく、ターンの最終コスト、または予算に達した作業の最終コストを確認できます。

支出上限を適用するには、使用量をポーリングして自分でセッションを停止するのではなく、セッション予算を設定してください。プラットフォームはセッションの消費量を継続的に価格計算し、セッションの定価コストが上限に達すると、各スレッドを次のモデルリクエストの前に一時停止します。ストリーム上でどのように見えるかについては、セッション予算への到達を参照してください。



Consoleでのオブザーバビリティ

Claude Consoleは、エージェントセッションの視覚的なタイムラインビューを提供します。ConsoleのClaude Managed Agentsセクションに移動すると、以下を確認できます。

• セッション一覧： すべてのセッションとそのステータス、作成時刻、エージェント
• トレーシングビュー： セッション内のイベント（コンテンツ、タイムスタンプ、トークン使用量）の時系列ビュー。トレーシングビューは開発者と管理者のみがアクセスできます。
• ツール実行： 各ツール呼び出しとその結果の詳細



デバッグのヒント

• セッションイベントを確認する： セッションエラーは session.error イベントを通じて伝達されます
• ツール結果を確認する： ツール実行の失敗は、予期しないエージェントの動作の原因を説明することがよくあります
• トークン使用量を追跡する： トークン消費量を監視してプロンプトを最適化し、コストを削減します
• システムプロンプトを使用する： システムプロンプトにログ記録の指示を追加して、エージェントに推論を説明させます
• プレビューのトラブルシューティング： イベントデルタをオプトインしたストリームが期待どおりに動作しない場合は、プレビューのトラブルシューティングを参照してください