AG-UI

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 adaptor, 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 agUiAdaptor and agUiDecoder from @ably/ai-transport/ag-ui. The agent publishes with an agent adaptor, and the client reads with a decoder. The codec works with version 1 of the @ag-ui/core package.

import { agUiAdaptor, agUiDecoder } from '@ably/ai-transport/ag-ui';

// Agent code. channel is a channel from your Ably Pub/Sub client.
const adaptor = agUiAdaptor(channel);

// Client code.
const decoder = agUiDecoder();
await channel.subscribe(decoder.receive);

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 database 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 rewind the channel to it, and receive the next run from its RUN_STARTED event.

Create an AG-UI adaptor

Create an agent adaptor that publishes AG-UI events on channel.

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

const adaptor = agUiAdaptor(channel);

Parameters

Returns

AgentAdaptor<AGUIEvent>. The adaptor, for events of the following type:

Create an AG-UI decoder

Create a decoder that decodes what an AG-UI adaptor publishes.

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

const decoder = agUiDecoder();
await channel.subscribe(decoder.receive);

Parameters

Returns

Decoder<AGUIEvent>. The decoder, for the events the adaptor publishes.

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 POSTs the user's message to the agent, which publishes it 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 in its UI.

// Client code.
decoder.onEvent(({ event }) => {
  if (event?.type === 'TEXT_MESSAGE_CONTENT') appendText(event.messageId, event.delta);
});
await channel.subscribe(decoder.receive);

const input = { threadId, runId, messages: [userMessage], tools: [], context: [] };
await fetch('/api/agent', { method: 'POST', body: JSON.stringify({ channelName: 'conversations:42', input }) });

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