Temporal

Temporal keeps your agent algorithm and processing safe by running it as a workflow. Ably AI Transport helps that agent communicate in realtime with connected clients. You don't have to write intermediate AI state to your database just so it can reach a client. With AI Transport you can send that output from your Temporal activities directly to subscribed clients.

Ably AI Transport support for Temporal is built on the features that Streaming gives you, and uses the pattern that Durable execution describes.

What you build

A Next.js chat app where:

• Each user message starts a Temporal workflow. The workflow runs one activity that calls the model and pipes the response to an Ably channel.
• The activity adds headers to every message it publishes to the channel, so that a client can identify retried and duplicated output and prefer the later output.
• The first attempt of the activity fails part way through the response, so you can watch Temporal retry the activity and the retry's response replace the partial one.
• A stop button cancels the workflow through Temporal, and the activity stops streaming.

Prerequisites

• Node.js 22 or later.
• Working knowledge of Temporal workflows and activities. If Temporal is new to you, start with Understanding Temporal.
• The Temporal CLI installed locally (brew install temporal on macOS).
• An Ably account with an API key.
• An OpenAI API key, if you follow the OpenAI version of this guide.
• An Anthropic API key, or any other model provider the Vercel AI SDK supports, if you follow the Vercel version.

Install dependencies

Install the AI Transport SDK, the Ably client, the OpenAI SDK, the Temporal packages, and Next.js:

npm install @ably/ai-transport ably openai \
  @temporalio/client @temporalio/worker @temporalio/workflow @temporalio/activity \
  next react react-dom dotenv
npm install -D tsx

The Temporal worker runs as a separate Node process from the Next.js app. Add a script for it in package.json:

{
  "scripts": {
    "dev": "next dev",
    "worker": "tsx workflow/worker.js"
  }
}

Next.js needs to load the Temporal client from node_modules instead of bundling it. In next.config.mjs:

/** @type {import('next').NextConfig} */
export default {
  serverExternalPackages: ['@temporalio/client'],
};

Add your keys to .env.local:

# Ably API key in "keyName:keySecret" form, from your Ably dashboard.
ABLY_API_KEY=
# OpenAI API key used by the worker's activity.
OPENAI_API_KEY=

Set up authentication

Create an auth endpoint at /api/auth/token that returns an Ably token to the browser, with the client id and the channel capabilities the user needs, as described in Set up authentication. The client needs the publish, subscribe and history capabilities on the conversation channels.

The client fetches its token with authUrl: '/api/auth/token'. The worker runs on your own infrastructure, so it uses the API key directly.

Configure the channel rule

AI Transport streams a response's text by appending each delta to a channel message. That needs the Message annotations, updates, deletes, and appends rule (mutableMessages: true in the Control API) on the channels your conversations use. Enable it once per Ably app through the dashboard, Control API, or CLI.

Store the conversation

Create app/store.js. The Next.js app and the Temporal worker both read the conversation, and they run as separate processes, so this stand-in for your database writes to a JSON file. The Responses API takes its history as an input array of items, so the store keeps the items of each turn under the turn's requestId, with the serial of the last stored response.

// Stand-in for your own database. Replace these functions with real queries.
import fs from 'node:fs';

const file = '.data/conversations.json';
const readAll = () => (fs.existsSync(file) ? JSON.parse(fs.readFileSync(file, 'utf8')) : {});
const writeAll = (all) => {
  fs.mkdirSync('.data', { recursive: true });
  fs.writeFileSync(file, JSON.stringify(all));
};

export function loadConversation(channelName) {
  const { turns = [], serial } = readAll()[channelName] ?? {};
  return { items: turns.flatMap((turn) => [...turn.input, ...turn.output]), serial };
}

export function recordInput(channelName, requestId, input) {
  const all = readAll();
  const conversation = all[channelName] ?? { turns: [] };
  conversation.turns.push({ requestId, input, output: [] });
  all[channelName] = conversation;
  writeAll(all);
}

export function recordOutput(channelName, requestId, output, serial) {
  const all = readAll();
  const conversation = all[channelName];
  const turn = conversation?.turns.find((t) => t.requestId === requestId);
  if (!turn) return;
  turn.output = output;
  conversation.serial = serial;
  writeAll(all);
}

recordOutput replaces the output of the turn, so a second write of the same response replaces the first.

Define the workflow

