# Streaming

> Return Web Streams responses from functions: Server-Sent Events, LLM pass-through, and NDJSON, with Lambda response streaming in production.

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, where it sets up Lambda response streaming for you. 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.

```typescript
// 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

Proxy an authenticated LLM API call and pipe the upstream stream straight through to the client. The auth alias injects the API key, so the credential never appears in your code, and the response is never buffered:

```typescript
// 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({
  messages: z.array(z.object({
    role: z.enum(['system', 'user', 'assistant']),
    content: z.string(),
  })),
  model: z.string().default('gpt-4o'),
});

export default defineFunction<typeof input>(async (ctx) => {
  // The auth module injects the OpenAI API key, never in your code
  const response = await ctx.http
    .authed('openai')
    .post('https://api.openai.com/v1/chat/completions', {
      json: {
        model: ctx.input.model,
        messages: ctx.input.messages,
        stream: true,
      },
    });

  // Pipe the upstream response body directly to the client, no buffering
  return new Response(response.body, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
    },
  });
});
```

`ctx.http.authed('openai').post(...)` returns a `Response` whose `.body` is a `ReadableStream`. Passing it into a new `Response` lets upstream chunks flow through as they arrive, which matters for LLM output that can be large and takes several seconds. Calling models through a managed gateway (`ctx.ai()`) rather than your own credentials is platform design, not something you can do today.

### Transforming chunks

When you need to inspect or modify chunks on the way through, insert a `TransformStream`:

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

export const input = z.object({
  messages: z.array(z.object({
    role: z.enum(['system', 'user', 'assistant']),
    content: z.string(),
  })),
});

export default defineFunction<typeof input>(async (ctx) => {
  const response = await ctx.http
    .authed('openai')
    .post('https://api.openai.com/v1/chat/completions', {
      json: { model: 'gpt-4o', messages: ctx.input.messages, stream: true },
    });

  let tokenCount = 0;

  const transform = new TransformStream({
    transform(chunk, controller) {
      // chunk is a Uint8Array
      const text = new TextDecoder().decode(chunk);
      const lines = text.split('\n').filter(l => l.startsWith('data: '));
      tokenCount += lines.length;

      // Pass the chunk through unchanged
      controller.enqueue(chunk);
    },
    flush() {
      ctx.log.info('Stream complete', { tokenCount });
    },
  });

  return new Response(response.body.pipeThrough(transform), {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
    },
  });
});
```

## NDJSON

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

```typescript
// 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:

```typescript
// 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 the runtime sets up Lambda response streaming automatically; you return a `Response` with a `ReadableStream` body and the runtime does the rest, with no `awslambda.streamifyResponse` wrapper in your code.

| Behavior | Production detail |
|----------|-------------------|
| Response size | Up to 20 MB streamed, versus 6 MB for buffered responses |
| First byte | Can reach the client before the function finishes |
| Billing | Billed per ms of streaming duration |
| Timeout | The function timeout still applies (default 30 s, max 15 min) |
| Disconnects | `ctx.request.signal` fires on client disconnect |

:::note{title="Output schemas don't apply to streams"}
When a function returns a streaming `Response`, the `output` schema is not applied. A stream can't be validated before it has been fully produced, so output validation runs only on non-streaming responses (plain objects and JSON).
:::

---

Canonical: https://helix.tray.ai/documentation/guides/streaming/
Any link on this page is available as markdown by appending .md to its URL.
Full corpus: https://helix.tray.ai/documentation/llms-full.txt