Write a custom codec

What you build

A Next.js chat app for a model provider that AI Transport does not ship a codec for, where:

• Your own event types stream from the agent to every connected client in realtime.
• A second tab shows the same conversation, because every subscribed client receives the same messages.
• Reloading the page reads the conversation from your own store, then replays what arrived on the channel after it.

The codec is the only part specific to your events. The transport takes any codec, so the agent and the client use the same createTransport calls as with a codec that ships with the SDK.

Prerequisites

• Node.js 22 or later.
• An Ably account with an API key.
• A model provider that gives you a stream of events. This guide uses an invented one, so you can follow it without a provider.

Install dependencies

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

npm install @ably/ai-transport ably 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 below fetches its token with authUrl: '/api/auth/token'. The agent route 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. The rule also persists the channel's messages, so a client that reloads can replay them.

Write the codec

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

• 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 a reason.

defineCodec builds a codec from a table with one row for each event type. Each row's encode returns the message body and where the message goes, and its decode rebuilds the event from the body.

The text-delta row streams. Its data is the delta's text, and append keys every delta of a response to one message, so the response's text grows in one message on the channel. The first delta publishes the message. The response id goes in fields, because an append carries only data and the fields travel beside it. The text-end row uses ends to close the stream. Every other row is a plain publish.

// Shared by the agent and the client.
import { defineCodec } from '@ably/ai-transport';

export const chatCodec = defineCodec({
  adapterTag: 'my-provider',
  typeOf: (event) => event.type,
  events: {
    '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: { reason: event.reason } }),
      decode: ({ fields }) => ({ type: 'done', reason: String(fields.reason) }),
    },
  },
});

The codec handles the three ways Ably delivers a streamed message. A live client receives each delta. A client that joins late, or that replays history, receives one text-delta with the response's text so far, and then the rest live. The codec drops a delivery it has already decoded, so history and live delivery can overlap without duplicating text.

Create the agent route

Create app/api/chat/route.js. The POST handler is the agent. fakeModel stands in for your provider: it yields the response as text-start, a text-delta for each word, text-end and done. The route pipes those events to the channel, and a small generator collects the response's text as the events pass to the pipe. When the pipe resolves, the route stores the user's message and the response, with the serial of the last message.

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

// Agent code. This runs on your own infrastructure, so it holds the API key.
import * as Ably from 'ably';
import { createTransport } from '@ably/ai-transport';
import { chatCodec } from '../../codec';
import { loadConversation, recordMessages } from '../../store';

const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });

// Stand-in for your provider's stream of events.
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', reason: 'complete' };
}

// Pass each event to the pipe, and collect the response's text for your store.
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) {
  const { channelName, message } = await req.json();
  const responseId = crypto.randomUUID();

  const transport = createTransport({ client: ably, channelName, codec: chatCodec });
  try {
    let text = '';
    const { serial } = await transport.pipe(keepText(fakeModel(responseId, message.text), (t) => (text = t)), {
      signal: req.signal,
    });
    recordMessages(channelName, [message, { id: responseId, role: 'assistant', text }], serial);
    return new Response(null, { status: 204 });
  } finally {
    await transport.close();
  }
}

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

Create the chat component

Create app/chat.js. When the transport is ready, the component reads the stored conversation from the agent route and subscribes from its serial with fromSerial. The transport replays the messages that arrived after the serial, in channel order, then delivers live messages. With no stored serial, nothing is stored yet, so the component replays the whole channel with untilEvent, using a predicate that never matches.

Each message the handler receives moves lastSeen forward. After a discontinuity, the component subscribes again from lastSeen, so it receives the messages it missed.

To send, the component adds the user's message to its list, then wakes the agent with a POST that carries the message. The agent reads the prompt from the request body.

'use client';

// Client code. The browser fetches a token, so it never holds an API key.
import { useCallback, useEffect, useRef, useState } from 'react';
import { fromSerial, untilEvent } from '@ably/ai-transport';
import { useTransport } from '@ably/ai-transport/react';
import { mergeEvent } from './merge';

// With no stored serial, nothing is stored yet, so replay the whole channel.
const allHistory = untilEvent(() => false);

