Get started with OpenAI

import { after } from 'next/server';
import OpenAI from 'openai';
import * as Ably from 'ably';
import { createAgentSession, Invocation } from '@ably/ai-transport';
import { ResponsesCodec, toResponsesInput } from '@ably/ai-transport/openai';

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

export async function POST(req) {
  const invocation = Invocation.fromJSON(await req.json());

  const session = createAgentSession({
    client: ably,
    channelName: invocation.sessionName,
    codec: ResponsesCodec,
  });

  await session.connect();

  // No identity is pinned: this run is not retried, so a generated run id and
  // invocation id are correct.
  const run = session.createRun(invocation, {}, { signal: req.signal });

  // Load the full conversation from history before starting the run.
  // run.start() needs to have seen this run's triggering input; draining it
  // from history means run.start() resolves at once instead of waiting for
  // that input to arrive live on the channel.
  while (run.view.hasOlder()) {
    await run.view.loadOlder();
  }

  // Start before returning the response. A continuation re-keys run.runId from
  // the triggering input's headers, so the id is only final once start resolves.
  await run.start();

  // Flatten the conversation into the Responses `input` array. Each stored
  // message already holds valid model input, so no conversion is needed.
  const input = toResponsesInput(run.view.getMessages().map(({ message }) => message));

  after(async () => {
    try {
      const stream = await openai.responses.create(
        { model: 'gpt-5.5', input, stream: true },
        { signal: run.abortSignal },
      );

      // The Responses stream is an async iterable, and each raw event is valid
      // codec output, so pipe it straight through. run.pipe watches
      // run.abortSignal, so a cancel ends the run with no extra handling here.
      const { reason } = await run.pipe(stream);

      await run.end({ reason });
    } catch (err) {
      await run.end({ reason: 'error' });
      throw err;
    } finally {
      await session.end();
    }
  });

  return Response.json({ runId: run.runId, invocationId: run.invocationId });
}

'use client';

import { useState } from 'react';
import { useClientSession, useView } from '@ably/ai-transport/react';
import { ResponsesCodec } from '@ably/ai-transport/openai';

async function wakeAgent(run) {
  await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(run.toInvocation().toJSON()),
  });
}

// Flatten an OpenAI message's items into rendered text.
function messageText(message) {
  let text = '';
  for (const item of message.items) {
    if (item.type !== 'message') continue;
    for (const part of item.content) {
      if (part.type === 'output_text' || part.type === 'input_text') text += part.text;
    }
  }
  return text;
}

export function Chat() {
  const [input, setInput] = useState('');

  const { session } = useClientSession();
  const view = useView({ limit: 30 });
  const { messages, runOf } = view;

  // The Stop button shows only while the latest message's Run is streaming.
  const latestRun = runOf(messages.at(-1)?.codecMessageId ?? '');
  const isStreaming = latestRun?.status === 'active';

  const handleSubmit = async (e) => {
    e.preventDefault();
    if (!input.trim()) return;
    const text = input;
    setInput('');
    const run = await view.send(
      ResponsesCodec.createUserMessage({
        role: 'user',
        items: [{ type: 'message', role: 'user', content: [{ type: 'input_text', text }] }],
      }),
    );
    await wakeAgent(run);
  };

  const stop = () => {
    if (latestRun) void session.cancel(latestRun.runId);
  };

  return (
    <div>
      {messages.map(({ codecMessageId, message }) => (
        <div key={codecMessageId}>
          <strong>{message.role}:</strong> {messageText(message)}
        </div>
      ))}
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={(e) => setInput(e.target.value)} placeholder="Type a message..." />
        {isStreaming ? (
          <button type="button" onClick={stop}>Stop</button>
        ) : (
          <button type="submit">Send</button>
        )}
      </form>
    </div>
  );
}

'use client';

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

function Providers({ children }) {
  const [client, setClient] = useState(null);

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

  if (!client) return null;
  return <AblyProvider client={client}>{children}</AblyProvider>;
}

export default function Page() {
  const channelName = 'conversations:my-chat-session';
  return (
    <Providers>
      <ClientSessionProvider channelName={channelName} codec={ResponsesCodec}>
        <Chat />
      </ClientSessionProvider>
    </Providers>
  );
}