Agent adaptor

An AgentAdaptor publishes the events of one codec on an Ably channel. Your agent gets the channel from its client in the Ably Pub/Sub SDK. 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 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.

The following agent code runs on your server, and creates an adaptor for one conversation's channel:

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

Create an adaptor that publishes the events of codec on channel. For a codec that AI Transport ships, use its provider factory. Each factory takes the same channel and options, and returns an AgentAdaptor:

• vercelAdaptor
• openaiAdaptor
• agUiAdaptor

The following agent code creates an adaptor with a default header and a logger:

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:

Returns

AgentAdaptor<E>. Returns the adaptor. The function throws an ErrorInfo object with InvalidArgument (40003) when a default header value is not a string, number, boolean, or null.

Send an event

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 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 to that point:

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:

Returns

Promise<SendResult>. Returns a promise. The promise is fulfilled with the serial of the last message published, or rejected with an ErrorInfo object.

The promise rejects with the following error codes:

• 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

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:

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:

Returns

Promise<PipeResult>. 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 object.

The promise rejects with the following error codes:

• 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 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

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:

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:

Returns

AsyncIterable<Published<E>>. Returns an async iterable of the events, each with the serial of the message it was written to.

The method and the loop throw an ErrorInfo object with the following error codes:

• 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.
• 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:

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();
  }
}