Streaming

Functions stream responses, and consume streaming request bodies, with the Web Streams API (ReadableStream, WritableStream, TransformStream). Return a Response whose body is a ReadableStream and the runtime streams it to the client, both in local dev and in production. Chunks reach the client as they’re produced instead of buffering in memory.

Server-sent events

The most common case: push real-time updates to a client over one long-lived response.

// functions/api/v1/events.get.ts, SSE stream
import { defineFunction } from '@trayai/helix-sdk';

export default defineFunction(async (ctx) => {
  const stream = new ReadableStream({
    start(controller) {
      const interval = setInterval(() => {
        const data = JSON.stringify({ time: Date.now(), status: 'ok' });
        controller.enqueue(`data: ${data}\n\n`);
      }, 1000);

      // Clean up when the client disconnects
      ctx.request.signal.addEventListener('abort', () => {
        clearInterval(interval);
        controller.close();
      });
    },
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    },
  });
});

Streaming an LLM response

Use ctx.ai(alias) with streamText. The platform injects the model provider’s API key, so the credential never appears in your code, and the tokens reach the client as the model produces them:

// functions/api/v1/chat.post.ts, stream an LLM response to the client
import { z } from 'zod';
import { defineFunction } from '@trayai/helix-sdk';

export const input = z.object({
  prompt: z.string().min(1),
});

export default defineFunction<typeof input>(async (ctx) => {
  const result = ctx.ai('openaiKey').streamText({
    model: 'gpt-4o',
    prompt: ctx.input.prompt,
  });

  return result.toUIMessageStreamResponse();
});

See AI for the providers and methods ctx.ai supports.

Transforming chunks

To inspect or modify chunks on the way through, read result.textStream and write into your own stream:

// functions/api/v1/chat-with-logging.post.ts, stream and count chunks
import { z } from 'zod';
import { defineFunction } from '@trayai/helix-sdk';

export const input = z.object({
  prompt: z.string().min(1),
});

export default defineFunction<typeof input>(async (ctx) => {
  const result = ctx.ai('openaiKey').streamText({
    model: 'gpt-4o',
    prompt: ctx.input.prompt,
  });

  const encoder = new TextEncoder();
  let chunkCount = 0;

  const stream = new ReadableStream({
    async start(controller) {
      for await (const text of result.textStream) {
        chunkCount += 1;
        controller.enqueue(encoder.encode(text));
      }
      ctx.log.info('Stream complete', { chunkCount });
      controller.close();
    },
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/plain; charset=utf-8',
      'Cache-Control': 'no-cache',
    },
  });
});

NDJSON

Newline-delimited JSON streams structured data one complete JSON object per line:

// functions/api/v1/export.get.ts, stream stored records as NDJSON
import { defineFunction } from '@trayai/helix-sdk';

export default defineFunction(async (ctx) => {
  const kv = ctx.kv();
  const ids = (await kv.getItem<string[]>('user:index')) ?? [];

  const stream = new ReadableStream({
    async start(controller) {
      const encoder = new TextEncoder();
      // Read one record at a time so the whole set is never held in memory
      for (const id of ids) {
        const user = await kv.getItem(`user:${id}`);
        if (user) controller.enqueue(encoder.encode(JSON.stringify(user) + '\n'));
      }
      controller.close();
    },
  });

  return new Response(stream, {
    headers: { 'Content-Type': 'application/x-ndjson' },
  });
});

Streaming request bodies

Incoming requests stream too. ctx.request.body is a ReadableStream of the raw request, useful for large uploads you want to process chunk by chunk instead of holding in memory. Helix ships no file storage, so there is nowhere to write the bytes; what you can do today is consume each chunk as it arrives and keep the result:

// functions/api/v1/upload.post.ts, hash a large upload chunk by chunk
import { createHash } from 'node:crypto';
import { defineFunction } from '@trayai/helix-sdk';

export default defineFunction(async (ctx) => {
  const reader = ctx.request.body.getReader();
  const hash = createHash('sha256');
  let totalBytes = 0;

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    totalBytes += value.byteLength;
    // Process each chunk as it arrives, never holding the whole body
    hash.update(value);
  }

  const sha256 = hash.digest('hex');
  await ctx.kv().setItem(`upload:${sha256}`, {
    bytes: totalBytes,
    receivedAt: new Date().toISOString(),
  });

  ctx.log.info('Upload complete', { totalBytes, sha256 });
  return { uploaded: true, bytes: totalBytes, sha256 };
});

Client disconnects

ctx.request.signal is an AbortSignal that fires when the client disconnects, in local dev and in production alike. Listen for abort to stop timers, cancel upstream requests, and close stream controllers, as the SSE example above does. Without cleanup, a disconnected client leaves your function producing chunks nobody will receive.

Streaming in production

Locally, helix dev serves streams over Node.js with no size limits. In production you return a Response with a ReadableStream body and the runtime streams it for you.