Get started with the Vercel AI SDK

// Agent code. This runs on your own infrastructure, so it holds the API key.
import { anthropic } from '@ai-sdk/anthropic';
import { convertToModelMessages, streamText, toUIMessageStream } from 'ai';
import { createRealtimeClient, ErrorInfo } from '@ably/pubsub-server';
import { ErrorCode } from '@ably/ai-transport';
import { vercelAdaptor } from '@ably/ai-transport/vercel';
import { loadConversation, recordMessages } from '../../store';

const ably = createRealtimeClient({ key: process.env.ABLY_API_KEY });

export async function POST(req) {
  const { channelName, message } = await req.json();
  const headers = { requestId: message.id };
  const responseId = crypto.randomUUID();

  const adaptor = vercelAdaptor(ably.channels.get(channelName));

  // Publish the user's message to the channel, so every subscribed client shows it.
  const { serial: promptSerial } = await adaptor.send({ type: 'user-message', message }, { headers });
  recordMessages(channelName, [message], promptSerial);
  const { messages } = loadConversation(channelName);

  try {
    const result = streamText({
      model: anthropic('claude-sonnet-4-20250514'),
      messages: await convertToModelMessages(messages),
      abortSignal: req.signal,
    });

    let response;
    const { serial } = await adaptor.pipe(
      toUIMessageStream({
        stream: result.stream,
        originalMessages: messages,
        generateMessageId: () => responseId,
        onEnd: ({ responseMessage }) => {
          response = responseMessage;
        },
      }),
      { signal: req.signal, headers },
    );

    if (response && serial) recordMessages(channelName, [response], serial);
    return new Response(null, { status: 204 });
  } catch (error) {
    // Best effort: publishing the abort chunk can fail for the same reason the pipe did.
    await adaptor.send({ type: 'abort' }, { headers }).catch((sendError) => console.error('abort publish failed', sendError));
    if (error instanceof ErrorInfo && error.code === ErrorCode.OperationCancelled) return new Response(null, { status: 204 });
    throw error;
  }
}

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, useMemo, useState } from 'react';
import { vercelDecoder } from '@ably/ai-transport/vercel';
import { usePubSubClient } from '@ably/pubsub-device/react';
import { createMerge } from './merge';

export function Chat({ channelName }) {
  const client = usePubSubClient();
  const [messages, setMessages] = useState([]);
  const [input, setInput] = useState('');

  // Put a message in the list by id, so each fuller state of a response replaces the last one.
  const upsert = useCallback((message) => {
    setMessages((prev) =>
      prev.some((m) => m.id === message.id) ? prev.map((m) => (m.id === message.id ? message : m)) : [...prev, message],
    );
  }, []);
  const merge = useMemo(() => createMerge(upsert), [upsert]);

  const apply = useCallback(
    ({ event, message }) => {
      if (event === undefined) return;
      if (event.type === 'user-message') upsert(event.message);
      else merge(event, message.extras?.headers?.requestId);
    },
    [merge, upsert],
  );

  useEffect(() => {
    let channel;
    let cancelled = false;
    const decoder = vercelDecoder();
    decoder.onEvent(apply);

    (async () => {
      const res = await fetch(`/api/chat?channel=${encodeURIComponent(channelName)}`);
      const stored = await res.json();
      if (cancelled) return;
      setMessages(stored.messages);
      // Rewind applies when the channel attaches, so set it before subscribing.
      channel = client.channels.get(channelName, stored.serial ? { params: { rewind: stored.serial } } : undefined);
      await channel.subscribe(decoder.receive);
    })();

    return () => {
      cancelled = true;
      channel?.unsubscribe(decoder.receive);
      // Detach, so the channel rewinds again when the effect runs again.
      channel?.detach().catch(() => {});
    };
  }, [client, channelName, apply]);

  const send = async (e) => {
    e.preventDefault();
    const text = input.trim();
    if (!text) return;
    setInput('');
    const message = { id: crypto.randomUUID(), role: 'user', parts: [{ type: 'text', text }] };
    upsert(message);
    await fetch('/api/chat', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ channelName, message }) });
  };

  return (
    <div>
      {messages.map((m) => (
        <p key={m.id}>
          <strong>{m.role}:</strong> {m.parts.map((p) => (p.type === 'text' ? p.text : '')).join('')}
        </p>
      ))}
      <form onSubmit={send}>
        <input value={input} onChange={(e) => setInput(e.target.value)} />
        <button type="submit">Send</button>
      </form>
    </div>
  );
}

// Client code.
import { readUIMessageStream } from 'ai';

export function createMerge(onMessage) {
  const responses = new Map();

  return (event, requestId) => {
    if (event.type === 'start') {
      if (event.messageId) onMessage({ id: event.messageId, role: 'assistant', parts: [] });
      const stream = new ReadableStream({ start: (controller) => responses.set(requestId, controller) });
      (async () => {
        for await (const message of readUIMessageStream({ stream })) onMessage(message);
      })();
    }
    responses.get(requestId)?.enqueue(event);
    if (event.type === 'finish' || event.type === 'abort') {
      responses.get(requestId)?.close();
      responses.delete(requestId);
    }
  };
}

// Stand-in for your own database. Replace both functions with real queries.
const conversations = new Map();

export function loadConversation(channelName) {
  return conversations.get(channelName) ?? { messages: [] };
}

export function recordMessages(channelName, messages, serial) {
  const current = conversations.get(channelName) ?? { messages: [] };
  const merged = [...current.messages];
  for (const message of messages) {
    const at = merged.findIndex((m) => m.id === message.id);
    if (at === -1) merged.push(message);
    else merged[at] = message;
  }
  conversations.set(channelName, { messages: merged, serial: serial ?? current.serial });
}

'use client';

// Client code.
import { useEffect, useState } from 'react';
import { createClient } from '@ably/pubsub-device';
import { PubSubClientProvider } from '@ably/pubsub-device/react';
import { Chat } from './chat';

const channelName = 'conversations:demo';

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

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

  if (!client) return null;

  return (
    <PubSubClientProvider client={client}>
      <Chat channelName={channelName} />
    </PubSubClientProvider>
  );
}