Chapter 17 · Streaming Data, Metadata, and Message Persistence

Chapter goals:

• Master createUIMessageStream + writer.write to stream custom data to the frontend (data parts, sources, transient parts)
• Understand data part reconciliation (same-ID update) mechanism and onData callback handling
• Learn to attach per-message metadata (timestamps, model info, token usage) at the message level
• Implement chat message persistence: store/load/validate UIMessages, and server-side message ID generation

17.1 Why Stream Custom Data Is Needed

Beyond the model's response, you often need to send additional data to the frontend. For example: sending status updates, echoing back a message id after database storage, or referencing content the language model is currently discussing.

AI SDK provides several helper functions to stream additional data to the client and attach it to the UIMessage parts array:

• createUIMessageStream: Creates a data stream;
• createUIMessageStreamResponse: Creates a streaming response object;
• pipeUIMessageStreamToResponse: Pipes a data stream to a server response object.

Data is transmitted as Server-Sent Events within the response stream.

Type-Safe Data Stream Definition

First define a custom message type with data part schemas:

import { UIMessage } from 'ai';

// Define your custom message type with data part schemas
export type MyUIMessage = UIMessage<
  never, // metadata type
  {
    weather: {
      city: string;
      weather?: string;
      status: 'loading' | 'success';
    };
    notification: {
      message: string;
      level: 'info' | 'warning' | 'error';
    };
  } // data parts type
>;

17.2 Streaming Data from the Server

In the server-side route, create a UIMessageStream and pass it to createUIMessageStreamResponse for return:

import {
  convertToModelMessages,
  createUIMessageStream,
  createUIMessageStreamResponse,
  streamText,
  toUIMessageStream,
} from 'ai';
import { createGateway } from 'ai';
import type { MyUIMessage } from '@/ai/types';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

export async function POST(req: Request) {
  const { messages } = await req.json();

  const stream = createUIMessageStream<MyUIMessage>({
    execute: ({ writer }) => {
      // 1. Send initial status (transient - won't be added to message history)
      writer.write({
        type: 'data-notification',
        data: { message: 'Processing your request...', level: 'info' },
        transient: true, // This part won't be added to message history
      });

      // 2. Send sources (useful for RAG use cases)
      writer.write({
        type: 'source',
        value: {
          type: 'source',
          sourceType: 'url',
          id: 'source-1',
          url: 'https://weather.com',
          title: 'Weather Data Source',
        },
      });

      // 3. Send data parts with loading state
      writer.write({
        type: 'data-weather',
        id: 'weather-1',
        data: { city: 'San Francisco', status: 'loading' },
      });

      const result = streamText({
        model: gateway('openai/gpt-5'),
        messages: await convertToModelMessages(messages),
        onEnd() {
          // 4. Update the same data part (reconciliation)
          writer.write({
            type: 'data-weather',
            id: 'weather-1', // Same ID = update existing part
            data: {
              city: 'San Francisco',
              weather: 'sunny',
              status: 'success',
            },
          });

          // 5. Send completion notification (transient)
          writer.write({
            type: 'data-notification',
            data: { message: 'Request completed', level: 'info' },
            transient: true, // Won't be added to message history
          });
        },
      });

      writer.merge(toUIMessageStream({ stream: result.stream }));
    },
  });

  return createUIMessageStreamResponse({ stream });
}

💡 Custom backends (e.g., Python/FastAPI) can also stream data per the UI Message Stream protocol.

Three Categories of Streamable Data

Persistent data parts — enter message history, appear in message.parts:

writer.write({
  type: 'data-weather',
  id: 'weather-1', // Optional: enables reconciliation
  data: { city: 'San Francisco', status: 'loading' },
});

Sources — suitable for RAG scenarios to display referenced documents or links (see step 2 above).

Transient (transient) data parts — sent to the client but not added to message history; only accessible via useChat's onData callback:

// server
writer.write({
  type: 'data-notification',
  data: { message: 'Processing...', level: 'info' },
  transient: true, // Won't be added to message history
});

// client
const [notification, setNotification] = useState();

