Run a session and stream its events

The scripts p2-run-stream.sh, p2-run-stream.py and p2-run-stream.ts are in full under Complete script.

Before you start

Set CONTEXT_BASE_URL to your deployment’s host, without /v1 (for example, https://api.context.ai).
Set CONTEXT_API_KEY to a personal Task API key. The default capabilities are enough.
Set AGENT_ID to an agent id from GET /v1/agents.

Client setup used by every step

#!/usr/bin/env bash
set -euo pipefail
: "${CONTEXT_BASE_URL:?Set CONTEXT_BASE_URL, for example https://api.context.ai}"
: "${CONTEXT_API_KEY:?Set CONTEXT_API_KEY to a personal Task API key}"
: "${AGENT_ID:?Set AGENT_ID to an id from GET /v1/agents}"
api() {
  local path=$1; shift
  curl -sS --fail-with-body "$CONTEXT_BASE_URL/v1$path" -H "x-api-key: $CONTEXT_API_KEY" "$@"
}
fail() { echo "error: $*" >&2; exit 1; }
# Opens the session's event stream; exec makes the background job's pid the curl process itself.
stream() {
  exec curl -sS --fail-with-body -N "$CONTEXT_BASE_URL/v1/sessions/$1/events/stream" \
    -H "x-api-key: $CONTEXT_API_KEY" "${@:2}"
}
# The JSON of every event received so far on a stream saved to a file. Ping frames carry no id and are skipped.
events_in() { sed -n 's/^data: //p' "$1" | jq -c 'select(.id)'; }
# Waits until the stream file holds a session.status_idle that ends the turn (not requires_action).
wait_for_end_of_turn() {
  for _ in $(seq 1 300); do
    grep -q '^event: error' "$1" && fail "the stream ended with an error: $(grep -A2 '^event: error' "$1" | sed -n 's/^data: //p')"
    events_in "$1" | jq -se 'any(.[]; .type == "session.status_idle" and .stop_reason.type != "requires_action")' >/dev/null && return 0
    sleep 1
  done
  fail "the turn did not end within 300 seconds"
}
WORK=$(mktemp -d)

import os

from anthropic import Anthropic

client = Anthropic(
    base_url=os.environ["CONTEXT_BASE_URL"],
    api_key=os.environ["CONTEXT_API_KEY"],
    max_retries=0,
)
agent_id = os.environ["AGENT_ID"]


def check(ok, message):
    if not ok:
        raise SystemExit(f"error: {message}")


def ends_turn(event):
    return event.type == "session.status_idle" and event.stop_reason.type != "requires_action"

import Anthropic from "@anthropic-ai/sdk"
import type { BetaManagedAgentsStreamSessionEvents } from "@anthropic-ai/sdk/resources/beta/sessions/events"

const client = new Anthropic({
  baseURL: process.env.CONTEXT_BASE_URL,
  apiKey: process.env.CONTEXT_API_KEY,
  maxRetries: 0,
})
const agentId = process.env.AGENT_ID!

function check(ok: unknown, message: string): asserts ok {
  if (!ok) throw new Error(message)
}
function endsTurn(event: BetaManagedAgentsStreamSessionEvents): boolean {
  return event.type === "session.status_idle" && event.stop_reason.type !== "requires_action"
}

Create a session

A new session is idle until you send it a message.

SESSION=$(api /sessions -H "content-type: application/json" \
  -d "$(jq -n --arg agent "$AGENT_ID" '{agent: $agent, environment_id: "env_default", title: "Streaming demo"}')")
SESSION_ID=$(jq -r .id <<<"$SESSION")
[[ $(jq -r .status <<<"$SESSION") == "idle" ]] || fail "expected an idle session: $SESSION"
echo "step 1: session $SESSION_ID idle"

session = client.beta.sessions.create(agent=agent_id, environment_id="env_default", title="Streaming demo")
check(session.status == "idle", f"expected an idle session, got {session.status}")
print(f"step 1: session {session.id} idle")

const session = await client.beta.sessions.create({ agent: agentId, environment_id: "env_default", title: "Streaming demo" })
check(session.status === "idle", `expected an idle session, got ${session.status}`)
console.log(`step 1: session ${session.id} idle`)

Sample output:

step 1: session dd13d728-93e3-440a-874c-4efeb58c97e0 idle

Open the stream before sending

The stream carries only the events that happen after it opens. Open it before you send the message so you don’t miss any part of the turn.

stream "$SESSION_ID" > "$WORK/stream.txt" &
STREAM_PID=$!
sleep 1
kill -0 "$STREAM_PID" 2>/dev/null || fail "the stream did not open: $(cat "$WORK/stream.txt")"
echo "step 2: stream open"

stream = client.beta.sessions.events.stream(session.id)
check(stream.response.status_code == 200, f"stream answered {stream.response.status_code}")
print("step 2: stream open")

const { data: stream, response } = await client.beta.sessions.events.stream(session.id).withResponse()
check(response.ok, `stream answered ${response.status}`)
console.log("step 2: stream open")

Sample output:

step 2: stream open

Send a message

Sending returns one receipt for each event you send, with a sevt_receipt_ id. The stored event, with its own id, arrives on the stream. Don’t use a receipt id as Last-Event-ID. To resume, use the id of a stored event from the stream.

SENT=$(api "/sessions/$SESSION_ID/events" -H "content-type: application/json" -d '{"events": [
  {"type": "user.message", "content": [{"type": "text", "text": "Reply with one sentence about Tokyo."}]}]}')
