AG-UI codec

The AG-UI codec carries AG-UI events over an Ably channel, plus a run-input event for the input a client sends to start a run. The agent pipes its AG-UI event stream into the transport, and every subscribed client decodes the same events. A client can merge them with AG-UI's own reducer, such as an AbstractAgent from @ag-ui/client.

Import agui from @ably/ai-transport/ag-ui and pass it to createTransport. The codec works with version 1 of the @ag-ui/core package.

import { createTransport } from '@ably/ai-transport';
import { agui } from '@ably/ai-transport/ag-ui';

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

Properties

The codec writes each event type in one of three ways:

• Streams: TEXT_MESSAGE_CONTENT, TOOL_CALL_ARGS and REASONING_MESSAGE_CONTENT each stream into one message. The first delta of a stream publishes the message, later deltas append to it, and the matching *_END event ends the stream.
• MESSAGES_SNAPSHOT is not published, and an inbound one does not decode to an event.
• Every other event is a plain publish with the whole event as the message's data field.
• A run-input event is a plain publish with its input as the message's data field.

The codec does not write any headers. An agent can carry the runId on every message of a run by passing it in the pipe() headers.

Events that change on the wire

Some events reach a subscriber in a different form from the one the agent piped:

• The codec does not publish MESSAGES_SNAPSHOT, and an inbound snapshot does not decode to an event. A snapshot holds the whole conversation, so it grows past the Ably message size limit, and AG-UI's reducer treats a snapshot as the complete message list, so it would drop every message the snapshot leaves out. Your application can load the conversation from its own store instead.
• RUN_STARTED travels with input.messages empty, for the same reason. Its other input fields travel as they are.
• History and a client that joins late read one delta event per stream, with the text joined. That event carries the timestamp and metadata of the last delta only.
• The codec strips rawEvent from every event. The field carries the provider event that the AG-UI event was translated from, so it repeats the event's content. AG-UI's reducer does not read it.
• Every *_CHUNK event is its own message, and the codec does not append to it. A chunk stream does not have an end event, so Ably could deliver an appended chunk after the event that followed it. An agent that wants one message per stream can emit the START, CONTENT and END events instead.
• The codec does not check sizes, so a large STATE_SNAPSHOT, ACTIVITY_SNAPSHOT, TOOL_CALL_RESULT or RUN_FINISHED fails its publish, and pipe() rejects with PipeFailed.

Join at a run boundary

AG-UI's reducer rejects a stream that does not start with RUN_STARTED, and its state deltas patch the state the client already holds. A client needs to join the channel at a run boundary, where the first event the AG-UI reducer receives is a RUN_STARTED. If you are not using the AG-UI reducer to create messages from the event stream then this doesn't apply.

pipe() resolves with the serial of the last publish of the run. If you store that serial with the run's messages and state, a client can replay the channel from it with fromSerial, and receive the next run from its RUN_STARTED event.

Create an AG-UI codec

Build an AG-UI codec, with its own decoder table.

import { createAGUICodec } from '@ably/ai-transport/ag-ui';

const codec = createAGUICodec();

Returns

Codec<AGUIEvent>. The codec, for events of the following type:

A run-input message whose data does not have threadId and runId as strings, and messages, tools and context as arrays, fails to decode. It reaches a subscriber with event set to undefined.

Example

The client sends the user's message as a run-input event. The agent pipes its AG-UI event stream with the runId header, and the client appends each text delta to the message it renders.

// Client code.
await transport.send({
  type: 'run-input',
  input: { threadId, runId, messages: [userMessage], tools: [], context: [] },
});

transport.subscribe(({ event }) => {
  if (event?.type === 'TEXT_MESSAGE_CONTENT') appendText(event.messageId, event.delta);
});

// Agent code. events is your agent's AG-UI event stream, as an async iterable.
await transport.pipe(events, { signal: req.signal, headers: { runId } });