Chapter 22 · Capstone 3: Full-Stack Streaming Chat Application

Chapter goals:

• Connect the full链路: server-side streamText + UI Message Stream + client-side useChat
• Master the server-side usage of toUIMessageStream / createUIMessageStreamResponse
• Connect to a custom backend with DefaultChatTransport
• Implement the key要点 for message persistence and历史 session恢复

22.1 Goal and Tech Stack

We will deliver a production-ready streaming chat application:

Layer	Technology	Responsibility
Frontend	React + useChat (@ai-sdk/react)	Message rendering, sending, streaming updates
Transport	UIMessage Stream Protocol (SSE)	Structured message stream
Server	Node.js + streamText	Model calls and stream transcoding
Storage	Any KV/DB (example uses in-memory Map)	Session persistence

The core data structure is UIMessage: a structured message with a parts array (text, tool calls, etc. are all parts); the frontend and backend exchange data uniformly using it.

22.2 Server: streamText and UI Message Stream

Start with a原生 Node.js HTTP server implementing the chat endpoint (in a Next.js project, put the same logic in a route handler):

import {
  streamText,
  toUIMessageStream,
  convertToModelMessages,
  createUIMessageStreamResponse,
  type UIMessage,
} from 'ai';
import { getLanguageModel } from './provider'; // Compatible with AI Gateway and custom OpenAI-compatible provider, see Chapter 20
import { createServer } from 'node:http';

export async function handleChat(req: Request): Promise<Response> {
  // 1. Frontend sends UIMessage[] (structured messages with parts)
  const { messages }: { messages: UIMessage[] } = await req.json();

  // 2. UIMessage → ModelMessage (strip UI-specific fields)
  const modelMessages = await convertToModelMessages(messages);

  const result = streamText({
    model: getLanguageModel(),
    instructions: 'You are a helpful assistant.',
    messages: modelMessages,
  });

  // 3. Transcode the model output stream to UI Message Stream and wrap as Response
  return createUIMessageStreamResponse({
    stream: toUIMessageStream({ stream: result.stream }),
  });
}

createServer(async (req, res) => {
  if (req.method === 'POST' && req.url === '/api/chat') {
    const response = await handleChat(
      new Request('http://localhost/api/chat', {
        method: 'POST',
        headers: { 'content-type': 'application/json' },
        body: JSON.stringify({
          messages: [], // Initial empty messages
        }),
      }),
    );
    res.writeHead(response.status, Object.fromEntries(response.headers));
    response.body?.pipeTo(new WritableStream({
      write(chunk) { res.write(chunk); },
      close() { res.end(); },
    }));
  } else {
    res.writeHead(404);
    res.end();
  }
}).listen(3000);

22.3 Frontend: useChat Integration

'use client';

import { useChat } from '@ai-sdk/react';
import { DefaultChatTransport } from 'ai';
import { useState } from 'react';

export default function Chat() {
  const { messages, sendMessage, status, stop } = useChat({
    transport: new DefaultChatTransport({
      api: 'http://localhost:3000/api/chat',
    }),
  });
  const [input, setInput] = useState('');

  return (
    <div className="chat-container">
      <div className="messages">
        {messages.map(message => (
          <div key={message.id} className={`message ${message.role}`}>
            <div className="role">{message.role === 'user' ? 'You' : 'AI'}</div>
            <div className="content">
              {message.parts.map((part, index) => {
                if (part.type === 'text') {
                  return <span key={index}>{part.text}</span>;
                }
                return null;
              })}
            </div>
          </div>
        ))}
      </div>

      <div className="controls">
        {status === 'streaming' && (
          <button onClick={stop}>Stop</button>
        )}
        <form onSubmit={e => {
          e.preventDefault();
          if (input.trim()) {
            sendMessage({ text: input });
            setInput('');
          }
        }}>
          <input
            value={input}
            onChange={e => setInput(e.target.value)}
            disabled={status !== 'ready'}
            placeholder="Type a message..."
          />
          <button type="submit" disabled={status !== 'ready'}>Send</button>
        </form>
      </div>
    </div>
  );
}

22.4 Message Persistence and History Recovery

Store Messages on the Server

Extend the server to save and load messages:

import { createServer } from 'node:http';
import { MapStore } from './store'; // In-memory store

const store = new MapStore();

export async function handleChat(req: Request): Promise<Response> {
  const { messages, sessionId } = await req.json();

  // Load history
  const history = await store.getMessages(sessionId);
  const allMessages = [...history, ...messages];

  const modelMessages = await convertToModelMessages(allMessages);

  const result = streamText({
    model: getLanguageModel(),
    messages: modelMessages,
  });

  // Save after completion
  result.then(async final => {
    await store.saveMessage(sessionId, final.messages);
  });

  return createUIMessageStreamResponse({
    stream: toUIMessageStream({ stream: result.stream }),
  });
}

