Codec

A Codec turns your framework's events into Ably messages, and turns Ably messages back into events. A transport uses one codec for everything it sends, pipes and receives. The Vercel AI SDK, OpenAI and AG-UI codecs ship with the SDK, and defineCodec builds a codec for any other event type.

A codec only defines the wire format. The transport owns the channel and delivers the decoded events.

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

const chatCodec = defineCodec({
  typeOf: (event) => event.type,
  events: {
    prompt: {
      encode: (event) => ({ data: event.text }),
      decode: ({ data }) => ({ type: 'prompt', text: String(data) }),
    },
  },
});

// ably is your Ably.Realtime client.
const transport = createTransport({ client: ably, channelName: 'conversations:42', codec: chatCodec });

Properties

The transport handles a decode that throws: it reports the error on the transport's error event, and delivers the message with no event.

A message with none of publish, append or update is a plain publish that nothing appends to. send() has no keys to append to, so it rejects an event the codec encodes as an append or an update. Use pipe() for streams.

Build a codec

Build a codec from a table with one row for each event type. Each row has an encode function that returns the message body and where it goes, and a decode function that rebuilds the event from the body. A row produces at most one message and at most one event.

The builder writes the row's data as the message body, and the event type and the row's fields under extras.ai. It writes the row's headers under extras.headers. On decode it reads them back and picks the row by the event type.

An append only adds to a message's data, so a field that a streaming event needs on every delta goes in fields.

The builder also handles the three ways Ably delivers a streamed message. A live append carries only its new text. The first delivery after an attach carries the full content so far, as an update. A history read returns the whole message. The builder reduces all three to the text that delivery adds before the row's decode runs, and drops a message it has already seen, so history and live delivery can overlap without duplicates.

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

const chatCodec = defineCodec({
  adapterTag: 'my-chat',
  typeOf: (event) => event.type,
  events: {
    prompt: {
      encode: (event) => ({ data: event.text }),
      decode: ({ data }) => ({ type: 'prompt', text: String(data) }),
    },
    '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) }),
    },
  },
});

Parameters

Returns

Codec<E>. The codec. defineCodec throws an ErrorInfo with InvalidArgument when a row key is a wildcard with no prefix. A built codec's encode throws InvalidArgument for an event type with no row, or for a row that sets more than one of publish, append and update. Its decode throws InvalidArgument for a message whose type has no row.

Create a decoder core

You only need a decoder core for a codec you write without defineCodec, which creates one for you. Codec architecture explains what the decoder core does.

Create the decoder core that defineCodec uses, for a codec you write yourself. The core keeps, for each streamed message, the text it has passed on and the newest version it has seen. Its prepare method reduces an inbound message to the text that delivery adds, and drops a message the core has already seen.

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

const core = createDecoderCore();

const decode = (message) => {
  const prepared = core.prepare(message);
  if (prepared === undefined) return [];
  return [toEvent(prepared)];
};

Parameters

Returns

DecoderCore. The decoder core.

Example

A codec for a chat application whose agent streams text. A prompt is a plain publish. Each text delta appends to one message per stream, keyed by the stream id, and the end event forgets the key.

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

const chatCodec = defineCodec({
  adapterTag: 'my-chat',
  typeOf: (event) => event.type,
  events: {
    prompt: {
      encode: (event) => ({ data: event.text }),
      decode: ({ data }) => ({ type: 'prompt', text: String(data) }),
    },
    '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) }),
    },
  },
});

const ably = new Ably.Realtime({ authUrl: '/auth' });
const transport = createTransport({ client: ably, channelName: 'conversations:42', codec: chatCodec });

transport.subscribe(({ event }) => {
  if (event?.type === 'text-delta') appendText(event.id, event.delta);
});