Get started with OpenAI

// Agent code. This runs on your own infrastructure, so it holds the API key.
import * as Ably from 'ably';
import OpenAI from 'openai';
import { createTransport } from '@ably/ai-transport';
import { openai } from '@ably/ai-transport/openai';
import { loadConversation, recordItems } from '../../store';

const ably = new Ably.Realtime({ key: process.env.ABLY_API_KEY });
const client = new OpenAI();

// Pass each event to the pipe, and keep the finished output items for your store.
async function* keepOutput(stream, onOutput) {
  for await (const event of stream) {
    if (event.type === 'response.completed') onOutput(event.response.output);
    yield event;
  }
}

export async function POST(req) {
  const { channelName, items } = await req.json();
  const { items: history } = loadConversation(channelName);

  const transport = createTransport({ client: ably, channelName, codec: openai });

  // Optionally: publish the user's input to the channel
  await transport.send({ type: 'input', items });

  try {
    const stream = await client.responses.create(
      { model: 'gpt-5.5', input: [...history, ...items], stream: true },
      { signal: req.signal },
    );

    let output = [];
    const { serial } = await transport.pipe(keepOutput(stream, (finished) => (output = finished)), {
      signal: req.signal,
    });

    recordItems(channelName, [...items, ...output], serial);
    return new Response(null, { status: 204 });
  } catch (error) {
    // The client closed the request, so the model call or the pipe was cancelled.
    if (req.signal.aborted) return new Response(null, { status: 204 });
    throw error;
  } finally {
    await transport.close();
  }
}

export function GET(req) {
  const channelName = new URL(req.url).searchParams.get('channel');
  return Response.json(loadConversation(channelName));
}

'use client';

// Client code. The browser fetches a token, so it never holds an API key.
import { useCallback, useEffect, useRef, useState } from 'react';
import { fromSerial, untilEvent } from '@ably/ai-transport';
import { useTransport } from '@ably/ai-transport/react';
import { mergeEvent, toEntries } from './merge';

// With no stored serial, nothing is stored yet, so replay the whole channel.
const allHistory = untilEvent(() => false);

export function Chat({ channelName }) {
  const { transport } = useTransport();
  const [entries, setEntries] = useState([]);
  const [input, setInput] = useState('');
  const lastSeen = useRef(undefined);

  const apply = useCallback(({ event, message }) => {
    if (lastSeen.current === undefined || message.serial > lastSeen.current) lastSeen.current = message.serial;
    if (event !== undefined) setEntries((prev) => mergeEvent(prev, event));
  }, []);

  useEffect(() => {
    if (!transport) return;
    let unsubscribe;
    let cancelled = false;
    const subscribeFrom = (serial) => {
      unsubscribe?.();
      unsubscribe = transport.subscribe(apply, { history: { replay: serial ? fromSerial(serial) : allHistory } });
    };

    (async () => {
      const res = await fetch(`/api/chat?channel=${encodeURIComponent(channelName)}`);
      const stored = await res.json();
      if (cancelled) return;
      setEntries(toEntries(stored.items));
      lastSeen.current = stored.serial;
      subscribeFrom(stored.serial);
    })();

    const off = transport.on('discontinuity', () => subscribeFrom(lastSeen.current));
    return () => {
      cancelled = true;
      off();
      unsubscribe?.();
    };
  }, [transport, channelName, apply]);

  const send = async (e) => {
    e.preventDefault();
    const text = input.trim();
    if (!text || !transport) return;
    setInput('');
    const items = [{ role: 'user', content: text }];
    setEntries((prev) => [...prev, ...toEntries(items)]);
    await fetch('/api/chat', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ channelName, items }) });
  };

  return (
    <div>
      {entries.map((entry) => (
        <p key={entry.id}>
          <strong>{entry.role}:</strong> {entry.text}
        </p>
      ))}
      <form onSubmit={send}>
        <input value={input} onChange={(e) => setInput(e.target.value)} />
        <button type="submit" disabled={!transport}>Send</button>
      </form>
    </div>
  );
}

// Client code.
const textOf = (content) =>
  typeof content === 'string' ? content : content.map((part) => part.text ?? '').join('');

export function toEntries(items) {
  return items
    .filter((item) => item.role === 'user' || item.role === 'assistant')
    .map((item) => ({ id: item.id ?? crypto.randomUUID(), role: item.role, text: textOf(item.content) }));
}

export function mergeEvent(entries, event) {
  if (event.type !== 'response.output_text.delta') return entries;
  if (!entries.some((entry) => entry.id === event.item_id)) {
    return [...entries, { id: event.item_id, role: 'assistant', text: event.delta }];
  }
  return entries.map((entry) => (entry.id === event.item_id ? { ...entry, text: entry.text + event.delta } : entry));
}

'use client';

// Client code.
import { useEffect, useState } from 'react';
import * as Ably from 'ably';
import { AblyProvider } from 'ably/react';
import { TransportProvider } from '@ably/ai-transport/react';
import { openai } from '@ably/ai-transport/openai';
import { Chat } from './chat';

const channelName = 'conversations:demo';

export default function Page() {
  const [client, setClient] = useState();

  useEffect(() => {
    const realtime = new Ably.Realtime({ authUrl: '/api/auth/token' });
    setClient(realtime);
    return () => realtime.close();
  }, []);

  if (!client) return null;

  return (
    <AblyProvider client={client}>
      <TransportProvider channelName={channelName} codec={openai}>
        <Chat channelName={channelName} />
      </TransportProvider>
    </AblyProvider>
  );
}