工作階段事件串流
傳送事件、串流回應,並在執行過程中中斷或重新導向您的工作階段。
與 Claude Managed Agents 的通訊是以事件為基礎的。您開啟一個串流來追蹤工作階段、傳送事件來引導它,並在它暫停等待輸入時回應它。
事件類型
事件以兩個方向流動:您向代理傳送事件,而工作階段則將事件傳回給您。
- 您傳送的事件:
user.*事件會啟動工作階段,並在其進行過程中引導它。system.message事件會附加系統層級的上下文。 - 您接收的事件: 工作階段事件、span 事件和代理事件會回報工作階段的狀態和代理的進度。
這些事件類型字串遵循 {domain}.{action} 命名慣例。如需完整目錄,請參閱參考資料中的事件類型。
Webhook 事件類型是分開的,其中部分名稱與串流中的不同。例如,webhook 使用 session.status_idled 而非 session.status_idle。
串流事件
從工作階段串流事件,以便在代理工作時接收即時更新。串流只會傳遞在其開啟之後發出的事件,因此請在傳送事件之前開啟串流,以避免「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":
# 此連線未啟用 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 | 一個或多個工具呼叫需要您的回應,例如自訂工具呼叫或確認請求。 | 回應每個阻擋中的工具呼叫。 |
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,並附上結果內容。
您可以在 agent.custom_tool_use 事件送達後立即傳送結果,無需等待 session.status_idle。工作階段仍會針對該呼叫發出帶有 requires_action 停止原因的 session.status_idle,您的用戶端可以忽略它。針對同一呼叫的第二個結果會被接受,但不會產生任何效果。
當您重新連線到已暫停的工作階段時,最新 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 時等待您的確認。請不要等待閒置事件:當 agent.tool_use 或 agent.mcp_tool_use 事件送達,且其 evaluated_permission 為 ask 時,就回應它。追蹤執行中的範例以這種方式回應自訂工具呼叫,其簡介也說明了如何加入確認。
恢復閒置的工作階段
工作階段會在互動之間持續存在。若要恢復工作階段,請照常向其傳送 user.message 事件:
# 透過傳送新的 user.message 事件來恢復先前建立的 session。
# 在正式環境中,請傳入您想要恢復之 session 的已儲存 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 事件來重新導向它:
# Agent 目前正在分析檔案...
# 以新的指示中斷:
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 prompt」(系統提示),系統提示是由代理定義上的 system 欄位設定的。
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 中檢查工作階段,並查看其 token 用量和牌價成本。
在重大事件發生時收到通知,無需輪詢。
查詢工作階段可以傳送或接收的每種事件類型。
Was this page helpful?