Get started with custom events

What you build

A Next.js chat app for your custom event structure, where:

• Your custom events stream from the agent to every subscribed client in realtime.
• The prompt appears in every subscribed client, because the agent publishes it to the Ably channel.
• Reloading a tab while a response streams shows the stored conversation, then the response so far, then the rest of it live.
• A Stop button ends the response in every subscribed client.

The app builds the conversation, which consists of the prompts and responses in order. The same channel can also carry awareness, such as what the agent is doing now, and artifacts, the things the agent makes.

The custom codec is the only part of the app specific to your events. The agent passes it to createAgentAdaptor, and the client passes it to createDecoder. Your database stores the conversation, and you build the UI. The chat component in this guide is one example.

Prerequisites

• Node.js 22 or later.
• An Ably account with an API key.
• The agent streams from fakeModel, a stand-in that you replace with your model provider's stream.

Install dependencies

Install the AI Transport SDK, the Ably Pub/Sub server and device SDKs and Next.js, then add the scripts that run the app:

npm install @ably/ai-transport @ably/pubsub-server @ably/pubsub-device next react react-dom
npm pkg set scripts.dev="next dev" scripts.build="next build"

This guide uses Next.js as an example, and AI Transport does not depend on it.

Set up authentication

The browser fetches a short-lived Ably token from your server, so it never holds your API key. Create app/auth/route.js. This server code signs a token request for a clientId and the capabilities the client needs on the conversation channels:

// This runs on your server, so it holds the API key.
import { createHttpClient } from '@ably/pubsub-server';

export async function GET() {
  const ably = createHttpClient({ key: process.env.ABLY_API_KEY });
  // Replace the random id with the user your own auth identifies.
  const tokenRequest = await ably.auth.createTokenRequest({
    clientId: `user-${crypto.randomUUID().slice(0, 8)}`,
    capability: { 'conversations:*': ['subscribe', 'history'] },
  });
  return Response.json(tokenRequest);
}

The client does not need the publish capability, because the agent publishes the prompt. The agent uses your API key, and you set the key's capabilities in the Ably dashboard. The key needs publish to publish a response, and message-update-any to append to it. You can scope the token to the channels each user may access.

Configure the channel rule

AI Transport streams a response's text by appending each delta to a channel message. Appending needs the Message annotations, updates, deletes, and appends channel rule on the channels your conversations use. You enable it once per Ably app, through the dashboard, the Control API, or the CLI. The rule also persists the channel's messages, so a tab that reloads can receive them with a rewind.

Define your events

Create app/codec.js. The events in this guide are:

• prompt, which carries the user's message.
• text-start, text-delta and text-end, which stream the response's text, each with the response's id.
• done, which ends the response, with the response's id and a reason.

defineCodec builds a custom codec from a mapping of event names to the encode and decode functions that convert those events to and from Ably messages.

In the custom codec code below, the prompt, text-start, text-end, and done events publish as single discrete Ably messages. The text-delta events append the data of those events together, which creates a live stream of deltas for live subscribed clients, but a single message in history containing the full text-delta content. The response's events carry its id in fields, because an append adds only to data.

The agent and the client share this code:

import { defineCodec } from '@ably/ai-transport';

export const chatCodec = defineCodec({
  id: 'my-chat',
  typeOf: (event) => event.type,
  events: {
    prompt: {
      encode: (event) => ({ data: event.message }),
      decode: ({ data }) => ({ type: 'prompt', message: data }),
    },
    'text-start': {
      encode: (event) => ({ fields: { id: event.id } }),
      decode: ({ fields }) => ({ type: 'text-start', id: String(fields.id) }),
    },
    'text-delta': {
      encode: (event) => ({ data: event.delta, fields: { id: event.id }, append: event.id }),
      decode: ({ data, fields }) => ({ type: 'text-delta', id: String(fields.id), delta: String(data) }),
    },
    'text-end': {
      encode: (event) => ({ fields: { id: event.id }, ends: event.id }),
      decode: ({ fields }) => ({ type: 'text-end', id: String(fields.id) }),
    },
    done: {
      encode: (event) => ({ fields: { id: event.id, reason: event.reason } }),
      decode: ({ fields }) => ({ type: 'done', id: String(fields.id), reason: String(fields.reason) }),
    },
  },
});

