セッションイベントストリーム
イベントを送信し、レスポンスをストリーミングし、実行中のセッションを中断またはリダイレクトします。
Claude Managed Agentsとの通信はイベントベースです。ストリームを開いてセッションを追跡し、イベントを送信してセッションを指示し、セッションが入力待ちで一時停止したときに応答します。
イベントタイプ
イベントは2つの方向に流れます。あなたがエージェントにイベントを送信し、セッションがあなたにイベントを送り返します。
- 送信するイベント：
user.*イベントはセッションを開始し、進行に合わせてセッションを誘導します。system.messageイベントはシステムレベルのコンテキストを追加します。 - 受信するイベント： セッションイベント、スパンイベント、エージェントイベントは、セッションの状態とエージェントの進捗を報告します。
これらのイベントタイプ文字列は {domain}.{action} という命名規則に従います。完全な一覧については、リファレンスのイベントタイプを参照してください。
Webhookイベントタイプは別物であり、一部の名前はストリームのものと異なります。たとえば、Webhookでは session.status_idle ではなく session.status_idled を使用します。
イベントをストリーミングする
セッションからイベントをストリーミングして、エージェントの作業中にリアルタイムの更新を受け取ります。ストリームは開かれた後に発行されたイベントのみを配信するため、「race condition」（競合状態）を避けるために、イベントを送信する前にストリームを開いてください。
# 先にストリームを開いてから、ユーザーメッセージを送信します
with client.beta.sessions.events.stream(session.id) as stream:
client.beta.sessions.events.send(
session.id,
events=[
{
"type": "user.message",
"content": [{"type": "text", "text": "Summarize the repo README"}],
},
],
)
for event in stream:
match event.type:
case "agent.message":
for block in event.content:
if block.type == "text":
print(block.text, end="")
case "session.status_idle":
break
case "session.error":
error_message = event.error.message if event.error else "unknown"
print(f"\n[Error: {error_message}]")
breakセッションは session.error イベントを通じてエラーを報告します。そのフィールドについてはイベントタイプを参照してください。
エージェントのレスポンステキストはバッファリングされた agent.message イベントとして届き、各イベントはそれを生成したモデルリクエストが完了した後にのみ発行されます。モデルがまだ生成している間にテキストをレンダリングするには、イベントデルタでレスポンスをプレビューするを参照してください。
イベントを見逃さずに再接続する
イベントを見逃さずに既存のセッションに再接続するには、新しいストリームとイベント履歴を組み合わせます。
- 新しいストリームを開きます。
- 完全なイベント履歴を一覧表示して、確認済みイベントIDのセットを初期化します。
- ライブストリームを追跡し、履歴一覧ですでに返されたイベントはスキップします。
with client.beta.sessions.events.stream(session.id) as stream:
# ストリームは開いてバッファリング中です。ライブ追跡の前に履歴を一覧取得します。
history = client.beta.sessions.events.list(session.id)
seen_event_ids = {past_event.id for past_event in history}
# ライブイベントを追跡し、既に確認済みのものはスキップします
for event in stream:
if event.type == "event_start" or event.type == "event_delta":
# この接続ではデルタプレビューが有効になっていません。
continue
if event.id in seen_event_ids:
continue
seen_event_ids.add(event.id)
match event.type:
case "agent.message":
for block in event.content:
if block.type == "text":
print(block.text, end="")
case "session.status_idle":
breakイベントを送信する
user.message イベントを送信して、エージェントの作業を開始または継続します。
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",
},
],
},
],
)セッション履歴内のすべてのイベントには processed_at タイムスタンプが含まれ、イベントの処理が完了した時点で設定されます。送信したイベントでは、イベントがまだ先行するイベントの後ろでキューに入っている間、processed_at は None です。例外は user.define_outcome、user.custom_tool_result、user.tool_result で、これらは受信時に処理され、processed_at がすでに設定された状態でエコーバックされます。
セッションがアイドルになったときに応答する
session.status_idle イベントは、エージェントが停止して入力を待っていることを意味します。その stop_reason.type が理由を示します。
stop_reason.type | セッションが停止した理由 | 対処方法 |
|---|---|---|
end_turn | エージェントがターンを終了したか、あなたが中断しました。 | エージェントに追加の作業がある場合はuser.message を送信します。 |
requires_action | カスタムツール呼び出しや確認リクエストなど、1つ以上のツール呼び出しにあなたからの応答が必要です。 | ブロックしている各ツール呼び出しに応答します。 |
budget_reached | セッションの追跡された定価コストが予算に達しました。 | 予算を変更または削除します。 |
予算で一時停止したセッションを、送信するイベントで再開することはできません。予算によって一時停止された作業は、予算を消費済みの定価コストを上回る値に変更するか、予算を削除すると自動的に再開されます。作業が再開されると、セッションは予算によって一時停止されたワークフロー実行ごとに workflow_run.status_running イベントを発行します。中断によって一時停止された実行は一時停止したままです。予算に達したセッションを再開するを参照してください。一時停止を示すイベントと、セッションが引き続き受け付けるイベントについては、セッションが予算に達したときを参照してください。
セッションを一時停止させるツール呼び出しに応答する
セッションは、エージェントがカスタムツールを呼び出したとき、および権限ポリシーのもとでツール呼び出しにあなたの確認が必要なときに一時停止します。どちらの一時停止も同じ手順に従います。
- セッションはツール呼び出しを
agent.custom_tool_use、agent.tool_use、またはagent.mcp_tool_useイベントとして発行します。 - セッションは、
stop_reason.typeがrequires_actionであるsession.status_idleイベントとともに一時停止します。ブロックしているイベントIDはstop_reason.event_ids配列に含まれます。 - ブロックしている各イベントIDについて、
user.custom_tool_resultまたはuser.tool_confirmationイベントを送信します。カスタムツールの結果は手順2を待つ必要はありません。 - ブロックしているすべてのイベントが解決されると、セッションは
runningに戻ります。
マルチエージェントセッションでは、サブエージェントのブロックしているイベントはプライマリスレッドにもクロスポストされます。ツール権限とカスタムツールを参照してください。
カスタムツールの結果を返す
agent.custom_tool_use イベントにはツール名と入力が含まれます。あなたのシステムでツールを実行します。次に、user.custom_tool_result イベントを送信し、custom_tool_use_id パラメータにイベントIDを、結果のコンテンツとともに渡します。
session.status_idle を待たずに、agent.custom_tool_use イベントが届いた時点ですぐに結果を送信できます。セッションはその呼び出しに対して requires_action の停止理由を持つ session.status_idle を引き続き発行しますが、クライアントはそれを無視できます。同じ呼び出しに対する2回目の結果は受け付けられますが、効果はありません。
一時停止したセッションに再接続した場合、最新の session.status_idle イベントの stop_reason.event_ids に応答すべき呼び出しが一覧表示されます。
次の例では、各呼び出しが届くたびに応答します。
with client.beta.sessions.events.stream(session.id) as stream:
for event in stream:
match event.type:
case "agent.custom_tool_use":
# ツールを実行します
result = call_tool(event.name, 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 "session.status_idle":
if event.stop_reason and event.stop_reason.type == "end_turn":
breakエージェントに開いているワークフロー実行がある場合、セッションが running のままでもカスタムツール呼び出しが届くことがあります。requires_action のアイドルイベントは、セッションのスレッドがどれも作業していないときにのみ届きます。それを待たずに、各呼び出しの agent.custom_tool_use イベントが届いたときに応答してください。その方法は実行を追跡するのサンプルで示されています。
ツール呼び出しを確認する
user.tool_confirmation イベントを送信し、tool_use_id パラメータにイベントIDを渡します。result を "allow" または "deny" に設定します。どの呼び出しが確認を待つか、および拒否の理由を説明する方法については、確認リクエストに応答するを参照してください。
次の例では、保留中のすべての呼び出しを承認します。
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前の例では、セッションがアイドルになった後に各呼び出しを承認します。エージェントに開いているワークフロー実行がある場合、セッションが running のままでもツール呼び出しがあなたの確認を待つことがあります。アイドルイベントを待たずに、evaluated_permission が ask である agent.tool_use または agent.mcp_tool_use イベントが届いたときに応答してください。実行を追跡するのサンプルはこの方法でカスタムツール呼び出しに応答しており、その導入部で確認を追加する方法を説明しています。
アイドル状態のセッションを再開する
セッションはインタラクション間で保持されます。セッションを再開するには、通常どおり user.message イベントを送信します。
# 新しい user.message イベントを送信して、以前に作成したセッションを再開します。
# 本番環境では、再開したいセッションの保存済み ID を渡してください。
client.beta.sessions.events.send(
session.id,
events=[
{
"type": "user.message",
"content": [
{
"type": "text",
"text": "Now run the tests against the changes you made earlier.",
},
],
},
],
)会話履歴は、セッションが明示的に削除されない限り保持されます。セッションがアイドルになると、そのサンドボックスはチェックポイントされます。チェックポイントは、ファイルシステム、インストール済みパッケージ、エージェントが作成したファイルを含む、サンドボックスの完全な状態を保持します。
エージェントを中断する
user.interrupt イベントを送信して実行中のエージェントを停止し、続けて user.message イベントを送信してリダイレクトします。
# エージェントは現在ファイルを分析中...
# 新しい指示で中断:
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.",
},
],
},
],
)呼び出しはイベントがキューに入るとすぐに返ります。その後、中断は次の順序で有効になります。
- エージェントが中断を適用します。それまでの間、中断の
processed_atはNoneのままで、セッションはrunningのままです。進行中のモデルレスポンスは即座に停止しますが、ツール呼び出しの実行中は中断の適用に時間がかかることがあります。 user.interruptイベントがストリームに現れ、中断されたターンはsession.status_idleイベントで終了します。- エージェントは、中断の後に送信した
user.messageで次のターンを開始します。
アイドルイベントの stop_reason.type は end_turn で、自然に終了したターンと同じ値です。ワークフロー実行が開いている場合、中断によってそれらが終了することはありません。実行中の実行があるとセッションが running のままになることがあるため、手順2の session.status_idle イベントが届かない場合があります。実行が開いているセッションを中断するを参照してください。
過去のイベントを一覧表示する
セッションの完全なイベント履歴を取得します。
events = client.beta.sessions.events.list(session.id)
for event in events.data:
print(f"[{event.type}] {event.processed_at}")特定のイベントタイプのみを返すには、types フィルターを渡します。
events = client.beta.sessions.events.list(
session.id,
types=["agent.tool_use", "agent.tool_result"],
)
for event in events.data:
print(f"[{event.type}] {event.processed_at}")システムメッセージを送信する
サポートされているモデルでは、system.message イベントを送信して、エージェントに特権的なシステムレベルのコンテキストを与えることができます。このコンテキストは、付随するターンとそれ以降のすべてのターンに適用されます。異なるペルソナ、改訂された制約、実行時に取得したコンテキストなど、セッションの途中でエージェントに更新されたシステムレベルのガイダンスが必要な場合に使用します。
イベントの content は1〜1000個のテキスト項目を受け付けます。コンテンツは role: "system" ターンとしてセッションのシステムコンテキストに追加されます。これは、エージェント定義の system フィールドで設定されるトップレベルの「system prompt」（システムプロンプト）を置き換えるものではありません。
client.beta.sessions.events.send(
session.id,
events=[
{
"type": "system.message",
"content": [
{
"type": "text",
"text": "The user's current timezone is America/New_York.",
},
],
},
],
)サポートされているモデル
system.message は次のモデルでサポートされています。
- Claude Fable 5.1
- Claude Mythos 5.1
- Claude Fable 5
- Claude Mythos 5
- Claude Opus 5.5
- Claude Opus 5
- Claude Opus 4.8
- Claude Sonnet 5.5
- Claude Haiku 5.5
エージェントのプライマリモデルが会話途中のシステム挿入をサポートしていない場合、イベントは model_does_not_support_mid_conversation_system 検証エラーで拒否されます。system.message はプライマリスレッドにのみ届くため、サブエージェントのモデルはチェックされません。
ツール呼び出しが保留中のシステムメッセージ
セッションが requires_action の停止理由でアイドル状態の間、system.message は同じリクエスト内でツール結果イベントの後に続ける必要があります。単独で、または user.message とともに送信した場合、保留中のツールイベントが解決されるまで拒否されます。
次のステップ
モデルがまだ生成している間に、エージェントのレスポンステキストをレンダリングします。
Claude Consoleでセッションを検査し、そのトークン使用量と定価コストを確認します。
ポーリングせずに、主要なイベントが発生したときに通知を受け取ります。
セッションが送受信できるすべてのイベントタイプを確認します。
Was this page helpful?