스트리밍 받기
SSE로 세션 이벤트를 받고, 끊기면 Last-Event-ID로 이어 받는다
sessions.events.stream(GET /v1/sessions/{id}/events)은 세션의 이벤트를 Server-Sent Events로 받는다. 이미 저장된 이벤트를 처음부터 다시 보내고, 그 뒤로 새 이벤트가 생기는 대로 이어 보낸다. scope는 sessions:read다. 보내기의 세션(SDK는 sessionId, HTTP API는 SID)을 쓴다.
받기
스트림은 15초마다 keepalive를 보내므로 정상일 때도 닫히지 않는다. 아래 예제는 5초 뒤 끊는다. SDK는 signal이나 stream.close()로, curl은 --max-time으로 끊는다.
import { AbortedError, type SessionEvent } from "@werft-ai/sdk";
const seen: SessionEvent[] = [];
const stream = werft.sessions.events.stream(sessionId, {
signal: AbortSignal.timeout(5_000),
});
try {
for await (const event of stream) seen.push(event);
} catch (error) {
if (!(error instanceof AbortedError)) throw error;
}
const counts = new Map<string, number>();
for (const event of seen) counts.set(event.kind, (counts.get(event.kind) ?? 0) + 1);
console.log(counts);Map(6) {
'status' => 5,
'system' => 8,
'assistant' => 2,
'result' => 2,
'tool_use' => 1,
'tool_result' => 1
}SDK의 이벤트는 id, turn_id, attempt_id, occurred_at, kind(아래 표의 event), payload(안쪽 data)를 가진다.
이 세션의 turn 두 개(준비 메시지와 ls)의 이벤트가 모두 왔다. frame 하나는 event:, data:, id: 세 줄과 빈 줄이다.
event: tool_use
data: {"schema_version":1,"session_id":"…","turn_id":"2","attempt_id":"att_…","occurred_at":"…","data":{…}}
id: ev_3mdata는 JSON이다. 어느 세션·turn·attempt의 이벤트인지와 시각이 있고, 내용은 안쪽 data에 있다.
| event | 안쪽 data |
|---|---|
system | 엔진이 알리는 시작·설정 정보 |
assistant | 모델의 메시지. 텍스트는 text, 생각은 thinking |
tool_use | 모델이 부른 도구와 입력(tool_call의 id·name·input) |
tool_result | 도구의 결과(tool_result의 tool_call_id·text). 실패면 is_error: true |
question | 권한 요청이나 질문이 생겼다. 답하는 법은 승인 대기와 재개 |
result | 엔진이 turn을 마쳤다. 완료 판정이 아니다 |
status | 세션 상태가 바뀌었다(phase: running, idle 등) |
error | 실행 오류 |
이 표에 없는 안쪽 필드(message, session_id 등)는 엔진마다 다르다. 엔진이 바뀌면 사라지거나 모양이 바뀔 수 있으니 읽지 않는다.
텍스트와 도구 결과 뽑기
for (const event of seen) {
if (event.kind === "assistant")
console.log((event.payload as { text?: string }).text);
if (event.kind === "tool_result") {
const result = (event.payload as { tool_result?: { is_error?: boolean } })
.tool_result;
console.log({ is_error: result?.is_error });
}
}ready
{ is_error: false }
listedturn이 끝났는지는 이벤트가 아니라 turn의 status로 본다(보내기). 일이 됐는지는 assistant와 tool_result의 내용으로 판단한다.
끊긴 뒤 이어 받기
마지막으로 받은 id를 넘기면 그다음 이벤트부터 받는다. SDK는 stream.lastEventId를 lastEventId 옵션으로, HTTP API는 Last-Event-ID 헤더로 넘긴다. SDK는 연결이 끊기면 스스로 이 값으로 다시 붙는다. 받은 이벤트는 다시 오지 않는다.
const more = werft.sessions.events.stream(sessionId, {
lastEventId: stream.lastEventId,
signal: AbortSignal.timeout(3_000),
});
let count = 0;
try {
for await (const _event of more) count++;
} catch (error) {
if (!(error instanceof AbortedError)) throw error;
}
console.log(count);0그 뒤로 새 이벤트가 없어서 아무것도 오지 않았다. 그사이 다음 메시지를 보냈다면 그 turn의 이벤트가 온다.
| 응답 | 뜻 |
|---|---|
410 CURSOR_EXPIRED | 그 id가 보존 범위 밖이다. 세션과 turn을 HTTP로 다시 읽고 새로 연결한다. 지금 서버는 이벤트를 지우지 않아 이 응답을 보내지 않지만, 클라이언트는 처리해 둔다 |
| 429 | 동시에 연 스트림이 너무 많다 |