# Custom agents

> Source: https://trigger.dev/docs/ai-chat/custom-agents

The AI Agents and Prompts surface ships as part of the **v4.5 release candidate**. Install with `@trigger.dev/sdk@rc` (or pin `4.5.0-rc.0` or later) to use these features — they aren’t yet on the latest stable, and APIs may still change before the 4.5.0 GA. See [supported AI SDK versions](https://trigger.dev/docs/ai-chat/reference#compatibility)
 and the [AI chat changelog](https://trigger.dev/docs/ai-chat/changelog)
 for details.

**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](https://trigger.dev/docs/ai-chat/backend)
 before dropping down. The frontend is unchanged either way: all levels speak the same wire protocol, so [`useTriggerChatTransport`](https://trigger.dev/docs/ai-chat/frontend)
 points at a custom agent exactly like a `chat.agent()`.

[​](https://trigger.dev/docs/ai-chat/custom-agents#chat-customagent)

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](https://trigger.dev/docs/ai-chat/sessions)
 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](https://trigger.dev/docs/ai-chat/custom-agents#managed-loop-chatcreatesession)
    ** — `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](https://trigger.dev/docs/ai-chat/custom-agents#hand-rolled-loop-with-primitives)
    ** — 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.

[​](https://trigger.dev/docs/ai-chat/custom-agents#managed-loop-chat-createsession)

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:

trigger/my-chat.ts

    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",
      run: async (payload: ChatTaskWirePayload, { signal }) => {
        // One-time initialization — plain code, no hooks. Upsert, not create:
        // continuation runs boot with the row already in place.
        const clientData = payload.metadata as { userId: string };
        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 },
          });
        }
      },
    });
    

If you pass `compaction` or `pendingMessages` to `chat.createSession()`, you must also pass `prepareStep: turn.prepareStep()` to `streamText` (or spread `chat.toStreamTextOptions()`, which wires it automatically). Without it, both features silently no-op.

### 

[​](https://trigger.dev/docs/ai-chat/custom-agents#chatsessionoptions)

ChatSessionOptions

| Option | Type | Default | Description |
| --- | --- | --- | --- |
| `signal` | `AbortSignal` | required | Run-level cancel signal (from task context) |
| `idleTimeoutInSeconds` | `number` | `30` | Seconds to stay idle between turns before suspending |
| `timeout` | `string` | `"1h"` | Duration string for suspend timeout |
| `maxTurns` | `number` | `100` | Max turns before ending |
| `compaction` | `ChatAgentCompactionOptions` | `undefined` | Automatic context [compaction](https://trigger.dev/docs/ai-chat/compaction)<br> — same options as on `chat.agent()` |
| `pendingMessages` | `PendingMessagesOptions` | `undefined` | Mid-execution [message injection](https://trigger.dev/docs/ai-chat/pending-messages)<br> — same options as on `chat.agent()` |

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.

### 

[​](https://trigger.dev/docs/ai-chat/custom-agents#chatturn)

ChatTurn

Each turn yielded by the iterator provides:

| Field | Type | Description |
| --- | --- | --- |
| `number` | `number` | Turn number (0-indexed) |
| `chatId` | `string` | Chat session ID |
| `trigger` | `string` | What triggered this turn |
| `clientData` | `unknown` | Client data from the transport |
| `messages` | `ModelMessage[]` | Full accumulated model messages — pass to `streamText` |
| `uiMessages` | `UIMessage[]` | Full accumulated UI messages — use for persistence |
| `signal` | `AbortSignal` | Combined stop+cancel signal (fresh each turn) |
| `stopped` | `boolean` | Whether the user stopped generation this turn |
| `continuation` | `boolean` | Whether this is a continuation run |
| `previousTurnUsage` | `LanguageModelUsage \| undefined` | Token usage from the previous turn (undefined on turn 0) |
| `totalUsage` | `LanguageModelUsage` | Cumulative token usage across all completed turns |
| `handover` | `{ isFinal: boolean } \| null` | The [`chat.headStart`](https://trigger.dev/docs/ai-chat/fast-starts#handover-with-custom-agents)<br> handover for this turn (turn 0 only); `null` otherwise |

| Method | Description |
| --- | --- |
| `turn.complete(source?)` | Pipe stream, capture response, accumulate, and signal turn-complete. Call with no source on a final head-start handover (`turn.handover.isFinal`), where the warm step-1 partial is already the response |
| `turn.done()` | Signal turn-complete only (when you have piped manually) |
| `turn.addResponse(response)` | Add a response to the accumulator manually |
| `turn.setMessages(uiMessages)` | Replace the accumulated messages — continuation seeding and on-demand compaction |
| `turn.prepareStep()` | `prepareStep` callback wiring compaction + injection — pass to `streamText` when not spreading `chat.toStreamTextOptions()` |

### 

[​](https://trigger.dev/docs/ai-chat/custom-agents#continuation-runs-and-history-seeding)

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.

### 

[​](https://trigger.dev/docs/ai-chat/custom-agents#turn-complete-vs-manual-control)

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 response = await chat.pipeAndCapture(result, { signal: turn.signal });
    
      if (response) {
        // Custom processing before accumulating
        await turn.addResponse(response);
      }
    
      // Custom persistence, analytics, etc.
      await db.chat.update({ ... });
    
      // Must call done() when not using complete()
      await turn.done();
    }
    

[​](https://trigger.dev/docs/ai-chat/custom-agents#stopping-generation)

Stopping generation
----------------------------------------------------------------------------------------------

The frontend stops a turn with [`transport.stopGeneration(chatId)`](https://trigger.dev/docs/ai-chat/frontend#stop-generation)
, 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. `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:

trigger/my-chat.ts

    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](https://trigger.dev/docs/ai-chat/custom-agents#hand-rolled-loop-checklist)
.

[​](https://trigger.dev/docs/ai-chat/custom-agents#hand-rolled-loop-with-primitives)

Hand-rolled loop with primitives
------------------------------------------------------------------------------------------------------------------------

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

| Primitive | Description |
| --- | --- |
| `chat.messages` | Input stream for incoming messages — use `.waitWithIdleTimeout()` to wait for the next turn |
| `chat.createStopSignal()` | Create a managed stop signal wired to the stop input stream |
| `chat.pipeAndCapture(result)` | Pipe a `StreamTextResult` to the chat stream and capture the response |
| `chat.writeTurnComplete()` | Signal the frontend that the current turn is complete |
| `chat.MessageAccumulator` | Accumulates conversation messages across turns |
| `chat.pipe(stream)` | Pipe a stream to the frontend (no response capture) |
| `chat.cleanupAbortedParts(msg)` | Clean up incomplete parts from a stopped response |

A complete loop:

trigger/my-chat-raw.ts

    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),
          });
    
          let response;
          try {
            response = await chat.pipeAndCapture(result, { signal: combinedSignal });
          } catch (error) {
            if (error instanceof Error && error.name === "AbortError") {
              if (runSignal.aborted) break;
              // Stop — fall through to accumulate partial
            } else {
              throw error;
            }
          }
    
          if (response) {
            const cleaned =
              stop.signal.aborted && !runSignal.aborted ? chat.cleanupAbortedParts(response) : response;
            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();
      },
    });
    

