### Shell
```
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](https://ably.com/docs/auth/capabilities.md) the client needs on the conversation channels:
### Javascript
```
// 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](https://ably.com/dashboard). The key needs `publish` to publish a response, and `message-update-any` to append to it. You can [scope the token](https://ably.com/docs/ai-transport/setup/authentication.md#capabilities) 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](https://ably.com/docs/ai-transport/setup/channel-rules.md) 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](https://ably.com/docs/channels/options/rewind.md).
> **Important**
>
> Without the rule, the first append fails with error `93002` and the response does not stream. The channel name you use must match the rule's expression.
## 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`](https://ably.com/docs/ai-transport/api/javascript/codec.md#define-codec) 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:
### Javascript
```
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](https://ably.com/docs/ai-transport/setup/authentication.md#agent-post) with your application's own auth before the agent publishes anything. The agent creates a [realtime client](https://ably.com/docs/basics.md#server-realtime-client) with your API key, gets the conversation's channel, and passes it to [`createAgentAdaptor`](https://ably.com/docs/ai-transport/api/javascript/agent-adaptor.md#create-agent-adaptor) with your custom codec.
The agent publishes the [prompt](https://ably.com/docs/ai-transport/conversation.md#prompt) to the channel with `send`, as a `prompt` event, so every subscribed client shows it. Each message on the channel has a [serial](https://ably.com/docs/ai-transport/conversation/history.md#how-it-works), 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`](https://ably.com/docs/ai-transport/api/javascript/agent-adaptor.md#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:
### Javascript
```
// 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](https://ably.com/docs/ai-transport/api/javascript/decoder.md). 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](https://ably.com/docs/ai-transport/conversation/reconnection-and-recovery.md#detect). 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:
### Javascript
```
'use client';
// The browser fetches a token, so it never holds an API key.
import { useCallback, useEffect, useRef, useState } from 'react';
import { createDecoder } from '@ably/ai-transport';
import { useChannelStateListener, usePubSubClient } from '@ably/pubsub-device/react';
import { chatCodec } from './codec';
import { mergeEvent } from './merge';
export function Chat({ channelName, stored, onDiscontinuity }) {
const [messages, setMessages] = useState(stored);
const [input, setInput] = useState('');
// The AbortController of this tab's request while the agent works on it.
const [request, setRequest] = useState();
const client = usePubSubClient();
const channel = client.channels.get(channelName);
const hasAttached = useRef(channel.state === 'attached');
useChannelStateListener(channelName, (change) => {
if (change.current === 'attached' && !hasAttached.current) {
hasAttached.current = true;
return;
}
if (!hasAttached.current) return;
const lost =
change.current === 'detached' || change.current === 'failed' || (change.current === 'attached' && !change.resumed);
if (lost) onDiscontinuity();
});
const apply = useCallback(({ event }) => {
if (event === undefined) return;
setMessages((prev) => mergeEvent(prev, event));
}, []);
useEffect(() => {
const decoder = createDecoder(chatCodec);
decoder.onEvent(apply);
decoder.onError((error) => console.error('decode failed', error));
const start = async () => {
await channel.subscribe(decoder.receive);
await channel.attach();
};
start().catch((error) => console.error('attach failed', error));
return () => {
channel.unsubscribe(decoder.receive);
};
}, [channel, apply]);
const send = async (e) => {
e.preventDefault();
const text = input.trim();
if (!text) return;
setInput('');
const message = { id: crypto.randomUUID(), role: 'user', text };
setMessages((prev) => mergeEvent(prev, { type: 'prompt', message }));
const controller = new AbortController();
setRequest(controller);
try {
await fetch('/api/chat', {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ channelName, message }),
signal: controller.signal,
});
} catch (error) {
if (error.name !== 'AbortError') console.error('agent request failed', error);
} finally {
setRequest((current) => (current === controller ? undefined : current));
}
};
return (
{messages.map((m) => (
{m.role}: {m.text}
{m.reason === 'cancelled' ? ' (stopped)' : ''}
))}
);
}
```
## 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](https://ably.com/docs/ai-transport/conversation/token-streaming.md#rollup) 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`:
### Javascript
```
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:
#### Javascript
```
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.
#### Javascript
```
// 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](https://ably.com/docs/ai-transport/conversation/history.md#catch-up) from the newer serial and a new decoder receives the messages. This client code is the page:
#### Javascript
```
'use client';
import { useCallback, useEffect, useRef, useState } from 'react';
import { createClient } from '@ably/pubsub-device';
import { ChannelProvider, PubSubClientProvider } from '@ably/pubsub-device/react';
import { Chat } from './chat';
const channelName = 'conversations:demo';
const loadConversation = async () => {
const res = await fetch(`/api/messages?channel=${encodeURIComponent(channelName)}`);
if (!res.ok) throw new Error(`unable to load the conversation: ${res.status}`);
return res.json();
};
export default function Page() {
const [client, setClient] = useState();
const [stored, setStored] = useState();
// Counts the remounts, so each one mounts a new chat component with a new decoder.
const [generation, setGeneration] = useState(0);
const [reloading, setReloading] = useState(false);
const reloadingRef = useRef(false);
useEffect(() => {
const ably = createClient({ authUrl: '/auth' });
setClient(ably);
return () => ably.close();
}, []);
useEffect(() => {
let cancelled = false;
loadConversation()
.then((conversation) => {
if (!cancelled) setStored(conversation);
})
.catch((error) => console.error('load failed', error));
return () => {
cancelled = true;
};
}, []);
const reinit = useCallback(async () => {
if (!client || reloadingRef.current) return;
reloadingRef.current = true;
setReloading(true);
try {
// Detach first, so the channel rewinds when it attaches again. A FAILED channel rejects detach(), and attaching again recovers it.
await client.channels.get(channelName).detach().catch(() => {});
setStored(await loadConversation());
setGeneration((g) => g + 1);
} catch (error) {
console.error('reload failed', error);
} finally {
reloadingRef.current = false;
setReloading(false);
}
}, [client]);
if (!client || !stored || reloading) return Connecting
;
return (
);
}
```
Create `app/layout.js`. Next.js needs a root layout to build the app:
#### Javascript
```
export default function RootLayout({ children }) {
return (
{children}
);
}
```
### Run the app
Add your Ably API key to `.env.local`:
#### Shell
```
ABLY_API_KEY=your-ably-api-key
```
Start the dev server:
#### Shell
```
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](https://ably.com/docs/ai-transport/internals/codec-architecture.md) 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](https://ably.com/docs/ai-transport/conversation.md#boundary).
## Explore next
- [Custom codec](https://ably.com/docs/ai-transport/frameworks/custom-codec.md): headers, ephemeral messages, wildcard rows, and sharing a channel with another codec.
- [Codec API reference](https://ably.com/docs/ai-transport/api/javascript/codec.md): `defineCodec` and the fields a row returns.
- [Agent adaptor API reference](https://ably.com/docs/ai-transport/api/javascript/agent-adaptor.md): `createAgentAdaptor`, `send`, `pipe` and `pipeEach`.
- [Decoder API reference](https://ably.com/docs/ai-transport/api/javascript/decoder.md): `createDecoder`, `receive`, `decode` and the events it emits.
- [Token streaming](https://ably.com/docs/ai-transport/conversation/token-streaming.md): how a response grows as appends to one message, and how to stop it.
- [Awareness](https://ably.com/docs/ai-transport/awareness.md): show users what the agent is doing now, and let the agent check whether the user is still on the channel.
- [Artifacts](https://ably.com/docs/ai-transport/artifacts.md): keep the thing the agent makes as shared state on the channel, which the agent and the user can both edit.
## Related Topics
- [Vercel AI SDK](https://ably.com/docs/ai-transport/get-started/vercel-ai-sdk.md): Build a streaming chat app with the Vercel AI SDK and Ably AI Transport. Responses reach every connected client, and a client that reloads reads the conversation from your database, then replays what arrived on the channel after it.
- [OpenAI responses](https://ably.com/docs/ai-transport/get-started/openai.md): Build a streaming chat app with the OpenAI Responses API and Ably AI Transport. Responses reach every connected client, and a client that reloads reads the conversation from your database, then replays what arrived on the channel after it.
## Documentation Index
To discover additional Ably documentation:
1. Fetch [llms.txt](https://ably.com/llms.txt) for the canonical list of available pages.
2. Identify relevant URLs from that index.
3. Fetch target pages as needed.
Avoid using assumed or outdated documentation paths.