[[ $(jq -r '.data[0].id' <<<"$SENT") == sevt_receipt_* ]] || fail "unexpected receipt: $SENT"
echo "step 3: sent, receipt $(jq -r '.data[0].id' <<<"$SENT")"

sent = client.beta.sessions.events.send(
    session.id,
    events=[{"type": "user.message", "content": [{"type": "text", "text": "Reply with one sentence about Tokyo."}]}],
)
check(sent.data[0].id.startswith("sevt_receipt_"), f"unexpected receipt {sent}")
print(f"step 3: sent, receipt {sent.data[0].id}")

const sent = await client.beta.sessions.events.send(session.id, {
  events: [{ type: "user.message", content: [{ type: "text", text: "Reply with one sentence about Tokyo." }] }],
})
const receiptId = sent.data?.[0]?.id ?? ""
check(receiptId.startsWith("sevt_receipt_"), `unexpected receipt ${JSON.stringify(sent)}`)
console.log(`step 3: sent, receipt ${receiptId}`)

Sample output:

step 3: sent, receipt sevt_receipt_219b3afe-157c-452d-981b-c71353fb80bc

Read events until the turn ends

Events arrive as server-sent events. The turn is over at a session.status_idle whose stop_reason.type is not requires_action (usually end_turn).

wait_for_end_of_turn "$WORK/stream.txt"
kill "$STREAM_PID"
ANSWER=$(events_in "$WORK/stream.txt" | jq -r 'select(.type == "agent.message") | .content[] | select(.type == "text") | .text')
[[ -n $ANSWER ]] || fail "no agent.message on the stream"
FIRST_ID=$(events_in "$WORK/stream.txt" | jq -rs '.[0].id')
LAST_ID=$(events_in "$WORK/stream.txt" | jq -rs '.[-1].id')
echo "step 4: $(events_in "$WORK/stream.txt" | wc -l) events, last $LAST_ID, agent said: $ANSWER"

seen, answer = [], ""
for event in stream:
    seen.append(event.id)
    if event.type == "agent.message":
        answer += "".join(block.text for block in event.content if block.type == "text")
    if ends_turn(event):
        break
stream.close()
check(answer, "no agent.message on the stream")
print(f"step 4: {len(seen)} events, last {seen[-1]}, agent said: {answer}")

const seen: string[] = []
let answer = ""
for await (const event of stream) {
  if ("id" in event) seen.push(event.id)
  if (event.type === "agent.message") answer += event.content.map(b => (b.type === "text" ? b.text : "")).join("")
  if (endsTurn(event)) break
}
stream.controller.abort()
check(answer, "no agent.message on the stream")
console.log(`step 4: ${seen.length} events, last ${seen.at(-1)}, agent said: ${answer}`)

Sample output:

step 4: 6 events, last sevt_4_1, agent said: Tokyo is the most populous metropolitan area in the world.

