Streaming | Kizuna.js

An assistant's reply arrives one token at a time, and a caller who has to wait for the last token before seeing the first stops reading. A long export, a progress report, and a feed of events that has no end share the shape. The route declares what each piece looks like, the handler yields the pieces, and the client reads them as they come.

Declare the stream

A stream takes the place of a body on the response:

import { z } from 'zod';
import { ProblemDetailsSchema } from 'kizunajs/schemas';
import { k } from '../k';

export const replyRoute = k.route({
    method: 'POST',
    path: '/assistant/reply',
    body: z.object({
        prompt: z.string().min(1),
    }),
    responses: {
        200: {
            stream: {
                delta: z.object({
                    text: z.string(),
                }),
                done: z.object({
                    inputTokens: z.int(),
                    outputTokens: z.int(),
                }),
            },
        },
        400: ProblemDetailsSchema,
    },
});

Two bodys meet on a route like this. The route's own body is the request body, the prompt the caller posts, and a streaming route takes one like any other route. The body a status would declare under responses is the response body, and stream takes its place there: a status is sent at once or piece by piece. The 400 is an ordinary Problem Details body, returned or thrown as always. The 200 alone streams.

stream is a record of event name to schema, so every message the route sends has a name and a shape. Give it one schema instead when the messages are all alike.

Write the handler

The handler returns the status, and for body an async generator function. Kizuna sends the status and headers, calls the function, and writes each yield the moment it happens:

import Anthropic from '@anthropic-ai/sdk';

const anthropic = new Anthropic();

export const assistant = k.routes.assistant({
    reply: replyRoute.handler(async ({ body }) => ({
        status: 200,
        body: async function* ({ signal }) {
            const stream = anthropic.messages.stream(
                {
                    model: 'claude-opus-5',
                    max_tokens: 64000,
                    messages: [
                        {
                            role: 'user',
                            content: body.prompt,
                        },
                    ],
                },
                {
                    signal,
                }
            );
            for await (const event of stream) {
                if (event.type === 'content_block_delta' && event.delta.type === 'text_delta') {
                    yield {
                        event: 'delta',
                        data: {
                            text: event.delta.text,
                        },
                    };
                }
            }
            const message = await stream.finalMessage();
            yield {
                event: 'done',
                data: {
                    inputTokens: message.usage.input_tokens,
                    outputTokens: message.usage.output_tokens,
                },
            };
        },
    })),
});

Each yield names its event and carries data matching that event's schema, and TypeScript checks the pair, so a delta carrying a done payload is a compile error. The provider's own events stay inside the handler. What you declared describes your API, and the client on the other side reads delta and done without knowing which model answered.

The wire is text/event-stream, one block per yield, with data as JSON:

event: delta
data: {"text":"Hel"}

event: delta
data: {"text":"lo"}

event: done
data: {"inputTokens":12,"outputTokens":2}

Read it

On the streamed status the fetch client hands back body as an async iterable. Read it with for await, and each message narrows on event to its own data:

const result = await apiClient.assistant.reply({
    body: {
        prompt: 'Explain server-sent events in one paragraph',
    },
});

if (result.status === 200) {
    let reply = '';
    for await (const message of result.body) {
        if (message.event === 'delta') reply += message.data.text;
        if (message.event === 'done') console.log(`${message.data.outputTokens} tokens`);
    }
}

Every other status arrives as it does today, so result.status === 400 gives a parsed Problem Details body.

In React, append each delta to state and the reply types itself out:

const [reply, setReply] = useState('');

const ask = async (prompt: string) => {
    setReply('');
    const result = await apiClient.assistant.reply({
        body: {
            prompt,
        },
    });
    if (result.status !== 200) return;
    for await (const message of result.body) {
        if (message.event === 'delta') setReply((current) => current + message.data.text);
    }
};

Cancelling

Pass a signal as you would to fetch. Aborting it closes the connection:

const controller = new AbortController();

const result = await apiClient.assistant.reply({
    body: {
        prompt,
    },
    fetchOptions: {
        signal: controller.signal,
    },
});

stopButton.onclick = () => controller.abort();

On the server the generator's signal fires, so an upstream call that took it stops too, and Kizuna returns the generator so a finally block runs. The handler above hands signal to the Anthropic SDK for this reason.

Messages without names

Give stream one schema when every message has the same shape. The handler yields { data }, the wire carries data: lines alone, and the client reads message.data:

watchEvents: k
    .route({
        method: 'GET',
        path: '/events/watch',
        responses: {
            200: {
                stream: EventRecord,
            },
        },
    })
    .handler(async () => ({
        status: 200,
        body: async function* ({ signal }) {
            for await (const record of db.events.watch(signal)) {
                yield {
                    data: record,
                };
            }
        },
    })),

