Custom agents

A custom agent is a task you register with chat.customAgent() and drive yourself — either with the managed turn iterator from chat.createSession(), or with a fully hand-rolled loop over the raw chat primitives. You give up chat.agent()'s lifecycle hooks and automatic continuation recovery; you gain inline control over every turn, and (at the lowest level) full control over the stream conversion.

See the comparison table before dropping down. The frontend is unchanged either way: all levels speak the same wire protocol, so useTriggerChatTransport points at a custom agent exactly like a chat.agent().

chat.customAgent()

chat.customAgent() is a thin wrapper around task() that does two things: it registers the task as an agent (so it appears in the agent dashboard, the playground, and the MCP server's list_agents), and it binds the run to its backing Session so the chat.* primitives resolve to the right .in/.out channels. There is no managed lifecycle — no turn loop, no hooks, no preload handling.

A plain task() works with the same primitives but stays invisible to the agent surfaces, so prefer customAgent unless you specifically don't want the task listed as an agent.

Inside the wrapper, pick one of two loop styles:

• Managed loop — chat.createSession() yields turns; the SDK handles stop signals, accumulation, idle suspend/resume, and turn-complete signaling. You write the turn body.
• Hand-rolled loop — you write the loop itself with chat.messages, MessageAccumulator, pipeAndCapture, and writeTurnComplete. The right choice when you need complete control over .toUIMessageStream() (e.g. onFinish, originalMessages) beyond what chat.setUIMessageStreamOptions() provides, or you're implementing a custom protocol.

Validating client data

Use chat.withClientData({ schema }) to validate payload.metadata. Custom agents parse the metadata on the initial payload and every later non-close input frame before passing it to run, chat.messages, or chat.createSession. Schema defaults and transforms are included in the value your code receives.

This only validates metadata. A raw custom agent does not expose an action schema, so payload.action remains unknown. Validate the full frame or action payload in your own loop when you need that boundary.

If validation fails for a submitted turn or an async read such as wait(), the SDK consumes and skips the invalid frame, writes an Invalid client data error followed by turn-complete, then waits for the next valid frame. The invalid value is not returned to the raw caller. The detailed validator error is available in the task log and onClientDataValidationError, but it is not sent to the client.

This convenience path settles the invalid input before the read returns. If your raw loop needs to coordinate validation with persistence or settlement, omit withClientData({ schema }) and validate the full wire frame in the loop instead. A messageless preload or continuation boot has no submitted turn to complete, so the SDK reports the error through the task log and callback while it waits.

An invalid head-start handover boot fails closed. The SDK waits for the warm handler to finish so stream ordering stays intact. A handover skip ends the run. A real handover writes the validation error and turn-complete after the warm output, then ends the run. Without a schema, metadata is passed through unchanged.

chat.messages.on() is different because a subscribed frame can arrive while the current response is still streaming. Ending the turn at that point would cut off the response. While the subscription is active, the SDK skips an invalid frame, logs the validation error, and calls onClientDataValidationError if you set it. A raw subscription has no turn boundary the SDK can key a later write to, so it reports through the callback and the task log only and never writes to the stream.

The steering subscription created by chat.createSession({ pendingMessages }) skips an invalid frame the same way, but the session does own the turn boundary, so it can write the client-visible error once the turn has closed. reportErrorAt governs that write and applies to steering frames only, not to chat.messages.on().

By default the Invalid client data error for a steering frame is held until the turn ends, so a bad send cannot truncate an answer the user is already reading. Pass reportErrorAt: "arrival" to withClientData if you would rather surface it as soon as validation fails, accepting that it ends the response in progress:

chat.withClientData({
  schema: z.object({ userId: z.string() }),
  reportErrorAt: "arrival",
  onValidationError: ({ error, payload }) => logger.warn("bad client data", { error }),
});

The validation callback and the task log fire on arrival in both modes, and the frame is never delivered as a turn either way.

Calling off() stops the subscription from accepting new frames. A valid frame accepted before off() still finishes validation and is delivered to the handler. An invalid frame that finishes validation after off() is logged without calling the handler or error callback.

import { chat } from "@trigger.dev/sdk/ai";
import { z } from "zod";

export const myChat = chat
  .withClientData({ schema: z.object({ userId: z.string() }) })
  .customAgent({
    id: "my-chat",
    onClientDataValidationError: ({ error, payload }) => {
      console.warn("Invalid client data", { error, trigger: payload.trigger });
    },
    run: async (payload) => {
      // ...
    },
  });

chat.messages.peek() validates synchronously and throws validation errors to the caller. If your schema only supports asynchronous parsing, use once(), wait(), or waitWithIdleTimeout() instead.

Managed loop: chat.createSession()

chat.createSession() gives you an async iterator of ChatTurn objects. Each turn arrives with the accumulated history, a combined stop+cancel signal, and helpers to finish the turn:

import { chat } from "@trigger.dev/sdk/ai";
import { streamText, stepCountIs } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
import { z } from "zod";

export const myChat = chat
  .withClientData({ schema: z.object({ userId: z.string() }) })
  .customAgent({
    id: "my-chat",
    run: async (payload, { signal }) => {
      // One-time initialization — plain code, no hooks. Upsert, not create:
      // continuation runs boot with the row already in place.
      const clientData = payload.metadata!;
      await db.chat.upsert({
        where: { id: payload.chatId },
        create: { id: payload.chatId, userId: clientData.userId },
        update: {},
      });

      const session = chat.createSession(payload, {
        signal,
        idleTimeoutInSeconds: 60,
        timeout: "1h",
      });

      for await (const turn of session) {
        // Persist the incoming user message BEFORE streaming — this is your
        // onTurnStart equivalent. Without it, a page reload mid-stream
        // restores the assistant text (replayed from the session) but loses
        // the user message that prompted it.
        await db.chat.update({
          where: { id: turn.chatId },
          data: { messages: turn.uiMessages },
        });

        const result = streamText({
          model: anthropic("claude-sonnet-4-5"),
          messages: turn.messages,
          abortSignal: turn.signal,
          stopWhen: stepCountIs(15),
        });

        // Pipe, capture, accumulate, and signal turn-complete — all in one call
        await turn.complete(result);

        // Persist the full exchange after the turn — your onTurnComplete equivalent
        await db.chat.update({
          where: { id: turn.chatId },
          data: { messages: turn.uiMessages },
        });
      }
    },
  });

ChatSessionOptions

Between turns the run idles on waitWithIdleTimeout: after idleTimeoutInSeconds with no message it suspends (compute is freed), and the next message restores it on the same run — the same warm/suspended pipeline chat.agent() uses.

ChatTurn

Each turn yielded by the iterator provides:

Continuation runs and history seeding

chat.agent() rebuilds conversation history automatically when a chat continues on a fresh run (after a cancel, crash, version upgrade, or TTL expiry) — via its snapshot/replay boot or your hydrateMessages hook. Custom agents do none of that: a continuation run starts with an empty accumulator, and history restoration is your job.

With createSession, check turn.continuation on the first turn and seed from your store with turn.setMessages():

for await (const turn of session) {
  if (turn.continuation && turn.number === 0) {
    const row = await db.chat.findUnique({ where: { id: turn.chatId } });
    const stored = (row?.messages ?? []) as UIMessage[];
    if (stored.length > 0) {
      // Keep any incoming message that isn't already persisted
      const incoming = turn.uiMessages.filter((m) => !stored.some((s) => s.id === m.id));
      await turn.setMessages([...stored, ...incoming]);
    }
  }

  // ... streamText + turn.complete as usual
}

Without this, a resumed chat silently loses its history: the model sees only the message that triggered the continuation. In a hand-rolled loop, seed by passing the stored history into the turn-0 addIncoming call — shown in the example below.

Rotating to a new deployment

With chat.createSession(), use chat.requestUpgrade() and let the iterator exit normally. For an immediate handoff, close the iterator before calling chat.endAndContinue(); the method rejects until the iterator and any active next() call have settled. In a fully hand-rolled custom agent, call it directly to hand the Session to a fresh run.

Close the iterator between reads. If return() races a next() that is already waiting for input, it waits for that read to settle before releasing the handoff guard. Input dispatched while the iterator is closing is not yielded as a turn and remains available to the continuation unless you write another turn-complete boundary.

Call it between turns, after detaching the old run's input listeners. If the old run completed its current turn, persist its state and write the turn-complete boundary before the handoff:

// Detach any chat.messages.on() subscriptions you created.
stop.cleanup();
await persistMessages(conversation.uiMessages);
await chat.writeTurnComplete();
await chat.endAndContinue();
return;

The server starts a continuation run using the Session's existing trigger configuration and atomically makes it the current run. The Session and its streams stay open, so input that the old run has not consumed remains on .in for the continuation run. The new run uses the latest deployed task version unless the Session's trigger configuration sets lockToVersion.

If input has been dispatched to the old run but should be processed by the continuation, detach the listeners and do not write another turn-complete boundary before handing off. chat.writeTurnComplete() acknowledges the latest input dispatched to the old run; writing it after that dispatch would make the continuation resume after the input.

turn.complete() vs manual control

turn.complete(result) is the one-call path — it handles piping, capturing the response, accumulating messages, cleaning up aborted parts on a stop, and writing the turn-complete chunk.

For more control, you can do each step manually:

for await (const turn of session) {
  const result = streamText({
    model: anthropic("claude-sonnet-4-5"),
    messages: turn.messages,
    abortSignal: turn.signal,
    stopWhen: stepCountIs(15),
  });

  // Manual: pipe and capture separately
  const { message } = await chat.pipeAndCapture(result, { signal: turn.signal });

  if (message) {
    // Custom processing before accumulating
    await turn.addResponse(message);
  }

  // Custom persistence, analytics, etc.
  await db.chat.update({ ... });

  // Must call done() when not using complete()
  await turn.done();
}

Stopping generation

The frontend stops a turn with transport.stopGeneration(chatId), which writes a stop signal to the session's input stream. It aborts the current turn's generation but keeps the run alive, so the next message continues on the same session.

A stop only applies to the turn that was live when it arrived. If the run crashes and a later run recovers a message that had not been answered yet, a stop that was already applied before the crash is not applied again, so the turn answering the recovered message runs to completion. A stop sent after the recovery is live and aborts that turn as normal.

turn.signal is a combined stop-and-cancel AbortSignal, fresh each turn. Pass it to streamText so the stop reaches the model, then let turn.complete() finish the turn:

for await (const turn of session) {
  const result = streamText({
    model: anthropic("claude-sonnet-4-5"),
    messages: turn.messages,
    abortSignal: turn.signal, // fires on a user stop OR a run cancel
    stopWhen: stepCountIs(15),
  });

  await turn.complete(result);

  if (turn.stopped) {
    // user stopped this turn — the partial response is already accumulated
  }
}

On a stop, turn.complete() cleans up the aborted parts of the partial response, accumulates it as its own assistant message, and writes turn-complete. The run does not end — the loop continues to the next turn.

Read turn.stopped to tell a user stop from a full run cancel:

• User stop (transport.stopGeneration): turn.signal aborts, turn.stopped is true, the partial response is accumulated, and the run stays alive for the next message.
• Run cancel (cancelled, expired, or maxDuration exceeded): turn.signal aborts, turn.stopped is false, and turn.complete() returns without accumulating because the run is ending.

A hand-rolled loop wires this itself with chat.createStopSignal() and chat.cleanupAbortedParts(). Two things createSession handles for you are easy to get wrong there — see the hand-rolled loop checklist.

Hand-rolled loop with primitives

For full control, skip createSession and compose the primitives directly:

chat.messages mailbox

chat.messages exposes the incoming message mailbox for hand-rolled loops:

hasPending() checks whether a message has already been delivered locally and is waiting for next() to take it. It does not query the remote Session channel or start a subscription. Use waitWithIdleTimeout() when the loop needs to idle until future input arrives.

next({ timeoutInSeconds: 0 }) is also a local, non-blocking read. Call next() without a timeout, or with a positive timeout, to subscribe for future input.

next() returns a readonly record envelope:

const record = await chat.messages.next({ timeoutInSeconds: 5 });
if (record) {
  console.log(record.id, record.seqNum);
  currentPayload = record.payload;
}

• id is the append's stable idempotency key.
• seqNum is the monotonic sequence on this Session's .in channel.
• payload is the existing ChatTaskWirePayload delivered by the other mailbox methods.

Both identifiers remain the same if the record is delivered again after a reconnect. Each next() call commits only the record it returns, so a loop that owns its own turn sequencing never advances past input it has not taken. By contrast, on() commits a record as soon as it dispatches the handler; avoid mixing on() and next() when a single loop owns mailbox consumption.

The Session .in channel also carries control records such as stops and handovers. Those are routed to their own consumers and never block messages: a message that arrived behind one is still reported by hasPending() and still returned by next(), in channel order. The same holds for a record kind this version of the SDK does not recognise, which is discarded rather than left where it would make every message behind it undeliverable.

next() returns undefined when no message became consumable before the timeout.

A complete loop:

import { chat, type ChatTaskWirePayload } from "@trigger.dev/sdk/ai";
import { streamText, stepCountIs } from "ai";
import { anthropic } from "@ai-sdk/anthropic";

export const myChat = chat.customAgent({
  id: "my-chat-raw",
  run: async (payload: ChatTaskWirePayload, { signal: runSignal }) => {
    let currentPayload = payload;

    // Handle preload — wait for the first real message
    if (currentPayload.trigger === "preload") {
      const result = await chat.messages.waitWithIdleTimeout({
        idleTimeoutInSeconds: 60,
        timeout: "1h",
        spanName: "waiting for first message",
      });
      if (!result.ok) return;
      currentPayload = result.output;
    }

    const stop = chat.createStopSignal();
    const conversation = new chat.MessageAccumulator();

    // Continuation runs (cancel, crash, upgrade) start with an empty
    // accumulator — fetch stored history so turn 0 can seed it.
    let continuationSeed: UIMessage[] = [];
    if (currentPayload.continuation) {
      const row = await db.chat.findUnique({ where: { id: currentPayload.chatId } });
      continuationSeed = (row?.messages ?? []) as UIMessage[];
    }

    for (let turn = 0; turn < 100; turn++) {
      stop.reset();

      // The wire payload carries at most one new message per turn. Turn 0
      // REPLACES the accumulator, so seed stored history through
      // addIncoming together with the incoming message — a setMessages
      // call before the loop would be wiped here.
      const incoming = currentPayload.message ? [currentPayload.message] : [];
      const turnInput =
        turn === 0 && continuationSeed.length > 0
          ? [...continuationSeed.filter((s) => !incoming.some((m) => m.id === s.id)), ...incoming]
          : incoming;
      const messages = await conversation.addIncoming(turnInput, currentPayload.trigger, turn);

      // Persist the incoming user message before streaming so a
      // mid-stream reload doesn't lose it.
      await db.chat.update({
        where: { id: currentPayload.chatId },
        data: { messages: conversation.uiMessages },
      });

      const combinedSignal = AbortSignal.any([runSignal, stop.signal]);

      const result = streamText({
        model: anthropic("claude-sonnet-4-5"),
        messages,
        abortSignal: combinedSignal,
        stopWhen: stepCountIs(15),
      });

      const { message, status, error } = await chat.pipeAndCapture(result, {
        signal: combinedSignal,
      });
      // pipeAndCapture never throws: a user stop returns status "aborted" with
      // the partial message, and a failure returns status "error" with `error`.
      if (status === "error") throw error;

      if (message) {
        const cleaned =
          stop.signal.aborted && !runSignal.aborted ? chat.cleanupAbortedParts(message) : message;
        await conversation.addResponse(cleaned);
      }

      if (runSignal.aborted) break;

      // Persist, analytics, etc.
      await db.chat.update({
        where: { id: currentPayload.chatId },
        data: { messages: conversation.uiMessages },
      });

      await chat.writeTurnComplete();

      // Wait for the next message
      const next = await chat.messages.waitWithIdleTimeout({
        idleTimeoutInSeconds: 60,
        timeout: "1h",
        spanName: "waiting for next message",
      });
      if (!next.ok) break;
      currentPayload = next.output;
    }

    stop.cleanup();
  },
});

MessageAccumulator

addIncoming(messages, trigger, turn) has two modes:

• Turn 0 or trigger === "regenerate-message": replaces the accumulator with exactly what you pass. This is why continuation seeding goes through addIncoming (above), and why a regenerate needs you to slice your own history — the wire omits the message on regenerate, so pass the stored history minus the last assistant message.
• Every other turn: appends what you pass (the wire carries at most the one new user message).

const conversation = new chat.MessageAccumulator();

// Returns full accumulated ModelMessage[] for streamText
const messages = await conversation.addIncoming(
  payload.message ? [payload.message] : [],
  payload.trigger,
  turn
);

// After piping, add the response
const { message } = await chat.pipeAndCapture(result);
if (message) await conversation.addResponse(message);

// Access accumulated messages for persistence
conversation.uiMessages; // UIMessage[]
conversation.modelMessages; // ModelMessage[]

The constructor also accepts compaction and pendingMessages options (same shapes as on chat.agent()); pass prepareStep: conversation.prepareStep() to streamText to activate them. See pending messages for the manual steering wiring.

Hand-rolled loop checklist

Things the managed levels do for you that a raw loop has to get right:

• Don't bare-await result.totalUsage. On a stop-abort the AI SDK's totalUsage promise never settles, which wedges the loop forever. Race it with a timeout:

  tsCopy

  const turnUsage = await Promise.race([
    result.totalUsage,
    new Promise((resolve) => setTimeout(() => resolve(undefined), 2000)),
  ]);
  

• Persist the user message before streaming (shown in the example above). The session replay restores the assistant's streamed text after a page reload, but nothing restores a user message you haven't written down.

• Seed history on continuation runs through the turn-0 addIncoming (shown above). payload.continuation is true when this run picked up an existing chat; the accumulator starts empty — and because turn 0 replaces the accumulator, a setMessages call before the loop gets wiped.

• Clean up aborted parts on a stop with chat.cleanupAbortedParts() before accumulating, or the partial response carries half-open tool calls into the next turn's prompt.

• Read payload.message (singular). The wire payload carries at most one new message per turn; there is no messages array on the payload.

Next steps