Resume the stream from an event id

If the connection drops, reopen the stream with a Last-Event-ID header set to the last event id you received. The server replays every event after it.

# After a dropped connection, send the id of the last event you received. Here we resume from the first one.
stream "$SESSION_ID" -H "Last-Event-ID: $FIRST_ID" > "$WORK/resumed.txt" &
RESUME_PID=$!
wait_for_end_of_turn "$WORK/resumed.txt"
kill "$RESUME_PID"
[[ $(events_in "$WORK/resumed.txt" | jq -rs '.[0].id') != "$FIRST_ID" ]] || fail "the resumed stream repeated $FIRST_ID"
echo "step 5: resumed after $FIRST_ID, replayed $(events_in "$WORK/resumed.txt" | wc -l) events up to $(events_in "$WORK/resumed.txt" | jq -rs '.[-1].id')"

# After a dropped connection, send the id of the last event you received. Here we resume from the first one.
resumed = []
with client.beta.sessions.events.stream(session.id, extra_headers={"Last-Event-ID": seen[0]}) as replay:
    for event in replay:
        resumed.append(event.id)
        if ends_turn(event):
            break
check(resumed and resumed[0] != seen[0] and resumed[-1] == seen[-1], f"unexpected replay {resumed}")
print(f"step 5: resumed after {seen[0]}, replayed {len(resumed)} events up to {resumed[-1]}")

// After a dropped connection, send the id of the last event you received. Here we resume from the first one.
const replay = await client.beta.sessions.events.stream(session.id, {}, { headers: { "Last-Event-ID": seen[0] } })
const resumed: string[] = []
for await (const event of replay) {
  if ("id" in event) resumed.push(event.id)
  if (endsTurn(event)) break
}
replay.controller.abort()
check(resumed.length > 0 && resumed[0] !== seen[0] && resumed.at(-1) === seen.at(-1), `unexpected replay ${resumed}`)
console.log(`step 5: resumed after ${seen[0]}, replayed ${resumed.length} events up to ${resumed.at(-1)}`)

Sample output:

step 5: resumed after sevt_1_0, replayed 5 events up to sevt_4_1

List the session's events

The events endpoint returns the stored history with the same ids the stream carried, oldest first. The script prints how many events it listed and which event types they include.

LISTED=$(api "/sessions/$SESSION_ID/events")
jq -e --arg id "$LAST_ID" 'any(.data[]; .id == $id)' <<<"$LISTED" >/dev/null || fail "$LAST_ID is not listed"
echo "step 6: $(jq '.data | length' <<<"$LISTED") events listed: $(jq -r '[.data[].type] | unique | join(", ")' <<<"$LISTED")"

listed = client.beta.sessions.events.list(session.id).data
check(any(e.id == seen[-1] for e in listed), f"{seen[-1]} is not listed")
print(f"step 6: {len(listed)} events listed: {', '.join(sorted({e.type for e in listed}))}")

const listed = (await client.beta.sessions.events.list(session.id)).data
check(listed.some(e => e.id === seen.at(-1)), `${seen.at(-1)} is not listed`)
console.log(`step 6: ${listed.length} events listed: ${[...new Set(listed.map(e => e.type))].sort().join(", ")}`)

Sample output:

step 6: 6 events listed: agent.message, session.status_idle, session.status_running, session.usage, user.message

Archive the session

Archiving ends the session. Its status becomes terminated and its history stays readable. Sending it a message returns 400.

ARCHIVED=$(api "/sessions/$SESSION_ID/archive" -X POST)
[[ $(jq -r .status <<<"$ARCHIVED") == "terminated" ]] || fail "expected terminated: $ARCHIVED"
[[ $(jq -r .archived_at <<<"$ARCHIVED") != null ]] || fail "archived_at is not set: $ARCHIVED"
echo "step 7: session terminated at $(jq -r .archived_at <<<"$ARCHIVED")"

archived = client.beta.sessions.archive(session.id)
check(archived.status == "terminated", f"expected terminated, got {archived.status}")
check(archived.archived_at is not None, "archived_at is not set")
print(f"step 7: session {archived.status} at {archived.archived_at}")