A subscribed client receives each text-delta event, live. A client that joins while a response streams, or that joins late and reads history, receives a single text-delta event with the response's text so far, and then the rest of the text-delta events live.

Create the agent's HTTP endpoint

Create app/api/chat/route.js. You are responsible for authenticating the request with your application's own auth before the agent publishes anything. The agent creates a realtime client with your API key, gets the conversation's channel, and passes it to createAgentAdaptor with your custom codec.

The agent publishes the prompt to the channel with send, as a prompt event, so every subscribed client shows it. Each message on the channel has a serial, which is its position on the channel, and a client can rewind the channel to a serial to receive every message after it. send resolves with the serial of the published message, and the agent stores the prompt with that serial. A tab that loads while the response streams rewinds the channel to just after the prompt to allow that late joining client to merge what's in the database with what is streaming live on the channel.

fakeModel is the stand-in for your own LLM provider. It yields the response as text-start, a text-delta for each word, text-end and done. The agent pipes those events to the channel with pipe, through keepText, which collects the response's text for your database as the events pass to the pipe. The requestId header holds the prompt's message id, and the agent sets it on the prompt and on every message of the response, so a client can match each response to its prompt. When the pipe resolves, the agent stores the response with the serial of the pipe's last message.

If the browser aborts its request, req.signal cancels the pipe, and the pipe rejects with OperationCancelled. A cancelled or failed pipe stops reading the source, so the text-end and done events never reach the channel. The agent sends its own done event, with the reason cancelled or error, to tell every subscribed client the response has ended. With a real provider, you must also pass the same signal to the model call, so the call stops with the pipe.

This agent code creates one realtime client per request, and closes it when the request ends:

// This runs on your server, so it holds the API key.
import { createAgentAdaptor, ErrorCode } from '@ably/ai-transport';
import { createRealtimeClient, ErrorInfo } from '@ably/pubsub-server';
import { chatCodec } from '../../codec';
import { recordMessages } from '../../store';

async function* fakeModel(responseId, prompt) {
  yield { type: 'text-start', id: responseId };
  for (const word of `You said: ${prompt}`.split(' ')) {
    await new Promise((resolve) => setTimeout(resolve, 100));
    yield { type: 'text-delta', id: responseId, delta: `${word} ` };
  }
  yield { type: 'text-end', id: responseId };
  yield { type: 'done', id: responseId, reason: 'complete' };
}

async function* keepText(events, onText) {
  let text = '';
  for await (const event of events) {
    if (event.type === 'text-delta') text += event.delta;
    yield event;
  }
  onText(text);
}

export async function POST(req) {
  // Check the user with your own auth here, before the agent publishes anything.
  const { channelName, message } = await req.json();
  const headers = { requestId: message.id };
  const responseId = crypto.randomUUID();

  const ably = createRealtimeClient({ key: process.env.ABLY_API_KEY });
  const adaptor = createAgentAdaptor(ably.channels.get(channelName), chatCodec);

  try {
    const { serial: promptSerial } = await adaptor.send({ type: 'prompt', message }, { headers });
    recordMessages(channelName, [message], promptSerial);

    let text = '';
    try {
      const { serial } = await adaptor.pipe(
        keepText(fakeModel(responseId, message.text), (collected) => {
          text = collected;
        }),
        { signal: req.signal, headers },
      );
      recordMessages(channelName, [{ id: responseId, role: 'assistant', text }], serial);
      return new Response(null, { status: 204 });
    } catch (error) {
      const cancelled = error instanceof ErrorInfo && error.code === ErrorCode.OperationCancelled;
      // Best effort: publishing the done event can fail for the same reason the pipe did.
      try {
        await adaptor.send({ type: 'done', id: responseId, reason: cancelled ? 'cancelled' : 'error' }, { headers });
      } catch (sendError) {
        console.error('done publish failed', sendError);
      }
      if (cancelled) return new Response(null, { status: 204 });
      console.error('pipe failed', error);
      return new Response(null, { status: 500 });
    }
  } finally {
    ably.close();
  }
}