const { messages } = useChat({
  onData: ({ data, type }) => {
    if (type === 'data-notification') {
      setNotification({ message: data.message, level: data.level });
    }
  },
});

Reconciliation: Automatic Update for Same-ID Data Parts

Writing to the same id triggers automatic client-side reconciliation and update. This enables powerful dynamic experiences:

• Collaborative artifacts — real-time updated code, docs, or designs;
• Progressive loading — smooth transition from loading state to final result;
• Real-time status — update progress bars, counters, status indicators;
• Interactive components — UI elements that evolve with user interaction.

Reconciliation happens自动: just use the same id when writing to the stream.

17.3 Handling Streaming Data on the Client

onData Callback

onData is the key callback for handling streaming data, especially for transient parts:

import { useChat } from '@ai-sdk/react';
import type { MyUIMessage } from '@/ai/types';

const { messages } = useChat<MyUIMessage>({
  api: '/api/chat',
  onData: dataPart => {
    // Handle all data parts as they arrive (including transient parts)
    console.log('Received data part:', dataPart);

    // Handle different data part types
    if (dataPart.type === 'data-weather') {
      console.log('Weather update:', dataPart.data);
    }

    // Handle transient notifications (ONLY available here, not in message.parts)
    if (dataPart.type === 'data-notification') {
      showToast(dataPart.data.message, dataPart.data.level);
    }
  },
});