Client-Side Session ID Management

import { useState, useEffect } from 'react';

export default function Chat() {
  const [sessionId] = useState(() => `session-${Date.now()}-${Math.random().toString(36).slice(2)}`);

  const { messages, sendMessage, status, stop } = useChat({
    id: sessionId,
    transport: new DefaultChatTransport({
      api: 'http://localhost:3000/api/chat',
    }),
  });
  // ...
}

Recovery: Loading History on Re-visit

export async function handleGetSession(req: Request, sessionId: string): Promise<Response> {
  const messages = await store.getMessages(sessionId);
  return new Response(JSON.stringify({ messages }), {
    headers: { 'content-type': 'application/json' },
  });
}

useEffect(() => {
  fetch(`http://localhost:3000/api/session/${sessionId}`)
    .then(res => res.json())
    .then(data => {
      // Pre-populate messages
      setMessages(data.messages);
    });
}, [sessionId]);

22.5 Advanced: Custom Transport for Non-HTTP Backends

For backends that do not use HTTP (e.g., WebSocket, gRPC), implement a custom transport:

import { ChatTransport, UIMessage } from 'ai';

export class WebSocketChatTransport implements ChatTransport {
  private ws: WebSocket;

  constructor(private url: string) {
    this.ws = new WebSocket(url);
  }

  async sendMessage(options: {
    messages: UIMessage[];
    onRequest: (request: Request) => void;
  }): Promise<void> {
    this.ws.send(JSON.stringify({ messages: options.messages }));
    // Listen for responses...
  }

  abort(): void {
    this.ws.close();
  }
}

const { messages } = useChat({
  transport: new WebSocketChatTransport('ws://localhost:8080/ws'),
});

22.6 Error Handling and Retry

const { messages, sendMessage, status, error, regenerate } = useChat({
  transport: new DefaultChatTransport({
    api: 'http://localhost:3000/api/chat',
    onError: (error) => {
      console.error('Chat error:', error);
      // Show error to user
    },
  }),
});

// Retry on error
{error && (
  <button onClick={() => regenerate()}>Regenerate</button>
)}

22.7 Performance Optimization

Throttle UI Updates

const { messages } = useChat({
  throttle: 100, // Update UI at most every 100ms
});

Lazy Load History

For long conversations, load history in batches:

const PAGE_SIZE = 20;
const start = Math.max(0, messages.length - PAGE_SIZE);
const recentMessages = messages.slice(start);

Streaming Protocol Selection

Choose between text stream and data stream based on needs:

// Text stream: simpler, only text
import { TextStreamChatTransport } from 'ai';
const { messages } = useChat({
  transport: new TextStreamChatTransport({ api: '/api/chat' }),
});

// Data stream (default): supports tool calls, sources, data parts
import { DefaultChatTransport } from 'ai';
const { messages } = useChat({
  transport: new DefaultChatTransport({ api: '/api/chat' }),
});

22.8 Deployment Checklist

• [ ] Server runs on a persistent process (Node.js, Docker, PM2);
• [ ] Environment variables for model API keys are injected securely;
• [ ] Session storage is persistent (database, not in-memory Map for production);
• [ ] CORS headers are configured for the frontend domain;
• [ ] Rate limiting is applied to prevent abuse;
• [ ] Error logging is set up (Sentry, Prometheus, etc.);
• [ ] Health check endpoint (/healthz) returns 200.

22.9 Chapter Summary

• Full-stack链路: frontend useChat → HTTP POST → server streamText → toUIMessageStream → createUIMessageStreamResponse → SSE → frontend renders parts;
• convertToModelMessages strips UI fields for the model; toUIMessageStream re-adds them for the client;
• Persistence: server stores messages by sessionId; client passes sessionId; recovery loads history on re-visit;
• Custom transports enable non-HTTP backends (WebSocket, gRPC);
• Performance: throttle UI updates, lazy-load history, choose the right stream protocol.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What does convertToModelMessages do?

2. What is the role of createUIMessageStreamResponse?

3. Why is sessionId important for message persistence?

4. When should you use TextStreamChatTransport instead of DefaultChatTransport?

🛠️ Hands-On Exercises

1. Build the full-stack streaming chat app: Node.js server with streamText + toUIMessageStream, React frontend with useChat + DefaultChatTransport. Verify streaming works end-to-end.
2. Add session persistence: save messages to a JSON file keyed by sessionId; verify that refreshing the page restores the conversation.
3. Add error handling: when the server returns an error, show a retry button; verify regenerate() resends the last message.

Capstone 3 complete. Course complete!