会话事件流

# 先打开流，再发送用户消息
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}<&-

# 此工作流不适合用一次性的 shell 命令表达。
# 请改用此代码组中的某个 SDK 示例。

# 先打开流，再发送用户消息
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

// 先打开流，再发送用户消息
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;
  }
}

// 先打开流，然后再发送用户消息
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;
    }
}

    // 先打开流，然后发送用户消息
    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)
    }

// 先打开流，然后发送用户消息
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<BetaManagedAgentsStreamSessionEvents> 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;
        }
    }
}

// 先打开流，再发送用户消息
$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();

# 先打开流，再发送用户消息
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

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}<&-

# 此工作流不适合用一次性的 shell 命令表达。
# 请改用此代码组中的某个 SDK 示例。

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

const seenEventIds = new Set<string>();
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;
  }
}

using var stream = await client.Beta.Sessions.Events.WithRawResponse.StreamStreaming(session.ID);

// 流已打开并正在缓冲。在追踪实时事件之前先列出历史记录。
HashSet<string> 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;
    }
}

    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)
    }

try (var stream = client.beta().sessions().events().streamStreaming(session.id())) {
    // 流已打开并正在缓冲。先列出历史记录，再追踪实时事件。
    // 每个事件变体都带有 `id`；从原始 JSON 中读取它以跨变体去重。
    var seenEventIds = new HashSet<String>();
    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()))));
}

$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();

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

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)"'

ant beta:sessions:events list --session-id "$SESSION_ID" \
  --format jsonl --transform '{type,processed_at}'

events = client.beta.sessions.events.list(session.id)
for event in events.data:
    print(f"[{event.type}] {event.processed_at}")

const events = await client.beta.sessions.events.list(session.id);
for (const event of events.data) {
  console.log(`[${event.type}] ${event.processed_at}`);
}

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}");
}

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)
}

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"));
}

$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";
}

events = client.beta.sessions.events.list(session.id)
events.data.each { puts "[#{it.type}] #{it.processed_at}" }

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)"'

ant beta:sessions:events list --session-id "$SESSION_ID" \
  --type agent.tool_use --type agent.tool_result \
  --format jsonl --transform '{type,processed_at}'

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}")

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}`);
}

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}");
}

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)
}

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 中，通过 EventListParams 传入您需要的类型；请参阅 OMA PHP SDK。

events = client.beta.sessions.events.list(
  session.id,
  types: ["agent.tool_use", "agent.tool_result"]
)
events.data.each { puts "[#{it.type}] #{it.processed_at}" }