Create the chat component

Create app/chat.js. The component shows the stored conversation, then subscribes to the channel through a decoder. The page mounts the channel with a rewind to the stored serial, so when the channel attaches, Ably delivers every message after that serial, then live messages, in channel order.

Each time the effect runs, it creates a decoder with createDecoder(chatCodec). You must not reuse a decoder that has already seen the channel's messages, so a remount in React Strict Mode or after Fast Refresh gets a new decoder. The effect subscribes the decoder with channel.subscribe(decoder.receive) and then calls channel.attach(), because subscribing does not attach the channel under ChannelProvider. The effect's cleanup function unsubscribes the decoder.

The decoder passes each event to apply, which passes it to mergeEvent. A message on the channel that your custom codec did not encode arrives with event set to undefined, and apply ignores it.

To send, the component adds the prompt to its list, then POSTs it to the agent's HTTP endpoint. The prompt this client sent comes back from the channel as a prompt event with the same id, and replaces the copy the component added. The Stop button aborts the POST request, which cancels the agent's pipe.

The highlighted lines listen for the channel's state changes. The listener treats a change as a loss of continuity when the channel goes DETACHED or FAILED, or reaches ATTACHED or gets an UPDATE with resumed: false, and then calls onDiscontinuity. The listener ignores the first ATTACHED change, because the component's own channel.attach() call on mount causes that change. The page passes in reinit as onDiscontinuity. This client code is the whole component:

1'use client';2 3// The browser fetches a token, so it never holds an API key.4import { useCallback, useEffect, useRef, useState } from 'react';5import { createDecoder } from '@ably/ai-transport';6import { useChannelStateListener, usePubSubClient } from '@ably/pubsub-device/react';7import { chatCodec } from './codec';8import { mergeEvent } from './merge';9 10export function Chat({ channelName, stored, onDiscontinuity }) {11  const [messages, setMessages] = useState(stored);12  const [input, setInput] = useState('');13  // The AbortController of this tab's request while the agent works on it.14  const [request, setRequest] = useState();15  const client = usePubSubClient();16  const channel = client.channels.get(channelName);17 18  const hasAttached = useRef(channel.state === 'attached');19  useChannelStateListener(channelName, (change) => {20    if (change.current === 'attached' && !hasAttached.current) {21      hasAttached.current = true;22      return;23    }24    if (!hasAttached.current) return;25    const lost =26      change.current === 'detached' || change.current === 'failed' || (change.current === 'attached' && !change.resumed);27    if (lost) onDiscontinuity();28  });29 30  const apply = useCallback(({ event }) => {31    if (event === undefined) return;32    setMessages((prev) => mergeEvent(prev, event));33  }, []);34 35  useEffect(() => {36    const decoder = createDecoder(chatCodec);37    decoder.onEvent(apply);38    decoder.onError((error) => console.error('decode failed', error));39    const start = async () => {40      await channel.subscribe(decoder.receive);41      await channel.attach();42    };43    start().catch((error) => console.error('attach failed', error));44    return () => {45      channel.unsubscribe(decoder.receive);46    };47  }, [channel, apply]);48 49  const send = async (e) => {50    e.preventDefault();51    const text = input.trim();52    if (!text) return;53    setInput('');54    const message = { id: crypto.randomUUID(), role: 'user', text };55    setMessages((prev) => mergeEvent(prev, { type: 'prompt', message }));56    const controller = new AbortController();57    setRequest(controller);58    try {59      await fetch('/api/chat', {60        method: 'POST',61        headers: { 'content-type': 'application/json' },62        body: JSON.stringify({ channelName, message }),63        signal: controller.signal,64      });65    } catch (error) {66      if (error.name !== 'AbortError') console.error('agent request failed', error);67    } finally {68      setRequest((current) => (current === controller ? undefined : current));69    }70  };71 72  return (73    <div>74      {messages.map((m) => (75        <p key={m.id}>76          <strong>{m.role}:</strong> {m.text}77          {m.reason === 'cancelled' ? ' (stopped)' : ''}78        </p>79      ))}80      <form onSubmit={send}>81        <input value={input} onChange={(e) => setInput(e.target.value)} />82        <button type="submit">Send</button>83        <button type="button" onClick={() => request?.abort()} disabled={!request}>84          Stop85        </button>86      </form>87    </div>88  );89}