Ids, retry, and comments

A yield may carry the other two fields a server-sent event has. id names the message so a client that reconnects can say where it left off, and retry tells it how many milliseconds to wait before it does:

yield {
    event: 'delta',
    data: {
        text,
    },
    id: `${messageId}:${index}`,
    retry: 5000,
};

A comment yields a comment line. It carries no data, and it keeps an idle connection open through proxies that close quiet ones:

yield {
    comment: 'keep-alive',
};

The client hands id and retry back on each message and drops comments.

Resuming

A client that reconnects sends the last id it saw as Last-Event-ID. Declare the header and the handler reads it like any other:

watchEvents: k
    .route({
        method: 'GET',
        path: '/events/watch',
        headers: z.object({
            'last-event-id': z.string().optional(),
        }),
        responses: {
            200: {
                stream: EventRecord,
            },
        },
    })
    .handler(/* ... */),

The fetch client reads one connection and leaves reconnecting to you. The browser's EventSource reconnects on its own and can open a GET route like this one directly, with the limits it has always had: GET only, no request body, and no headers of your own, so it reaches the POST assistant route above through the fetch client alone.

Raw chunks

A z.string() schema with a text/* content type, or BinarySchema with any other, writes each yield as it is, with no framing around it:

exportEvents: k
    .route({
        method: 'GET',
        path: '/events/export.csv',
        responses: {
            200: {
                stream: z.string(),
                contentType: 'text/csv',
            },
        },
    })
    .handler(async () => ({
        status: 200,
        body: async function* () {
            yield 'id,kind,occurredAt\n';
            for await (const record of db.events.all()) {
                yield `${record.id},${record.kind},${record.occurredAt}\n`;
            }
        },
    })),

For bytes, hand over a ReadableStream in place of the generator. That is how a route passes a file from storage through without holding it in memory:

downloadArchive: k
    .route({
        method: 'GET',
        path: '/reports/:id/archive',
        responses: {
            200: {
                stream: BinarySchema,
                contentType: 'application/zip',
            },
        },
    })
    .handler(async ({ params }) => {
        const upstream = await fetch(storage.archiveUrl(params.id));
        return {
            status: 200,
            body: upstream.body,
        };
    }),

The client iterates strings for a text stream and Uint8Array chunks for a binary one.

Building the generator elsewhere

An inline generator function sits inside the route's return type, so its event names infer as literals. One declared on its own has nothing to infer from, and TypeScript widens event to string. Annotate it with StreamBody, naming the route and the status:

import type { StreamBody } from 'kizunajs';
import { replyRoute } from './assistant';

const cannedReply: StreamBody<typeof replyRoute, 200> = async function* () {
    yield {
        event: 'delta',
        data: {
            text: 'Hello',
        },
    };
    yield {
        event: 'done',
        data: {
            inputTokens: 0,
            outputTokens: 1,
        },
    };
};

export const assistant = k.routes.assistant({
    reply: replyRoute.handler(async () => ({
        status: 200,
        body: cannedReply,
    })),
});

Errors

Before the first message, everything works as on any route: validation answers 400, a guard 401 or 403, and throwError any status the route declares, all as Problem Details. throwError takes no streamed status. A stream is a body to produce, and bailing out is what throwError is for.

Once the handler has returned, the status and headers are on the wire. A throw inside the generator after that cannot become a 500, so Kizuna ends the connection. The client's loop rejects instead of finishing quietly, and the adapter's error callback is told. Declare a terminal event like done when the reader must know the reply is whole, and an error event of your own when it must know why it stopped short.

With responseValidation on, Kizuna validates each message against its schema before writing it, and a message that fails ends the stream the same way.

Headers

A streamed response goes out with Content-Type: text/event-stream and Cache-Control: no-store, since a cache stores a complete representation and a stream has none. Return your own headers to add to what Kizuna sends, the way you would on any response. X-Accel-Buffering: no tells nginx to pass each message through rather than collect them.

Every GET route answers HEAD, a stream route included: the status and headers, no body, and no Content-Length, since the length is unknown. The handler runs, and the generator is never called.

Tools

A streamed response can name the tools the model may call, adding tool_call, tool_result and tool_error to the events it declares:

200: {
    stream: {
        delta: z.object({
            text: z.string(),
        }),
    },
    tools,
},

Each is discriminated on the tool's dotted key. The handler receives tools, whose run calls one of those routes as the person streaming and yields its events. See tools.