Create workflow/workflows.js. A Temporal workflow has to be deterministic, so it does not call the model or publish to the channel from the workflow definition. A Temporal activity is configured with a number of allowed retries. That activity will automatically retry until the configured number is exhausted, only then will it return an error to the workflow. The two example activities in this workflow first stream a response and second handle the error when the first activity exhausts its retries or is cancelled by Temporal directly:

// Workflow code. Temporal replays it, so it does no I/O of its own.
import { ActivityCancellationType, CancellationScope, proxyActivities } from '@temporalio/workflow';

const { streamResponse, abortResponse } = proxyActivities({
  startToCloseTimeout: '5 minutes',
  heartbeatTimeout: '10 seconds',
  retry: { maximumAttempts: 3 },
  cancellationType: ActivityCancellationType.WAIT_CANCELLATION_COMPLETED,
});

export async function chatWorkflow({ channelName, requestId }) {
  try {
    await streamResponse({ channelName, requestId });
  } catch (error) {
    await CancellationScope.nonCancellable(() => abortResponse({ channelName, requestId }));
    throw error;
  }
}

OpenAI gives each output item its own item_id, so a retry publishes its response under new ids. The client uses the headers to remove the output items of an attempt that a retry has replaced.

With heartbeatTimeout, Temporal retries the activity when ten seconds pass without a heartbeat from it, for example after the worker crashes.

With cancellationType set to WAIT_CANCELLATION_COMPLETED, the workflow waits for a cancelled activity to stop piping before its catch block runs. The event that ends the response then arrives after the activity's last messages. Without the option, the activity is cancelled as TRY_CANCEL, and the catch block can run first. The workflow runs abortResponse inside CancellationScope.nonCancellable, so Temporal still runs the activity after it cancels the workflow.

Write the activities

Create workflow/activities.js:

// Activity code. It runs on the worker, so it holds the API key.
import * as Ably from 'ably';
import OpenAI from 'openai';
import { Context } from '@temporalio/activity';
import { createTransport } from '@ably/ai-transport';
import { openai } from '@ably/ai-transport/openai';
import { loadConversation, recordOutput } from '../app/store';

const client = new OpenAI();

export async function streamResponse({ channelName, requestId }) {
  const context = Context.current();
  const { activityId, attempt, workflowExecution } = context.info;
  const headers = { requestId, activityId: `${workflowExecution.workflowId}/${activityId}`, attempt };
  const { items } = loadConversation(channelName);

  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const transport = createTransport({ client: ably, channelName, codec: openai });
  const heartbeat = setInterval(() => context.heartbeat(), 1000);

  try {
    if (attempt > 1) {
      await transport.send(
        { type: 'error', code: 'retried', message: 'A retry replaced this attempt.', param: null, sequence_number: 0 },
        { headers },
      );
    }

    const stream = await client.responses.create(
      { model: 'gpt-5.5', input: items, stream: true },
      { signal: context.cancellationSignal },
    );

    let output = [];
    const { serial } = await transport.pipe(
      failFirstAttempt(keepOutput(stream, (finished) => (output = finished)), attempt),
      { signal: context.cancellationSignal, headers },
    );

    if (serial) recordOutput(channelName, requestId, output, serial);
  } catch (error) {
    // Temporal cancelled the activity, so report the cancel rather than the error it caused.
    if (context.cancellationSignal.aborted) throw context.cancellationSignal.reason;
    throw error;
  } finally {
    clearInterval(heartbeat);
    await transport.close();
    ably.close();
  }
}

export async function abortResponse({ channelName, requestId }) {
  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const transport = createTransport({ client: ably, channelName, codec: openai });
  try {
    await transport.send(
      { type: 'error', code: 'failed', message: 'The response failed.', param: null, sequence_number: 0 },
      { headers: { requestId } },
    );
  } finally {
    await transport.close();
    ably.close();
  }
}

// Pass each event to the pipe, and keep the finished output items for your store.
async function* keepOutput(stream, onOutput) {
  for await (const event of stream) {
    if (event.type === 'response.completed') onOutput(event.response.output);
    yield event;
  }
}

// Throws part way through the first attempt's response, so you can watch Temporal retry it.
async function* failFirstAttempt(stream, attempt) {
  let events = 0;
  for await (const event of stream) {
    if (attempt === 1 && ++events > 20) throw new Error('simulated crash part way through the response');
    yield event;
  }
}

Each attempt creates its own Ably Realtime client and its own transport, and closes both before it returns, because Temporal can run each attempt on a different worker.

The headers go on every message the activity pipes, including each append:

A Temporal activity id is unique inside one workflow execution only, so the activity uses a composite of the workflow id and activity id. The composite keeps the value unique on the channel, so that a client can identify retried and duplicated output and prefer the later output.

When the activity runs with an attempt number above one, it sends an error event with its own headers before it pipes, to mark the previous attempt's output on the channel. In this example, error is the Responses API's ResponseErrorEvent, which the OpenAI codec publishes as it is. It is not part of the AI Transport specification. You can send any event the codec accepts, and you should structure it with content and headers that help your UI clients update their own state according to the retry or failure.

The activity heartbeats every second, and passes context.cancellationSignal to the model call as abortSignal and to pipe() as signal. A Temporal activity learns that Temporal cancelled it only when a heartbeat reaches Temporal. The worker throttles heartbeats to 80% of heartbeatTimeout, which is eight seconds here, so the worker below sets maxHeartbeatThrottleInterval to one second. When the signal fires, pipe() rejects and leaves whole messages on the channel. Once the signal has fired, the activity throws context.cancellationSignal.reason, which is Temporal's CancelledFailure, so Temporal records the activity as cancelled. Temporal records any other error from the activity as a failure.

When the pipe resolves, the activity stores the response with the serial of its last message, so a client that loads the conversation later can replay the channel from it. failFirstAttempt is there for the demo, and you can remove it in your own app.

Host the worker

Create workflow/worker.js. The worker hosts the workflow and the activities:

import path from 'node:path';
import { config as loadDotenv } from 'dotenv';
import { NativeConnection, Worker } from '@temporalio/worker';
import * as activities from './activities';

// tsx does not load .env.local the way next dev does.
loadDotenv({ path: path.resolve(__dirname, '../.env.local') });

async function main() {
  const connection = await NativeConnection.connect({ address: 'localhost:7233' });
  const worker = await Worker.create({
    connection,
    taskQueue: 'ai-transport-demo',
    workflowsPath: require.resolve('./workflows'),
    activities,
    // Send each activity heartbeat to Temporal within a second, so a cancel reaches the activity quickly.
    maxHeartbeatThrottleInterval: '1 second',
  });
  await worker.run();
}

main().catch((error) => {
  console.error(error);
  process.exit(1);
});

Create the agent route

Create app/api/chat/route.js:

// Agent code. This runs on your own infrastructure.
import { Client, Connection } from '@temporalio/client';
import { loadConversation, recordInput } from '../../store';

let temporal;
async function temporalClient() {
  temporal ??= new Client({ connection: await Connection.connect({ address: 'localhost:7233' }) });
  return temporal;
}

export async function POST(req) {
  const { channelName, items, requestId } = await req.json();
  recordInput(channelName, requestId, items);
  const client = await temporalClient();
  await client.workflow.start('chatWorkflow', {
    workflowId: requestId,
    taskQueue: 'ai-transport-demo',
    args: [{ channelName, requestId }],
  });
  return new Response(null, { status: 202 });
}

export async function DELETE(req) {
  const { requestId } = await req.json();
  const client = await temporalClient();
  await client.workflow.getHandle(requestId).cancel();
  return new Response(null, { status: 204 });
}

export function GET(req) {
  const channelName = new URL(req.url).searchParams.get('channel');
  return Response.json(loadConversation(channelName));
}

The POST handler records the user's input items under the requestId, starts a workflow with the requestId as its workflow id, and returns as soon as Temporal has started the workflow.

The DELETE handler cancels the workflow with Temporal's cancel feature. The workflow id is the requestId, so the handler can find the workflow from the request.

The GET handler returns the stored conversation and its serial to the page.

This app cancels through the workflow engine. A client can also cancel through the channel, with a subscriber on your own infrastructure that cancels the workflow.

Create the chat component

Create app/chat.js, app/merge.js, app/page.js and app/layout.js as in the OpenAI quickstart. Then, in app/chat.js, replace apply and send with these versions, and add attempts, lastRequest and stop:

// Client code, in app/chat.js.
const attempts = useRef(new Map());
const lastRequest = useRef(undefined);

