Streaming chat ergonomics: typed streaming-query helpers, validators, and known-good AI SDK versions

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## Unreleased
+
+- Add `listUIMessagesWithStreams` / `listMessagesWithStreams` helpers and
+  `vStreamUIMessagesReturnValue` / `vSyncStreamsReturnValue` / `vUIMessage`
+  validators so a from-scratch streaming chat query compiles without
+  reconstructing the paginated + streams return shape (fixes a TS2719 landmine
+  when combining `listUIMessages` with `returns: vStreamMessagesReturnValue`)
+- Accept queries with a `returns` validator (where `streams` is optional) in
+  the hooks' `stream: true` type guard (`StreamQuery`)
+- Document the compatible `ai`/`@ai-sdk/*` provider versions (AI SDK v6 pairs
+  with v3.x providers) and add a complete streaming chat quickstart to the
+  README
+
 ## 0.6.4
 
 - Fix streaming UI message dedupe (#281)

README.md

@@ -60,6 +60,110 @@ workflows.
 
 [Read the associated Stack post here](https://stack.convex.dev/ai-agents).
 
+## Streaming chat quickstart
+
+Install the component alongside the AI SDK v6 and a **v3.x** provider package
+(provider majors from other AI SDK generations will cause peer dependency
+conflicts with `ai@^6`):
+
+```sh
+npm i @convex-dev/agent ai@^6 @ai-sdk/anthropic@^3 convex-helpers
+```
+
+Define an agent and expose a mutation to send a message, an action that
+streams the response as deltas, and a query the client subscribes to:
+
+```ts
+// convex/chat.ts
+import { paginationOptsValidator } from "convex/server";
+import { v } from "convex/values";
+import { components, internal } from "./_generated/api";
+import { internalAction, mutation, query } from "./_generated/server";
+import {
+  Agent,
+  listUIMessagesWithStreams,
+  vStreamArgs,
+  vStreamUIMessagesReturnValue,
+} from "@convex-dev/agent";
+import { anthropic } from "@ai-sdk/anthropic";
+
+const agent = new Agent(components.agent, {
+  name: "Chat Agent",
+  languageModel: anthropic("claude-sonnet-4-5"),
+});
+
+export const sendMessage = mutation({
+  args: { threadId: v.string(), prompt: v.string() },
+  handler: async (ctx, { threadId, prompt }) => {
+    const { messageId } = await agent.saveMessage(ctx, {
+      threadId,
+      prompt,
+      skipEmbeddings: true,
+    });
+    await ctx.scheduler.runAfter(0, internal.chat.streamResponse, {
+      threadId,
+      promptMessageId: messageId,
+    });
+  },
+});
+
+export const streamResponse = internalAction({
+  args: { threadId: v.string(), promptMessageId: v.string() },
+  handler: async (ctx, { threadId, promptMessageId }) => {
+    const result = await agent.streamText(
+      ctx,
+      { threadId },
+      { promptMessageId },
+      { saveStreamDeltas: true },
+    );
+    await result.consumeStream();
+  },
+});
+
+export const listMessages = query({
+  args: {
+    threadId: v.string(),
+    paginationOpts: paginationOptsValidator,
+    streamArgs: vStreamArgs,
+  },
+  returns: vStreamUIMessagesReturnValue,
+  handler: async (ctx, args) => {
+    // Add your own auth check here, e.g. authorizeThreadAccess(ctx, args.threadId)
+    return listUIMessagesWithStreams(ctx, components.agent, args);
+  },
+});
+```
+
+Then subscribe from React, with tokens streaming in live:
+
+```tsx
+import { useUIMessages, useSmoothText } from "@convex-dev/agent/react";
+import { api } from "../convex/_generated/api";
+
+function Chat({ threadId }: { threadId: string }) {
+  const { results: messages } = useUIMessages(
+    api.chat.listMessages,
+    { threadId },
+    { initialNumItems: 10, stream: true },
+  );
+  return (
+    <div>
+      {messages.map((m) => (
+        <Message key={m.key} text={m.text} streaming={m.status === "streaming"} />
+      ))}
+    </div>
+  );
+}
+
+function Message({ text, streaming }: { text: string; streaming: boolean }) {
+  const [visibleText] = useSmoothText(text, { startStreaming: streaming });
+  return <p>{visibleText}</p>;
+}
+```
+
+See the [streaming docs](https://docs.convex.dev/agents/streaming) for
+customizing chunking/throttling, aborting streams, and HTTP streaming.
+
 [![Powerful AI Apps Made Easy with the Agent Component](https://thumbs.video-to-markdown.com/b323ac24.jpg)](https://youtu.be/tUKMPUlOCHY)
 **Read the [docs](https://docs.convex.dev/agents) for more details.**
 

docs/getting-started.mdx

@@ -13,10 +13,22 @@ Run `npm create convex` or follow any of the
 
 ## Installation
 
-Install the component package:
+Install the component package, along with the AI SDK and the provider(s) you
+want to use:
 
-```ts
-npm install @convex-dev/agent
+```sh
+npm install @convex-dev/agent ai @ai-sdk/openai
+```
+
+Note on versions: this component requires AI SDK v6 (`ai@^6`), which pairs
+with **v3.x** of the AI SDK provider packages, e.g. `@ai-sdk/openai@^3`,
+`@ai-sdk/anthropic@^3`, `@ai-sdk/google@^3` (and `@ai-sdk/provider-utils@^4`).
+If npm reports an unresolvable peer dependency conflict on `ai`, you likely
+installed a provider major version from a different AI SDK generation - fix it
+by installing the provider at `^3`, e.g.:
+
+```sh
+npm install ai@^6 @ai-sdk/anthropic@^3
 ```
 
 Create a `convex.config.ts` file in your app's `convex/` folder and install the

docs/streaming.mdx

@@ -64,10 +64,16 @@ stream deltas. This is very similar to
 [retrieving messages](./messages.mdx#retrieving-messages), with a few changes:
 
 ```ts
+import { v } from "convex/values";
 import { paginationOptsValidator } from "convex/server";
 // highlight-next-line
-import { vStreamArgs, listUIMessages, syncStreams } from "@convex-dev/agent";
+import {
+  listUIMessagesWithStreams,
+  vStreamArgs,
+  vStreamUIMessagesReturnValue,
+} from "@convex-dev/agent";
 import { components } from "./_generated/api";
+import { query } from "./_generated/server";
 
 export const listThreadMessages = query({
   args: {
@@ -77,21 +83,48 @@ export const listThreadMessages = query({
     // highlight-next-line
     streamArgs: vStreamArgs,
   },
+  // Validates the paginated messages and stream deltas being returned.
+  // highlight-next-line
+  returns: vStreamUIMessagesReturnValue,
+  handler: async (ctx, args) => {
+    await authorizeThreadAccess(ctx, args.threadId);
+
+    // Fetches both the regular non-streaming messages (as UIMessages)
+    // and the stream deltas, in the shape `useUIMessages` expects.
+    // highlight-next-line
+    return listUIMessagesWithStreams(ctx, components.agent, args);
+  },
+});
+```
+
+If you want to filter or modify the messages or deltas, you can compose the
+underlying functions yourself and return `{ ...paginated, streams }`:
+
+```ts
+export const listThreadMessagesCustom = query({
+  args: {
+    threadId: v.string(),
+    paginationOpts: paginationOptsValidator,
+    streamArgs: vStreamArgs,
+  },
   handler: async (ctx, args) => {
-    await authorizeThreadAccess(ctx, threadId);
+    await authorizeThreadAccess(ctx, args.threadId);
 
     // Fetches the regular non-streaming messages.
     const paginated = await listUIMessages(ctx, components.agent, args);
 
-    // highlight-next-line
     const streams = await syncStreams(ctx, components.agent, args);
 
-    // highlight-next-line
+    // Here you could filter out / modify the messages & stream deltas.
     return { ...paginated, streams };
   },
 });
 ```
 
+Note: if you return `MessageDoc`s (from `listMessagesWithStreams` or
+`listMessages`) instead of `UIMessage`s, use `vStreamMessagesReturnValue` as
+the `returns` validator and the `useThreadMessages` hook on the client.
+
 Similar to with [non-streaming messages](./messages.mdx#useuimessages-hook), you
 can use the `useUIMessages` hook to fetch the messages, passing in
 `stream: true` to enable streaming.

example/convex/chat/streaming.ts

@@ -3,8 +3,10 @@ import { paginationOptsValidator } from "convex/server";
 import {
   createThread,
   listUIMessages,
+  listUIMessagesWithStreams,
   syncStreams,
   vStreamArgs,
+  vStreamUIMessagesReturnValue,
 } from "@convex-dev/agent";
 import { components, internal } from "../_generated/api";
 import {
@@ -89,6 +91,24 @@ export const listThreadMessages = query({
     paginationOpts: paginationOptsValidator, // Used to paginate the messages.
     streamArgs: vStreamArgs, // Used to stream messages.
   },
+  // Optional, but recommended: it matches what the handler returns.
+  returns: vStreamUIMessagesReturnValue,
+  handler: async (ctx, args) => {
+    await authorizeThreadAccess(ctx, args.threadId);
+    // This fetches both the paginated (finished) messages and the stream
+    // deltas, in the shape the `useUIMessages` React hook expects.
+    return listUIMessagesWithStreams(ctx, components.agent, args);
+  },
+});
+
+// If you want to filter or enrich the messages or deltas, you can compose
+// `listUIMessages` and `syncStreams` yourself instead:
+export const listThreadMessagesCustom = query({
+  args: {
+    threadId: v.string(),
+    paginationOpts: paginationOptsValidator,
+    streamArgs: vStreamArgs,
+  },
   handler: async (ctx, args) => {
     const { threadId, streamArgs } = args;
     await authorizeThreadAccess(ctx, threadId);

src/UIMessages.ts

@@ -13,7 +13,7 @@ import {
   type UIDataTypes,
   type UITools,
 } from "ai";
-import type { Infer } from "convex/values";
+import { v, type Infer } from "convex/values";
 import { toModelMessage, fromModelMessage, toUIFilePart } from "./mapping.js";
 import {
   extractReasoning,
@@ -22,12 +22,13 @@ import {
   joinText,
   sorted,
 } from "./shared.js";
-import type {
-  MessageDoc,
-  MessageStatus,
-  ProviderOptions,
-  SourcePart,
-  vSource,
+import {
+  vMessageStatus,
+  type MessageDoc,
+  type MessageStatus,
+  type ProviderOptions,
+  type SourcePart,
+  type vSource,
 } from "./validators.js";
 import { omit, pick } from "convex-helpers";
 
@@ -48,6 +49,37 @@ export type UIMessage<
   _creationTime: number;
 };
 
+/**
+ * A validator matching the {@link UIMessage} type, e.g. to use in the
+ * `returns` validator of a query that returns UIMessages
+ * (see {@link vStreamUIMessagesReturnValue} for the common paginated +
+ * streaming shape).
+ *
+ * Note: the AI SDK's message `parts` (and `metadata`) are generic over the
+ * app's tools and data parts, so they can't be expressed precisely as Convex
+ * validators and are validated as `any`.
+ */
+export const vUIMessage = v.object({
+  // Fields from the AI SDK's UIMessage:
+  id: v.string(),
+  role: v.union(
+    v.literal("system"),
+    v.literal("user"),
+    v.literal("assistant"),
+  ),
+  parts: v.array(v.any()),
+  metadata: v.optional(v.any()),
+  // Fields added by the Agent component:
+  key: v.string(),
+  order: v.number(),
+  stepOrder: v.number(),
+  status: v.union(v.literal("streaming"), vMessageStatus),
+  agentName: v.optional(v.string()),
+  userId: v.optional(v.string()),
+  text: v.string(),
+  _creationTime: v.number(),
+});
+
 /**
  * Converts a list of UIMessages to MessageDocs, along with extra metadata that
  * may be available to associate with the MessageDocs.

src/client/index.ts

@@ -168,9 +168,13 @@ export {
   DeltaStreamer,
   abortStream,
   compressUIMessageChunks,
+  listMessagesWithStreams,
   listStreams,
+  listUIMessagesWithStreams,
   syncStreams,
   vStreamMessagesReturnValue,
+  vStreamUIMessagesReturnValue,
+  vSyncStreamsReturnValue,
 } from "./streaming.js";
 export {
   createThread,
@@ -179,7 +183,12 @@ export {
   updateThreadMetadata,
 } from "./threads.js";
 export type { ContextHandler } from "./types.js";
-export { toUIMessages, fromUIMessages, type UIMessage } from "../UIMessages.js";
+export {
+  toUIMessages,
+  fromUIMessages,
+  vUIMessage,
+  type UIMessage,
+} from "../UIMessages.js";
 
 export type {
   AgentComponent,