### 

[​](https://trigger.dev/docs/ai-chat/custom-agents#messageaccumulator)

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 response = await chat.pipeAndCapture(result);
    if (response) await conversation.addResponse(response);
    
    // 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](https://trigger.dev/docs/ai-chat/pending-messages#backend-messageaccumulator-raw-task)
 for the manual steering wiring.

### 

[​](https://trigger.dev/docs/ai-chat/custom-agents#hand-rolled-loop-checklist)

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:
    
        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.

[​](https://trigger.dev/docs/ai-chat/custom-agents#next-steps)

Next steps
----------------------------------------------------------------------------

Backend overview
----------------

The three abstraction levels compared, and everything chat.agent() adds on top.

Sessions
--------

The durable stream pair every agent — managed or custom — is built on.

Compaction
----------

Automatic context compression — works with createSession and MessageAccumulator.

Client protocol
---------------

The wire format your loop is speaking, chunk by chunk.

Was this page helpful?

YesNo

[Previous](https://trigger.dev/docs/ai-chat/types)
[How it worksEnd-to-end mechanics of a chat.agent turn: the two durable channels per session, the long-lived task that reads and writes them, and how a chat survives refreshes, deploys, and idle gaps.\
\
Next](https://trigger.dev/docs/ai-chat/how-it-works)

⌘I

Assistant

Responses are generated using AI and may contain mistakes.