Vercel WDK

The Vercel Workflow Development Kit (WDK) keeps your agent algorithm and processing safe by running it as a workflow inside your Next.js app. 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 WDK steps directly to subscribed clients.

Ably AI Transport support for Vercel WDK 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 WDK workflow in the same Next.js deployment. The workflow runs one step that calls the model and pipes the response to an Ably channel.
• The step adds headers to every message it pipes, so that a client can identify retried and duplicated output and prefer the later output.
• The first attempt of the step fails part way through the response, so you can watch WDK retry the step and the retry's response replace the partial one.
• A stop button cancels the response through WDK, and the step stops streaming.

Prerequisites

• Node.js 22 or later.
• An Ably account with an API key.
• An Anthropic API key, or any other model provider the Vercel AI SDK supports.

Install dependencies

Install the AI Transport SDK, the Ably client, the Vercel AI SDK, the workflow runtime, and Next.js, then add the scripts that run the app:

npm install @ably/ai-transport ably ai @ai-sdk/anthropic workflow next react react-dom
npm pkg set scripts.dev="next dev" scripts.build="next build"

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 workflow steps run on your own infrastructure, so they use 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.

Configure Next.js for workflows

Wrap your Next.js config with withWorkflow. It installs the bundler transforms for the 'use workflow' and 'use step' directives, and serves the workflow runtime's endpoints under /.well-known/workflow/v1/*.

// next.config.mjs
import { withWorkflow } from 'workflow/next';

export default withWorkflow({});

In local development the runtime stores run state in .workflow-data/. Add that directory to .gitignore.

Store the conversation

Create app/store.js. The agent route and the workflow steps both read the conversation, and WDK can run each step as its own invocation, so this stand-in for your database writes to a JSON file. Each conversation holds its messages and the serial of the last stored response.

// Stand-in for your own database. Replace both 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')) : {});

export function loadConversation(channelName) {
  return readAll()[channelName] ?? { messages: [] };
}

export function recordMessages(channelName, messages, serial) {
  const all = readAll();
  const current = all[channelName] ?? { messages: [] };
  const merged = [...current.messages];
  for (const message of messages) {
    const at = merged.findIndex((m) => m.id === message.id);
    if (at === -1) merged.push(message);
    else merged[at] = message;
  }
  all[channelName] = { messages: merged, serial: serial ?? current.serial };
  fs.mkdirSync('.data', { recursive: true });
  fs.writeFileSync(file, JSON.stringify(all));
}

recordMessages saves each message under its id, so a second write of the same response replaces the first.

Define the workflow

Create app/workflows/chat.js. A 'use workflow' function has to be deterministic, so it does not call the model or publish to the channel from the workflow definition.

WDK retries a step that throws, up to the step's maxRetries. The default is three retries, so a step runs at most four times. WDK queues each retry as soon as the previous attempt fails. The workflow's await on the step rejects only after the last retry fails. A step that throws a FatalError skips its remaining retries. The two example steps in this workflow first stream a response and second handle the error when the first step exhausts its retries:

// Workflow code. WDK replays it, so it does no I/O of its own.
import { createHook } from 'workflow';
import { abortResponse, streamResponse } from './steps';

export async function chatWorkflow({ channelName, requestId }) {
  'use workflow';
  const responseId = crypto.randomUUID();
  const controller = new AbortController();
  const cancelHook = createHook({ token: `cancel:${requestId}` });
  await cancelHook.getConflict();
  cancelHook.then(() => controller.abort());

  try {
    await streamResponse({ channelName, requestId, responseId }, controller.signal);
  } catch (error) {
    await abortResponse({ channelName, requestId });
    throw error;
  }
}

The workflow generates responseId and passes it to the step, so every attempt publishes its response under the same message id. This is useful when the message id is used to represent a single message or single element on the UI. Keeping the same message id allows any second retried response to more easily replace an earlier one.

Write the steps

Create app/workflows/steps.js. A WDK step is a function marked 'use step', and WDK retries each step on its own when it fails:

// Step code. It runs on your own infrastructure, so it holds the API key.
import * as Ably from 'ably';
import { getStepMetadata } from 'workflow';
import { anthropic } from '@ai-sdk/anthropic';
import { convertToModelMessages, streamText, toUIMessageStream } from 'ai';
import { createTransport } from '@ably/ai-transport';
import { vercel } from '@ably/ai-transport/vercel';
import { loadConversation, recordMessages } from '../store';

export async function streamResponse({ channelName, requestId, responseId }, signal) {
  'use step';
  const { stepId, attempt } = getStepMetadata();
  const headers = { requestId, activityId: stepId, attempt };
  const { messages } = loadConversation(channelName);

  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const transport = createTransport({ client: ably, channelName, codec: vercel });

  try {
    if (attempt > 1) await transport.send({ type: 'abort' }, { headers });

    const result = streamText({
      model: anthropic('claude-sonnet-4-20250514'),
      messages: await convertToModelMessages(messages),
      abortSignal: signal,
    });

    let response;
    const { serial } = await transport.pipe(
      failFirstAttempt(
        toUIMessageStream({
          stream: result.fullStream,
          originalMessages: messages,
          generateMessageId: () => responseId,
          onEnd: ({ responseMessage }) => {
            response = responseMessage;
          },
        }),
        attempt,
      ),
      { signal, headers },
    );

    if (response && serial) recordMessages(channelName, [response], serial);
  } catch (error) {
    // The workflow aborted the signal, so the model call or the pipe was cancelled.
    if (signal.aborted) {
      await transport.send({ type: 'abort' }, { headers });
      return;
    }
    throw error;
  } finally {
    await transport.close();
    ably.close();
  }
}

