Input Streams - Trigger.dev

The Input Streams API allows you to send data into running Trigger.dev tasks from your backend code. This enables bidirectional communication — while output streams let you read data from tasks, input streams let you push data into them.

To learn how to receive input stream data inside your tasks, see Input Streams in the Streams doc.

Input streams are keyed by runId — they’re correct for sending data to a specific live run. If you need a bidirectional channel that survives run boundaries (e.g. a chat that resumes tomorrow, an agent coordinated across many runs), look at chat.agent: it’s built on a durable Session row that owns its runs and exposes the same consumer-side API (on / once / wait / waitWithIdleTimeout) on its .in channel.

Sending data to a running task

Using defined input streams (Recommended)

The recommended approach is to use defined input streams for full type safety:

import { cancelSignal, approval } from "./trigger/streams";

// Cancel a running AI stream
await cancelSignal.send(runId, { reason: "User clicked stop" });

// Approve a draft
await approval.send(runId, { approved: true, reviewer: "alice@example.com" });

The .send() method is fully typed — the data parameter must match the generic type you defined on the input stream.

.send() works the same regardless of how the task is listening — whether it uses .wait() (suspending), .once() (non-suspending), or .on() (continuous). The sender doesn’t need to know how the task is consuming the data. See Input Streams for details on each receiving method.

Practical examples

Cancel from a Next.js API route

app/api/cancel/route.ts

import { cancelStream } from "@/trigger/streams";

export async function POST(req: Request) {
  const { runId } = await req.json();

  await cancelStream.send(runId, { reason: "User clicked stop" });

  return Response.json({ cancelled: true });
}

Approval workflow API

app/api/approve/route.ts

import { approval } from "@/trigger/streams";

export async function POST(req: Request) {
  const { runId, approved, reviewer } = await req.json();

  await approval.send(runId, {
    approved,
    reviewer,
  });

  return Response.json({ success: true });
}

Remix action handler

app/routes/api.approve.ts

import { json, type ActionFunctionArgs } from "@remix-run/node";
import { approval } from "~/trigger/streams";

export async function action({ request }: ActionFunctionArgs) {
  const formData = await request.formData();
  const runId = formData.get("runId") as string;
  const approved = formData.get("approved") === "true";
  const reviewer = formData.get("reviewer") as string;

  await approval.send(runId, { approved, reviewer });

  return json({ success: true });
}

Express handler

import express from "express";
import { cancelSignal } from "./trigger/streams";

const app = express();
app.use(express.json());

app.post("/api/cancel", async (req, res) => {
  const { runId, reason } = req.body;

  await cancelSignal.send(runId, { reason });

  res.json({ cancelled: true });
});

Sending from another task

You can send input stream data from one task to another running task:

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

export const reviewerTask = task({
  id: "auto-reviewer",
  run: async (payload: { targetRunId: string }) => {
    // Perform automated review logic...
    const isApproved = await performReview();

    // Send approval to the waiting task
    await approval.send(payload.targetRunId, {
      approved: isApproved,
      reviewer: "auto-reviewer",
    });
  },
});

Error handling

The .send() method will throw if:

The run has already completed, failed, or been canceled
The payload exceeds the 1MB size limit
The run ID is invalid

import { cancelSignal } from "./trigger/streams";

try {
  await cancelSignal.send(runId, { reason: "User clicked stop" });
} catch (error) {
  console.error("Failed to send:", error);
  // Handle the error — the run may have already completed
}

Important notes

Maximum payload size per .send() call is 1MB
You cannot send data to a completed, failed, or canceled run
Data sent before a listener is registered inside the task is buffered and delivered when a listener attaches
Input streams require the current streams implementation (v2 is the default in SDK 4.1.0+). See Streams for details.

​Sending data to a running task

​Using defined input streams (Recommended)

​Practical examples

​Cancel from a Next.js API route

​Approval workflow API

​Remix action handler

​Express handler

​Sending from another task

​Error handling

​Important notes

Sending data to a running task

Using defined input streams (Recommended)

Practical examples

Cancel from a Next.js API route

Approval workflow API

Remix action handler

Express handler

Sending from another task

Error handling

Important notes