const archived = await client.beta.sessions.archive(session.id)
check(archived.status === "terminated", `expected terminated, got ${archived.status}`)
check(archived.archived_at, "archived_at is not set")
console.log(`step 7: session ${archived.status} at ${archived.archived_at}`)

Sample output:

step 7: session terminated at 2026-10-21T01:47:50.964Z

If something goes wrong

What you see	Why	What to do
The stream opens but no events arrive	The message was sent before the stream was opened, and the turn finished first.	Open the stream, then send. To catch up, list the events or reopen the stream with `Last-Event-ID`.
The loop stops while the agent still waits	The loop treated a `session.status_idle` whose `stop_reason.type` is `requires_action` as the end of the turn.	Stop only on an idle whose `stop_reason.type` is not `requires_action`. Answer `requires_action` as shown in Approve or deny tool calls.
`429 rate_limit_error` when opening a stream	The key already holds 10 open streams on the server instance that answered, which is the limit for one key.	Close streams you no longer read (abort the controller or close the stream object), then open the new one.

Complete script

#!/usr/bin/env bash
set -euo pipefail
: "${CONTEXT_BASE_URL:?Set CONTEXT_BASE_URL, for example https://api.context.ai}"
: "${CONTEXT_API_KEY:?Set CONTEXT_API_KEY to a personal Task API key}"
: "${AGENT_ID:?Set AGENT_ID to an id from GET /v1/agents}"
api() {
  local path=$1; shift
  curl -sS --fail-with-body "$CONTEXT_BASE_URL/v1$path" -H "x-api-key: $CONTEXT_API_KEY" "$@"
}
fail() { echo "error: $*" >&2; exit 1; }
# Opens the session's event stream; exec makes the background job's pid the curl process itself.
stream() {
  exec curl -sS --fail-with-body -N "$CONTEXT_BASE_URL/v1/sessions/$1/events/stream" \
    -H "x-api-key: $CONTEXT_API_KEY" "${@:2}"
}
# The JSON of every event received so far on a stream saved to a file. Ping frames carry no id and are skipped.
events_in() { sed -n 's/^data: //p' "$1" | jq -c 'select(.id)'; }
# Waits until the stream file holds a session.status_idle that ends the turn (not requires_action).
wait_for_end_of_turn() {
  for _ in $(seq 1 300); do
    grep -q '^event: error' "$1" && fail "the stream ended with an error: $(grep -A2 '^event: error' "$1" | sed -n 's/^data: //p')"
    events_in "$1" | jq -se 'any(.[]; .type == "session.status_idle" and .stop_reason.type != "requires_action")' >/dev/null && return 0
    sleep 1
  done
  fail "the turn did not end within 300 seconds"
}
WORK=$(mktemp -d)

# @step 1 Create a session
SESSION=$(api /sessions -H "content-type: application/json" \
  -d "$(jq -n --arg agent "$AGENT_ID" '{agent: $agent, environment_id: "env_default", title: "Streaming demo"}')")
SESSION_ID=$(jq -r .id <<<"$SESSION")
[[ $(jq -r .status <<<"$SESSION") == "idle" ]] || fail "expected an idle session: $SESSION"
echo "step 1: session $SESSION_ID idle"

# @step 2 Open the stream before sending
stream "$SESSION_ID" > "$WORK/stream.txt" &
STREAM_PID=$!
sleep 1
kill -0 "$STREAM_PID" 2>/dev/null || fail "the stream did not open: $(cat "$WORK/stream.txt")"
echo "step 2: stream open"

# @step 3 Send a message
SENT=$(api "/sessions/$SESSION_ID/events" -H "content-type: application/json" -d '{"events": [
  {"type": "user.message", "content": [{"type": "text", "text": "Reply with one sentence about Tokyo."}]}]}')
[[ $(jq -r '.data[0].id' <<<"$SENT") == sevt_receipt_* ]] || fail "unexpected receipt: $SENT"
echo "step 3: sent, receipt $(jq -r '.data[0].id' <<<"$SENT")"

