> ## Documentation Index
> Fetch the complete documentation index at: https://oma-codex-339-workspace-permissions.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
# 会话事件流
> 发送事件、流式传输响应，并在执行过程中中断或重定向您的会话。
与 Open Managed Agents的通信是基于事件的。您向智能体发送用户事件，并接收返回的智能体事件和会话事件以跟踪状态。
托管智能体 API 请求需要 `managed-agents-2026-04-01` Beta 请求头，但记忆存储端点除外，它们使用 `agent-memory-2026-07-22`。SDK 会自动设置正确的 Beta 请求头。请参阅[Beta 请求头](/docs/zh/api/versioning-beta)。
## 事件类型
事件以两个方向流动。
* **用户事件**和**系统事件**是您发送给智能体的内容：`user.*` 事件用于启动会话并在会话进行过程中对其进行引导；`system.message` 用于追加系统级上下文，该上下文适用于随附的轮次及所有后续轮次。
* **会话事件**、**跨度事件**和**智能体事件**会发送给您，以便您观察会话状态和智能体进度。选择加入的流连接还会接收[事件增量](/docs/zh/events-and-streaming#event-deltas)。
会话、跨度、智能体、用户和系统事件类型字符串遵循 `{domain}.{action}` 命名约定。仅限流的增量预览事件（`event_start`、`event_delta`）是例外。请参阅参考文档中的[事件类型](/docs/zh/reference#event-types)以获取完整目录。
每个持久化事件都包含一个 `processed_at` 时间戳，该时间戳在事件完成处理时设置。对于您发送的事件，当事件仍在排队等待处理先前事件时，`processed_at` 为 null。例外情况是 `user.define_outcome`、`user.custom_tool_result` 和 `user.tool_result`，它们在接收时即被处理，并在回显时已填充 `processed_at`。
## 集成事件
发送 `user.message` 事件以启动或继续智能体的工作：
```bash cURL theme={null}
curl --fail-with-body -sS "http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_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
```
```bash CLI theme={null}
ant beta:sessions:events send --session-id "$SESSION_ID" <<'YAML'
events:
- type: user.message
content:
- type: text
text: Analyze the performance of the sort function in utils.py
YAML
```
```python Python theme={null}
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",
},
],
},
],
)
```
```typescript TypeScript theme={null}
await 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",
},
],
},
],
});
```
```csharp C# theme={null}
await client.Beta.Sessions.Events.Send(session.ID, new()
{
Events =
[
new BetaManagedAgentsUserMessageEventParams
{
Type = BetaManagedAgentsUserMessageEventParamsType.UserMessage,
Content =
[
new BetaManagedAgentsTextBlock
{
Type = BetaManagedAgentsTextBlockType.Text,
Text = "Analyze the performance of the sort function in utils.py",
},
],
},
],
});
```
```go Go theme={null}
if _, err := client.Beta.Sessions.Events.Send(ctx, session.ID, anthropic.BetaSessionEventSendParams{
Events: []anthropic.BetaManagedAgentsEventParamsUnion{{
OfUserMessage: &anthropic.BetaManagedAgentsUserMessageEventParams{
Type: anthropic.BetaManagedAgentsUserMessageEventParamsTypeUserMessage,
Content: []anthropic.BetaManagedAgentsUserMessageEventParamsContentUnion{{
OfText: &anthropic.BetaManagedAgentsTextBlockParam{
Type: anthropic.BetaManagedAgentsTextBlockTypeText,
Text: "Analyze the performance of the sort function in utils.py",
},
}},
},
}},
}); err != nil {
panic(err)
}
```
```java Java theme={null}
client.beta().sessions().events().send(
session.id(),
EventSendParams.builder()
.addEvent(BetaManagedAgentsUserMessageEventParams.builder()
.type(BetaManagedAgentsUserMessageEventParams.Type.USER_MESSAGE)
.addTextContent("Analyze the performance of the sort function in utils.py")
.build())
.build());
```
```php PHP theme={null}
$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',
],
],
],
],
);
```
```ruby Ruby theme={null}
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"
}
]
}
]
)
```
发送 `user.interrupt` 事件以在执行过程中停止智能体，然后跟进发送 `user.message` 事件以重定向它：
```bash cURL theme={null}
# 智能体当前正在分析文件...
# 用新的指令中断：
curl --fail-with-body -sS "http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_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.interrupt"},
{
"type": "user.message",
"content": [
{"type": "text", "text": "Instead, focus on fixing the bug in line 42."}
]
}
]
}
EOF
```
```bash CLI theme={null}
# 智能体当前正在分析一个文件……
# 用新的指令中断：
ant beta:sessions:events send --session-id "$SESSION_ID" <<'YAML'
events:
- type: user.interrupt
- type: user.message
content:
- type: text
text: Instead, focus on fixing the bug in line 42.
YAML
```
```python Python theme={null}
# 智能体当前正在分析文件...
# 用新的指令中断：
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.",
},
],
},
],
)
```
```typescript TypeScript theme={null}
// 智能体当前正在分析文件……
// 用新的指令中断：
await 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.",
},
],
},
],
});
```
```csharp C# theme={null}
// 智能体当前正在分析文件...
// 用新的指令进行中断：
await client.Beta.Sessions.Events.Send(session.ID, new()
{
Events =
[
new BetaManagedAgentsUserInterruptEventParams
{
Type = BetaManagedAgentsUserInterruptEventParamsType.UserInterrupt,
},
new BetaManagedAgentsUserMessageEventParams
{
Type = BetaManagedAgentsUserMessageEventParamsType.UserMessage,
Content =
[
new BetaManagedAgentsTextBlock
{
Type = BetaManagedAgentsTextBlockType.Text,
Text = "Instead, focus on fixing the bug in line 42.",
},
],
},
],
});
```
```go Go theme={null}
// 智能体当前正在分析文件...
// 以新的指示中断：
if _, err := client.Beta.Sessions.Events.Send(ctx, session.ID, anthropic.BetaSessionEventSendParams{
Events: []anthropic.BetaManagedAgentsEventParamsUnion{
{
OfUserInterrupt: &anthropic.BetaManagedAgentsUserInterruptEventParams{
Type: anthropic.BetaManagedAgentsUserInterruptEventParamsTypeUserInterrupt,
},
},
{
OfUserMessage: &anthropic.BetaManagedAgentsUserMessageEventParams{
Type: anthropic.BetaManagedAgentsUserMessageEventParamsTypeUserMessage,
Content: []anthropic.BetaManagedAgentsUserMessageEventParamsContentUnion{{
OfText: &anthropic.BetaManagedAgentsTextBlockParam{
Type: anthropic.BetaManagedAgentsTextBlockTypeText,
Text: "Instead, focus on fixing the bug in line 42.",
},
}},
},
},
},
}); err != nil {
panic(err)
}
```
```java Java theme={null}
// 智能体当前正在分析文件……
// 以新的指示进行中断：
client.beta().sessions().events().send(
session.id(),
EventSendParams.builder()
.addEvent(BetaManagedAgentsUserInterruptEventParams.builder()
.type(BetaManagedAgentsUserInterruptEventParams.Type.USER_INTERRUPT)
.build())
.addEvent(BetaManagedAgentsUserMessageEventParams.builder()
.type(BetaManagedAgentsUserMessageEventParams.Type.USER_MESSAGE)
.addTextContent("Instead, focus on fixing the bug in line 42.")
.build())
.build());
```
```php PHP theme={null}
// 智能体当前正在分析文件……
// 用新的指示中断：
$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.',
],
],
],
],
);
```
```ruby Ruby theme={null}
# 智能体当前正在分析一个文件……
# 用新的指令中断：
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."}
]
}
]
)
```
智能体会确认中断并切换到新任务。被中断的轮次以 `session.status_idle` 事件结束，其 `stop_reason` 为 `end_turn`，与自行完成的轮次的值相同；没有专门针对中断的停止原因。
从会话流式传输事件，以便在智能体工作时接收实时更新。只有在流打开后发出的事件才会被传递，因此请在发送事件之前打开流，以避免竞态条件。
```bash cURL theme={null}
# 先打开流，再发送用户消息
exec {stream}< <(
curl --fail-with-body -sS -N \
"http://localhost:38080/v1/sessions/$SESSION_ID/events/stream?beta=true" \
-H "x-api-key: $OMA_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 \
"http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_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}<&-
```
```bash CLI theme={null}
# 此工作流不适合用一次性的 shell 命令表达。
# 请改用此代码组中的某个 SDK 示例。
```
```python Python theme={null}
# 先打开流，再发送用户消息
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
```
```typescript TypeScript theme={null}
// 先打开流，再发送用户消息
const stream = await client.beta.sessions.events.stream(session.id);
await client.beta.sessions.events.send(session.id, {
events: [
{
type: "user.message",
content: [{ type: "text", text: "Summarize the repo README" }]
}
]
});
for await (const event of stream) {
if (event.type === "agent.message") {
for (const block of event.content) {
if (block.type === "text") {
process.stdout.write(block.text);
}
}
} else if (event.type === "session.status_idle") {
break;
} else if (event.type === "session.error") {
console.log(`\n[Error: ${event.error?.message ?? "unknown"}]`);
break;
}
}
```
```csharp C# theme={null}
// 先打开流，然后再发送用户消息
using var stream = await client.Beta.Sessions.Events.WithRawResponse.StreamStreaming(session.ID);
await client.Beta.Sessions.Events.Send(session.ID, new()
{
Events =
[
new BetaManagedAgentsUserMessageEventParams
{
Type = BetaManagedAgentsUserMessageEventParamsType.UserMessage,
Content =
[
new BetaManagedAgentsTextBlock
{
Type = BetaManagedAgentsTextBlockType.Text,
Text = "Summarize the repo README",
},
],
},
],
});
await foreach (var streamEvent in stream.Enumerate())
{
if (streamEvent.Value is BetaManagedAgentsAgentMessageEvent message)
{
foreach (var block in message.Content)
{
Console.Write(block.Text);
}
}
else if (streamEvent.Value is BetaManagedAgentsSessionStatusIdleEvent)
{
break;
}
else if (streamEvent.Value is BetaManagedAgentsSessionErrorEvent error)
{
Console.WriteLine($"\n[Error: {error.Error?.Message ?? "unknown"}]");
break;
}
}
```
```go Go theme={null}
// 先打开流，然后发送用户消息
stream := client.Beta.Sessions.Events.StreamEvents(ctx, session.ID, anthropic.BetaSessionEventStreamParams{})
defer stream.Close()
if _, err := client.Beta.Sessions.Events.Send(ctx, session.ID, anthropic.BetaSessionEventSendParams{
Events: []anthropic.BetaManagedAgentsEventParamsUnion{{
OfUserMessage: &anthropic.BetaManagedAgentsUserMessageEventParams{
Type: anthropic.BetaManagedAgentsUserMessageEventParamsTypeUserMessage,
Content: []anthropic.BetaManagedAgentsUserMessageEventParamsContentUnion{{
OfText: &anthropic.BetaManagedAgentsTextBlockParam{
Type: anthropic.BetaManagedAgentsTextBlockTypeText,
Text: "Summarize the repo README",
},
}},
},
}},
}); err != nil {
panic(err)
}
events:
for stream.Next() {
switch event := stream.Current().AsAny().(type) {
case anthropic.BetaManagedAgentsAgentMessageEvent:
// 具体类型列表：BetaManagedAgentsTextBlock
for _, block := range event.Content {
fmt.Print(block.Text)
}
case anthropic.BetaManagedAgentsSessionStatusIdleEvent:
break events
case anthropic.BetaManagedAgentsSessionErrorEvent:
fmt.Printf("\n[Error: %s]\n", cmp.Or(event.Error.Message, "unknown"))
break events
}
}
if err := stream.Err(); err != nil {
panic(err)
}
```
```java Java theme={null}
// 先打开流，然后发送用户消息
try (var stream = client.beta().sessions().events().streamStreaming(session.id())) {
client.beta().sessions().events().send(
session.id(),
EventSendParams.builder()
.addEvent(BetaManagedAgentsUserMessageEventParams.builder()
.type(BetaManagedAgentsUserMessageEventParams.Type.USER_MESSAGE)
.addTextContent("Summarize the repo README")
.build())
.build()
);
Iterable events = stream.stream()::iterator;
for (var event : events) {
if (event.isAgentMessage()) {
event.asAgentMessage().content().forEach(block -> block.text().ifPresent(textBlock -> IO.print(textBlock.text())));
} else if (event.isSessionStatusIdle()) {
break;
} else if (event.isSessionError()) {
// `message` 字段在所有错误变体中都存在；从原始 JSON 中读取它。
var errorMessage =
event.asSessionError().error()._json().orElse(null) instanceof JsonObject json
? json.values().get("message").asStringOrThrow()
: "unknown";
IO.println("\n[Error: " + errorMessage + "]");
break;
}
}
}
```
```php PHP theme={null}
// 先打开流，再发送用户消息
$stream = $client->beta->sessions->events->streamStream($session->id);
$client->beta->sessions->events->send(
$session->id,
events: [
[
'type' => 'user.message',
'content' => [['type' => 'text', 'text' => 'Summarize the repo README']],
],
],
);
foreach ($stream as $event) {
match ($event->type) {
'agent.message' => array_walk(
$event->content,
static fn ($block) => $block->type === 'text' ? print($block->text) : null,
),
'session.error' => printf("\n[Error: %s]", $event->error?->message ?? 'unknown'),
default => null,
};
if ($event->type === 'session.status_idle' || $event->type === 'session.error') {
break;
}
}
$stream->close();
```
```ruby Ruby theme={null}
# 先打开流，再发送用户消息
stream = client.beta.sessions.events.stream_events(session.id)
client.beta.sessions.events.send_(
session.id,
events: [{
type: "user.message",
content: [{type: "text", text: "Summarize the repo README"}]
}]
)
stream.each do |event|
case event.type
in :"agent.message"
event.content.each { print it.text }
in :"session.status_idle"
break
in :"session.error"
puts "\n[Error: #{event.error&.message || "unknown"}]"
break
else
# 忽略其他事件类型
end
end
```
要重新连接到现有会话而不遗漏事件：
1. 打开一个新的流。
2. 列出完整的事件历史记录，以初始化一组已见事件 ID。
3. 跟踪实时流，跳过历史记录列表中已返回的任何事件。
```bash cURL theme={null}
exec {stream}< <(
curl --fail-with-body -sS -N \
"http://localhost:38080/v1/sessions/$SESSION_ID/events/stream?beta=true" \
-H "x-api-key: $OMA_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"
)
# 流已打开并正在缓冲。在跟踪实时事件之前先列出历史记录。
declare -A seen_event_ids
while IFS= read -r event_id; do
seen_event_ids[$event_id]=1
done < <(
curl --fail-with-body -sS \
"http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-H "anthropic-beta: managed-agents-2026-04-01" \
-H "content-type: application/json" | jq -r '.data[].id'
)
# 跟踪实时事件，跳过已见过的内容
while IFS= read -r -u "$stream" event_line; do
[[ $event_line == data:* ]] || continue
event_json=${event_line#data: }
event_id=$(jq -r '.id' <<<"$event_json")
[[ -n ${seen_event_ids[$event_id]+seen} ]] && continue
seen_event_ids[$event_id]=1
case $(jq -r '.type' <<<"$event_json") in
agent.message)
jq -j '.content[] | select(.type == "text") | .text' <<<"$event_json"
;;
session.status_idle)
break
;;
esac
done
exec {stream}<&-
```
```bash CLI theme={null}
# 此工作流不适合用一次性的 shell 命令表达。
# 请改用此代码组中的某个 SDK 示例。
```
```python Python theme={null}
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
```
```typescript TypeScript theme={null}
const seenEventIds = new Set();
const stream = await client.beta.sessions.events.stream(session.id);
// 流已打开并正在缓冲。先列出历史记录，再跟踪实时事件。
for await (const event of client.beta.sessions.events.list(session.id)) {
seenEventIds.add(event.id);
}
// 跟踪实时事件，跳过已见过的内容
for await (const event of stream) {
// 预览事件（event_start/event_delta）不携带顶层 id
if (event.type === "event_start" || event.type === "event_delta") continue;
if (seenEventIds.has(event.id)) continue;
seenEventIds.add(event.id);
if (event.type === "agent.message") {
for (const block of event.content) {
if (block.type === "text") {
process.stdout.write(block.text);
}
}
} else if (event.type === "session.status_idle") {
break;
}
}
```
```csharp C# theme={null}
using var stream = await client.Beta.Sessions.Events.WithRawResponse.StreamStreaming(session.ID);
// 流已打开并正在缓冲。在追踪实时事件之前先列出历史记录。
HashSet seenEventIds = [];
var history = await client.Beta.Sessions.Events.List(session.ID);
await foreach (var pastEvent in history.Paginate())
{
seenEventIds.Add(pastEvent.ID);
}
// 追踪实时事件，跳过任何已见过的事件
await foreach (var streamEvent in stream.Enumerate())
{
if (!seenEventIds.Add(streamEvent.ID))
{
continue;
}
if (streamEvent.Value is BetaManagedAgentsAgentMessageEvent message)
{
foreach (var block in message.Content)
{
Console.Write(block.Text);
}
}
else if (streamEvent.Value is BetaManagedAgentsSessionStatusIdleEvent)
{
break;
}
}
```
```go Go theme={null}
stream := client.Beta.Sessions.Events.StreamEvents(ctx, session.ID, anthropic.BetaSessionEventStreamParams{})
defer stream.Close()
// 流已打开并正在缓冲。先列出历史记录，再追踪实时事件。
seenEventIDs := map[string]struct{}{}
history := client.Beta.Sessions.Events.ListAutoPaging(ctx, session.ID, anthropic.BetaSessionEventListParams{})
for history.Next() {
seenEventIDs[history.Current().ID] = struct{}{}
}
if err := history.Err(); err != nil {
panic(err)
}
// 追踪实时事件，跳过已见过的任何内容
tail:
for stream.Next() {
event := stream.Current()
if _, seen := seenEventIDs[event.ID]; seen {
continue
}
seenEventIDs[event.ID] = struct{}{}
switch event := event.AsAny().(type) {
case anthropic.BetaManagedAgentsAgentMessageEvent:
// 具体类型列表：BetaManagedAgentsTextBlock
for _, block := range event.Content {
fmt.Print(block.Text)
}
case anthropic.BetaManagedAgentsSessionStatusIdleEvent:
break tail
}
}
if err := stream.Err(); err != nil {
panic(err)
}
```
```java Java theme={null}
try (var stream = client.beta().sessions().events().streamStreaming(session.id())) {
// 流已打开并正在缓冲。先列出历史记录，再追踪实时事件。
// 每个事件变体都带有 `id`；从原始 JSON 中读取它以跨变体去重。
var seenEventIds = new HashSet();
for (var pastEvent : client.beta().sessions().events().list(session.id()).autoPager()) {
if (pastEvent._json().orElseThrow() instanceof JsonObject json) {
seenEventIds.add(json.values().get("id").asStringOrThrow());
}
}
// 追踪实时事件；Set.add 对已见过的 ID 返回 false，从而跳过重放。
stream.stream()
.filter(event -> event._json().orElseThrow() instanceof JsonObject json
&& seenEventIds.add(json.values().get("id").asStringOrThrow()))
.takeWhile(event -> !event.isSessionStatusIdle())
.filter(BetaManagedAgentsStreamSessionEvents::isAgentMessage)
.forEach(event -> event.asAgentMessage().content()
.forEach(block -> block.text().ifPresent(textBlock -> IO.print(textBlock.text()))));
}
```
```php PHP theme={null}
$stream = $client->beta->sessions->events->streamStream($session->id);
// 流已打开并在缓冲。先列出历史记录，再跟踪实时事件。
$seenEventIds = [];
foreach ($client->beta->sessions->events->list($session->id)->pagingEachItem() as $event) {
$seenEventIds[$event->id] = true;
}
// 跟踪实时事件，跳过已见过的内容
foreach ($stream as $event) {
if (isset($seenEventIds[$event->id])) {
continue;
}
$seenEventIds[$event->id] = true;
match ($event->type) {
'agent.message' => array_walk(
$event->content,
static fn ($block) => $block->type === 'text' ? print($block->text) : null,
),
default => null,
};
if ($event->type === 'session.status_idle') {
break;
}
}
$stream->close();
```
```ruby Ruby theme={null}
stream = client.beta.sessions.events.stream_events(session.id)
# 流已打开并在缓冲。先列出历史记录，再跟踪实时事件。
seen_event_ids = Set.new
client.beta.sessions.events.list(session.id).auto_paging_each { seen_event_ids << it.id }
# 跟踪实时事件，跳过已见过的内容 — Set#add? 对重复项返回 nil
stream.each do |event|
next unless seen_event_ids.add?(event.id)
case event.type
in :"agent.message"
event.content.each { print it.text }
in :"session.status_idle"
break
else
# 忽略其他事件类型
end
end
```
检索会话的完整事件历史记录：
```bash cURL theme={null}
curl --fail-with-body -sS "http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-H "anthropic-beta: managed-agents-2026-04-01" \
-H "content-type: application/json" \
| jq -r '.data[] | "[\(.type)] \(.processed_at)"'
```
```bash CLI theme={null}
ant beta:sessions:events list --session-id "$SESSION_ID" \
--format jsonl --transform '{type,processed_at}'
```
```python Python theme={null}
events = client.beta.sessions.events.list(session.id)
for event in events.data:
print(f"[{event.type}] {event.processed_at}")
```
```typescript TypeScript theme={null}
const events = await client.beta.sessions.events.list(session.id);
for (const event of events.data) {
console.log(`[${event.type}] ${event.processed_at}`);
}
```
```csharp C# theme={null}
var events = await client.Beta.Sessions.Events.List(session.ID);
foreach (var sessionEvent in events.Items)
{
Console.WriteLine($"[{sessionEvent.Json.GetProperty("type").GetString()}] {sessionEvent.ProcessedAt}");
}
```
```go Go theme={null}
events, err := client.Beta.Sessions.Events.List(ctx, session.ID, anthropic.BetaSessionEventListParams{})
if err != nil {
panic(err)
}
for _, event := range events.Data {
fmt.Printf("[%s] %s\n", event.Type, event.ProcessedAt)
}
```
```java Java theme={null}
var events = client.beta().sessions().events().list(session.id());
for (var event : events.data()) {
var eventJson = event._json().orElseThrow().convert(JsonNode.class);
var processedAt = eventJson.path("processed_at");
IO.println("[" + eventJson.get("type").asText() + "] "
+ (processedAt.isTextual() ? processedAt.asText() : "null"));
}
```
```php PHP theme={null}
$events = $client->beta->sessions->events->list($session->id);
foreach ($events->data as $event) {
$processedAt = ($event->processedAt ?? null)?->format(DATE_RFC3339) ?? 'null';
echo "[{$event->type}] {$processedAt}\n";
}
```
```ruby Ruby theme={null}
events = client.beta.sessions.events.list(session.id)
events.data.each { puts "[#{it.type}] #{it.processed_at}" }
```
传递 `types` 过滤器以仅返回特定事件类型：
```bash cURL theme={null}
curl --fail-with-body -sS "http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true&types[]=agent.tool_use&types[]=agent.tool_result" \
-H "x-api-key: $OMA_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-H "anthropic-beta: managed-agents-2026-04-01" \
| jq -r '.data[] | "[\(.type)] \(.processed_at)"'
```
```bash CLI theme={null}
ant beta:sessions:events list --session-id "$SESSION_ID" \
--type agent.tool_use --type agent.tool_result \
--format jsonl --transform '{type,processed_at}'
```
```python Python theme={null}
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}")
```
```typescript TypeScript theme={null}
const events = await client.beta.sessions.events.list(session.id, {
types: ["agent.tool_use", "agent.tool_result"],
});
for (const event of events.data) {
console.log(`[${event.type}] ${event.processed_at}`);
}
```
```csharp C# theme={null}
var events = await client.Beta.Sessions.Events.List(session.ID, new()
{
Types = ["agent.tool_use", "agent.tool_result"],
});
foreach (var sessionEvent in events.Items)
{
Console.WriteLine($"[{sessionEvent.Json.GetProperty("type").GetString()}] {sessionEvent.ProcessedAt}");
}
```
```go Go theme={null}
events, err := client.Beta.Sessions.Events.List(ctx, session.ID, anthropic.BetaSessionEventListParams{
Types: []string{"agent.tool_use", "agent.tool_result"},
})
if err != nil {
panic(err)
}
for _, event := range events.Data {
fmt.Printf("[%s] %s\n", event.Type, event.ProcessedAt)
}
```
```java Java theme={null}
var events = client.beta().sessions().events().list(
session.id(),
EventListParams.builder()
.addType("agent.tool_use")
.addType("agent.tool_result")
.build());
for (var event : events.data()) {
event.agentToolUse().ifPresent(toolUse ->
IO.println("[" + toolUse.type() + "] " + toolUse.processedAt()));
event.agentToolResult().ifPresent(toolResult ->
IO.println("[" + toolResult.type() + "] " + toolResult.processedAt()));
}
```
```php PHP theme={null}
// 在 PHP 中，通过 EventListParams 传入您需要的类型；请参阅 OMA PHP SDK。
```
```ruby Ruby theme={null}
events = client.beta.sessions.events.list(
session.id,
types: ["agent.tool_use", "agent.tool_result"]
)
events.data.each { puts "[#{it.type}] #{it.processed_at}" }
```
## 事件增量
默认情况下，智能体的响应文本以缓冲的 `agent.message` 事件形式到达流，每个事件仅在生成它的模型请求完成后才发出。事件增量让您能够在模型仍在生成文本时，以实时预览的方式增量渲染该文本。预览不是响应本身：预览是尽力而为的显示辅助，缓冲的 `agent.message` 始终是权威记录。忽略预览的客户端仍会收到完整、正确的流。
### 选择加入预览
预览是按流连接选择加入的。将 `event_deltas[]` 查询参数添加到您正在读取的流中，为您希望预览的每种事件类型重复一次。由于 `[]` 是 shell 通配符模式，因此在 shell 中构建请求时请为 URL 加引号；示例中将方括号百分号编码为 `%5B%5D`，这同样有效。两个流端点都接受该参数：位于 `GET /v1/sessions/{session_id}/events/stream` 的会话级流，以及每个[会话线程](/docs/zh/multiagent-orchestration)自己的流，位于 `GET /v1/sessions/{session_id}/threads/{thread_id}/stream`。接受的值为 `agent.message` 和 `agent.thinking`；任何其他值都会返回 400 错误，包含超过 100 个值的请求也是如此。子智能体的预览出现在[该子智能体自己的线程流](/docs/zh/events-and-streaming#preview-session-thread-events)上。
当预览事件开始时，流会发出一个 `event_start`，其中携带即将到来的事件的类型和 `id`：
```json theme={null}
{
"type": "event_start",
"event": {
"type": "agent.message",
"id": "sevt_01abc..."
}
}
```
对于 `agent.message`，起始事件之后是携带增量文本的 `event_delta` 事件。每个增量在 `event_id` 中指明它所扩展的事件，在 `delta.index` 中指明它所扩展的内容块：
```json theme={null}
{
"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`。
事件增量使用与流式传输消息不同的传输格式，这种差异是有意为之的。预览的 `agent.message` 会得到一个 `event_start`，之后仅跟随 `event_delta` 事件。没有针对每个内容块的开始或停止事件，也没有针对预览事件本身的停止事件。增量类型是 `content_delta`，而非 `content_block_delta`。为消息 API 编写的累加器代码无法原封不动地沿用。
### 累加与协调
每个支持事件增量的 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`](/docs/zh/reference#event-types)（位于“跨度事件”选项卡中）。在传输层面，这是该序列的预览部分，与连接的其他缓冲事件交错出现：
```text wrap theme={null}
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. 收到 `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` 的前缀（是前缀，不一定是完整文本，因为在负载较高时增量可能被丢弃）。
* 一个连接对每个 `event_id` 最多发出一个 `event_start`，并且缓冲事件是该连接为该 `id` 传递的最后一项内容。
```bash cURL theme={null}
# 通过 event_deltas 选择启用 agent.message 预览，然后手动累积。
exec {stream}< <(
curl --fail-with-body -sS -N \
"http://localhost:38080/v1/sessions/$SESSION_ID/events/stream?beta=true&event_deltas%5B%5D=agent.message" \
-H "x-api-key: $OMA_API_KEY" \
-H "anthropic-version: 2023-06-01" \
-H "anthropic-beta: managed-agents-2026-04-01" \
-H "accept: text/event-stream"
)
curl --fail-with-body -sS \
"http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_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": "In one short sentence, describe what an event delta is."}]
}
]
}
EOF
# 以（消息 id、内容索引）为键累积增量；最终的
# agent.message 携带完整文本，因此会替换该 id 的所有预览。
declare -A preview
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_start)
preview_id=$(jq -r '.event.id' <<<"$event_json")
printf '[event_start id=%s]\n' "$preview_id"
;;
event_delta)
preview_key=$(jq -r '.event_id + ":" + (.delta.index | tostring)' <<<"$event_json")
preview[$preview_key]+=$(jq -r '.delta.content.text' <<<"$event_json")
printf '[event_delta] %s\n' "${preview[$preview_key]}"
;;
agent.message)
msg_id=$(jq -r '.id' <<<"$event_json")
for preview_key in "${!preview[@]}"; do
[[ $preview_key == "$msg_id":* ]] && unset "preview[$preview_key]"
done
printf '[agent.message id=%s] ' "$msg_id"
jq -j '.content[] | select(.type == "text") | .text' <<<"$event_json"
printf '\n'
;;
span.model_request_end)
for preview_key in "${!preview[@]}"; do
printf '[closing unreconciled preview for %s]\n' "${preview_key%%:*}"
done
preview=()
;;
session.status_idle)
break
;;
esac
done
exec {stream}<&-
```
```bash CLI theme={null}
# 此工作流不适合用一次性的 shell 命令表达。
# 请改用此代码组中的某个 SDK 示例。
```
```python Python theme={null}
# 预览快照，以事件 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
```
```typescript TypeScript theme={null}
// 预览快照，以事件 id 为键。`accumulateManagedAgentsEvent`
// 将 event_start / event_delta 预览折叠进一个 agent.message 快照。
const previews = new Map();
// 仅为此连接启用 agent.message 预览
const stream = await client.beta.sessions.events.stream(session.id, {
event_deltas: ["agent.message"],
});
await client.beta.sessions.events.send(session.id, {
events: [
{
type: "user.message",
content: [{ type: "text", text: "Summarize the repo README" }]
}
]
});
for await (const event of stream) {
if (event.type === "event_start") {
// 1. 记下宣告的 id 并打开快照。增量和
// 缓冲事件携带相同的 id。
const preview = accumulateManagedAgentsEvent(undefined, event);
if (preview) previews.set(event.event.id, preview);
console.log(`event_start ${event.event.type} ${event.event.id}`);
} else if (event.type === "event_delta") {
// 2. 将片段折叠进快照并渲染
const preview = accumulateManagedAgentsEvent(previews.get(event.event_id), event);
if (preview) {
previews.set(event.event_id, preview);
const text = preview.content.map((block) => block.text).join("");
console.log(`event_delta preview: ${JSON.stringify(text)}`);
}
} else if (event.type === "agent.message") {
// 3. 缓冲事件才是正式记录：它会替换并关闭预览
const message = accumulateManagedAgentsEvent(previews.get(event.id), event);
previews.delete(event.id);
const text = message.content.map((block) => block.text).join("");
console.log(`agent.message ${event.id} ${JSON.stringify(text)}`);
} else if (event.type === "span.model_request_end") {
// 4. 不会再有增量到来。关闭所有未被最终确认的预览。
for (const eventId of previews.keys()) {
console.log(`span.model_request_end closing preview for ${eventId}`);
}
previews.clear();
} else if (event.type === "session.status_idle") {
break;
}
}
stream.controller.abort();
```
```csharp C# theme={null}
// 选择启用事件增量：agent.message 事件在生成时即被预览。
using var stream = await client.Beta.Sessions.Events.WithRawResponse.StreamStreaming(
session.ID,
new() { EventDeltas = [BetaManagedAgentsDeltaType.AgentMessage] }
);
await client.Beta.Sessions.Events.Send(session.ID, new()
{
Events =
[
new BetaManagedAgentsUserMessageEventParams
{
Type = BetaManagedAgentsUserMessageEventParamsType.UserMessage,
Content =
[
new BetaManagedAgentsTextBlock
{
Type = BetaManagedAgentsTextBlockType.Text,
Text = "Write a haiku about event streams.",
},
],
},
],
});
// 按 (event id, content index) 累积预览片段。随后到达的缓冲
// agent.message 携带完整内容，因此它会替换
// 累积的预览，而不是追加到其后。
Dictionary> previews = [];
await foreach (var streamEvent in stream.Enumerate())
{
if (streamEvent.TryPickStartEvent(out var start))
{
// 已为具有此 id 的事件打开预览。此流仅选择启用
// agent.message 增量；TryPick* 返回 false 而非抛出异常，
// 因此其他预览类型（包括后续新增的类型）会被跳过。
if (start.Event.TryPickAgentMessage(out var preview))
{
Console.WriteLine($"event_start {preview.Type.Raw()} {preview.ID}");
}
}
else if (streamEvent.TryPickDeltaEvent(out var delta))
{
// 在新索引处插入，在现有索引处追加
if (!previews.TryGetValue(delta.EventID, out var fragments))
{
previews[delta.EventID] = fragments = [];
}
var index = delta.Delta.Index ?? 0;
fragments[index] = fragments.GetValueOrDefault(index, "") + delta.Delta.Content.Text;
Console.WriteLine($"event_delta preview: {fragments[index]}");
}
else if (streamEvent.TryPickAgentMessageEvent(out var message))
{
// 增量是尽力而为的：丢弃预览并使用缓冲的事件
previews.Remove(message.ID);
Console.WriteLine($"agent.message {message.ID} {string.Concat(message.Content.Select(block => block.Text))}");
}
else if (streamEvent.TryPickSpanModelRequestEndEvent(out _))
{
// 不会再有增量到来；关闭任何从未被协调的预览。
foreach (var eventId in previews.Keys)
{
Console.WriteLine($"span.model_request_end closing preview for {eventId}");
}
previews.Clear();
}
else if (streamEvent.TryPickSessionStatusIdleEvent(out _))
{
break;
}
}
```
```go Go theme={null}
// 选择启用 agent.message 事件的增量预览
stream := client.Beta.Sessions.Events.StreamEvents(ctx, session.ID, anthropic.BetaSessionEventStreamParams{
EventDeltas: []anthropic.BetaManagedAgentsDeltaType{
anthropic.BetaManagedAgentsDeltaTypeAgentMessage,
},
})
if _, err := client.Beta.Sessions.Events.Send(ctx, session.ID, anthropic.BetaSessionEventSendParams{
Events: []anthropic.BetaManagedAgentsEventParamsUnion{{
OfUserMessage: &anthropic.BetaManagedAgentsUserMessageEventParams{
Type: anthropic.BetaManagedAgentsUserMessageEventParamsTypeUserMessage,
Content: []anthropic.BetaManagedAgentsUserMessageEventParamsContentUnion{{
OfText: &anthropic.BetaManagedAgentsTextBlockParam{
Type: anthropic.BetaManagedAgentsTextBlockTypeText,
Text: "Write a haiku about the ocean.",
},
}},
},
}},
}); err != nil {
panic(err)
}
// 累加器将 event_start / event_delta 片段折叠为
// 按 event-id 分组的 agent.message 快照。零值即可直接使用。
var previews anthropic.BetaManagedAgentsEventAccumulator
deltas:
for stream.Next() {
event := stream.Current()
previews.Accumulate(event)
switch event := event.AsAny().(type) {
case anthropic.BetaManagedAgentsStartEvent:
fmt.Printf("event_start %s %s\n", event.Event.Type, event.Event.ID)
case anthropic.BetaManagedAgentsDeltaEvent:
fmt.Printf("event_delta preview: %q\n", previews.AgentMessageText(event.EventID))
case anthropic.BetaManagedAgentsAgentMessageEvent:
// 缓冲的事件携带完整内容：累加器
// 用它替换预览
fmt.Printf("agent.message %s %q\n", event.ID, previews.AgentMessageText(event.ID))
case anthropic.BetaManagedAgentsSpanModelRequestEndEvent:
// 此请求不会再有增量到达。累加器
// 在此处丢弃其快照，关闭任何从未被
// 缓冲的 agent.message 协调一致的预览。
fmt.Println("span.model_request_end no more deltas for this request")
case anthropic.BetaManagedAgentsSessionStatusIdleEvent:
break deltas
}
}
if err := stream.Err(); err != nil {
panic(err)
}
stream.Close()
```
```java Java theme={null}
// 预览文本，先按事件 ID 再按内容索引作为键。缓冲的 agent.message 会替换它。
Map> previews = new HashMap<>();
// 在此连接上选择启用 agent.message 预览
try (var stream = client.beta().sessions().events().streamStreaming(
session.id(),
EventStreamParams.builder()
.addEventDelta(BetaManagedAgentsDeltaType.AGENT_MESSAGE)
.build()
)) {
client.beta().sessions().events().send(
session.id(),
EventSendParams.builder()
.addEvent(BetaManagedAgentsUserMessageEventParams.builder()
.type(BetaManagedAgentsUserMessageEventParams.Type.USER_MESSAGE)
.addTextContent("Describe the repo in one sentence.")
.build())
.build()
);
Iterable events = stream.stream()::iterator;
for (var event : events) {
if (event.isEventStart() && event.asEventStart().event().isAgentMessage()) {
var preview = event.asEventStart().event().asAgentMessage();
IO.println("event_start " + preview.type().asString() + " " + preview.id());
} else if (event.isEventDelta()) {
var eventDelta = event.asEventDelta();
var fragment = eventDelta.delta();
var buffer = previews
.computeIfAbsent(eventDelta.eventId(), _ -> new HashMap<>())
.computeIfAbsent(fragment.index().orElse(0L), _ -> new StringBuilder());
buffer.append(fragment.content().text());
IO.println("event_delta preview: " + buffer);
} else if (event.isAgentMessage()) {
// 缓冲的事件才是正式记录：丢弃其预览，渲染其内容
var message = event.asAgentMessage();
previews.remove(message.id());
var text = message.content().stream()
.flatMap(block -> block.text().stream())
.map(textBlock -> textBlock.text())
.collect(Collectors.joining());
IO.println("agent.message " + message.id() + " " + text);
} else if (event.isSpanModelRequestEnd()) {
// 不会再有增量到来。关闭所有缓冲事件始终未到达的预览。
previews.keySet().forEach(eventId ->
IO.println("span.model_request_end closing preview for " + eventId));
previews.clear();
} else if (event.isSessionStatusIdle()) {
break;
}
}
}
```
```php PHP theme={null}
// 在 PHP 中，在 EventStreamParams 上设置 eventDeltas，并使用 Anthropic\Lib\Sessions\EventAccumulator 进行累积。
```
```ruby Ruby theme={null}
# 启用事件增量：agent.message 预览以增量片段的形式流式传输。
stream = client.beta.sessions.events.stream_events(
session.id,
event_deltas: [Anthropic::Beta::BetaManagedAgentsDeltaType::AGENT_MESSAGE]
)
client.beta.sessions.events.send_(
session.id,
events: [{
type: "user.message",
content: [{type: "text", text: "Give a one-sentence project tagline."}]
}]
)
# 按 (event_id, index) 将预览片段累积到显式可变的
# （`+""`）缓冲区中，以便 `<<` 可以就地追加。具有相同 id 的已缓冲
# agent.message 是权威版本，会替换增量所累积的全部内容。
buffers = Hash.new do |by_event, event_id|
by_event[event_id] = Hash.new { |fragments, index| fragments[index] = +"" }
end
stream.each do |event|
case event.type
in :event_start
puts "event_start #{event.event.type} #{event.event.id}"
in :event_delta
delta = event.delta
fragment = delta.content.text
buffers[event.event_id][delta.index || 0] << fragment
puts "event_delta preview: #{buffers[event.event_id][delta.index || 0].inspect}"
in :"agent.message"
# 替换：丢弃累积的预览并渲染完整事件。
buffers.delete(event.id)
puts "agent.message #{event.id} #{event.content.map(&:text).join.inspect}"
in :"span.model_request_end"
# 不会再有增量到来。关闭所有从未被最终确认的预览。
buffers.each_key { |event_id| puts "span.model_request_end closing preview for #{event_id}" }
buffers.clear
in :"session.status_idle"
break
else
# 忽略其他事件类型
end
end
```
### 预览会话线程事件
在[多智能体](/docs/zh/multiagent-orchestration)会话中，每个会话线程在 `GET /v1/sessions/{session_id}/threads/{thread_id}/stream` 处都有自己的事件流，并且它接受相同的 `event_deltas[]` 参数和相同的值。预览在设计上是线程范围的：一个连接仅预览它正在读取的线程。子线程的预览在该子线程自己的流上传递，永远不会交叉发布到会话级流，后者的预览始终限定于主线程。要在模型生成时观察子智能体的文本，请打开该子智能体的线程流。
线程流的路径很容易弄错：它是 `/threads/{thread_id}/stream`，而不是 `/events/stream`（后者仅存在于会话级别），并且不存在 `/threads/{thread_id}/events/stream` 端点。
预览事件本身不会改变。`event_start` 和 `event_delta` 在线程流上的结构与在会话级流上相同，[累加与协调](/docs/zh/events-and-streaming#accumulate-and-reconcile)模式可按原样应用。唯一的调整是记录方式：为每个流连接运行一个累加器实例。
```bash cURL theme={null}
# 列出会话的线程并选择一个子线程：子线程带有非空的
# parent_thread_id，而主线程的 parent_thread_id 为 null。
THREAD_ID=$(
curl --fail-with-body -sS \
"http://localhost:38080/v1/sessions/$SESSION_ID/threads?beta=true" \
-H "x-api-key: $OMA_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 \
"http://localhost:38080/v1/sessions/$SESSION_ID/threads/$THREAD_ID/stream?beta=true&event_deltas%5B%5D=agent.message" \
-H "x-api-key: $OMA_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}<&-
```
```bash CLI theme={null}
# 列出会话的线程并选择一个子线程：子线程带有非空的
# parent_thread_id，而主线程的 parent_thread_id 为 null
# （--transform 的 #(parent_thread_id!=~null) 查询匹配非空值）。
THREAD_ID=$(ant beta:sessions:threads list \
--session-id "$SESSION_ID" \
--format raw --transform 'data.#(parent_thread_id!=~null).id' --raw-output)
# 子线程的流接受与会话流相同的 event_deltas 参数，
# 每个要预览的事件类型对应一个 --event-delta 标志。@tostr
# 将每个文本字段重新编码为 JSON 字符串，因此每个值都保持在一行
# YAML 中，jq 的 fromjson 可以恢复原始文本。
transform='{type,frag:delta.content.text|@tostr,text:content.#(type=="text").text|@tostr}'
exec {stream}< <(ant beta:sessions:threads:events stream \
--session-id "$SESSION_ID" \
--thread-id "$THREAD_ID" \
--event-delta agent.message \
--transform "$transform" \
--format yaml)
type=
while IFS= read -r -u "$stream" line; do
case "$line" in
type:\ session.thread_status_idle) break ;;
type:\ *) type=${line#type: } ;;
frag:*)
[[ $type == event_delta ]] || continue
jq -j fromjson <<<"${line#frag: }" ;;
text:*)
[[ $type == agent.message ]] || continue
# 缓冲的事件是权威记录；渲染其内容。
printf '\n'
jq -r fromjson <<<"${line#text: }" ;;
esac
done
exec {stream}<&-
```
```python Python theme={null}
# 列出会话的线程并选择一个子线程：子线程带有非空的
# parent_thread_id，而主线程的 parent_thread_id 为 null。
child_thread = next(
thread
for thread in client.beta.sessions.threads.list(session.id)
if thread.parent_thread_id is not None
)
# 子线程的流接受与会话流相同的
# event_deltas 参数。
with client.beta.sessions.threads.events.stream(
child_thread.id,
session_id=session.id,
event_deltas=["agent.message"],
) as stream:
for event in stream:
match event.type:
case "event_delta":
print(event.delta.content.text, end="")
case "agent.message":
# 缓冲事件是权威记录；渲染其内容
print()
for block in event.content:
if block.type == "text":
print(block.text, end="")
print()
case "session.thread_status_idle":
break
```
```typescript TypeScript theme={null}
// 列出会话的线程并选择一个子线程：子线程带有非空的
// parent_thread_id，而主线程的 parent_thread_id 为 null。
let childThreadId: string | undefined;
for await (const thread of client.beta.sessions.threads.list(session.id)) {
if (thread.parent_thread_id !== null) {
childThreadId = thread.id;
break;
}
}
if (!childThreadId) throw new Error("No child thread found");
// 子线程的流接受与会话流相同的
// event_deltas 参数。
const stream = await client.beta.sessions.threads.events.stream(childThreadId, {
session_id: session.id,
event_deltas: ["agent.message"],
});
for await (const event of stream) {
if (event.type === "event_delta") {
process.stdout.write(event.delta.content.text);
} else if (event.type === "agent.message") {
// 缓冲事件是权威记录；渲染其内容。
process.stdout.write("\n");
const text = event.content.map((block) => block.text).join("");
console.log(text);
} else if (event.type === "session.thread_status_idle") {
break;
}
}
stream.controller.abort();
```
```csharp C# theme={null}
// 列出会话的线程并选取一个子线程：子线程带有非空的
// parent_thread_id，而主线程的 parent_thread_id 为 null。
var threads = await client.Beta.Sessions.Threads.List(session.ID);
var childThread = threads.Items.First(thread => thread.ParentThreadID is not null);
// 子线程的流接受与会话流相同的 event_deltas
// 参数。
using var stream = await client.Beta.Sessions.Threads.Events.WithRawResponse.StreamStreaming(
childThread.ID,
new() { SessionID = session.ID, EventDeltas = [BetaManagedAgentsDeltaType.AgentMessage] }
);
await foreach (var streamEvent in stream.Enumerate())
{
if (streamEvent.TryPickDeltaEvent(out var delta))
{
Console.Write(delta.Delta.Content.Text);
}
else if (streamEvent.TryPickAgentMessageEvent(out var message))
{
// 缓冲的事件是权威记录；渲染其内容。
Console.WriteLine();
Console.WriteLine(string.Concat(message.Content.Select(block => block.Text)));
}
else if (streamEvent.TryPickSessionThreadStatusIdleEvent(out _))
{
break;
}
}
```
```go Go theme={null}
// 列出会话的线程并选择一个子线程：子线程带有非空的
// parent_thread_id，而主线程的 parent_thread_id 为 null。
var childThreadID string
threads := client.Beta.Sessions.Threads.ListAutoPaging(ctx, session.ID, anthropic.BetaSessionThreadListParams{})
for threads.Next() {
if thread := threads.Current(); thread.ParentThreadID != "" {
childThreadID = thread.ID
break
}
}
if err := threads.Err(); err != nil {
panic(err)
}
// 子线程的流接受与会话流相同的 event_deltas 参数；
// 每个流连接运行一个读取循环。
stream := client.Beta.Sessions.Threads.Events.StreamEvents(ctx, childThreadID, anthropic.BetaSessionThreadEventStreamParams{
SessionID: session.ID,
EventDeltas: []anthropic.BetaManagedAgentsDeltaType{
anthropic.BetaManagedAgentsDeltaTypeAgentMessage,
},
})
threadDeltas:
for stream.Next() {
switch event := stream.Current().AsAny().(type) {
case anthropic.BetaManagedAgentsDeltaEvent:
fmt.Print(event.Delta.Content.Text)
case anthropic.BetaManagedAgentsAgentMessageEvent:
// 缓冲的事件是权威记录；渲染其内容。
fmt.Println()
// 具体类型列表：BetaManagedAgentsTextBlock
for _, block := range event.Content {
fmt.Print(block.Text)
}
fmt.Println()
case anthropic.BetaManagedAgentsSessionThreadStatusIdleEvent:
break threadDeltas
}
}
if err := stream.Err(); err != nil {
panic(err)
}
stream.Close()
```
```java Java theme={null}
// 列出会话的线程并选择一个子线程：子线程带有非空的
// parent_thread_id，而主线程的 parent_thread_id 为 null。
var childThread = client.beta().sessions().threads().list(session.id()).autoPager().stream()
.filter(thread -> thread.parentThreadId().isPresent())
.findFirst()
.orElseThrow();
// 子线程的流接受与会话流相同的 event_deltas 参数。
// 其 params 类与会话级别的类共享简单名称，因此需限定它。
try (var stream = client.beta().sessions().threads().events().streamStreaming(
childThread.id(),
com.anthropic.models.beta.sessions.threads.events.EventStreamParams.builder()
.sessionId(session.id())
.addEventDelta(BetaManagedAgentsDeltaType.AGENT_MESSAGE)
.build()
)) {
Iterable events = stream.stream()::iterator;
for (var event : events) {
if (event.isEventDelta()) {
IO.print(event.asEventDelta().delta().content().text());
} else if (event.isAgentMessage()) {
// 缓冲的事件是权威记录；渲染其内容。
IO.println();
event.asAgentMessage().content().forEach(block -> block.text().ifPresent(textBlock -> IO.print(textBlock.text())));
IO.println();
} else if (event.isSessionThreadStatusIdle()) {
break;
}
}
}
```
```php PHP theme={null}
// 在 PHP 中，在线程的 EventStreamParams 上设置 eventDeltas，并使用 Anthropic\Lib\Sessions\EventAccumulator 进行累积。
```
```ruby Ruby theme={null}
# 列出会话的线程并选择一个子线程：子线程带有非空的
# parent_thread_id，而主线程的 parent_thread_id 为 null。
child_thread = client.beta.sessions.threads.list(session.id).to_enum.find { it.parent_thread_id }
# 子线程的流接受与会话流相同的
# event_deltas 参数。
stream = client.beta.sessions.threads.events.stream_events(
child_thread.id,
session_id: session.id,
event_deltas: [Anthropic::Beta::BetaManagedAgentsDeltaType::AGENT_MESSAGE]
)
stream.each do |event|
case event.type
in :event_delta
print event.delta.content.text
in :"agent.message"
# 已缓冲的事件是权威记录；渲染其内容。
puts
event.content.each { print it.text }
puts
in :"session.thread_status_idle"
break
else
# 忽略其他事件类型
end
end
```
读取循环在 [`session.thread_status_idle`](/docs/zh/reference#event-types) 时退出，该事件在会话线程的轮次完成且线程进入空闲状态时发出。
### 限制
预览针对响应速度进行了优化。请基于以下约束进行构建：
* **尽力而为：** 在负载较高时，服务器可能会丢弃某个事件的增量。发生这种情况时，您会收到文本的连续前缀，之后不再收到该事件的任何增量。缓冲的 `agent.message` 仍会完整到达。切勿将累加的预览视为最终结果。
* **重连时不重放：** 增量仅在选择加入的连接打开期间传递给该连接。这同样适用于会话级流和每个会话线程流，并且在模型请求开始后打开的连接不会收到该进行中事件的任何增量。如果流断开，请按照"流式传输事件"选项卡中的[重连步骤](/docs/zh/events-and-streaming#integrating-events)操作：重新打开流并列出事件历史记录。历史记录包含您断开连接期间发出的所有缓冲事件，包括您的预览正在等待的 `agent.message`。无法重新请求已错过的增量。
* **单线程，仅文本：** 预览涵盖连接正在读取的线程上的助手文本。工具使用、工具结果、MCP 结果以及任何其他[会话线程](/docs/zh/multiagent-orchestration)上的活动永远不会在该连接上被预览。
* **`agent.thinking` 仅有起始事件：** `agent.thinking` 预览仅发出 `event_start` 作为思考块已开始的信号；不会跟随任何 `event_delta` 事件。
* **从不持久化：** `event_start` 和 `event_delta` 仅存在于实时流中。它们不会出现在会话的事件历史记录（`GET /v1/sessions/{session_id}/events`）中，也不会出现在任何会话线程的事件历史记录中。
### 预览故障排查
如果流的行为与您的预期不符：
| 您看到的现象 | 含义 |
| ---------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- |
| 流中有缓冲事件但没有 `event_start` 或 `event_delta` | 您正在读取的连接未选择加入（`event_deltas[]` 按连接生效，而非按会话），或者该轮次从未触及您正在流式传输的线程。预览是线程范围的，因此请列出会话的线程（`GET /v1/sessions/{session_id}/threads`）以查找实际运行的线程。 |
| 流 URL 返回 404 | 路径或某个 ID 有误，或者请求根本未携带 managed-agents Beta 请求头。线程端点受 beta 门控，因此没有该标头时它们不存在。 |
| 指明 `event_deltas` 的 400 错误 | 仅接受 `agent.message` 和 `agent.thinking`。 |
## 其他场景
### 处理自定义工具调用
当智能体调用[自定义工具](/docs/zh/tools#custom-tools)时：
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` 状态。
```bash cURL theme={null}
exec {stream_fd}< <(curl --fail-with-body -sS -N \
"http://localhost:38080/v1/sessions/$SESSION_ID/events/stream?beta=true" \
-H "x-api-key: $OMA_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 \
"http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_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}<&-
```
```bash CLI theme={null}
# 此工作流不适合转换为一次性的 shell 命令。
# 请改用此代码组中的某个 SDK 示例。
```
```python Python theme={null}
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
```
```typescript TypeScript theme={null}
const stream = await client.beta.sessions.events.stream(session.id);
for await (const event of stream) {
if (event.type !== "session.status_idle") continue;
if (event.stop_reason.type === "end_turn") break;
if (event.stop_reason.type !== "requires_action") continue;
for (const eventId of event.stop_reason.event_ids) {
// 查找该自定义工具使用事件并执行它
const toolEvent = eventsById.get(eventId);
if (!toolEvent) continue;
const result = await callTool(toolEvent.name, toolEvent.input);
// 将结果发送回去
await client.beta.sessions.events.send(session.id, {
events: [
{
type: "user.custom_tool_result",
custom_tool_use_id: eventId,
content: [{ type: "text", text: result }],
},
],
});
}
}
```
```csharp C# theme={null}
await foreach (var streamEvent in client.Beta.Sessions.Events.StreamStreaming(session.ID))
{
if (streamEvent.Value is not BetaManagedAgentsSessionStatusIdleEvent idle) continue;
if (idle.StopReason?.Value is BetaManagedAgentsSessionRequiresAction requiresAction)
{
foreach (var eventId in requiresAction.EventIds)
{
// 查找自定义工具使用事件并执行它
var toolEvent = eventsById[eventId];
var result = await CallTool(toolEvent.Name, toolEvent.Input);
// 将结果发送回去
await client.Beta.Sessions.Events.Send(session.ID, new()
{
Events =
[
new BetaManagedAgentsUserCustomToolResultEventParams
{
Type = BetaManagedAgentsUserCustomToolResultEventParamsType.UserCustomToolResult,
CustomToolUseID = eventId,
Content =
[
new BetaManagedAgentsTextBlock
{
Type = BetaManagedAgentsTextBlockType.Text,
Text = result,
},
],
},
],
});
}
}
else if (idle.StopReason?.Value is BetaManagedAgentsSessionEndTurn)
{
break;
}
}
```
```go Go theme={null}
stream := client.Beta.Sessions.Events.StreamEvents(ctx, session.ID, anthropic.BetaSessionEventStreamParams{})
defer stream.Close()
loop:
for stream.Next() {
event, ok := stream.Current().AsAny().(anthropic.BetaManagedAgentsSessionStatusIdleEvent)
if !ok {
continue
}
switch stopReason := event.StopReason.AsAny().(type) {
case anthropic.BetaManagedAgentsSessionRequiresAction:
for _, eventID := range stopReason.EventIDs {
// 查找自定义工具使用事件并执行它
toolEvent := eventsByID[eventID]
result := callTool(toolEvent.Name, toolEvent.Input)
// 将结果发送回去
if _, err := client.Beta.Sessions.Events.Send(ctx, session.ID, anthropic.BetaSessionEventSendParams{
Events: []anthropic.BetaManagedAgentsEventParamsUnion{{
OfUserCustomToolResult: &anthropic.BetaManagedAgentsUserCustomToolResultEventParams{
Type: anthropic.BetaManagedAgentsUserCustomToolResultEventParamsTypeUserCustomToolResult,
CustomToolUseID: eventID,
Content: []anthropic.BetaManagedAgentsUserCustomToolResultEventParamsContentUnion{{
OfText: &anthropic.BetaManagedAgentsTextBlockParam{
Type: anthropic.BetaManagedAgentsTextBlockTypeText,
Text: result,
},
}},
},
}},
}); err != nil {
panic(err)
}
}
case anthropic.BetaManagedAgentsSessionEndTurn:
break loop
}
}
if err := stream.Err(); err != nil {
panic(err)
}
```
```java Java theme={null}
try (var stream = client.beta().sessions().events().streamStreaming(session.id())) {
stream.stream()
.filter(BetaManagedAgentsStreamSessionEvents::isSessionStatusIdle)
.map(idleEvent -> idleEvent.asSessionStatusIdle().stopReason())
.takeWhile(stopReason -> !stopReason.isEndTurn())
.filter(stopReason -> stopReason.isRequiresAction())
.flatMap(stopReason -> stopReason.asRequiresAction().eventIds().stream())
.forEach(eventId -> {
// 查找自定义工具使用事件并执行它
var toolEvent = eventsById.get(eventId);
var result = callTool(toolEvent.name(), toolEvent.input());
// 将结果发送回去
client.beta().sessions().events().send(
session.id(),
EventSendParams.builder()
.addEvent(BetaManagedAgentsUserCustomToolResultEventParams.builder()
.type(BetaManagedAgentsUserCustomToolResultEventParams.Type.USER_CUSTOM_TOOL_RESULT)
.customToolUseId(eventId)
.addTextContent(result)
.build())
.build());
});
}
```
```php PHP theme={null}
$stream = $client->beta->sessions->events->streamStream($session->id);
foreach ($stream as $event) {
if ($event->type === 'session.status_idle' && $event->stopReason) {
if ($event->stopReason->type === 'requires_action') {
foreach ($event->stopReason->eventIDs as $eventId) {
// 查找自定义工具使用事件并执行它
$toolEvent = $eventsById[$eventId];
$result = callTool($toolEvent->name, $toolEvent->input);
// 将结果发送回去
$client->beta->sessions->events->send(
$session->id,
events: [
[
'type' => 'user.custom_tool_result',
'custom_tool_use_id' => $eventId,
'content' => [['type' => 'text', 'text' => $result]],
],
],
);
}
} elseif ($event->stopReason->type === 'end_turn') {
break;
}
}
}
```
```ruby Ruby theme={null}
client.beta.sessions.events.stream_events(session.id).each do |event|
case event
in {type: :"session.status_idle", stop_reason: {type: :requires_action, event_ids:}}
event_ids.each do |event_id|
# 查找自定义工具使用事件并执行它
tool_event = events_by_id[event_id]
result = call_tool.call(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}]
}
]
)
end
in {type: :"session.status_idle", stop_reason: {type: :end_turn}}
break
else
end
end
```
### 工具确认
当[权限策略](/docs/zh/permission-policies)要求在工具执行前进行确认时：
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` 状态。
```bash cURL theme={null}
exec {stream_fd}< <(curl --fail-with-body -sS -N \
"http://localhost:38080/v1/sessions/$SESSION_ID/events/stream?beta=true" \
-H "x-api-key: $OMA_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
# 批准待处理的工具调用
jq -n --arg id "$event_id" \
'{events: [{type: "user.tool_confirmation", tool_use_id: $id, result: "allow"}]}' |
curl --fail-with-body -sS \
"http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_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}<&-
```
```bash CLI theme={null}
# 此工作流不适合转换为一次性的 shell 命令。
# 请改用此代码组中的某个 SDK 示例。
```
```python Python theme={null}
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
```
```typescript TypeScript theme={null}
const stream = await client.beta.sessions.events.stream(session.id);
for await (const event of stream) {
if (event.type !== "session.status_idle") continue;
if (event.stop_reason.type === "end_turn") break;
if (event.stop_reason.type !== "requires_action") continue;
for (const eventId of event.stop_reason.event_ids) {
// 批准待处理的工具调用
await client.beta.sessions.events.send(session.id, {
events: [
{
type: "user.tool_confirmation",
tool_use_id: eventId,
result: "allow",
},
],
});
}
}
```
```csharp C# theme={null}
await foreach (var streamEvent in client.Beta.Sessions.Events.StreamStreaming(session.ID))
{
if (streamEvent.Value is not BetaManagedAgentsSessionStatusIdleEvent idle) continue;
if (idle.StopReason?.Value is BetaManagedAgentsSessionRequiresAction requiresAction)
{
foreach (var eventId in requiresAction.EventIds)
{
// 批准待处理的工具调用
await client.Beta.Sessions.Events.Send(session.ID, new()
{
Events =
[
new BetaManagedAgentsUserToolConfirmationEventParams
{
Type = BetaManagedAgentsUserToolConfirmationEventParamsType.UserToolConfirmation,
ToolUseID = eventId,
Result = BetaManagedAgentsUserToolConfirmationEventParamsResult.Allow,
},
],
});
}
}
else if (idle.StopReason?.Value is BetaManagedAgentsSessionEndTurn)
{
break;
}
}
```
```go Go theme={null}
stream := client.Beta.Sessions.Events.StreamEvents(ctx, session.ID, anthropic.BetaSessionEventStreamParams{})
defer stream.Close()
loop:
for stream.Next() {
event, ok := stream.Current().AsAny().(anthropic.BetaManagedAgentsSessionStatusIdleEvent)
if !ok {
continue
}
switch stopReason := event.StopReason.AsAny().(type) {
case anthropic.BetaManagedAgentsSessionRequiresAction:
for _, eventID := range stopReason.EventIDs {
// 批准待处理的工具调用
if _, err := client.Beta.Sessions.Events.Send(ctx, session.ID, anthropic.BetaSessionEventSendParams{
Events: []anthropic.BetaManagedAgentsEventParamsUnion{{
OfUserToolConfirmation: &anthropic.BetaManagedAgentsUserToolConfirmationEventParams{
Type: anthropic.BetaManagedAgentsUserToolConfirmationEventParamsTypeUserToolConfirmation,
ToolUseID: eventID,
Result: anthropic.BetaManagedAgentsUserToolConfirmationEventParamsResultAllow,
},
}},
}); err != nil {
panic(err)
}
}
case anthropic.BetaManagedAgentsSessionEndTurn:
break loop
}
}
if err := stream.Err(); err != nil {
panic(err)
}
```
```java Java theme={null}
try (var stream = client.beta().sessions().events().streamStreaming(session.id())) {
stream.stream()
.filter(BetaManagedAgentsStreamSessionEvents::isSessionStatusIdle)
.map(idleEvent -> idleEvent.asSessionStatusIdle().stopReason())
.takeWhile(stopReason -> !stopReason.isEndTurn())
.filter(stopReason -> stopReason.isRequiresAction())
.flatMap(stopReason -> stopReason.asRequiresAction().eventIds().stream())
// 批准每个待处理的工具调用
.forEach(toolUseId -> client.beta().sessions().events().send(
session.id(),
EventSendParams.builder()
.addEvent(BetaManagedAgentsUserToolConfirmationEventParams.builder()
.type(BetaManagedAgentsUserToolConfirmationEventParams.Type.USER_TOOL_CONFIRMATION)
.toolUseId(toolUseId)
.result(BetaManagedAgentsUserToolConfirmationEventParams.Result.ALLOW)
.build())
.build()));
}
```
```php PHP theme={null}
$stream = $client->beta->sessions->events->streamStream($session->id);
foreach ($stream as $event) {
if ($event->type === 'session.status_idle' && $event->stopReason) {
if ($event->stopReason->type === 'requires_action') {
foreach ($event->stopReason->eventIDs as $eventId) {
// 批准待处理的工具调用
$client->beta->sessions->events->send(
$session->id,
events: [
[
'type' => 'user.tool_confirmation',
'tool_use_id' => $eventId,
'result' => 'allow',
],
],
);
}
} elseif ($event->stopReason->type === 'end_turn') {
break;
}
}
}
```
```ruby Ruby theme={null}
client.beta.sessions.events.stream_events(session.id).each do |event|
case event
in {type: :"session.status_idle", stop_reason: {type: :requires_action, event_ids:}}
event_ids.each do |event_id|
# 批准待处理的工具调用
client.beta.sessions.events.send_(
session.id,
events: [
{type: "user.tool_confirmation", tool_use_id: event_id, result: "allow"}
]
)
end
in {type: :"session.status_idle", stop_reason: {type: :end_turn}}
break
else
end
end
```
### 恢复空闲会话
会话在交互之间持久存在。除非显式删除会话，否则对话历史记录会被保留。当会话进入空闲状态时，其沙箱会被检查点保存，保留完整的沙箱状态，包括文件系统、已安装的软件包以及智能体创建的任何文件。这使您能够从非活动状态干净地恢复。
虽然会话历史记录会一直保留直到被删除，但沙箱状态仅在沙箱创建后保留 30 天。活动不会延长此窗口期：30 天后，沙箱状态（文件、已安装的工具等）将无法恢复，恢复的会话将从全新的沙箱开始。如果您的工作流程依赖于沙箱内容，请让智能体在窗口期结束前将重要产物写入[输出](/docs/zh/define-outcomes#retrieving-deliverables)。
要恢复会话，像往常一样向其发送 `user.message` 事件：
```bash cURL theme={null}
# 在生产环境中，请传入您想要恢复的会话的已存储 ID。
curl --fail-with-body -sS "http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_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": "Now run the tests against the changes you made earlier."}
]
}
]
}
EOF
```
```bash CLI theme={null}
# 在生产环境中，传入您想要恢复的会话的已存储 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
```
```python Python theme={null}
# 通过发送新的 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.",
},
],
},
],
)
```
```typescript TypeScript theme={null}
// 通过向先前创建的会话发送新的用户事件来恢复该会话。
// 在生产环境中，请传入您想要恢复的会话的已存储 ID。
await 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.",
},
],
},
],
});
```
```csharp C# theme={null}
// 通过 ID 恢复先前创建的会话。在生产环境中，请传入
// 您在创建会话时存储的会话 ID。
await client.Beta.Sessions.Events.Send(session.ID, new()
{
Events =
[
new BetaManagedAgentsUserMessageEventParams
{
Type = BetaManagedAgentsUserMessageEventParamsType.UserMessage,
Content =
[
new BetaManagedAgentsTextBlock
{
Type = BetaManagedAgentsTextBlockType.Text,
Text = "Now run the tests against the changes you made earlier.",
},
],
},
],
});
```
```go Go theme={null}
// 恢复先前创建的会话：向其发送新的 user.message
// 事件。在生产环境中，请传入要恢复的会话的已存储 ID。
if _, err := client.Beta.Sessions.Events.Send(ctx, session.ID, anthropic.BetaSessionEventSendParams{
Events: []anthropic.BetaManagedAgentsEventParamsUnion{{
OfUserMessage: &anthropic.BetaManagedAgentsUserMessageEventParams{
Type: anthropic.BetaManagedAgentsUserMessageEventParamsTypeUserMessage,
Content: []anthropic.BetaManagedAgentsUserMessageEventParamsContentUnion{{
OfText: &anthropic.BetaManagedAgentsTextBlockParam{
Type: anthropic.BetaManagedAgentsTextBlockTypeText,
Text: "Now run the tests against the changes you made earlier.",
},
}},
},
}},
}); err != nil {
panic(err)
}
```
```java Java theme={null}
// 通过 ID 恢复先前创建的会话。在生产环境中，请传入
// 您在创建会话时存储的会话 ID。
client.beta().sessions().events().send(
session.id(),
EventSendParams.builder()
.addEvent(BetaManagedAgentsUserMessageEventParams.builder()
.type(BetaManagedAgentsUserMessageEventParams.Type.USER_MESSAGE)
.addTextContent("Now run the tests against the changes you made earlier.")
.build())
.build());
```
```php PHP theme={null}
// 通过发送新的 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.',
],
],
],
],
);
```
```ruby Ruby theme={null}
# 恢复会话只需向其发送下一个事件。在生产环境中，
# 请传入您在创建会话时存储的会话 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."}
]
}
]
)
```
### 达到会话预算
使用[预算](/docs/zh/budgets)创建的会话会暂停而不是超支。当会话的跟踪标价成本达到上限时，平台会在每个线程的下一个模型请求之前暂停该线程，会话进入空闲状态，其 `stop_reason` 为 `budget_reached`，而不是终止。使总额超过上限的那个请求会运行至完成，因此 `session.usage` 快照报告的 `list_cost` 可能显示为[等于或略微超过上限](/docs/zh/budgets#when-a-session-reaches-its-budget)。在流上，暂停以三个事件的形式依次到达：
1. `session.thread_status_idle`，带有 `stop_reason: budget_reached`，每个线程暂停时各发出一个。
2. `session.usage`，会话累计使用量和跟踪标价成本的快照。
3. `session.status_idle`，带有 `stop_reason: budget_reached`。`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` 是 `requires_action` 而非 `budget_reached`：结清该确认请求不会触发模型请求，因此请照常响应它。
没有任何事件能恢复在上限处暂停的会话。取而代之的是更新会话的预算：将上限更改为任何高于已消耗标价成本的值，或通过使用 `"budget": null` 更新会话来移除预算，都会自动恢复暂停的工作。有关标价成本的跟踪方式和完整的预算更新语义，请参阅[会话预算](/docs/zh/budgets)。
### 发送系统消息
`system.message` 目前受 `claude-opus-4-8`、`claude-fable-5`、`claude-mythos-5` 和 `claude-opus-5` 支持。如果智能体的主模型不支持对话中途的系统注入，该事件将被拒绝并返回 `model_does_not_support_mid_conversation_system` 验证错误；子智能体模型不会被检查，因为 `system.message` 仅落在主线程上。
发送 `system.message` 事件，为智能体提供特权系统级上下文，该上下文适用于随附的轮次及所有后续轮次。与智能体定义上的 `system` 字段（用于设置顶层系统提示）不同，`system.message` 的内容会作为 `role: "system"` 轮次追加到会话的系统上下文中，而不是替换该提示。当智能体在会话中途需要更新的系统级指导时使用它：不同的角色设定、修订的约束条件，或在运行时获取的、应影响模型后续行为的上下文。
```bash cURL theme={null}
curl --fail-with-body -sS "http://localhost:38080/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $OMA_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": "system.message",
"content": [
{"type": "text", "text": "The user's current timezone is America/New_York."}
]
}
]
}
EOF
```
```bash CLI theme={null}
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
```
```python Python theme={null}
client.beta.sessions.events.send(
session.id,
events=[
{
"type": "system.message",
"content": [
{
"type": "text",
"text": "The user's current timezone is America/New_York.",
},
],
},
],
)
```
```typescript TypeScript theme={null}
await client.beta.sessions.events.send(session.id, {
events: [
{
type: "system.message",
content: [
{
type: "text",
text: "The user's current timezone is America/New_York.",
},
],
},
],
});
```
```csharp C# theme={null}
await client.Beta.Sessions.Events.Send(session.ID, new()
{
Events =
[
new BetaManagedAgentsSystemMessageEventParams
{
Type = BetaManagedAgentsSystemMessageEventParamsType.SystemMessage,
Content =
[
new BetaManagedAgentsSystemContentBlock
{
Type = BetaManagedAgentsSystemContentBlockType.Text,
Text = "The user's current timezone is America/New_York.",
},
],
},
],
});
```
```go Go theme={null}
if _, err := client.Beta.Sessions.Events.Send(ctx, session.ID, anthropic.BetaSessionEventSendParams{
Events: []anthropic.BetaManagedAgentsEventParamsUnion{{
OfSystemMessage: &anthropic.BetaManagedAgentsSystemMessageEventParams{
Type: anthropic.BetaManagedAgentsSystemMessageEventParamsTypeSystemMessage,
Content: []anthropic.BetaManagedAgentsSystemContentBlockParam{{
Type: anthropic.BetaManagedAgentsSystemContentBlockTypeText,
Text: "The user's current timezone is America/New_York.",
}},
},
}},
}); err != nil {
panic(err)
}
```
```java Java theme={null}
client.beta().sessions().events().send(
session.id(),
EventSendParams.builder()
.addEvent(BetaManagedAgentsSystemMessageEventParams.builder()
.type(BetaManagedAgentsSystemMessageEventParams.Type.SYSTEM_MESSAGE)
.addTextContent("The user's current timezone is America/New_York.")
.build())
.build());
```
```php PHP theme={null}
$client->beta->sessions->events->send(
$session->id,
events: [
[
'type' => 'system.message',
'content' => [
[
'type' => 'text',
'text' => "The user's current timezone is America/New_York.",
],
],
],
],
);
```
```ruby Ruby theme={null}
client.beta.sessions.events.send_(
session.id,
events: [
{
type: "system.message",
content: [
{type: "text", text: "The user's current timezone is America/New_York."}
]
}
]
)
```
当会话处于空闲状态且 `stop_reason: requires_action` 时，`system.message` 仅在同一请求中跟随在工具结果事件之后时才会被接受；如果单独发送或与 `user.message` 一起发送，它将被拒绝，直到待处理的工具事件得到解决。`content` 接受 1–1000 个文本项。
### 跟踪使用情况
会话对象包含一个 `usage` 字段，其中记录了会话的累计使用情况：令牌计数、服务器工具使用、活跃时间以及按标价计算的成本。在会话进入空闲状态后获取会话，即可读取最新的总计数据。
```json theme={null}
{
"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` 是会话中至少有一个线程在运行的累计时间；并发线程的重叠活动只计算一次，这与会话 `stats` 对象中的 `active_seconds` 不同，后者是对每个线程各自活跃时间的求和。这个去重后的数值即为会话运行时成本的计价时长。`server_tool_use` 统计用于计价的服务器端执行的工具请求数：网络搜索请求按每次请求计入标价成本，而网络抓取请求不收取每次请求费用且不计量，因此 `web_fetch_requests` 显示为 `0`。每个[会话线程](/docs/zh/multiagent-orchestration)自身的 `usage` 也包含 `list_cost` 和 `active_seconds`。各线程的数值是独立四舍五入的，且不包含会话的运行时间成本，因此它们的总和不会与会话的 `list_cost` 完全相等；会话级别的数值才是权威数据。
您无需轮询会话即可观察这些总计数据。`session.usage` 事件会在会话流和事件历史记录中携带相同的累计快照（即 `usage` 对象，以及会话的 `budget`，当会话没有预算时该值为 `null`）。该事件在空闲状态转换时发出，而非按定时器发出：无论停止原因为何，会话都会在进入空闲状态之前立即发出一个此事件；当某个线程在[会话预算](/docs/zh/budgets)处暂停时，也会发出一个。因此，流读取器无需额外获取即可看到某个轮次的最终成本，或触及预算的那部分工作的最终成本。
如需强制执行支出限制，请设置[会话预算](/docs/zh/budgets)，而不是自行轮询使用情况并停止会话。平台会持续对会话的消费进行计价，一旦会话的标价成本达到上限，就会在每个线程的下一次模型请求之前将其暂停；有关这在流中的表现形式，请参阅[达到会话预算](/docs/zh/events-and-streaming#reaching-a-session-budget)。
## 控制台可观测性
OMA 控制台提供了智能体会话的可视化时间线视图。导航至 Open Managed Agents 部分即可查看：
* **会话列表：** 所有会话及其状态、创建时间和智能体
* **追踪视图：** 会话内事件（内容、时间戳、令牌使用情况）的时间顺序视图。追踪视图仅对开发者和管理员开放。
* **工具执行：** 每次工具调用及其结果的详细信息
## 调试技巧
* **检查会话事件：** 会话错误通过 `session.error` 事件传达
* **查看工具结果：** 工具执行失败通常可以解释智能体的异常行为
* **跟踪令牌使用情况：** 监控令牌消耗以优化提示并降低成本
* **使用系统提示：** 在系统提示中添加日志记录指令，让智能体解释其推理过程
* **排查预览问题：** 如果选择接收事件增量的流未按预期运行，请参阅[排查预览问题](/docs/zh/events-and-streaming#troubleshoot-previews)