세션 이벤트 스트림
세션 이벤트 스트림 (Session event stream)
Claude Managed Agents와의 통신은 이벤트 기반이에요. 에이전트에 사용자 이벤트를 보내고, 에이전트와 세션 이벤트를 받아서 상태를 추적해요.
출처: 문서
본문
Claude Managed Agents와의 통신은 이벤트 기반이에요. 에이전트에 사용자 이벤트를 보내고, 에이전트와 세션 이벤트를 받아서 상태를 추적해요.
이벤트 유형 (Event types)
이벤트는 두 방향으로 흘러요.
- **사용자 이벤트(user events)**와 **시스템 이벤트(system events)**는 에이전트에 보내는 것이에요:
user.*이벤트는 세션을 시작하고 진행됨에 따라 조종하며,system.message는 수반되는 턴과 이후 모든 턴에 적용되는 시스템 수준 컨텍스트를 추가해요. - 세션 이벤트(session events), 스팬 이벤트(span events), **에이전트 이벤트(agent events)**는 세션 상태와 에이전트 진행 상황에 대한 관찰을 위해 여러분에게 보내져요. 옵트인한 스트림 연결은 이벤트 델타도 받아요.
세션, 스팬, 에이전트, 사용자, 시스템 이벤트 유형 문자열은 {domain}.{action} 명명 규칙을 따라요. 스트림 전용 델타 미리보기 이벤트(event_start, event_delta)가 예외예요. 전체 카탈로그는 reference의 이벤트 유형을 참고하세요. 웹훅 이벤트 유형은 별개이며 일부 이름이 스트림과 달라요(예: session.status_idle이 아니라 session.status_idled).
모든 지속 이벤트는 처리가 끝났을 때 설정되는 processed_at 타임스탬프를 포함해요. 여러분이 보내는 이벤트에서는, 이벤트가 앞선 이벤트들 뒤에 대기 중인 동안 processed_at은 null이에요. 예외는 user.define_outcome, user.custom_tool_result, user.tool_result로, 수신 즉시 처리되고 processed_at이 이미 채워진 채로 에코돼요.
이벤트 통합하기 (Integrating events)
<CodeGroup>
```bash cURL
curl --fail-with-body -sS "https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $ANTHR...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
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
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
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#
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
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
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
$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
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"
}
]
}
]
)
```
</CodeGroup>
에이전트를 실행 중에 멈추려면 `user.interrupt` 이벤트를 보내고, 그다음 `user.message` 이벤트로 방향을 바꾸세요:
<CodeGroup>
```bash cURL
# Agent is currently analyzing a file...
# Interrupt with a new direction:
curl --fail-with-body -sS "https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $ANTHR...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
# Agent is currently analyzing a file...
# Interrupt with a new direction:
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
# Agent is currently analyzing a file...
# Interrupt with a new direction:
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
// Agent is currently analyzing a file...
// Interrupt with a new direction:
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#
// Agent is currently analyzing a file...
// Interrupt with a new direction:
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
// Agent is currently analyzing a file...
// Interrupt with a new direction:
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
// Agent is currently analyzing a file...
// Interrupt with a new direction:
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
// Agent is currently analyzing a file...
// Interrupt with a new direction:
$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
# Agent is currently analyzing a file...
# Interrupt with a new direction:
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."}
]
}
]
)
```
</CodeGroup>
호출은 이벤트가 큐에 들어가는 즉시 반환되고, 중단의 `processed_at`은 에이전트가 그것을 적용할 때까지 null로 남아요. 진행 중인 모델 응답은 즉시 멈춰요. 도구 호출이 실행되는 동안에는 중단 적용에 더 오래 걸릴 수 있고, 적용될 때까지 세션은 `running`으로 남아요. 그러면 `user.interrupt` 이벤트가 스트림에 나타나고, 중단된 턴은 `session.status_idle` 이벤트로 끝나요. 그 `stop_reason`은 스스로 끝난 턴과 같은 값인 `end_turn`이에요. 중단 특유의 정지 이유는 없어요. 에이전트는 중단 후 보낸 `user.message`로 다음 턴을 시작해요.
<CodeGroup>
```bash cURL
# Open the stream first, then send the user message
exec {stream}< <(
curl --fail-with-body -sS -N \
"https://api.anthropic.com/v1/sessions/$SESSION_ID/events/stream?beta=true" \
-H "x-api-key: $ANTHR...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 \
"https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $ANTHR...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
# This workflow does not translate well to a one-off shell command.
# Use one of the SDK examples in this code group instead.
```
```python Python
# Open the stream first, then send the user message
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
// Open the stream first, then send the user message
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" }]
}
]
});
events: for await (const event of stream) {
switch (event.type) {
case "agent.message":
for (const block of event.content) {
if (block.type === "text") {
process.stdout.write(block.text);
}
}
break;
case "session.status_idle":
break events;
case "session.error":
console.log(`\n[Error: ${event.error?.message ?? "unknown"}]`);
break events;
}
}
```
```csharp C#
// Open the stream first, then send the user message
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)
{
if (block.Value is BetaManagedAgentsTextBlock textBlock)
{
Console.Write(textBlock.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
// Open the stream first, then send the user message
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:
// concrete-typed list: 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
// Open the stream first, then send the user message
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;
events:
for (var event : events) {
switch (event.type().value()) {
case AGENT_MESSAGE -> event.asAgentMessage().content().forEach(block -> block.text().ifPresent(textBlock -> IO.print(textBlock.text())));
case SESSION_STATUS_IDLE -> {
break events;
}
case SESSION_ERROR -> {
// The `message` field spans all error variants; read it from the raw JSON.
var errorMessage =
event.asSessionError().error()._json().orElse(null) instanceof JsonObject json
? json.values().get("message").asStringOrThrow()
: "unknown";
IO.println("\n[Error: " + errorMessage + "]");
break events;
}
}
}
}
```
```php PHP
// Open the stream first, then send the user message
$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 (true) {
$event instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsAgentMessageEvent => array_walk(
$event->content,
static fn ($block) => $block instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsTextBlock ? print($block->text) : null,
),
$event instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsSessionErrorEvent => 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
# Open the stream first, then send the user message
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
when Anthropic::Beta::Sessions::BetaManagedAgentsAgentMessageEvent
event.content.each { print it.text }
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionStatusIdleEvent
break
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionErrorEvent
puts "\n[Error: #{event.error&.message || "unknown"}]"
break
else
# ignore other event types
end
end
```
</CodeGroup>
이벤트를 놓치지 않고 기존 세션에 다시 연결하려면:
1. 새 스트림을 여세요.
2. 전체 이벤트 기록을 나열해 이미 본 이벤트 ID 세트를 시드하세요.
3. 실시간 스트림을 tail하면서 기록 목록이 이미 반환한 이벤트는 건너뛰세요.
<CodeGroup>
```bash cURL
exec {stream}< <(
curl --fail-with-body -sS -N \
"https://api.anthropic.com/v1/sessions/$SESSION_ID/events/stream?beta=true" \
-H "x-api-key: $ANTHR...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"
)
# Stream is open and buffering. List history before tailing live.
declare -A seen_event_ids
while IFS= read -r event_id; do
seen_event_ids[$event_id]=1
done < <(
curl --fail-with-body -sS \
"https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $ANTHR...KEY" \
-H "anthropic-version: 2023-06-01" \
-H "anthropic-beta: managed-agents-2026-04-01" \
-H "content-type: application/json" | jq -r '.data[].id'
)
# Tail live events, skipping anything already seen
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
# This workflow does not translate well to a one-off shell command.
# Use one of the SDK examples in this code group instead.
```
```python Python
with client.beta.sessions.events.stream(session.id) as stream:
# Stream is open and buffering. List history before tailing live.
history = client.beta.sessions.events.list(session.id)
seen_event_ids = {past_event.id for past_event in history}
# Tail live events, skipping anything already seen
for event in stream:
if event.type == "event_start" or event.type == "event_delta":
# Delta previews aren't enabled on this connection.
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
const seenEventIds = new Set<string>();
const stream = await client.beta.sessions.events.stream(session.id);
// Stream is open and buffering. List history before tailing live.
for await (const event of client.beta.sessions.events.list(session.id)) {
seenEventIds.add(event.id);
}
// Tail live events, skipping anything already seen
tail: for await (const event of stream) {
// Preview events (event_start/event_delta) carry no top-level id
if (event.type === "event_start" || event.type === "event_delta") continue;
if (seenEventIds.has(event.id)) continue;
seenEventIds.add(event.id);
switch (event.type) {
case "agent.message":
for (const block of event.content) {
if (block.type === "text") {
process.stdout.write(block.text);
}
}
break;
case "session.status_idle":
break tail;
}
}
```
```csharp C#
using var stream = await client.Beta.Sessions.Events.WithRawResponse.StreamStreaming(session.ID);
// Stream is open and buffering. List history before tailing live.
HashSet<string> seenEventIds = [];
var history = await client.Beta.Sessions.Events.List(session.ID);
await foreach (var pastEvent in history.Paginate())
{
seenEventIds.Add(pastEvent.ID);
}
// Tail live events, skipping anything already seen
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)
{
if (block.Value is BetaManagedAgentsTextBlock textBlock)
{
Console.Write(textBlock.Text);
}
}
}
else if (streamEvent.Value is BetaManagedAgentsSessionStatusIdleEvent)
{
break;
}
}
```
```go Go
stream := client.Beta.Sessions.Events.StreamEvents(ctx, session.ID, anthropic.BetaSessionEventStreamParams{})
defer stream.Close()
// Stream is open and buffering. List history before tailing live.
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 live events, skipping anything already seen
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:
// concrete-typed list: 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
try (var stream = client.beta().sessions().events().streamStreaming(session.id())) {
// Stream is open and buffering. List history before tailing live.
// Every event variant carries `id`; read it from the raw JSON to dedup across variants.
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());
}
}
// Tail live events; Set.add returns false for already-seen IDs, skipping the replay.
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
$stream = $client->beta->sessions->events->streamStream($session->id);
// Stream is open and buffering. List history before tailing live.
$seenEventIds = [];
foreach ($client->beta->sessions->events->list($session->id)->pagingEachItem() as $event) {
$seenEventIds[$event->id] = true;
}
// Tail live events, skipping anything already seen
foreach ($stream as $event) {
if (isset($seenEventIds[$event->id])) {
continue;
}
$seenEventIds[$event->id] = true;
match (true) {
$event instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsAgentMessageEvent => array_walk(
$event->content,
static fn ($block) => $block instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsTextBlock ? print($block->text) : null,
),
default => null,
};
if ($event->type === 'session.status_idle') {
break;
}
}
$stream->close();
```
```ruby Ruby
stream = client.beta.sessions.events.stream_events(session.id)
# Stream is open and buffering. List history before tailing live.
seen_event_ids = Set.new
client.beta.sessions.events.list(session.id).auto_paging_each { seen_event_ids << it.id }
# Tail live events, skipping anything already seen — Set#add? returns nil for duplicates
stream.each do |event|
next unless seen_event_ids.add?(event.id)
case event
when Anthropic::Beta::Sessions::BetaManagedAgentsAgentMessageEvent
event.content.each { print it.text }
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionStatusIdleEvent
break
else
# ignore other event types
end
end
```
</CodeGroup>
<CodeGroup>
```bash cURL
curl --fail-with-body -sS "https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true" \
-H "x-api-key: $ANTHR...KEY" \
-H "anthropic-version: 2023-06-01" \
-H "anthropic-beta: managed-agents-2026-04-01" \
-H "content-type: application/json"
```
```bash CLI
ant beta:sessions:events list --session-id "$SESSION_ID" --format jsonl
```
```python Python
events = client.beta.sessions.events.list(session.id)
for event in events.data:
print(f"[{event.type}] {event.processed_at}")
```
```typescript TypeScript
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#
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
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
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
$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
events = client.beta.sessions.events.list(session.id)
events.data.each { puts "[#{it.type}] #{it.processed_at}" }
```
</CodeGroup>
특정 이벤트 유형만 반환하려면 `types` 필터를 전달하세요:
<CodeGroup>
```bash cURL
curl --fail-with-body -sS "https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true&types[]=agent.tool_use&types[]=agent.tool_result" \
-H "x-api-key: $ANTHR...KEY" \
-H "anthropic-version: 2023-06-01" \
-H "anthropic-beta: managed-agents-2026-04-01"
```
```bash CLI
ant beta:sessions:events list --session-id "$SESSION_ID" \
--type agent.tool_use --type agent.tool_result \
--format jsonl
```
```python Python
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
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#
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
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
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
// In PHP, pass the types you want on EventListParams; see the Anthropic PHP SDK.
```
```ruby Ruby
events = client.beta.sessions.events.list(
session.id,
types: ["agent.tool_use", "agent.tool_result"]
)
events.data.each { puts "[#{it.type}] #{it.processed_at}" }
```
</CodeGroup>
이벤트 델타 (Event deltas)
기본적으로 에이전트의 응답 텍스트는 버퍼링된 agent.message 이벤트로 스트림에 도달하며, 각 이벤트는 그것을 생성한 모델 요청이 끝난 뒤에만 발행돼요. 이벤트 델타(event deltas)는 그 텍스트를 모델이 아직 생성하는 동안 증분적으로 라이브 미리보기로 렌더링할 수 있게 해줘요. 미리보기는 응답이 아니에요: 미리보기는 최선 노력(best-effort) 표시 보조 수단이고, 버퍼링된 agent.message가 항상 권위 있는 기록이에요. 미리보기를 무시하는 클라이언트도 여전히 완전하고 정확한 스트림을 받아요.
미리보기 옵트인 (Opt in to previews)
미리보기는 스트림 연결별로 옵트인돼요. 읽고 있는 스트림에 event_deltas[] 쿼리 파라미터를 추가하고, 미리보기할 각 이벤트 유형마다 한 번씩 반복하세요. []는 셸 glob 패턴이라 셸에서 요청을 만들 때마다 URL을 인용하고, 예시는 괄호를 %5B%5D로 퍼센트 인코딩하는데 이도 동작해요. 두 스트림 엔드포인트 모두 파라미터를 받아요: GET /v1/sessions/{session_id}/events/stream의 세션 수준 스트림과 각 세션 스레드의 자체 스트림 GET /v1/sessions/{session_id}/threads/{thread_id}/stream. 허용되는 값은 agent.message와 agent.thinking이고, 다른 값은 400 오류를 반환하며 100개 이상의 값을 가진 요청도 그래요. 하위 에이전트의 미리보기는 그 하위 에이전트의 자체 스레드 스트림에 나타나요.
미리보기된 이벤트가 시작되면 스트림이 다가오는 이벤트의 유형과 id를 담은 event_start를 발행해요:
{
"type": "event_start",
"event": {
"type": "agent.message",
"id": "sevt_01abc..."
}
}
agent.message의 경우 시작 뒤에 증분 텍스트를 담은 event_delta 이벤트가 따라와요. 각 델타는 event_id로 확장하는 이벤트를, delta.index로 확장하는 콘텐츠 블록을 이름 붙여요:
{
"type": "event_delta",
"event_id": "sevt_01abc...",
"delta": {
"type": "content_delta",
"index": 0,
"content": {
"type": "text",
"text": "Here is the summary"
}
}
}
agent.thinking 이벤트가 미리보기되면 event_start만 발행돼요. event_delta 이벤트는 뒤따르지 않고, 미리보기를 마무리하는 버퍼링된 agent.thinking 이벤트는 thinking 콘텐츠를 담지 않아요. 그것은 콘텐츠 운반자가 아니라 진행 신호예요.
지속 이벤트와 달리 event_start와 event_delta는 자체 id나 processed_at이 없어요. 그것들이 옮기는 유일한 식별자는 미리보기하는 이벤트의 id예요.
누적하고 조정하기 (Accumulate and reconcile)
이벤트 델타를 지원하는 모든 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를 생성해요. 와이어에서 이것은 그 시퀀스의 미리보기된 부분이며 연결의 다른 버퍼링된 이벤트들과 인터리브돼요:
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 줄은 텍스트 조각마다 한 번씩 반복돼요. 도착하는 각 이벤트를 처리하세요:
event_start에서 발표된id를 기록하세요. 식별자는 항상 정렬돼요:event_start.event.id, 모든event_delta.event_id, 버퍼링된agent.message의id가 같은 값이에요.- 각
event_delta에서delta.content.text를(event_id, delta.index)의 항목에 추가하고 진행 중인 텍스트를 렌더링하세요. 한index의 첫 델타가 그 항목을 만들어요. - 버퍼링된
agent.message가 도착하면id로 맞추고, 누적된 미리보기를 버리고 대신 메시지의 콘텐츠를 렌더링하세요. span.model_request_end에서 버퍼링된 이벤트로 조정되지 않은 미리보기를 닫으세요. 더 이상 그에 대한 델타가 오지 않아요. 턴이 오류나 중단으로 끝나면 버퍼링된 이벤트가 결코 도착하지 않을 수 있어도span.model_request_end는 여전히 도착해요.
패턴이 의존하는 보장:
- 미리보기의 델타를 도착 순서로 연결하고
(event_id, index)로 키잉하면 버퍼링된 이벤트의content[index].text의 접두사가 돼요(전체 텍스트가 아니라도 되는데, 부하 시 델타가 벗겨질 수 있기 때문이에요). - 연결은
event_id당 최대 하나의event_start를 발행하고, 버퍼링된 이벤트가 그 연결이 그id에 전달하는 마지막 것이에요.
curl --fail-with-body -sS
"https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true"
-H "x-api-key: $ANTHR...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
Accumulate deltas keyed by (message id, content index); the final
agent.message carries the full text, so it replaces every preview for that 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
# This workflow does not translate well to a one-off shell command.
# Use one of the SDK examples in this code group instead.
# Preview snapshots, keyed by event id. accumulate_managed_agents_event folds each
# event_start / event_delta into an agent.message snapshot; the buffered
# agent.message replaces it.
previews: dict[str, BetaManagedAgentsAgentMessageEvent] = {}
# Opt in to agent.message previews on this connection
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":
# The buffered event is the record: it replaces and closes the preview
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":
# No more deltas are coming. Close any preview whose
# buffered event never arrived.
for event_id in previews:
print(f"span.model_request_end closing preview for {event_id}")
previews.clear()
case "session.status_idle":
break
// Preview snapshots, keyed by event id. `accumulateManagedAgentsEvent`
// folds event_start / event_delta previews into an agent.message snapshot.
const previews = new Map<string, BetaManagedAgentsAgentMessageEvent>();
// Opt in to agent.message previews for this connection only
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" }]
}
]
});
deltas: for await (const event of stream) {
switch (event.type) {
case "event_start": {
// 1. Note the announced id and open the snapshot. Deltas and the
// buffered event carry the same id.
const preview = accumulateManagedAgentsEvent(undefined, event);
if (preview) previews.set(event.event.id, preview);
console.log(`event_start ${event.event.type} ${event.event.id}`);
break;
}
case "event_delta": {
// 2. Fold the fragment into the snapshot and render it
const preview = accumulateManagedAgentsEvent(previews.get(event.event_id), event);
if (preview) {
previews.set(event.event_id, preview);
const text = preview.content
.map((block) => (block.type === "text" ? block.text : ""))
.join("");
console.log(`event_delta preview: ${JSON.stringify(text)}`);
}
break;
}
case "agent.message": {
// 3. The buffered event is the record: it replaces and closes the preview
const message = accumulateManagedAgentsEvent(previews.get(event.id), event);
previews.delete(event.id);
const text = message.content
.map((block) => (block.type === "text" ? block.text : ""))
.join("");
console.log(`agent.message ${event.id} ${JSON.stringify(text)}`);
break;
}
case "span.model_request_end":
// 4. No more deltas are coming. Close any preview that was never reconciled.
for (const eventId of previews.keys()) {
console.log(`span.model_request_end closing preview for ${eventId}`);
}
previews.clear();
break;
case "session.status_idle":
break deltas;
}
}
stream.controller.abort();
// Opt in to event deltas: agent.message events are previewed as they are produced.
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.",
},
],
},
],
});
// Accumulate preview fragments per (event id, content index). The buffered
// agent.message that follows carries the complete content, so it replaces the
// accumulated preview rather than appending to it.
Dictionary<string, SortedDictionary<long, string>> previews = [];
await foreach (var streamEvent in stream.Enumerate())
{
if (streamEvent.TryPickStartEvent(out var start))
{
// A preview opened for the event with this id. This stream only opts in
// to agent.message deltas; TryPick* returns false instead of throwing,
// so other preview types (including ones added later) are skipped.
if (start.Event.TryPickAgentMessage(out var preview))
{
Console.WriteLine($"event_start {preview.Type.Raw()} {preview.ID}");
}
}
else if (streamEvent.TryPickDeltaEvent(out var delta))
{
// Insert at a new index, append at an existing one
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))
{
// Deltas are best-effort: discard the preview and use the buffered event
previews.Remove(message.ID);
var text = string.Concat(message.Content.Select(block =>
block.TryPickBetaManagedAgentsTextBlock(out var textBlock) ? textBlock.Text : ""));
Console.WriteLine($"agent.message {message.ID} {text}");
}
else if (streamEvent.TryPickSpanModelRequestEndEvent(out _))
{
// No more deltas are coming; close any preview that was never reconciled.
foreach (var eventId in previews.Keys)
{
Console.WriteLine($"span.model_request_end closing preview for {eventId}");
}
previews.Clear();
}
else if (streamEvent.TryPickSessionStatusIdleEvent(out _))
{
break;
}
}
// Opt in to incremental previews of agent.message events
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)
}
// The accumulator folds event_start / event_delta fragments into
// per-event-id agent.message snapshots. The zero value is ready to use.
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:
// The buffered event carries the complete content: the accumulator
// replaces the preview with it
fmt.Printf("agent.message %s %q\n", event.ID, previews.AgentMessageText(event.ID))
case anthropic.BetaManagedAgentsSpanModelRequestEndEvent:
// No more deltas are coming for this request. The accumulator
// drops its snapshots here, closing any preview that was never
// reconciled by a buffered 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()
// Preview text, keyed by event ID then content index. The buffered agent.message replaces it.
Map<String, Map<Long, StringBuilder>> previews = new HashMap<>();
// Opt in to agent.message previews on this connection
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<BetaManagedAgentsStreamSessionEvents> events = stream.stream()::iterator;
deltas:
for (var event : events) {
switch (event.type().value()) {
case EVENT_START -> {
if (event.asEventStart().event().isAgentMessage()) {
var preview = event.asEventStart().event().asAgentMessage();
IO.println("event_start " + preview.type().asString() + " " + preview.id());
}
}
case EVENT_DELTA -> {
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);
}
case AGENT_MESSAGE -> {
// The buffered event is the record: drop its preview, render its content
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);
}
case SPAN_MODEL_REQUEST_END -> {
// No more deltas are coming. Close any preview whose buffered event never arrived.
previews.keySet().forEach(eventId ->
IO.println("span.model_request_end closing preview for " + eventId));
previews.clear();
}
case SESSION_STATUS_IDLE -> {
break deltas;
}
}
}
}
// In PHP, set eventDeltas on EventStreamParams and accumulate with Anthropic\Lib\Sessions\EventAccumulator.
# Opt in to event deltas: agent.message previews stream as incremental fragments.
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."}]
}]
)
# Accumulate preview fragments by (event_id, index) into explicitly mutable
# (`+""`) buffers so `<<` can append in place. The buffered agent.message with
# the same id is authoritative and replaces whatever the deltas built up.
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
when Anthropic::Beta::BetaManagedAgentsStartEvent
puts "event_start #{event.event.type} #{event.event.id}"
when Anthropic::Beta::BetaManagedAgentsDeltaEvent
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}"
when Anthropic::Beta::Sessions::BetaManagedAgentsAgentMessageEvent
# Replace: drop the accumulated preview and render the complete event.
buffers.delete(event.id)
puts "agent.message #{event.id} #{event.content.map(&:text).join.inspect}"
when Anthropic::Beta::Sessions::BetaManagedAgentsSpanModelRequestEndEvent
# No more deltas are coming. Close any preview that was never reconciled.
buffers.each_key { |event_id| puts "span.model_request_end closing preview for #{event_id}" }
buffers.clear
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionStatusIdleEvent
break
else
# ignore other event types
end
end
세션 스레드 이벤트 미리보기 (Preview session thread events)
멀티에이전트 세션에서 모든 세션 스레드는 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는 스레드 스트림에서 세션 수준 스트림과 같은 형태를 가지며, 누적하고 조정하기 패턴이 그대로 적용돼요. 한 가지 조정은 장부에 있어요: 스트림 연결마다 누산기 인스턴스 하나를 실행하세요.
The child thread's stream takes the same event_deltas[] parameter as the
session stream. Percent-encode the brackets (%5B%5D) and quote the URL.
exec {stream}< <(
curl --fail-with-body -sS -N
"https://api.anthropic.com/v1/sessions/$SESSION_ID/threads/$THREAD_ID/stream?beta=true&event_deltas%5B%5D=agent.message"
-H "x-api-key: $ANTHR...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) # The buffered event is the authoritative record; render its content. printf '\n' jq -j '.content[] | select(.type == "text") | .text' <<<"$event_json" printf '\n' ;; session.thread_status_idle) break ;; esac done exec {stream}<&-
```bash CLI
# List the session's threads and pick a child: child threads carry a non-null
# parent_thread_id, and the primary thread's parent_thread_id is null
# (--transform's #(parent_thread_id!=~null) query matches non-null values).
THREAD_ID=$(ant beta:sessions:threads list \
--session-id "$SESSION_ID" \
--format raw --transform 'data.#(parent_thread_id!=~null).id' --raw-output)
# The child thread's stream takes the same event_deltas parameter as the
# session stream, one --event-delta flag per event type to preview. @tostr
# re-encodes each text field as a JSON string, so every value stays on one
# YAML line and jq's fromjson recovers the original text.
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
# The buffered event is the authoritative record; render its content.
printf '\n'
jq -r fromjson <<<"${line#text: }" ;;
esac
done
exec {stream}<&-
# List the session's threads and pick a child: child threads carry a non-null
# parent_thread_id, and the primary thread's parent_thread_id is null.
child_thread = next(
thread
for thread in client.beta.sessions.threads.list(session.id)
if thread.parent_thread_id is not None
)
# The child thread's stream takes the same event_deltas parameter as the
# session stream.
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":
# The buffered event is the authoritative record; render its content
print()
for block in event.content:
if block.type == "text":
print(block.text, end="")
print()
case "session.thread_status_idle":
break
// List the session's threads and pick a child: child threads carry a non-null
// parent_thread_id, and the primary thread's parent_thread_id is 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");
// The child thread's stream takes the same event_deltas parameter as the
// session stream.
const stream = await client.beta.sessions.threads.events.stream(childThreadId, {
session_id: session.id,
event_deltas: ["agent.message"],
});
threadDeltas: for await (const event of stream) {
switch (event.type) {
case "event_delta":
process.stdout.write(event.delta.content.text);
break;
case "agent.message": {
// The buffered event is the authoritative record; render its content.
process.stdout.write("\n");
const text = event.content
.map((block) => (block.type === "text" ? block.text : ""))
.join("");
console.log(text);
break;
}
case "session.thread_status_idle":
break threadDeltas;
}
}
stream.controller.abort();
// List the session's threads and pick a child: child threads carry a non-null
// parent_thread_id, and the primary thread's parent_thread_id is null.
var threads = await client.Beta.Sessions.Threads.List(session.ID);
var childThread = threads.Items.First(thread => thread.ParentThreadID is not null);
// The child thread's stream takes the same event_deltas parameter as the
// session stream.
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))
{
// The buffered event is the authoritative record; render its content.
Console.WriteLine();
var text = string.Concat(message.Content.Select(block =>
block.TryPickBetaManagedAgentsTextBlock(out var textBlock) ? textBlock.Text : ""));
Console.WriteLine(text);
}
else if (streamEvent.TryPickSessionThreadStatusIdleEvent(out _))
{
break;
}
}
// List the session's threads and pick a child: child threads carry a non-null
// parent_thread_id, and the primary thread's parent_thread_id is 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)
}
// The child thread's stream takes the same event_deltas parameter as the
// session stream; run one read loop per stream connection.
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:
// The buffered event is the authoritative record; render its content.
fmt.Println()
// concrete-typed list: 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()
// List the session's threads and pick a child: child threads carry a non-null
// parent_thread_id, and the primary thread's parent_thread_id is null.
var childThread = client.beta().sessions().threads().list(session.id()).autoPager().stream()
.filter(thread -> thread.parentThreadId().isPresent())
.findFirst()
.orElseThrow();
// The child thread's stream takes the same event_deltas parameter as the session
// stream. Its params class shares the session-level one's simple name, so qualify it.
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<BetaManagedAgentsStreamSessionThreadEvents> events = stream.stream()::iterator;
threadDeltas:
for (var event : events) {
switch (event.type().value()) {
case EVENT_DELTA -> IO.print(event.asEventDelta().delta().content().text());
case AGENT_MESSAGE -> {
// The buffered event is the authoritative record; render its content.
IO.println();
event.asAgentMessage().content().forEach(block -> block.text().ifPresent(textBlock -> IO.print(textBlock.text())));
IO.println();
}
case SESSION_THREAD_STATUS_IDLE -> {
break threadDeltas;
}
}
}
}
// In PHP, set eventDeltas on the thread EventStreamParams and accumulate with Anthropic\Lib\Sessions\EventAccumulator.
# List the session's threads and pick a child: child threads carry a non-null
# parent_thread_id, and the primary thread's parent_thread_id is null.
child_thread = client.beta.sessions.threads.list(session.id).to_enum.find { it.parent_thread_id }
# The child thread's stream takes the same event_deltas parameter as the
# session stream.
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
when Anthropic::Beta::BetaManagedAgentsDeltaEvent
print event.delta.content.text
when Anthropic::Beta::Sessions::BetaManagedAgentsAgentMessageEvent
# The buffered event is the authoritative record; render its content.
puts
event.content.each { print it.text }
puts
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionThreadStatusIdleEvent
break
else
# ignore other event types
end
end
읽기 루프는 세션 스레드의 턴이 끝나고 스레드가 idle이 될 때 발행되는 이벤트인 session.thread_status_idle에서 종료돼요.
제한 사항 (Limitations)
미리보기는 응답성에 맞춰 튜닝돼요. 이 제약들에 맞춰 구축하세요:
- 최선 노력 (Best effort): 부하 시 서버가 이벤트의 델타를 벗겨낼 수 있어요. 그럴 때 텍스트의 연속 접두사를 받고 그 이벤트에 대해 더 이상 델타가 오지 않아요. 버퍼링된
agent.message는 여전히 완전히 도착해요. 누적된 미리보기를 결코 최종으로 취급하지 마세요. - 재연결 시 재생 없음 (No replay on reconnect): 델타는 옵트인한 연결이 열려 있는 동안에만 그 연결에 전달돼요. 세션 수준 스트림과 각 세션 스레드 스트림 모두에 적용되며, 모델 요청이 시작된 뒤에 열린 연결은 그 진행 중인 이벤트에 대한 델타를 받지 않아요. 스트림이 끊기면 스트리밍 이벤트 탭의 재연결 절차를 따르세요: 스트림을 다시 열고 이벤트 기록을 나열하세요. 기록에는 연결이 끊긴 동안 발행된 버퍼링된 이벤트(여러분의 미리보기가 기다리던
agent.message포함)가 포함돼요. 놓친 델타를 다시 요청할 방법은 없어요. - 한 스레드, 텍스트 전용 (One thread, text only): 미리보기는 연결이 읽고 있는 스레드의 어시스턴트 텍스트를 다뤄요. 도구 사용, 도구 결과, MCP 결과, 다른 세션 스레드의 활동은 그 연결에서 결코 미리보기되지 않아요.
- 시작 전용
agent.thinking(Start-onlyagent.thinking):agent.thinking미리보기는 thinking 블록이 시작됐다는 신호로event_start만 발행해요.event_delta이벤트는 뒤따르지 않아요. - 결코 지속되지 않음 (Never persisted):
event_start와event_delta는 라이브 스트림에만 존재해요. 세션의 이벤트 기록(GET /v1/sessions/{session_id}/events)이나 어떤 세션 스레드의 이벤트 기록에도 나타나지 않아요.
미리보기 문제 해결 (Troubleshoot previews)
스트림이 예상대로 동작하지 않으면:
| 보이는 것 (You see) | 의미 (What it means) |
|---|---|
버퍼링된 이벤트는 있는데 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만 허용돼요. |
추가 시나리오 (Additional scenarios)
커스텀 도구 호출 처리 (Handling custom tool calls)
에이전트가 커스텀 도구를 호출할 때:
- 세션이 도구 이름과 입력을 담은
agent.custom_tool_use이벤트를 발행해요. - 세션이
stop_reason: requires_action을 담은session.status_idle이벤트로 멈춰요. 차단 이벤트 ID는stop_reason.event_ids배열에 있어요. - 여러분의 시스템에서 도구를 실행하고 각각에
user.custom_tool_result이벤트를 보내며,custom_tool_use_id파라미터에 이벤트 ID와 결과 콘텐츠를 전달하세요. - 모든 차단 이벤트가 해결되면 세션이
running으로 돌아가요.
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
# Execute the tool and send the result back
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
"https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true"
-H "x-api-key: $ANTHR...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
# This workflow does not translate well to a one-off shell command.
# Use one of the SDK examples in this code group instead.
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:
# Look up the custom tool use event and execute it
tool_event = events_by_id[event_id]
result = call_tool(tool_event.name, tool_event.input)
# Send the result back
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
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) {
// Look up the custom tool use event and execute it
const toolEvent = eventsById.get(eventId);
if (!toolEvent) continue;
const result = await callTool(toolEvent.name, toolEvent.input);
// Send the result back
await client.beta.sessions.events.send(session.id, {
events: [
{
type: "user.custom_tool_result",
custom_tool_use_id: eventId,
content: [{ type: "text", text: result }],
},
],
});
}
}
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)
{
// Look up the custom tool use event and execute it
var toolEvent = eventsById[eventId];
var result = await CallTool(toolEvent.Name, toolEvent.Input);
// Send the result back
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;
}
}
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 {
// Look up the custom tool use event and execute it
toolEvent := eventsByID[eventID]
result := callTool(toolEvent.Name, toolEvent.Input)
// Send the result back
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)
}
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 -> {
// Look up the custom tool use event and execute it
var toolEvent = eventsById.get(eventId);
var result = callTool(toolEvent.name(), toolEvent.input());
// Send the result back
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());
});
}
$stream = $client->beta->sessions->events->streamStream($session->id);
foreach ($stream as $event) {
if ($event instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsSessionStatusIdleEvent && $event->stopReason) {
switch (true) {
case $event->stopReason instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsSessionRequiresAction:
foreach ($event->stopReason->eventIDs as $eventId) {
// Look up the custom tool use event and execute it
$toolEvent = $eventsById[$eventId];
$result = callTool($toolEvent->name, $toolEvent->input);
// Send the result back
$client->beta->sessions->events->send(
$session->id,
events: [
[
'type' => 'user.custom_tool_result',
'custom_tool_use_id' => $eventId,
'content' => [['type' => 'text', 'text' => $result]],
],
],
);
}
break;
case $event->stopReason instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsSessionEndTurn:
break 2;
}
}
}
client.beta.sessions.events.stream_events(session.id).each do |event|
case event
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionStatusIdleEvent
stop_reason = event.stop_reason
case stop_reason
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionRequiresAction
stop_reason.event_ids.each do |event_id|
# Look up the custom tool use event and execute it
tool_event = events_by_id[event_id]
result = call_tool.call(tool_event.name, tool_event.input)
# Send the result back
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
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionEndTurn
break
end
end
end
도구 확인 (Tool confirmation)
도구 호출은 always_ask 권한 정책 아래에서, 또는 auto 아래에서 서버가 판단에 이르지 못할 때 여러분의 확인을 기다려요. 그런 일이 생기면:
- 세션이
agent.tool_use또는agent.mcp_tool_use이벤트를 발행해요. - 세션이
stop_reason.type이requires_action인session.status_idle이벤트로 멈춰요. 차단 이벤트 ID는stop_reason.event_ids배열에 있어요. - 각각에
user.tool_confirmation이벤트를 보내고tool_use_id파라미터에 이벤트 ID를 전달하세요.result를"allow"또는"deny"로 설정하세요. 거부를 설명하려면deny_message를 사용하세요. - 모든 차단 이벤트가 해결되면 세션이
running으로 돌아가요.
각 agent.tool_use와 agent.mcp_tool_use 이벤트는 evaluated_permission(allow, ask, deny)을 싣고, evaluated_permission이 "ask"인 이벤트만 확인을 기다려요. 대부분의 이벤트는 어떤 정책이 그 결과를 만들었는지 기록하는 evaluation 객체도 싣는데, 각 호출이 어떻게 평가됐는지 보기에 설명돼 있어요. 예를 들어 always_ask 정책 아래에서 멈춘 bash 호출은 스트림에 다음과 같이 나타나요:
{
"type": "agent.tool_use",
"id": "sevt_01def...",
"name": "bash",
"input": {
"command": "pip install -r requirements.txt"
},
"evaluated_permission": "ask",
"evaluation": {
"type": "always_ask"
},
"processed_at": "2026-03-25T14:01:45Z"
}
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
# Approve the pending tool call
jq -n --arg id "$event_id"
'{events: [{type: "user.tool_confirmation", tool_use_id: $id, result: "allow"}]}' |
curl --fail-with-body -sS
"https://api.anthropic.com/v1/sessions/$SESSION_ID/events?beta=true"
-H "x-api-key: $ANTHR...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
# This workflow does not translate well to a one-off shell command.
# Use one of the SDK examples in this code group instead.
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:
# Approve the pending tool call
client.beta.sessions.events.send(
session.id,
events=[
{
"type": "user.tool_confirmation",
"tool_use_id": event_id,
"result": "allow",
},
],
)
case "end_turn":
break
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) {
// Approve the pending tool call
await client.beta.sessions.events.send(session.id, {
events: [
{
type: "user.tool_confirmation",
tool_use_id: eventId,
result: "allow",
},
],
});
}
}
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)
{
// Approve the pending tool call
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;
}
}
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 {
// Approve the pending tool call
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)
}
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())
// Approve each pending tool call
.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()));
}
$stream = $client->beta->sessions->events->streamStream($session->id);
foreach ($stream as $event) {
if ($event instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsSessionStatusIdleEvent && $event->stopReason) {
switch (true) {
case $event->stopReason instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsSessionRequiresAction:
foreach ($event->stopReason->eventIDs as $eventId) {
// Approve the pending tool call
$client->beta->sessions->events->send(
$session->id,
events: [
[
'type' => 'user.tool_confirmation',
'tool_use_id' => $eventId,
'result' => 'allow',
],
],
);
}
break;
case $event->stopReason instanceof \Anthropic\Beta\Sessions\Events\ManagedAgentsSessionEndTurn:
break 2;
}
}
}
client.beta.sessions.events.stream_events(session.id).each do |event|
case event
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionStatusIdleEvent
stop_reason = event.stop_reason
case stop_reason
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionRequiresAction
stop_reason.event_ids.each do |event_id|
# Approve the pending tool call
client.beta.sessions.events.send_(
session.id,
events: [
{type: "user.tool_confirmation", tool_use_id: event_id, result: "allow"}
]
)
end
when Anthropic::Beta::Sessions::BetaManagedAgentsSessionEndTurn
break
end
end
end
유휴 세션 재개하기 (Resuming an idle session)
세션은 상호작용 사이에 지속돼요. 세션이 명시적으로 삭제되지 않는 한 대화 기록은 보존돼요. 세션이 idle이 되면 샌드박스가 체크포인트되어 파일시스템, 설치된 패키지, 에이전트가 만든 파일을 포함한 전체 샌드박스 상태를 보존해요. 이렇게 하면 비활동에서 깨끗하게 재개할 수 있어요.
세션을 재개하려면 평소처럼 그것에 user.message 이벤트를 보내세요:
# In production, pass the stored ID of the session you want to resume.
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
# Resume a previously created session by sending it a new user.message event.
# In production, pass the stored ID of the session you want to resume.
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.",
},
],
},
],
)
// Resume a previously created session by sending it a new user event.
// In production, pass the stored ID of the session you want to resume.
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.",
},
],
},
],
});
// Resume a previously created session by ID. In production, pass the
// session ID you stored when the session was created.
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.",
},
],
},
],
});
// Resume a previously created session by sending it a new user.message
// event. In production, pass the stored ID of the session to resume.
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)
}
// Resume a previously created session by ID. In production, pass the
// session ID you stored when the session was created.
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());
// Resume a previously created session by sending it a new user.message event.
// In production, pass the session ID you stored when the session was created.
$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.',
],
],
],
],
);
# Resuming a session is just sending the next event to it. In production,
# pass the session ID you stored when the session was created.
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."}
]
}
]
)
세션 예산 도달 (Reaching a session budget)
예산으로 만들어진 세션은 초과 지출 대신 멈춰요. 세션의 추적된 리스트 비용이 상한에 도달하면 플랫폼이 다음 모델 요청 전에 각 스레드를 멈추고, 세션은 종료하는 대신 stop_reason이 budget_reached인 채 idle이 돼요. 총액을 상한을 넘기게 한 요청은 완료까지 실행되므로, session.usage 스냅샷이 보고하는 list_cost는 상한과 같거나 조금 넘어 읽힐 수 있어요. 스트림에서 그 멈춤은 세 가지 이벤트로 순서대로 도착해요:
- 각 스레드가 멈출 때마다
stop_reason: budget_reached를 가진session.thread_status_idle. session.usage, 세션의 누적 사용량과 추적된 리스트 비용의 스냅샷.stop_reason: budget_reached를 가진session.status_idle.session.usage이벤트는 항상 이 idle 바로 앞에 와요.
마지막 요청이 상한을 넘으면서도 턴을 완료시키는 스레드는 자체 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은 budget_reached가 아니라 requires_action이에요: 요청을 정리하는 것이 모델 요청을 촉발하지 않으므로 평소처럼 응답하세요.
상한에 멈춘 세션을 재개하는 이벤트는 없어요. 대신 세션의 예산을 업데이트하세요: 상한을 소비된 리스트 비용보다 높은 값으로 바꾸거나, "budget": null으로 세션을 업데이트해 예산을 제거하면 멈춘 작업이 자동으로 재개돼요. 리스트 비용이 어떻게 추적되고 전체 예산 업데이트 의미가 무엇인지는 세션 예산을 참고하세요.
시스템 메시지 보내기 (Sending system messages)
에이전트에게 수반되는 턴과 이후 모든 턴에 적용되는 특권 시스템 수준 컨텍스트를 주려면 system.message 이벤트를 보내세요. 에이전트 정의의 system 필드(최상위 시스템 프롬프트를 설정해요)와 달리, system.message 콘텐츠는 그 프롬프트를 대체하지 않고 role: "system" 턴으로 세션의 시스템 컨텍스트에 추가돼요. 세션 중간에 업데이트된 시스템 수준 지침(다른 페르소나, 수정된 제약, 런타임에 가져온 컨텍스트)이 에이전트의 미래 행동을 형성해야 할 때 사용하세요.
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
client.beta.sessions.events.send(
session.id,
events=[
{
"type": "system.message",
"content": [
{
"type": "text",
"text": "The user's current timezone is America/New_York.",
},
],
},
],
)
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.",
},
],
},
],
});
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.",
},
],
},
],
});
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)
}
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());
$client->beta->sessions->events->send(
$session->id,
events: [
[
'type' => 'system.message',
'content' => [
[
'type' => 'text',
'text' => "The user's current timezone is America/New_York.",
],
],
],
],
);
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으로 idle인 동안 system.message는 같은 요청에서 도구 결과 이벤트 뒤에 올 때만 받아들여져요. 단독으로 또는 user.message와 함께 보내면 보류 도구 이벤트가 해결될 때까지 거부돼요. content는 1–1000개의 텍스트 항목을 받아들여요.
사용량 추적 (Tracking usage)
세션 객체에는 세션의 누적 사용량인 usage 필드(토큰 수, 서버 도구 사용, 활성 시간, 추적된 리스트 비용)가 있어요. 세션이 idle이 된 뒤 가져와 최신 합계를 읽으세요.
{
"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으로 읽혀요. 각 세션 스레드의 자체 usage도 list_cost와 active_seconds를 담아요. 스레드별 수치는 독립적으로 반올림되고 세션의 런타임 비용을 제외하므로 세션의 list_cost에 정확히 합산되지 않아요. 세션 수치가 권위 있는 것이에요.
이 합계를 관찰하기 위해 세션을 폴링할 필요는 없어요. session.usage 이벤트가 같은 누적 스냅샷(usage 객체와, 없는 경우 null인 세션의 budget 포함)을 세션 스트림과 이벤트 기록에 싣고 있어요. 타이머가 아니라 idle 전환 시 발행돼요: 세션은 정지 이유가 무엇이든 idle이 되기 직전에 하나를, 그리고 스레드가 세션 예산에 멈출 때 하나를 발행해요. 따라서 스트림 읽기는 추가 가져오기 없이 턴의 최종 비용이나 예산에 닿은 작업의 비용을 보게 돼요.
지출 한도를 강제하려면 사용량을 폴링하고 세션을 직접 멈추기보다 세션 예산을 설정하세요. 플랫폼이 세션의 소비를 지속적으로 책정하고 세션의 리스트 비용이 상한에 도달하면 다음 모델 요청 전에 각 스레드를 멈춰요. 스트림에서 그것이 어떻게 보이는지는 세션 예산 도달을 참고하세요.
Console 관찰성 (Console observability)
Claude Console에는 코드를 쓰지 않고 에이전트가 무엇을 했는지 검사하는 세션 뷰어가 있어요. Console 사이드바의 Managed Agents 아래에서 Sessions를 선택하면 워크스페이스의 모든 세션을 상태, 에이전트, 토큰 사용량, 비용, 생성 시간과 함께 볼 수 있고, 세션을 선택해 열 수 있어요. 세션 뷰어는 Developers와 Admins만 접근할 수 있어요. 다음을 보여줘요:
-
타임라인 미니맵 (Timeline minimap): 시간에 따른 세션 활동의 확대 가능한 개요로, 멀티에이전트 세션에서는 스레드당 하나의 레인이 있어요. 레인을 선택해 그 스레드를 보거나, 마크를 선택해 그 이벤트로 이동하세요.
-
트랜스크립트 (Transcript): 모델 요청별로 그룹화된 대화로, thinking, 입력과 결과가 있는 도구 호출, 스트리밍되는 메시지 텍스트를 포함해요. 이벤트를 필터링하고 JSON으로 복사하거나 다운로드할 수 있어요.
-
인스펙터 (Inspector): 다섯 탭으로 세션에 대한 세부 정보가 있는 크기 조절 가능한 사이드 패널이에요:
- Session은 세션의 세부 사항과 메타데이터, 시간에 따른 누적 비용, 예산이 설정된 경우 예산 대비 지출을 보여줘요.
- Events는 현재 스레드의 모든 원시 이벤트를 서버가 보낸 순서대로 나열해요. 이벤트를 선택해 JSON을 볼 수 있어요. 페이지가 열려 있는 동안 스트리밍된 메시지에는 이벤트 델타의 Deltas 뷰도 있어요.
- Tools는 세션 에이전트가 구성된 도구를 호출 횟수, 실패, 중간 지속 시간과 함께 나열해요. 도구를 선택해 호출을 보고 하나를 트랜스크립트에서 찾을 수 있어요.
- Resources는 마운트된 파일, 리포지토리, 메모리 스토어를 컨테이너 경로에서, 각 스토어의 메모리와 이 세션이 그것들에 한 변경, 에이전트가
/mnt/session/outputs에 쓴 파일, 세션 에이전트에 붙은 스킬과 함께 나열해요. - Threads는 모든 스레드를 상태, 컨텍스트 크기, 비용과 함께 나열해요. 스레드를 선택해 에이전트, 모델, 컨텍스트 사용량, 비용 같은 세부 사항을 봐요.
세션 URL에 ?event={event_id}를 붙여 특정 이벤트에서 세션을 열 수 있어요.
ant beta:sessions connect로 ant CLI에서 같은 뷰어를 열거나 터미널에서 세션을 따라갈 수 있어요. 터미널에서 Managed Agents 세션에 연결하기를 참고하세요.
디버깅 팁 (Debugging tips)
- 세션 이벤트 확인: 세션 오류는
session.error이벤트로 전달돼요. - 도구 결과 검토: 도구 실행 실패가 예상치 못한 에이전트 동작을 종종 설명해요.
- 토큰 사용량 추적: 프롬프트를 최적화하고 비용을 줄이려면 토큰 소비를 모니터링하세요.
- 시스템 프롬프트 사용: 시스템 프롬프트에 로깅 지침을 추가해 에이전트가 자신의 추론을 설명하게 하세요.
- 미리보기 문제 해결: 이벤트 델타에 옵트인한 스트림이 예상대로 동작하지 않으면 미리보기 문제 해결을 참고하세요.
더 알아보기 (Learn more)
- 웹훅 구독하기 — 상태 변경 알림 받기
- 멀티에이전트 오케스트레이션 — 스레드와 하위 에이전트
- 권한 정책 — 도구 확인
- 세션 예산 — 지출 한도