# @step 4 Read events until the turn ends
wait_for_end_of_turn "$WORK/stream.txt"
kill "$STREAM_PID"
ANSWER=$(events_in "$WORK/stream.txt" | jq -r 'select(.type == "agent.message") | .content[] | select(.type == "text") | .text')
[[ -n $ANSWER ]] || fail "no agent.message on the stream"
FIRST_ID=$(events_in "$WORK/stream.txt" | jq -rs '.[0].id')
LAST_ID=$(events_in "$WORK/stream.txt" | jq -rs '.[-1].id')
echo "step 4: $(events_in "$WORK/stream.txt" | wc -l) events, last $LAST_ID, agent said: $ANSWER"

# @step 5 Resume the stream from an event id
# After a dropped connection, send the id of the last event you received. Here we resume from the first one.
stream "$SESSION_ID" -H "Last-Event-ID: $FIRST_ID" > "$WORK/resumed.txt" &
RESUME_PID=$!
wait_for_end_of_turn "$WORK/resumed.txt"
kill "$RESUME_PID"
[[ $(events_in "$WORK/resumed.txt" | jq -rs '.[0].id') != "$FIRST_ID" ]] || fail "the resumed stream repeated $FIRST_ID"
echo "step 5: resumed after $FIRST_ID, replayed $(events_in "$WORK/resumed.txt" | wc -l) events up to $(events_in "$WORK/resumed.txt" | jq -rs '.[-1].id')"

# @step 6 List the session's events
LISTED=$(api "/sessions/$SESSION_ID/events")
jq -e --arg id "$LAST_ID" 'any(.data[]; .id == $id)' <<<"$LISTED" >/dev/null || fail "$LAST_ID is not listed"
echo "step 6: $(jq '.data | length' <<<"$LISTED") events listed: $(jq -r '[.data[].type] | unique | join(", ")' <<<"$LISTED")"

# @step 7 Archive the session
ARCHIVED=$(api "/sessions/$SESSION_ID/archive" -X POST)
[[ $(jq -r .status <<<"$ARCHIVED") == "terminated" ]] || fail "expected terminated: $ARCHIVED"
[[ $(jq -r .archived_at <<<"$ARCHIVED") != null ]] || fail "archived_at is not set: $ARCHIVED"
echo "step 7: session terminated at $(jq -r .archived_at <<<"$ARCHIVED")"
# @end

import os

from anthropic import Anthropic

client = Anthropic(
    base_url=os.environ["CONTEXT_BASE_URL"],
    api_key=os.environ["CONTEXT_API_KEY"],
    max_retries=0,
)
agent_id = os.environ["AGENT_ID"]


def check(ok, message):
    if not ok:
        raise SystemExit(f"error: {message}")


def ends_turn(event):
    return event.type == "session.status_idle" and event.stop_reason.type != "requires_action"


# @step 1 Create a session
session = client.beta.sessions.create(agent=agent_id, environment_id="env_default", title="Streaming demo")
check(session.status == "idle", f"expected an idle session, got {session.status}")
print(f"step 1: session {session.id} idle")

# @step 2 Open the stream before sending
stream = client.beta.sessions.events.stream(session.id)
check(stream.response.status_code == 200, f"stream answered {stream.response.status_code}")
print("step 2: stream open")

# @step 3 Send a message
sent = client.beta.sessions.events.send(
    session.id,
    events=[{"type": "user.message", "content": [{"type": "text", "text": "Reply with one sentence about Tokyo."}]}],
)
check(sent.data[0].id.startswith("sevt_receipt_"), f"unexpected receipt {sent}")
print(f"step 3: sent, receipt {sent.data[0].id}")

# @step 4 Read events until the turn ends
seen, answer = [], ""
for event in stream:
    seen.append(event.id)
    if event.type == "agent.message":
        answer += "".join(block.text for block in event.content if block.type == "text")
    if ends_turn(event):
        break
stream.close()
check(answer, "no agent.message on the stream")
print(f"step 4: {len(seen)} events, last {seen[-1]}, agent said: {answer}")

# @step 5 Resume the stream from an event id
# After a dropped connection, send the id of the last event you received. Here we resume from the first one.
resumed = []
with client.beta.sessions.events.stream(session.id, extra_headers={"Last-Event-ID": seen[0]}) as replay:
    for event in replay:
        resumed.append(event.id)
        if ends_turn(event):
            break