export function Chat({ channelName }) {
  const { transport } = useTransport();
  const [messages, setMessages] = useState([]);
  const [input, setInput] = useState('');
  const lastSeen = useRef(undefined);

  const apply = useCallback(({ event, message }) => {
    if (lastSeen.current === undefined || message.serial > lastSeen.current) lastSeen.current = message.serial;
    if (event !== undefined) setMessages((prev) => mergeEvent(prev, event));
  }, []);

  useEffect(() => {
    if (!transport) return;
    let unsubscribe;
    let cancelled = false;
    const subscribeFrom = (serial) => {
      unsubscribe?.();
      unsubscribe = transport.subscribe(apply, { history: { replay: serial ? fromSerial(serial) : allHistory } });
    };

    (async () => {
      const res = await fetch(`/api/chat?channel=${encodeURIComponent(channelName)}`);
      const stored = await res.json();
      if (cancelled) return;
      setMessages(stored.messages);
      lastSeen.current = stored.serial;
      subscribeFrom(stored.serial);
    })();

    const off = transport.on('discontinuity', () => subscribeFrom(lastSeen.current));
    return () => {
      cancelled = true;
      off();
      unsubscribe?.();
    };
  }, [transport, channelName, apply]);

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

  return (
    <div>
      {messages.map((m) => (
        <p key={m.id}>
          <strong>{m.role}:</strong> {m.text}
        </p>
      ))}
      <form onSubmit={send}>
        <input value={input} onChange={(e) => setInput(e.target.value)} />
        <button type="submit" disabled={!transport}>Send</button>
      </form>
    </div>
  );
}

Merge the model output into conversation messages

Create app/merge.js. Each client merges the events it receives into the messages it renders. text-start adds an empty response, and each text-delta appends its text to the response with its id. A message the list already holds is skipped, so a message that a reload replayed does not appear twice.

// Client code.
export function mergeEvent(messages, event) {
  const has = (id) => messages.some((m) => m.id === id);
  switch (event.type) {
    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));
    default:
      return messages;
  }
}

Integrate with your database

Create app/store.js. This Map stands in for the database your application already uses. 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.
const conversations = new Map();

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

export function recordMessages(channelName, messages, serial) {
  const current = loadConversation(channelName);
  conversations.set(channelName, { messages: [...current.messages, ...messages], serial: serial ?? current.serial });
}

Wire it together

Create app/page.js. The page creates the Ably client in the browser, and wraps the chat in AblyProvider and TransportProvider with your codec:

'use client';

// Client code.
import { useEffect, useState } from 'react';
import * as Ably from 'ably';
import { AblyProvider } from 'ably/react';
import { TransportProvider } from '@ably/ai-transport/react';
import { chatCodec } from './codec';
import { Chat } from './chat';

const channelName = 'conversations:demo';

export default function Page() {
  const [client, setClient] = useState();

  useEffect(() => {
    const realtime = new Ably.Realtime({ authUrl: '/api/auth/token' });
    setClient(realtime);
    return () => realtime.close();
  }, []);

  if (!client) return null;

  return (
    <AblyProvider client={client}>
      <TransportProvider channelName={channelName} codec={chatCodec}>
        <Chat channelName={channelName} />
      </TransportProvider>
    </AblyProvider>
  );
}

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

Start the dev server:

npm run dev

Open http://localhost:3000 in two tabs and send a message from one. The response streams into both, one word at a time. The prompt travels in the HTTP request that wakes the agent, so it is not on the channel, and the other tab does not receive it in realtime. Multi-device and fan-out shows how to deliver the prompt to every device. 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.

What happens when you send a message

1. The client adds the user's message to its list, and POSTs the message to the agent route.
2. The POST wakes the agent, and the agent reads the prompt from the request body. AI Transport does not send HTTP requests of its own.
3. The agent pipes the provider's events to the channel. The codec 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 messages with the serial of the last message. A client that loads the page later replays the channel from that serial.

Understand the architecture

Your provider owns the model call and its event types. The codec maps those events to Ably messages, and the transport carries them over one Ably channel. Codec architecture covers how a codec handles appends and late joiners.

Explore next

• Codec API reference: defineCodec, the row fields, and the Codec contract.
• History and replay: replay the channel from a stored serial, and read history backwards.
• Token streaming: how a response is appended to one message.
• Transport API reference: every method on the transport.