Merge responses

Create app/merge.js. Each client merges the events it receives into the messages it shows. mergeEvent handles each event type:

• A prompt event puts the user's message in the list by its id, so it replaces the copy the sending client added, and every other subscribed client shows the prompt once.
• A text-start event adds an empty response, unless the list already holds a message with its id.
• A text-delta event adds its text to the response with the same id.
• A done event marks the response with its reason. The component shows a stopped response with a label.

Ably holds a response's appends for a short rollup window before it delivers them, and a new message does not wait for that window. A done event can arrive before the last deltas of a stopped response, so mergeEvent marks the response and keeps adding the deltas that follow. This client code is mergeEvent:

export function mergeEvent(messages, event) {
  const has = (id) => messages.some((m) => m.id === id);
  switch (event.type) {
    case 'prompt':
      return has(event.message.id)
        ? messages.map((m) => (m.id === event.message.id ? event.message : m))
        : [...messages, event.message];
    case 'text-start':
      return has(event.id) ? messages : [...messages, { id: event.id, role: 'assistant', text: '' }];
    case 'text-delta':
      return messages.map((m) => (m.id === event.id ? { ...m, text: m.text + event.delta } : m));
    case 'done':
      return messages.map((m) => (m.id === event.id ? { ...m, reason: event.reason } : m));
    default:
      return messages;
  }
}

Store, connect and run

Serve the stored conversation

A tab that loads reads the stored conversation, then rewinds the channel to the stored serial for everything after it. Create app/api/messages/route.js. This server code returns the stored messages and their serial:

import { loadConversation } from '../../store';

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

Create app/store.js. This Map stands in for the database your application already uses. Each conversation holds its messages and the latest serial the database accounts for: the last prompt's serial, then its response's serial once the response is stored.

// Server code. Replace both functions with queries to your database.
// The Map is held on globalThis, so it outlives `next dev` reloading the route modules.
const conversations = (globalThis.conversations ??= new Map());

export function loadConversation(channelName) {
  return conversations.get(channelName) ?? { messages: [] };
}

export function recordMessages(channelName, messages, serial) {
  const current = loadConversation(channelName);
  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;
  }
  // The stored serial only moves forward. Serials sort as strings.
  const latest = serial !== undefined && (current.serial === undefined || serial > current.serial) ? serial : current.serial;
  conversations.set(channelName, { messages: merged, serial: latest });
}

Connect the page

Create app/page.js. The page creates the Ably client in the browser with createClient({ authUrl: '/auth' }), and reads the stored conversation from /api/messages. It then mounts the chat component inside PubSubClientProvider, and a ChannelProvider that sets the channel's rewind param to the stored serial. When the database has no serial, the agent has not yet published to the channel, so the channel attaches without a rewind.

Rewind applies only when the channel attaches, so the page sets it before the chat component subscribes. The highlighted lines are reinit, which runs when the channel loses continuity. reinit detaches the channel, reads the database again, and remounts the chat component under a new key, so the channel rewinds from the newer serial and a new decoder receives the messages. This client code is the page:

1'use client';2 3import { useCallback, useEffect, useRef, useState } from 'react';4import { createClient } from '@ably/pubsub-device';5import { ChannelProvider, PubSubClientProvider } from '@ably/pubsub-device/react';6import { Chat } from './chat';7 8const channelName = 'conversations:demo';9 10const loadConversation = async () => {11  const res = await fetch(`/api/messages?channel=${encodeURIComponent(channelName)}`);12  if (!res.ok) throw new Error(`unable to load the conversation: ${res.status}`);13  return res.json();14};15 16export default function Page() {17  const [client, setClient] = useState();18  const [stored, setStored] = useState();19  // Counts the remounts, so each one mounts a new chat component with a new decoder.20  const [generation, setGeneration] = useState(0);21  const [reloading, setReloading] = useState(false);22  const reloadingRef = useRef(false);23 24  useEffect(() => {25    const ably = createClient({ authUrl: '/auth' });26    setClient(ably);27    return () => ably.close();28  }, []);29 30  useEffect(() => {31    let cancelled = false;32    loadConversation()33      .then((conversation) => {34        if (!cancelled) setStored(conversation);35      })36      .catch((error) => console.error('load failed', error));37    return () => {38      cancelled = true;39    };40  }, []);41 42  const reinit = useCallback(async () => {43    if (!client || reloadingRef.current) return;44    reloadingRef.current = true;45    setReloading(true);46    try {47      // Detach first, so the channel rewinds when it attaches again. A FAILED channel rejects detach(), and attaching again recovers it.48      await client.channels.get(channelName).detach().catch(() => {});49      setStored(await loadConversation());50      setGeneration((g) => g + 1);51    } catch (error) {52      console.error('reload failed', error);53    } finally {54      reloadingRef.current = false;55      setReloading(false);56    }57  }, [client]);58 59  if (!client || !stored || reloading) return <p>Connecting</p>;60 61  return (62    <PubSubClientProvider client={client}>63      <ChannelProvider channelName={channelName} options={stored.serial ? { params: { rewind: stored.serial } } : undefined}>64        <Chat key={generation} channelName={channelName} stored={stored.messages} onDiscontinuity={reinit} />65      </ChannelProvider>66    </PubSubClientProvider>67  );68}

Create app/layout.js. Next.js needs a root layout to build the app:

export default function RootLayout({ children }) {
  return (
    <html lang="en">
      <body>{children}</body>
    </html>
  );
}

Run the app

Add your Ably API key to .env.local:

ABLY_API_KEY=your-ably-api-key

Start the dev server:

npm run dev

Open http://localhost:3000 in two tabs and send a message from one. The prompt appears in both tabs, and the response streams into both, one word at a time. Reload a tab while a response is streaming, and the tab shows the stored conversation, then the response so far, then the rest of it live. Send another message and press Stop while the response streams, and the response ends in both tabs.

What happens when you send a message

1. The client adds the prompt to its list, and POSTs it to the agent's HTTP endpoint.
2. The agent publishes the prompt to the channel as a prompt event with send, and stores it with the serial that send returned. Every subscribed client shows the prompt.
3. The agent pipes the provider's events to the channel with the prompt's requestId. The adaptor appends each text-delta to one message for the response, and every subscribed client decodes the deltas and merges them.
4. When the pipe resolves, the agent stores the response with the serial of its last message. A tab that loads later rewinds the channel from that serial.

Understand the architecture

Your provider makes the model call, and you define the event types. Your custom codec maps those events to Ably messages. The adaptor uses the custom codec to publish the events to the channel, and the decoder uses it to decode them in every client. Your application can also use the channel directly through the Ably Pub/Sub SDK, so you can use the full range of Ably channel features. The SDK, Ably and your application each implement part of the conversation.

Explore next

• Custom codec: headers, ephemeral messages, wildcard rows, and sharing a channel with another codec.
• Codec API reference: defineCodec and the fields a row returns.
• Agent adaptor API reference: createAgentAdaptor, send, pipe and pipeEach.
• Decoder API reference: createDecoder, receive, decode and the events it emits.
• Token streaming: how a response grows as appends to one message, and how to stop it.
• Awareness: show users what the agent is doing now, and let the agent check whether the user is still on the channel.
• Artifacts: keep the thing the agent makes as shared state on the channel, which the agent and the user can both edit.