check(resumed and resumed[0] != seen[0] and resumed[-1] == seen[-1], f"unexpected replay {resumed}")
print(f"step 5: resumed after {seen[0]}, replayed {len(resumed)} events up to {resumed[-1]}")

# @step 6 List the session's events
listed = client.beta.sessions.events.list(session.id).data
check(any(e.id == seen[-1] for e in listed), f"{seen[-1]} is not listed")
print(f"step 6: {len(listed)} events listed: {', '.join(sorted({e.type for e in listed}))}")

# @step 7 Archive the session
archived = client.beta.sessions.archive(session.id)
check(archived.status == "terminated", f"expected terminated, got {archived.status}")
check(archived.archived_at is not None, "archived_at is not set")
print(f"step 7: session {archived.status} at {archived.archived_at}")
# @end

import Anthropic from "@anthropic-ai/sdk"
import type { BetaManagedAgentsStreamSessionEvents } from "@anthropic-ai/sdk/resources/beta/sessions/events"

const client = new Anthropic({
  baseURL: process.env.CONTEXT_BASE_URL,
  apiKey: process.env.CONTEXT_API_KEY,
  maxRetries: 0,
})
const agentId = process.env.AGENT_ID!

function check(ok: unknown, message: string): asserts ok {
  if (!ok) throw new Error(message)
}
function endsTurn(event: BetaManagedAgentsStreamSessionEvents): boolean {
  return event.type === "session.status_idle" && event.stop_reason.type !== "requires_action"
}

// @step 1 Create a session
const session = await client.beta.sessions.create({ agent: agentId, environment_id: "env_default", title: "Streaming demo" })
check(session.status === "idle", `expected an idle session, got ${session.status}`)
console.log(`step 1: session ${session.id} idle`)

// @step 2 Open the stream before sending
const { data: stream, response } = await client.beta.sessions.events.stream(session.id).withResponse()
check(response.ok, `stream answered ${response.status}`)
console.log("step 2: stream open")

// @step 3 Send a message
const sent = await client.beta.sessions.events.send(session.id, {
  events: [{ type: "user.message", content: [{ type: "text", text: "Reply with one sentence about Tokyo." }] }],
})
const receiptId = sent.data?.[0]?.id ?? ""
check(receiptId.startsWith("sevt_receipt_"), `unexpected receipt ${JSON.stringify(sent)}`)
console.log(`step 3: sent, receipt ${receiptId}`)

// @step 4 Read events until the turn ends
const seen: string[] = []
let answer = ""
for await (const event of stream) {
  if ("id" in event) seen.push(event.id)
  if (event.type === "agent.message") answer += event.content.map(b => (b.type === "text" ? b.text : "")).join("")
  if (endsTurn(event)) break
}
stream.controller.abort()
check(answer, "no agent.message on the stream")
console.log(`step 4: ${seen.length} events, last ${seen.at(-1)}, agent said: ${answer}`)

// @step 5 Resume the stream from an event id
// After a dropped connection, send the id of the last event you received. Here we resume from the first one.
const replay = await client.beta.sessions.events.stream(session.id, {}, { headers: { "Last-Event-ID": seen[0] } })
const resumed: string[] = []
for await (const event of replay) {
  if ("id" in event) resumed.push(event.id)
  if (endsTurn(event)) break
}
replay.controller.abort()
check(resumed.length > 0 && resumed[0] !== seen[0] && resumed.at(-1) === seen.at(-1), `unexpected replay ${resumed}`)
console.log(`step 5: resumed after ${seen[0]}, replayed ${resumed.length} events up to ${resumed.at(-1)}`)

// @step 6 List the session's events
const listed = (await client.beta.sessions.events.list(session.id)).data
check(listed.some(e => e.id === seen.at(-1)), `${seen.at(-1)} is not listed`)
console.log(`step 6: ${listed.length} events listed: ${[...new Set(listed.map(e => e.type))].sort().join(", ")}`)

// @step 7 Archive the session
const archived = await client.beta.sessions.archive(session.id)
check(archived.status === "terminated", `expected terminated, got ${archived.status}`)
check(archived.archived_at, "archived_at is not set")
console.log(`step 7: session ${archived.status} at ${archived.archived_at}`)
// @end