ストリーミングの受信
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)を持つ。
このセッションの 2 つのターン(準備メッセージと ls)のイベントがすべて届いた。frame 1 つは、event:、data:、id: の 3 行と空行である。
event: tool_use
data: {"schema_version":1,"session_id":"…","turn_id":"2","attempt_id":"att_…","occurred_at":"…","data":{…}}
id: ev_3mdata は JSON である。イベントが属するセッション・ターン・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 | エンジンがターンを終えた。完了の判定ではない |
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 }
listedターンが終わったかどうかは、イベントではなくターンの 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その後は新しいイベントがなく、何も届かなかった。その間に次のメッセージを送っていれば、そのターンのイベントが届く。
| レスポンス | 意味 |
|---|---|
410 CURSOR_EXPIRED | その id が保持範囲の外にある。セッションとターンを HTTP で読み直し、新しく接続する。現在のサーバーはイベントを削除しないためこのレスポンスは返さないが、クライアントでは処理しておく |
| 429 | 同時に開いているストリームが多すぎる |