OpenAI

The OpenAI codec carries the OpenAI Responses API's stream events over an Ably channel, plus an input event for the input items a client sends. The agent pipes the Responses stream from the openai package directly into the adaptor, which encodes the stream into Ably messages, and clients decode the messages back into the same stream of LLM events.

Import openaiAdaptor and openaiDecoder from @ably/ai-transport/openai. The agent publishes with an agent adaptor, and the client reads with a decoder. The codec works with the openai package at version 6.

JavaScript

import { openaiAdaptor, openaiDecoder } from '@ably/ai-transport/openai';

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

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

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

Streams: the first *.delta event of a stream creates a single message, and later deltas append to that same message, accumulating all the deltas into a single Ably message.
response.created, response.in_progress and response.queued are not published, because they do not carry any output.
Every other event is a plain publish with the whole event as the message's data field. The input event is a plain publish with its items as the message's data field.

The Responses API duplicates content in the *.done events: these events carry the full streamed data a second time. The codec drops the duplicated fields from the done events, preferring the streamed deltas that were already sent over the channel.

Create an OpenAI adaptor

openaiAdaptor(channel: ChannelWriter, options?: AgentAdaptorOptions): AgentAdaptor<OpenAIEvent>

Create an agent adaptor that publishes OpenAI events on channel.

JavaScript

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

const adaptor = openaiAdaptor(channel);

Parameters

channelrequiredChannelWriter

The channel to publish on. A RealtimeChannel satisfies it. See createAgentAdaptor.

optionsoptionalAgentAdaptorOptions

Default headers and a logger. See createAgentAdaptor.

Returns

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

ResponseStreamEventResponseStreamEvent

Every stream event the Responses API produces, as the openai package declares it.

inputOpenAIInput

The input items a client sends: a user message to start a turn, or a function_call_output that answers a function call.

Create an OpenAI decoder

openaiDecoder(options?: DecoderOptions): Decoder<OpenAIEvent>

Create a decoder that decodes what an OpenAI adaptor publishes.

JavaScript

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

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

Parameters

optionsoptionalDecoderOptions

A logger. See createDecoder.

Returns

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

Example

The client POSTs the user's message to the agent, which publishes it as an input event. The agent pipes the Responses stream, and the client appends each output text delta to the response in its UI.

JavaScript

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

const items = [{ role: 'user', content: 'What is the weather?' }];
await fetch('/api/chat', { method: 'POST', body: JSON.stringify({ channelName: 'conversations:42', items }) });

JavaScript

// Agent code.
import OpenAI from 'openai';

await adaptor.send({ type: 'input', items });
const client = new OpenAI();
const stream = await client.responses.create({ model: 'gpt-4o-mini', input: items, stream: true });

await adaptor.pipe(stream, { signal: req.signal });