export async function abortResponse({ channelName, requestId }) {
  'use step';
  const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
  const transport = createTransport({ client: ably, channelName, codec: vercel });
  try {
    await transport.send({ type: 'abort' }, { headers: { requestId } });
  } finally {
    await transport.close();
    ably.close();
  }
}

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

Each attempt creates its own Ably Realtime client and its own transport, and closes both before it returns, because WDK can run each attempt as a separate invocation.

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

WDK keeps the step id the same across retries of one step, and unique across workflow runs, so the value is unique on the channel. A client can then identify retried and duplicated output and prefer the later output. The header keeps the name activityId, so the same client code works for any workflow engine.

When the step runs with an attempt number above one, it sends an abort event with its own headers before it pipes, to mark the previous attempt's output on the channel. In this example, abort is not part of the AI Transport specification. It is any type of event that you define that can be sent to a client. You should structure this event with content and headers that help your UI clients update their own state according to the retry or failure.

When the pipe resolves, the step 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.

Cancel a response

WDK has two ways to cancel. getRun(runId).cancel() from workflow/api cancels the whole run, and WDK stops the workflow at its next step call. A step that is already running streams to the end of its response. An AbortSignal that the workflow passes to a step stops the step while it runs, because the step passes the signal to the model call and to pipe(). The workflow passes the signal as its own argument. WDK serializes an object that holds a signal as an AbortController, so the object's other fields would not reach the step.

This app uses the signal. The workflow creates an AbortController and a hook with the token cancel:${requestId}, and aborts the controller when the hook resumes. createHook() does not register the hook on its own, so the workflow awaits cancelHook.getConflict(), which registers it before the step starts. Your endpoint resumes the hook with resumeHook() from workflow/api. The agent route below adds that endpoint.

When the signal fires, pipe() rejects and leaves whole messages on the channel. The step checks the signal, and sends an abort event with its headers, so every subscribed client sees the response end. The step returns without throwing, so WDK does not retry it.

A client can also cancel through the channel, with a subscriber on your own infrastructure that resumes the hook.

Create the agent route

Create app/api/chat/route.js:

// Agent code. This runs on your own infrastructure.
import { resumeHook, start } from 'workflow/api';
import { HookNotFoundError } from 'workflow/errors';
import { chatWorkflow } from '../../workflows/chat';
import { loadConversation, recordMessages } from '../../store';

export async function POST(req) {
  const { channelName, message } = await req.json();
  recordMessages(channelName, [message]);
  await start(chatWorkflow, [{ channelName, requestId: message.id }]);
  return new Response(null, { status: 202 });
}

export async function DELETE(req) {
  const { requestId } = await req.json();
  try {
    await resumeHook(`cancel:${requestId}`, {});
  } catch (error) {
    // The run has not registered its hook yet, or it has already ended.
    if (HookNotFoundError.is(error)) return new Response(null, { status: 409 });
    throw error;
  }
  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 message, starts a workflow, and returns as soon as WDK has started the workflow. The DELETE handler resumes the workflow's cancel hook, and the workflow then aborts the signal that the step is using. The hook's token holds the requestId, so the handler can find the hook from the request.

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

Create the chat component

Create app/chat.js, app/merge.js, app/page.js and app/layout.js as in the Vercel AI SDK quickstart. Then, in app/chat.js, replace apply with this version, 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 { requestId, activityId, attempt } = message.extras?.headers ?? {};
    if (activityId !== undefined) {
      if (attempt < (attempts.current.get(activityId) ?? 0)) return;
      attempts.current.set(activityId, attempt);
    }
    if (event !== undefined) merge(event, requestId);
  },
  [merge],
);

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

In send, set lastRequest.current = message.id after the message is created, and add <button type="button" onClick={stop}>Stop</button> to the form.

apply keeps the newest attempt number it has seen for each step, and drops messages from older attempts. An older attempt's invocation can keep running after WDK starts a retry, so its messages can arrive after the retry's messages.

In the example, every attempt uses the same responseId, so a new or retried response can more easily replace the existing UI elements. When the retry's start event arrives, the merge puts an empty response under that message id, and upsert replaces the failed attempt's partial response with it. The retry's output then fills the same response.

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

One process runs everything. Next.js serves the app, and the workflow runtime runs workflows and steps inside it:

npm run dev

To watch workflows run, open the inspector in a second terminal:

npx workflow web

Open the app at http://localhost:3000 in two tabs and send a message from one. The first attempt streams part of the response and fails, WDK retries the step, and the retry's response replaces the partial one in both tabs. Each message you send appears in the inspector as a new workflow run, with the failed attempt on its step.

What happens when you send a message

1. The client adds the user's message to its list, and POSTs the message to /api/chat.
2. The route reads the message from the request body, records it, and starts a WDK workflow.
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, WDK retries the step. The retry sends an abort event, then pipes the response again under the same responseId. Every subscribed client drops messages from the older attempt, and replaces the partial response with the retry's.
5. When the pipe resolves, the step stores the response with the serial of its last message.
6. If every attempt fails, the workflow runs abortResponse. If the user presses stop, the step sends the abort event itself. 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 a step invocation stops after recordMessages() and before WDK records the step as complete, the retry streams and stores the response again. recordMessages() saves the response under its responseId, so the second write replaces the first.
• A cancel that arrives before the workflow registers its hook, or after the run ends, makes resumeHook() throw HookNotFoundError, and the route returns 409. The client can try the cancel again while the response is still streaming.
• A header value that is not a string, number, boolean or null makes pipe() reject with InvalidArgument before it reads the stream.

Read next

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