const apply = useCallback(({ event, message }) => {
  if (lastSeen.current === undefined || message.serial > lastSeen.current) lastSeen.current = message.serial;
  const { activityId, attempt } = message.extras?.headers ?? {};
  if (activityId !== undefined) {
    const seen = attempts.current.get(activityId);
    if (seen && attempt < seen.attempt) return;
    if (!seen || attempt > seen.attempt) {
      // A newer attempt: remove the output items the older attempt rendered.
      const replaced = seen?.itemIds ?? new Set();
      setEntries((prev) => prev.filter((entry) => !replaced.has(entry.id)));
      attempts.current.set(activityId, { attempt, itemIds: new Set() });
    }
    if (event?.type === 'response.output_text.delta') attempts.current.get(activityId).itemIds.add(event.item_id);
  }
  if (event !== undefined) setEntries((prev) => mergeEvent(prev, event));
}, []);

const send = async (e) => {
  e.preventDefault();
  const text = input.trim();
  if (!text || !transport) return;
  setInput('');
  const items = [{ role: 'user', content: text }];
  const requestId = crypto.randomUUID();
  lastRequest.current = requestId;
  setEntries((prev) => [...prev, ...toEntries(items)]);
  await fetch('/api/chat', {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({ channelName, items, requestId }),
  });
};

const stop = () =>
  fetch('/api/chat', {
    method: 'DELETE',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({ requestId: lastRequest.current }),
  });

Add <button type="button" onClick={stop}>Stop</button> to the form.

apply keeps the newest attempt number it has seen for each activity, and drops messages from older attempts. An older attempt's process can keep running after Temporal starts a retry, for example when the attempt times out, so its messages can arrive after the retry's messages.

apply also keeps the output item ids that each attempt rendered. When a newer attempt of the same activity arrives, it removes the older attempt's items, so the user sees one response, with the retry's output. send posts a requestId with the input, and the agent route uses it as the workflow id.

If the subscription replays history, the live subscription handler receives the replayed messages and the live messages in the order they appear on the channel; this allows the same prefer-later-retries logic to be used for history and for live subscription to ongoing messages.

Run the app

Open three terminals. --db-filename persists Temporal state across restarts, so a workflow you inspect in the Web UI is still there after a reboot.

# Terminal 1: Temporal dev server
temporal server start-dev --db-filename ai-transport-demo.db

# Terminal 2: Temporal worker
npm run worker

# Terminal 3: Next.js
npm run dev

Open the app at http://localhost:3000 in two tabs, and the Temporal Web UI at http://localhost:8233. Send a message from one tab. The first attempt streams part of the response and fails, Temporal retries the activity, and the retry's response replaces the partial one in both tabs. Each message you send appears in the Web UI as a new workflow, with the failed attempt in its history.

What happens when you send a message

1. The client adds the user's input to its entries, and POSTs the input to /api/chat with a requestId.
2. The route reads the input from the request body, records it, and starts a workflow with the requestId as its workflow id.
3. The workflow runs streamResponse, which calls the model and pipes the output to the channel with the requestId, activityId and attempt headers.
4. If the attempt fails, Temporal retries the activity. The retry sends an event that ends the previous attempt's response, then pipes the response again. Every subscribed client drops messages from the older attempt, and replaces the partial response with the retry's.
5. When the pipe resolves, the activity stores the response with the serial of its last message.
6. If every attempt fails, or the user presses stop, the workflow runs abortResponse, and every subscribed client sees the response end.

Edge cases

• When an attempt fails before it publishes a message, clients see only the retry's response.
• If the worker stops after the activity stores the response and before Temporal records the activity as complete, the retry streams and stores the response again. The store saves the response under a key every attempt shares, the responseId or the turn's requestId, so the second write replaces the first.
• A cancel that arrives before the activity starts stops the workflow, and the event that ends the response reaches clients that have not received any of the response. The merge has nothing open for that response, so it ignores the event.
• A header value that is not a string, number, boolean or null makes pipe() reject with InvalidArgument before it reads the stream.

FAQ

Why not send the output out of the activity directly?

An HTTP request starts the workflow, and Temporal schedules its activities on a worker, which is usually a different process from the one that received the request. That process does not have a route back to the browser. Publishing to a channel gives every subscribed client the output, whichever process produced it.

What happens if no client is subscribed while the agent runs?

The workflow completes anyway. The activity stores the response, and a client that loads the conversation later reads it from your store, then replays anything newer from the channel.

Does the browser talk to Temporal?

No, the browser publishes to the transport and calls your agent route, and the route starts and cancels the workflow.

Read next

• Durable execution: the headers, retries, failure marking and cancellation for any workflow engine.
• Vercel WDK: the same app on Vercel's workflow runtime.
• History and replay: replay the channel from your own store.
• Transport API reference: every method on the transport.