⚠️ Important: Transient data parts are only accessible via the onData callback; they do not appear in message.parts (because they don't enter message history).

Rendering Persistent Data Parts

Filter and render data parts from the message parts array:

const result = (
  <>
    {messages?.map(message => (
      <div key={message.id}>
        {/* Render weather data parts */}
        {message.parts
          .filter(part => part.type === 'data-weather')
          .map((part, index) => (
            <div key={index} className="weather-widget">
              {part.data.status === 'loading' ? (
                <>Getting weather for {part.data.city}...</>
              ) : (
                <>
                  Weather in {part.data.city}: {part.data.weather}
                </>
              )}
            </div>
          ))}

        {/* Render sources */}
        {message.parts
          .filter(part => part.type === 'source')
          .map((part, index) => (
            <div key={index} className="source">
              Source: <a href={part.url}>{part.title}</a>
            </div>
          ))}
      </div>
    ))}
  </>
);

Typical application scenarios: RAG apps (streaming source documents), real-time status (loading progress), collaboration tools, usage analytics, temporary notifications.

17.4 Message Metadata: Per-Message Metadata

Message metadata lets you attach custom information at the message level, suitable for tracking timestamps, model info, token usage, user context, and other "about the entire message" data — this complements data parts that构成 the message content.

Define Metadata Type

First define the metadata schema for type safety:

import { UIMessage } from 'ai';
import { z } from 'zod';

// Define your metadata schema
export const messageMetadataSchema = z.object({
  createdAt: z.number().optional(),
  model: z.string().optional(),
  totalTokens: z.number().optional(),
});

export type MessageMetadata = z.infer<typeof messageMetadataSchema>;

// Create a typed UIMessage
export type MyUIMessage = UIMessage<MessageMetadata>;

Send Metadata from the Server

Use the messageMetadata callback of toUIMessageStream to send metadata at different streaming stages:

import {
  convertToModelMessages,
  createUIMessageStreamResponse,
  streamText,
  toUIMessageStream,
} from 'ai';
import { createGateway } from 'ai';
import type { MyUIMessage } from '@/types';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

export async function POST(req: Request) {
  const { messages }: { messages: MyUIMessage[] } = await req.json();

  const result = streamText({
    model: gateway('openai/gpt-5'),
    messages: await convertToModelMessages(messages),
  });

  return createUIMessageStreamResponse({
    stream: toUIMessageStream({
      stream: result.stream,
      originalMessages: messages, // pass this in for type-safe return objects
      messageMetadata: ({ part }) => {
        // Send metadata when streaming starts
        if (part.type === 'start') {
          return {
            createdAt: Date.now(),
            model: 'your-model-id',
          };
        }

        // Send additional metadata when streaming completes
        if (part.type === 'finish') {
          return {
            totalTokens: part.totalUsage.totalTokens,
          };
        }
      },
    }),
  });
}

💡 To get type-safe return objects in messageMetadata, pass the originalMessages parameter typed as your UIMessage type.

Access Metadata on the Client

Read via the message.metadata property:

'use client';

import { useChat } from '@ai-sdk/react';
import { DefaultChatTransport } from 'ai';
import type { MyUIMessage } from '@/types';

export default function Chat() {
  const { messages } = useChat<MyUIMessage>({
    transport: new DefaultChatTransport({
      api: '/api/chat',
    }),
  });

  return (
    <div>
      {messages.map(message => (
        <div key={message.id}>
          <div>
            {message.role === 'user' ? 'User: ' : 'AI: '}
            {message.metadata?.createdAt && (
              <span className="text-sm text-gray-500">
                {new Date(message.metadata.createdAt).toLocaleTimeString()}
              </span>
            )}
          </div>

          {/* Render message content */}
          {message.parts.map((part, index) =>
            part.type === 'text' ? <div key={index}>{part.text}</div> : null,
          )}

          {/* Display additional metadata */}
          {message.metadata?.totalTokens && (
            <div className="text-xs text-gray-400">
              {message.metadata.totalTokens} tokens
            </div>
          )}
        </div>
      ))}
    </div>
  );
}

Common uses for metadata: timestamps, model info, token usage, user context, performance metrics (time-to-first-token), quality metrics (finish reason). For data that changes dynamically during generation, use data parts instead.

17.5 Message Persistence

Being able to store and load chat messages is critical for most AI chat applications. This section demonstrates how to combine useChat with streamText to实现 message persistence.

Creating a New Chat

When a user访问s the chat page without a chat ID, create a new chat first then redirect:

import { redirect } from 'next/navigation';
import { createChat } from '@util/chat-store';

export default async function Page() {
  const id = await createChat(); // create a new chat
  redirect(`/chat/${id}`); // redirect to chat page, see below
}

The example chat store uses the file system; real applications should swap in a database or云 storage — the function interfaces are designed for seamless replacement:

import { generateId } from 'ai';
import { existsSync, mkdirSync } from 'fs';
import { writeFile } from 'fs/promises';
import path from 'path';

// Treat chat IDs as opaque tokens before using them in file paths.
const chatIdRegex = /^[A-Za-z0-9_-]+$/;

export async function createChat(): Promise<string> {
  const id = generateId(); // generate a unique chat ID
  await writeFile(getChatFile(id), '[]'); // create an empty chat file
  return id;
}

function getChatFile(id: string): string {
  if (!chatIdRegex.test(id)) {
    throw new Error('Invalid chat ID');
  }

  const chatDir = path.resolve(process.cwd(), '.chats');
  const chatFile = path.resolve(chatDir, `${id}.json`);

  // Defense in depth: keep the resolved file inside the chat directory.
  if (!chatFile.startsWith(`${chatDir}${path.sep}`)) {
    throw new Error('Invalid chat ID');
  }

  if (!existsSync(chatDir)) mkdirSync(chatDir, { recursive: true });
  return chatFile;
}

⚠️ Chat IDs may come from URLs or request bodies; validate them as opaque tokens before用作 file paths; the resolved path check ensures the file cannot escape the .chats directory.

Loading an Existing Chat

import { UIMessage } from 'ai';
import { readFile } from 'fs/promises';

export async function loadChat(id: string): Promise<UIMessage[]> {
  return JSON.parse(await readFile(getChatFile(id), 'utf8'));
}

// ... rest of the file

Server-Side Message Validation

When handling历史 messages containing tool calls, custom metadata, or data parts, validate before sending to the model:

import {
  convertToModelMessages,
  createUIMessageStreamResponse,
  streamText,
  toUIMessageStream,
  UIMessage,
  validateUIMessages,
  tool,
} from 'ai';
import { z } from 'zod';
import { loadChat, saveChat } from '@util/chat-store';
import { dataPartsSchema, metadataSchema } from '@util/schemas';
import { createGateway } from 'ai';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

// Define your tools
const tools = {
  weather: tool({
    description: 'Get weather information',
    inputSchema: z.object({
      location: z.string(),
      units: z.enum(['celsius', 'fahrenheit']),
    }),
    execute: async ({ location, units }) => {
      /* tool implementation */
    },
  }),
  // other tools
};

export async function POST(req: Request) {
  const { message, id } = await req.json();

  // Load previous messages from database
  const previousMessages = await loadChat(id);

  // Append new message to previousMessages messages
  const messages = [...previousMessages, message];

  // Validate loaded messages against
  // tools, data parts schema, and metadata schema
  const validatedMessages = await validateUIMessages({
    messages,
    tools, // Ensures tool calls in messages match current schemas
    dataPartsSchema,
    metadataSchema,
  });

  const result = streamText({
    model: gateway('openai/gpt-5-mini'),
    messages: convertToModelMessages(validatedMessages),
    tools,
  });

  return createUIMessageStreamResponse({
    stream: toUIMessageStream({
      stream: result.stream,
      originalMessages: messages,
      onEnd: ({ messages }) => {
        saveChat({ chatId: id, messages });
      },
    }),
  });
}

Old messages in the database may not match the current schema; handle validation errors gracefully:

import {
  convertToModelMessages,
  streamText,
  validateUIMessages,
  TypeValidationError,
} from 'ai';
import { type MyUIMessage } from '@/types';

export async function POST(req: Request) {
  const { message, id } = await req.json();

  // Load and validate messages from database
  let validatedMessages: MyUIMessage[];

  try {
    const previousMessages = await loadMessagesFromDB(id);
    validatedMessages = await validateUIMessages({
      // append the new message to the previous messages:
      messages: [...previousMessages, message],
      tools,
      metadataSchema,
    });
  } catch (error) {
    if (error instanceof TypeValidationError) {
      // Log validation error for monitoring
      console.error('Database messages validation failed:', error);
      // Could implement message migration or filtering here
      // For now, start with empty history
      validatedMessages = [];
    } else {
      throw error;
    }
  }

  // Continue with validated messages...
}

Displaying the Chat Page and Saving Messages

The page component loads history and passes it to the chat component, which initializes useChat with id and initialMessages:

import { loadChat } from '@util/chat-store';
import Chat from '@ui/chat';

export default async function Page(props: { params: Promise<{ id: string }> }) {
  const { id } = await props.params;
  const messages = await loadChat(id);
  return <Chat id={id} initialMessages={messages} />;
}

'use client';

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

export default function Chat({
  id,
  initialMessages,
}: { id?: string | undefined; initialMessages?: UIMessage[] } = {}) {
  const [input, setInput] = useState('');
  const { sendMessage, messages } = useChat({
    id, // use the provided chat ID
    messages: initialMessages, // load initial messages
    transport: new DefaultChatTransport({
      api: '/api/chat',
    }),
  });

  const handleSubmit = (e: React.FormEvent) => {
    e.preventDefault();
    if (input.trim()) {
      sendMessage({ text: input });
      setInput('');
    }
  };

  // simplified rendering code, extend as needed:
  return (
    <div>
      {messages.map(m => (
        <div key={m.id}>
          {m.role === 'user' ? 'User: ' : 'AI: '}
          {m.parts
            .map(part => (part.type === 'text' ? part.text : ''))
            .join('')}
        </div>
      ))}

      <form onSubmit={handleSubmit}>
        <input
          value={input}
          onChange={e => setInput(e.target.value)}
          placeholder="Type a message..."
        />
        <button type="submit">Send</button>
      </form>
    </div>
  );
}

The save action completes in the onEnd callback of toUIMessageStream — it receives the full message list including the new AI response (UIMessage[]):

import { saveChat } from '@util/chat-store';
import {
  convertToModelMessages,
  createUIMessageStreamResponse,
  streamText,
  toUIMessageStream,
  UIMessage,
} from 'ai';
import { createGateway } from 'ai';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

export async function POST(req: Request) {
  const { messages, chatId }: { messages: UIMessage[]; chatId: string } =
    await req.json();

  const result = streamText({
    model: gateway('openai/gpt-5-mini'),
    messages: await convertToModelMessages(messages),
  });

  return createUIMessageStreamResponse({
    stream: toUIMessageStream({
      stream: result.stream,
      originalMessages: messages,
      onEnd: ({ messages }) => {
        saveChat({ chatId, messages });
      },
    }),
  });
}

📌 Note: useChat's message format (front-end facing, with id, createdAt, etc.) differs from the ModelMessage format. The official recommends storing messages in useChat format.

Server-Side Message ID Generation

By default, user message IDs are generated client-side by useChat, and AI response message IDs are generated server-side by streamText. For persistence scenarios, use IDs that are stable before the message enters storage, to ensure cross-session consistency and avoid ID conflicts on恢复.

Method 1: Use generateMessageId in toUIMessageStream,配合 createIdGenerator() to control the ID format:

import {
  createIdGenerator,
  createUIMessageStreamResponse,
  streamText,
  toUIMessageStream,
} from 'ai';

export async function POST(req: Request) {
  // ...
  const result = streamText({
    // ...
  });

  return createUIMessageStreamResponse({
    stream: toUIMessageStream({
      stream: result.stream,
      originalMessages: messages,
      // Generate consistent server-side IDs for persistence:
      generateMessageId: createIdGenerator({
        prefix: 'msg',
        size: 16,
      }),
      onEnd: ({ messages }) => {
        saveChat({ chatId, messages });
      },
    }),
  });
}

Method 2: Use createUIMessageStream to manually write a start message part with a custom ID:

import {
  generateId,
  streamText,
  createUIMessageStream,
  createUIMessageStreamResponse,
  toUIMessageStream,
} from 'ai';
import { createGateway } from 'ai';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

export async function POST(req: Request) {
  const { messages, chatId } = await req.json();

  const stream = createUIMessageStream({
    execute: ({ writer }) => {
      // Write start message part with custom ID
      writer.write({
        type: 'start',
        messageId: generateId(), // Generate server-side ID for persistence
      });

      const result = streamText({
        model: gateway('openai/gpt-5-mini'),
        messages: await convertToModelMessages(messages),
      });

      writer.merge(
        toUIMessageStream({ stream: result.stream, sendStart: false }),
      ); // omit start message part
    },
    originalMessages: messages,
    onEnd: ({ responseMessage }) => {
      // save your chat here
    },
  });

  return createUIMessageStreamResponse({ stream });
}

For pure client-side apps that don't require persistence, you can also customize the client-side ID generation:

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

const { ... } = useChat({
  generateId: createIdGenerator({
    prefix: 'msgc',
    size: 16,
  }),
  // ...
});

17.6 Chapter Summary

• Use writer.write in createUIMessageStream to send three categories of data: persistent data parts (enter message history), sources (RAG citations), and transient parts (only visible via onData);
• Writing to the same id triggers client-side reconciliation for automatic updates, enabling progress bars, collaborative editing, and other dynamic experiences;
• Message metadata is per-message info (timestamp/model/token usage); sent via the messageMetadata callback of toUIMessageStream; read by the client from message.metadata;
• Persistence three-piece: file/DB存取 createChat/loadChat/saveChat + server-side validateUIMessages to validate历史 messages + onEnd callback to persist;
• For persistence scenarios,务必 use server-generated stable message IDs (generateMessageId or手写 start part).

🧪 随堂测验

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

1. What is the characteristic of transient data parts?

2. How is Data Part Reconciliation triggered?

3. Regarding the division of labor between message metadata and data parts, which is correct?

4. Why does the官方 recommend server-side message ID generation for persistence scenarios?

🛠️ Hands-On Exercises

1. Extend the Chapter 16 chat app: have the server send a data-status progress part (loading → success two states) via createUIMessageStream; use reconciliation on the frontend to render the progress indicator.
2. Attach message metadata (start time + totalTokens) to each AI response; display "served by model X · N tokens" in the bubble corner.
3. Persist chat history to a local JSON file: implement createChat/loadChat/saveChat, add validateUIMessages validation and server-side generateMessageId; the session fully restores on page refresh.

Next: Chapter 18 · Agent Introduction: ToolLoopAgent and Loop Control