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

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



イベントタイプ

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

• ユーザーイベントとシステムイベントは、あなたがエージェントに送信するものです。user.*イベントはセッションを開始し、進行に応じてセッションを誘導します。system.messageはターン間でエージェントのシステムプロンプトを更新します。
• セッションイベント、スパンイベント、エージェントイベントは、セッションの状態とエージェントの進捗を可視化するためにあなたに送信されます。オプトインしたストリーム接続は、イベントデルタも受信します。

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

永続化されたすべてのイベントには、イベントがサーバー側で記録された時刻を示すprocessed_atタイムスタンプが含まれます。processed_atがnullの場合、そのイベントはハーネスによってキューに入れられており、先行するイベントの処理が完了した後に処理されることを意味します。



イベントの統合



イベントデルタ

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



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

プレビューはストリーム接続ごとにオプトインします。GET /v1/sessions/{session_id}/events/streamにevent_deltas[]クエリパラメータを追加し、プレビューしたいイベントタイプごとに1回ずつ繰り返します。受け入れられる値はagent.messageとagent.thinkingです。それ以外の値は400エラーを返します。このパラメータをサポートするのはセッションレベルのイベントストリームのみです。セッションスレッドのイベントストリームはこれを拒否します。

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

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

agent.messageの場合、開始の後に段階的なテキストを含む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です。



蓄積と照合

Python、TypeScript、Go SDKには、イベントのidでプレビューをキー付けし、indexの管理を代行するアキュムレータヘルパーが含まれています。手動パターンはすべての言語で機能します。他のSDKでは、生成されたイベントタイプにこれを適用してください。

手動パターンでは、プレビューをスクラッチバッファとして扱い、バッファリングされたイベントを記録として扱います。バッファは(event_id, index)でキー付けします。モデルリクエストごとに照合します。ターンは単一のsession.status_runningイベントで開始され、正常に完了するターンでは、各モデルリクエストが順にspan.model_request_start、event_start、event_deltaイベント、バッファリングされたagent.message、最後にspan.model_request_end（スパンイベントタブ内）を生成します。各イベントが到着するたびに処理します。

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は届きます。

# イベント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



制限事項

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

• ベストエフォート： 負荷がかかると、サーバーはイベントのデルタを破棄する場合があります。その場合、テキストの連続したプレフィックスを受信した後、そのイベントに対するデルタはそれ以上届きません。バッファリングされたagent.messageは完全な形で届きます。蓄積されたプレビューを最終的なものとして扱わないでください。
• 再接続時のリプレイなし： デルタは、オプトインした接続が開いている間のみ、その接続に配信されます。ストリームが切断された場合は、「イベントのストリーミング」タブの再接続手順に従ってください。ストリームを再度開き、イベント履歴を一覧表示します。履歴には、切断中に発行されたバッファリングされたイベント（プレビューが待っていたagent.messageを含む）が含まれます。見逃したデルタを再リクエストする方法はありません。
• プライマリスレッド、テキストのみ： プレビューはセッションのプライマリスレッド上のアシスタントテキストをカバーします。ツール使用、ツール結果、MCP結果、および他のセッションスレッド上のアクティビティはプレビューされません。
• 開始のみの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



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

ターン間でエージェントのシステムプロンプトを更新するには、system.messageイベントを送信します。エージェント定義のsystemフィールド（セッション作成時に固定される）とは異なり、system.messageはセッションの進行に応じてシステムプロンプトを変更できます。セッション途中でエージェントに更新されたシステムレベルのガイダンス（異なるペルソナ、改訂された制約、または今後のモデルの動作を形成すべき実行時に取得されたコンテキスト）が必要な場合に使用します。

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

system.messageは、セッションがstop_reason: requires_actionでアイドル状態の間は送信できません。contentは1〜1000個のテキストアイテムを受け入れます。



使用量の追跡

セッションオブジェクトには、累積トークン統計を含むusageフィールドが含まれています。セッションがアイドル状態になった後にセッションを取得して最新の合計を読み取り、コストの追跡、予算の適用、または消費量の監視に使用します。

{
  "id": "sesn_01...",
  "status": "idle",
  "usage": {
    "input_tokens": 5000,
    "output_tokens": 3200,
    "cache_creation_input_tokens": 2000,
    "cache_read_input_tokens": 20000
  }
}

input_tokensはキャッシュされていない入力トークンを報告し、output_tokensはセッション内のすべてのモデル呼び出しにわたる合計出力トークンを報告します。cache_creation_input_tokensおよびcache_read_input_tokensフィールドは、プロンプトキャッシングのアクティビティを反映します。キャッシュエントリは5分のTTLを使用するため、そのウィンドウ内の連続したターンはキャッシュ読み取りの恩恵を受け、トークンあたりのコストが削減されます。



Consoleでの可観測性

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

• セッションリスト： ステータス、作成時刻、モデルを含むすべてのセッション
• トレーシングビュー： セッション内のイベント（コンテンツ、タイムスタンプ、トークン使用量）の時系列ビュー。トレーシングビューはDeveloperとAdminのみがアクセスできます。
• ツール実行： 各ツール呼び出しとその結果の詳細



デバッグのヒント

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