Realtime Streams

Pipe streaming data from Trigger.dev tasks to your frontend or backend in real-time. Perfect for AI completions, progress updates, or any continuous data flow.

Streams v2 requires SDK 4.1.0+

Streams v2 is a significant upgrade. See limits comparison below.

Limits: v1 vs v2

Limit	Streams v1	Streams v2
Max stream length	2,000 chunks	Unlimited
Active streams per run	5	Unlimited
Max streams per run	10	Unlimited
Stream TTL	1 day	28 days
Max stream size	10 MB	300 MiB

Defining Typed Streams (Recommended)

Define streams in a shared location for reuse across tasks and frontend:

// app/streams.ts
import { streams, InferStreamType } from "@trigger.dev/sdk";

export const aiStream = streams.define<string>({ id: "ai-output" });
export type AIStreamPart = InferStreamType<typeof aiStream>;

Using Streams in Tasks

Pipe a ReadableStream

import { task } from "@trigger.dev/sdk";
import { aiStream } from "./streams";

export const streamTask = task({
  id: "stream-task",
  run: async (payload: { prompt: string }) => {
    const stream = await getAIStream(payload.prompt); // Returns ReadableStream
    const { stream: readableStream, waitUntilComplete } = aiStream.pipe(stream);
    await waitUntilComplete();
    return { message: "Stream completed" };
  },
});

Append Individual Chunks

await logStream.append("Processing started");
await progressStream.append({ step: "Initialization", percent: 0 });

Write Multiple Chunks with Writer

const { waitUntilComplete } = logStream.writer({
  execute: ({ write, merge }) => {
    write("Chunk 1");
    write("Chunk 2");
    const additionalStream = ReadableStream.from(["Chunk 3", "Chunk 4"]);
    merge(additionalStream);
  },
});
await waitUntilComplete();

Reading Streams (Backend)

import { aiStream } from "./streams";

const stream = await aiStream.read(runId);
for await (const chunk of stream) {
  console.log(chunk);
}

React Hooks (Frontend)

"use client";
import { useRealtimeStream } from "@trigger.dev/react-hooks";
import { aiStream } from "@/app/streams";

export function StreamViewer({
  accessToken,
  runId,
}: {
  accessToken: string;
  runId: string;
}) {
  const { parts, error } = useRealtimeStream(aiStream, runId, {
    accessToken,
    timeoutInSeconds: 600,
    throttleInMs: 100, // prevent excessive re-renders
  });

  if (error) return <div>Error: {error.message}</div>;
  if (!parts) return <div>Loading...</div>;

  return (
    <div>
      {parts.map((part, i) => (
        <span key={i}>{part}</span>
      ))}
    </div>
  );
}

Targeting Different Runs

Pipe a stream to any run — not just the current one:

aiStream.pipe(stream, { target: "parent" }); // send to parent task
aiStream.pipe(stream, { target: "root" });   // send to root task
aiStream.pipe(stream, { target: "self" });   // default
aiStream.pipe(stream, { target: otherRunId }); // any run

Streaming from Outside a Task

Pipe AI output from a Next.js API route into a running task's stream:

import { streams } from "@trigger.dev/sdk";
import { streamText } from "ai";

export async function POST(req: Request) {
  const { messages, runId } = await req.json();
  const result = streamText({ model: openai("gpt-4o"), messages });
  const { stream } = streams.pipe("ai-stream", result.toUIMessageStream(), {
    target: runId,
  });
  return new Response(stream as any, {
    headers: { "Content-Type": "text/event-stream" },
  });
}

Best Practices

1. Use streams.define() for type safety and reuse
2. Export stream types using InferStreamType
3. Handle errors gracefully in UI components
4. Use throttleInMs in useRealtimeStream to prevent excessive re-renders
5. Target parent runs when orchestrating child tasks
6. Use descriptive stream IDs that reflect the data type