# Agent adaptor An `AgentAdaptor` publishes the events of one [codec](https://ably.com/docs/ai-transport/internals/codec-architecture.md) on an [Ably channel](https://ably.com/docs/channels.md). Your agent gets the channel from its client in the [Ably Pub/Sub SDK](https://ably.com/docs/pub-sub.md). Use `send` to publish one event as new messages, for example the prompt or a tool result. Use `pipe` to write a stream of events, such as a whole response, with no further input from your code. Use `pipeEach` to write the same stream and get back each event after the adaptor writes it. The adaptor publishes each discrete event in a stream as a new message. It [appends](https://ably.com/docs/messages/updates-deletes.md#append) the deltas of each streamed part, such as the response text, to one message for that part. A subscribed client receives the deltas as the model produces them, and history holds one message with the full text of the part. The adaptor only publishes, appends, and updates messages, so it does not attach the channel. The adaptor does not hold a resource, and has no `close` method. Use `createAgentAdaptor` with a codec you built with [`defineCodec`](https://ably.com/docs/ai-transport/api/javascript/codec.md). The following agent code runs on your server, and creates an adaptor for one conversation's channel: #### Javascript ``` import { createRealtimeClient } from '@ably/pubsub-server'; import { createAgentAdaptor } from '@ably/ai-transport'; import { codec } from './codec.js'; const client = createRealtimeClient({ key: process.env.ABLY_API_KEY }); const adaptor = createAgentAdaptor(client.channels.get('conversations:42'), codec); ``` ## Create an agent adaptor `createAgentAdaptor(channel: ChannelWriter, codec: Codec, options?: AgentAdaptorOptions): AgentAdaptor` Create an adaptor that publishes the events of `codec` on `channel`. For a codec that AI Transport ships, use its [provider factory](https://ably.com/docs/ai-transport/api.md#entry-points). Each factory takes the same `channel` and `options`, and returns an `AgentAdaptor`: - [`vercelAdaptor`](https://ably.com/docs/ai-transport/api/javascript/vercel.md) - [`openaiAdaptor`](https://ably.com/docs/ai-transport/api/javascript/openai.md) - [`agUiAdaptor`](https://ably.com/docs/ai-transport/api/javascript/ag-ui.md) The following agent code creates an adaptor with a default header and a logger: ### Javascript ``` import { createAgentAdaptor, makeLogger, LogLevel } from '@ably/ai-transport'; const adaptor = createAgentAdaptor(channel, codec, { headers: { agent: 'support-bot' }, logger: makeLogger({ logLevel: LogLevel.Debug }), }); ``` ### Parameters The `createAgentAdaptor()` function takes the following parameters: | Parameter | Required | Description | Type | | --- | --- | --- | --- | | channel | Required | The channel to publish on, from `channels.get` on your Pub/Sub SDK client. A `RealtimeChannel` satisfies it. |
| | codec | Required | The codec that encodes events as Ably messages. | [`Codec`](https://ably.com/docs/ai-transport/api/javascript/codec.md) | | options | Optional | Default headers and a logger. |
|
| Method | Description | Type | | --- | --- | --- | | publish | Publish a new message. The first serial of the result identifies the new message. | `(message: Message) => Promise` | | appendMessage | Append `data` to the message that `message.serial` identifies. | `(message: Message) => Promise` | | updateMessage | Replace the content of the message that `message.serial` identifies. | `(message: Message) => Promise` |
| Property | Required | Description | Type | | --- | --- | --- | --- | | logger | Optional | The logger for the adaptor's diagnostics, such as repairs, failed acks, and cancelled pipes. Create one with [`makeLogger`](https://ably.com/docs/ai-transport/api.md#logger). The adaptor does not log when you leave it out. | `Logger` | | headers | Optional | Headers for every message the adaptor publishes, under `extras.headers`. Where a call's headers set the same key, the call's value is used. Where a codec [row](https://ably.com/docs/ai-transport/api/javascript/codec.md#define-codec) sets the same key, the row's value is used over both. A call can replace a default with another value, `null` included. A call's key with an `undefined` value is dropped, so the default stays and the call cannot remove it. |
|
| Property | Description | Type | | --- | --- | --- | | [key] | A flat map from header name to value. Each value must be a string, number, boolean, or `null`, which are the values Ably stores under `extras.headers`. | String, Number, Boolean, or `null` |
### Returns `AgentAdaptor`. Returns the adaptor. The function throws an [`ErrorInfo`](https://ably.com/docs/ai-transport/api/errors.md#errorinfo) object with `InvalidArgument` (40003) when a default header value is not a string, number, boolean, or `null`. ## Send an event `adaptor.send(event: E, options?: SendOptions): Promise` `send` encodes one event with the codec, and publishes the messages the codec encodes it to, in order. Publishing stops at the first failure. The messages published before the failure stay on the channel. `send` publishes only new messages. If any message of the event appends to or updates a message, `send` rejects and does not publish any message. Use [`pipe`](#pipe) for events that stream. The following agent code publishes a prompt and stores its serial, so a client that loads the conversation from your database can [rewind the channel](https://ably.com/docs/ai-transport/conversation/history.md#joining) to that point: ### Javascript ``` const { serial } = await adaptor.send( { type: 'user-message', message }, { headers: { requestId: message.id } }, ); await store.savePrompt(message, serial); ``` ### Parameters The `send()` method takes the following parameters: | Parameter | Required | Description | Type | | --- | --- | --- | --- | | event | Required | The event to publish, of the codec's event type. | `E` | | options | Optional | The headers for this call. |
|
| Property | Required | Description | Type | | --- | --- | --- | --- | | headers | Optional | Headers for every message the call publishes, under `extras.headers`. They are used over the adaptor's default headers where both set a key, and a codec row's own headers are used over them. An `undefined` value is dropped. The shipped codecs do not put these headers on the decoded event. A subscribed client reads them from `delivery.message.extras.headers`. |
|
### Returns `Promise`. Returns a promise. The promise is fulfilled with the serial of the last message published, or rejected with an [`ErrorInfo`](https://ably.com/docs/ai-transport/api/errors.md#errorinfo) object. | Property | Description | Type | | --- | --- | --- | | serial | The ack serial of the last message published for the event, or `undefined` when the codec encoded the event to no messages. | String or `undefined` |
The promise rejects with the following [error codes](https://ably.com/docs/ai-transport/api/errors.md#reference): - `InvalidArgument` (40003) when a header value is not a string, number, boolean, or `null`, when a message the codec encoded lacks the codec's `name` or its `id` under `extras.ai.codec` or when a message of the event appends or updates. In each case `send` does not publish any message. - `InsufficientCapability` (40160) when Ably rejects a message for a capability reason. - `SessionSendFailed` (104005) for any other publish failure. The Ably error is the `cause`. - `InternalError` (50000) when the ack of a publish does not carry a serial. An error that the codec's `encode` throws reaches you unchanged. For a codec built with `defineCodec`, that is `InvalidArgument` for an event type with no row. ## Pipe a stream of events `adaptor.pipe(source: PipeSource, options?: PipeOptions): Promise` `pipe` reads a source of events and writes each one to the channel with the codec. The promise resolves when the source ends. The adaptor sends each append without waiting for Ably to acknowledge it, so the response streams at the speed of the connection. It waits for all the appends of a stream before it writes the message that ends that stream, so a subscribed client receives that message after all of the stream's text. If an append fails, the pipe carries on. The adaptor repairs each stream with a failed append before it writes the message that ends that stream, and before the pipe resolves or rejects. The repair is an update that carries the full text, so after a successful repair every message the pipe wrote holds its full text. The adaptor does not end a stream that the source did not end, so a subscribed client sees that stream stay open. After a rejection, your agent is responsible for publishing an end event with `send`. For the Vercel codec, that event is `abort`. The following agent code pipes a Vercel AI SDK stream, and publishes `abort` when the pipe rejects: ### Javascript ``` import { ErrorCode, errorInfoIs } from '@ably/ai-transport'; // uiMessages is the conversation so far, ending with the prompt; result is the streamText result. const headers = { requestId: message.id }; try { const stream = toUIMessageStream({ stream: result.fullStream, originalMessages: uiMessages, generateMessageId: () => crypto.randomUUID(), }); const { serial } = await adaptor.pipe(stream, { signal: req.signal, headers }); await store.saveResponseSerial(serial); } catch (error) { await adaptor.send({ type: 'abort' }, { headers }); if (!errorInfoIs(error, ErrorCode.OperationCancelled)) throw error; } ``` ### Parameters The `pipe()` method takes the following parameters: | Parameter | Required | Description | Type | | --- | --- | --- | --- | | source | Required | The events to write. A provider SDK stream that is async-iterable pipes in directly. The adaptor reads one event at a time. |
| | options | Optional | The abort signal and the headers for this call. |
|
| Value | Description | | --- | --- | | `ReadableStream` | The adaptor reads the stream with a reader, and releases the reader's lock when the pipe ends. Releasing the lock does not cancel the stream. | | `AsyncIterable` | The adaptor reads the iterator, and calls its `return()` when the pipe ends. |
| Property | Required | Description | Type | | --- | --- | --- | --- | | signal | Optional | Cancels the pipe when it fires. The adaptor abandons the read in progress, and does not write any event after the signal fires. The adaptor then repairs what it wrote, and does not end any open stream. | `AbortSignal` | | headers | Optional | Headers for every message the pipe writes: the message that opens a stream, each append, each update, each repair and the message that ends a stream. Ably keeps the `extras` of the last write on an appended message, so the adaptor puts the headers on every write, and you should keep them small. The adaptor merges and checks them as it does for `send`. |
|
### Returns `Promise`. Returns a promise. The promise is fulfilled once the source has ended and everything the pipe wrote is on the channel, or rejected with an [`ErrorInfo`](https://ably.com/docs/ai-transport/api/errors.md#errorinfo) object. | Property | Description | Type | | --- | --- | --- | | serial | The ack serial of the last message the pipe published, including the message that opens a stream, or `undefined` when it did not publish a message. | String or `undefined` |
The promise rejects with the following [error codes](https://ably.com/docs/ai-transport/api/errors.md#reference): - `InvalidArgument` (40003) when a header value is not a string, number, boolean, or `null`. The adaptor rejects before it reads the source. - `OperationCancelled` (40033) when the signal fires. - `PipeFailed` (104008) when the source throws, the codec cannot encode an event, a message the codec encoded lacks the codec's `name` or `id`, publishing or updating a message fails, or a repair fails. The failure is the `cause`. When publishing a message fails, the `cause` is the Ably error, such as 40160 for a missing capability. When a repair fails, the `cause` is `StreamedMessageFinalizeFailed` (104000), and its own `cause` is the first Ably error from the repair, such as [93002](https://ably.com/docs/ai-transport/troubleshooting.md#namespace-not-configured) when the channel rule does not enable appends and updates. If a repair fails after the signal fires, the pipe rejects with `PipeFailed`. If a repair fails after an earlier failure, the earlier failure is the rejection, and the adaptor logs the repair failure. ## Pipe a stream and act on each event `adaptor.pipeEach(source: PipeSource, options?: PipeOptions): AsyncIterable>` `pipeEach` reads a source of events, writes each one to the channel with the codec, and returns each event after the adaptor writes it. Use it in a `for await` loop when the agent acts on events as the response streams. The adaptor does not read the source until the loop asks for the first event. It reads the next event only when the loop body has finished with the previous one. The loop gets an event that the adaptor published as a new message after Ably acknowledges it, and gets an appended event immediately, before Ably acknowledges the append. When the loop ends early on `break`, `return`, or a throw in the loop body, or when the signal fires, the adaptor stops reading the source. The adaptor then repairs what it wrote before the loop exits, and does not end any open stream. Leaving the loop early releases a `ReadableStream` source without cancelling it, so the producer, such as a model call, keeps running. You can stop the model call by passing it the same signal. The loop does not return a final result. If you need the last serial after the loop, you can save it in a variable inside the loop. The following agent code saves the response serial when the adaptor writes the `finish` event: ### Javascript ``` for await (const { event, serial } of adaptor.pipeEach(stream, { signal })) { if (event.type === 'finish') await db.saveResponseSerial(serial); } ``` ### Parameters The `pipeEach()` method takes the following parameters: | Parameter | Required | Description | Type | | --- | --- | --- | --- | | source | Required | The events to write. A provider SDK stream that is async-iterable pipes in directly. The adaptor reads one event at a time. |
| | options | Optional | The abort signal and the headers for this call. |
|
### Returns `AsyncIterable>`. Returns an async iterable of the events, each with the serial of the message it was written to. | Property | Description | Type | | --- | --- | --- | | event | The event, as the source produced it. | `E` | | serial | The serial of the message the event was written to. For a new message, its ack serial. For an append or an update, the serial of the stream's message. When the codec encoded the event to several messages, the serial of the last one. When the codec did not encode the event to any message, `undefined`. | String or `undefined` |
The method and the loop throw an [`ErrorInfo`](https://ably.com/docs/ai-transport/api/errors.md#errorinfo) object with the following [error codes](https://ably.com/docs/ai-transport/api/errors.md#reference): - `InvalidArgument` (40003) when a header value is not a string, number, boolean, or `null`. The `pipeEach()` call throws it before the loop starts. - `OperationCancelled` (40033) from the loop when the signal fires. - `PipeFailed` (104008) from the loop for the same causes as [`pipe`](#pipe-returns). - `PipeFailed` (104008) from the loop's exit when the repair after a `break` or `return` fails. After a throw in the loop body, that throw is the error, and the adaptor logs the repair failure. ## Example The following agent code is a Next.js route. It creates one realtime client for each request, publishes the prompt, pipes the response and saves the response serial when the adaptor writes the `finish` event. The client sets `appendRollupWindow` to `0`, which turns append rollup off, so subscribed clients receive each append as its own message: ### Javascript ``` import { ErrorCode, errorInfoIs } from '@ably/ai-transport'; import { vercelAdaptor } from '@ably/ai-transport/vercel'; import { createRealtimeClient } from '@ably/pubsub-server'; import { convertToModelMessages, streamText, toUIMessageStream } from 'ai'; import { loadConversation, savePrompt, saveResponseSerial } from './store'; import { model } from './model'; export async function POST(req) { const { channelName, message } = await req.json(); const client = createRealtimeClient({ key: process.env.ABLY_API_KEY, transportParams: { appendRollupWindow: 0 }, }); const adaptor = vercelAdaptor(client.channels.get(channelName), { headers: { requestId: message.id }, }); try { const uiMessages = [...(await loadConversation(channelName)), message]; const { serial: promptSerial } = await adaptor.send({ type: 'user-message', message }); await savePrompt(channelName, message, promptSerial); const result = streamText({ model, messages: await convertToModelMessages(uiMessages), abortSignal: req.signal, }); const stream = toUIMessageStream({ stream: result.fullStream, originalMessages: uiMessages, generateMessageId: () => crypto.randomUUID(), }); try { for await (const { event, serial } of adaptor.pipeEach(stream, { signal: req.signal })) { if (event.type === 'finish') await saveResponseSerial(channelName, serial); } } catch (error) { await adaptor.send({ type: 'abort' }).catch((sendError) => console.error(sendError)); if (!errorInfoIs(error, ErrorCode.OperationCancelled)) throw error; } return new Response(null, { status: 204 }); } finally { client.close(); } } ``` ## Related Topics - [Decoder](https://ably.com/docs/ai-transport/api/javascript/decoder.md): API reference for the AI Transport decoder: createDecoder, receive, decode, onEvent, and onError, and the Delivery each message becomes. - [Codec](https://ably.com/docs/ai-transport/api/javascript/codec.md): API reference for the AI Transport Codec: the encode and decode contract an adaptor and a decoder use, EncodedMessage, and how defineCodec builds a codec from a table of event rows. - [Vercel](https://ably.com/docs/ai-transport/api/javascript/vercel.md): API reference for the AI Transport codec for the Vercel AI SDK: vercelAdaptor, vercelDecoder, and the VercelEvent union of UIMessageChunk events and the user message. - [OpenAI](https://ably.com/docs/ai-transport/api/javascript/openai.md): API reference for the AI Transport codec for the OpenAI Responses API: openaiAdaptor, openaiDecoder, and the OpenAIEvent union of Responses stream events and client input. - [AG-UI](https://ably.com/docs/ai-transport/api/javascript/ag-ui.md): API reference for the AI Transport codec for AG-UI: agUiAdaptor, agUiDecoder, and the AGUIEvent union of AG-UI events and the run-input a client sends. ## 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.