diff --git a/CHANGELOG.md b/CHANGELOG.md index 47ff4e1b..a4c6bbc4 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/README.md b/README.md index fe60aedf..9e13b59b 100644 --- a/README.md +++ b/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 ( +
+ {messages.map((m) => ( + + ))} +
+ ); +} + +function Message({ text, streaming }: { text: string; streaming: boolean }) { + const [visibleText] = useSmoothText(text, { startStreaming: streaming }); + return

{visibleText}

; +} +``` + +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.** diff --git a/docs/getting-started.mdx b/docs/getting-started.mdx index 647b5550..c215bb1a 100644 --- a/docs/getting-started.mdx +++ b/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 diff --git a/docs/streaming.mdx b/docs/streaming.mdx index de392f69..f4a45bc4 100644 --- a/docs/streaming.mdx +++ b/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. diff --git a/example/convex/chat/streaming.ts b/example/convex/chat/streaming.ts index 4318a470..3da8c64e 100644 --- a/example/convex/chat/streaming.ts +++ b/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); diff --git a/src/UIMessages.ts b/src/UIMessages.ts index 6fc6bc67..0faf8312 100644 --- a/src/UIMessages.ts +++ b/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. diff --git a/src/client/index.ts b/src/client/index.ts index fd8fe2f0..fc172e4d 100644 --- a/src/client/index.ts +++ b/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, diff --git a/src/client/listWithStreams.test.ts b/src/client/listWithStreams.test.ts new file mode 100644 index 00000000..f49a8521 --- /dev/null +++ b/src/client/listWithStreams.test.ts @@ -0,0 +1,201 @@ +/// +import { beforeEach, describe, expect, expectTypeOf, test } from "vitest"; +import type { + FunctionReference, + GenericSchema, + PaginationOptions, + PaginationResult, + SchemaDefinition, +} from "convex/server"; +import type { Infer } from "convex/values"; +import { parse } from "convex-helpers/validators"; +import type { TestConvex } from "convex-test"; +import { components, initConvexTest } from "./setup.test.js"; +import { createThread, saveMessages } from "./index.js"; +import { + DeltaStreamer, + listMessagesWithStreams, + listUIMessagesWithStreams, + vStreamMessagesReturnValue, + vStreamUIMessagesReturnValue, +} from "./streaming.js"; +import type { StreamQuery } from "../react/types.js"; +import type { + UIMessageLike, + UIMessagesQuery, +} from "../react/useUIMessages.js"; +import { vUIMessage, type UIMessage } from "../UIMessages.js"; +import type { StreamArgs } from "../validators.js"; + +/** + * Type-level tests: a from-scratch streaming query built with + * `listUIMessagesWithStreams` (and optionally the exported `returns` + * validators) must satisfy both the validators and the React hooks. + */ + +// The runtime UIMessage type matches the vUIMessage validator. +expectTypeOf().toExtend>(); +expectTypeOf>().toExtend(); + +// The helpers' return values satisfy the exported `returns` validators, +// so `returns: vStreamUIMessagesReturnValue` + `return +// listUIMessagesWithStreams(...)` compiles (this was the TS2719 landmine). +type UIMessagesWithStreams = Awaited< + ReturnType +>; +type MessagesWithStreams = Awaited>; +expectTypeOf().toExtend< + Infer +>(); +expectTypeOf().toExtend< + Infer +>(); + +type BaseArgs = { + threadId: string; + paginationOpts: PaginationOptions; + streamArgs?: StreamArgs; +}; +// The query type produced when using `returns: vStreamUIMessagesReturnValue`. +type QueryWithValidator = FunctionReference< + "query", + "public", + BaseArgs, + Infer +>; +// The query type produced when the handler returns +// `listUIMessagesWithStreams(...)` without a `returns` validator. +type QueryWithoutValidator = FunctionReference< + "query", + "public", + BaseArgs, + UIMessagesWithStreams +>; +// Both are accepted by `useUIMessages(..., { stream: true })`. +expectTypeOf().toExtend(); +expectTypeOf().toExtend(); +expectTypeOf().toExtend(); +expectTypeOf().toExtend(); + +// A query that doesn't return `streams` at all is still rejected by +// `stream: true` (the ErrorMessage guard remains useful). +type NonStreamingQuery = FunctionReference< + "query", + "public", + { threadId: string; paginationOpts: PaginationOptions }, + PaginationResult +>; +expectTypeOf().not.toExtend(); + +/** + * Runtime tests: the values the helpers return pass the exported + * `returns` validators, including the split-pagination fields and both + * `streams` variants ("list" and "deltas"). + */ + +const streamerOptions = { + throttleMs: 0, + abortSignal: undefined, + compress: null, + onAsyncAbort: async (_reason: string) => {}, +}; + +describe("listUIMessagesWithStreams / listMessagesWithStreams", () => { + let t: TestConvex>; + let threadId: string; + + beforeEach(async () => { + t = initConvexTest(); + await t.run(async (ctx) => { + threadId = await createThread(ctx, components.agent, {}); + }); + }); + + test("returns UIMessages and streams that satisfy the validator", async () => { + await t.run(async (ctx) => { + await saveMessages(ctx, components.agent, { + threadId, + messages: [ + { role: "user", content: "Hello" }, + { role: "assistant", content: "Hi there!" }, + ], + }); + // Simulate an in-progress stream with one delta. + const streamer = new DeltaStreamer( + components.agent, + ctx, + streamerOptions, + { + threadId, + order: 5, + stepOrder: 0, + format: "UIMessageChunk", + }, + ); + await streamer.addParts([ + { type: "text-delta", id: "t1", delta: "Once upon a time" }, + ]); + + // kind: "list" — the first call the client hook makes. + const result = await listUIMessagesWithStreams(ctx, components.agent, { + threadId, + paginationOpts: { numItems: 10, cursor: null }, + streamArgs: { kind: "list" }, + }); + expect(result.page.length).toBeGreaterThan(0); + expect(result.streams?.kind).toBe("list"); + if (result.streams?.kind === "list") { + expect(result.streams.messages).toHaveLength(1); + } + expect(parse(vStreamUIMessagesReturnValue, result)).toBeDefined(); + + // kind: "deltas" — the follow-up calls with per-stream cursors. + const withDeltas = await listUIMessagesWithStreams( + ctx, + components.agent, + { + threadId, + paginationOpts: { numItems: 10, cursor: null }, + streamArgs: { + kind: "deltas", + cursors: [{ streamId: streamer.streamId!, cursor: 0 }], + }, + }, + ); + expect(withDeltas.streams?.kind).toBe("deltas"); + if (withDeltas.streams?.kind === "deltas") { + expect(withDeltas.streams.deltas.length).toBeGreaterThan(0); + } + expect(parse(vStreamUIMessagesReturnValue, withDeltas)).toBeDefined(); + + // Without streamArgs (e.g. from a non-streaming pagination call). + const noStreams = await listUIMessagesWithStreams( + ctx, + components.agent, + { threadId, paginationOpts: { numItems: 10, cursor: null } }, + ); + expect(noStreams.streams).toBeUndefined(); + expect(parse(vStreamUIMessagesReturnValue, noStreams)).toBeDefined(); + }); + }); + + test("returns MessageDocs and streams that satisfy the validator", async () => { + await t.run(async (ctx) => { + await saveMessages(ctx, components.agent, { + threadId, + messages: [ + { role: "user", content: "Hello" }, + { role: "assistant", content: "Hi there!" }, + ], + }); + const result = await listMessagesWithStreams(ctx, components.agent, { + threadId, + paginationOpts: { numItems: 10, cursor: null }, + streamArgs: { kind: "list" }, + }); + expect(result.page).toHaveLength(2); + expect(result.streams?.kind).toBe("list"); + expect(parse(vStreamMessagesReturnValue, result)).toBeDefined(); + }); + }); +}); diff --git a/src/client/streaming.ts b/src/client/streaming.ts index 0a501e01..1449fca6 100644 --- a/src/client/streaming.ts +++ b/src/client/streaming.ts @@ -8,16 +8,21 @@ import { type UIMessageChunk, } from "ai"; import { v } from "convex/values"; +import type { PaginationOptions, PaginationResult } from "convex/server"; import { vMessageDoc, vPaginationResult, vStreamDelta, vStreamMessage, + type MessageDoc, + type MessageStatus, type ProviderOptions, type StreamArgs, type StreamDelta, type StreamMessage, } from "../validators.js"; +import { vUIMessage, type UIMessage } from "../UIMessages.js"; +import { listMessages, listUIMessages } from "./messages.js"; import type { ActionCtx, AgentComponent, @@ -26,16 +31,132 @@ import type { SyncStreamsReturnValue, } from "./types.js"; +/** + * A validator for the `streams` field returned from {@link syncStreams}, + * to use in the `returns` validator of a streaming query. + * Matches the {@link SyncStreamsReturnValue} type. + */ +export const vSyncStreamsReturnValue = v.optional( + v.union( + v.object({ kind: v.literal("list"), messages: v.array(vStreamMessage) }), + v.object({ kind: v.literal("deltas"), deltas: v.array(vStreamDelta) }), + ), +); + +/** + * The `returns` validator for a streaming query that returns MessageDocs, + * e.g. one whose handler returns {@link listMessagesWithStreams} + * (used with the `useThreadMessages` React hook). + */ export const vStreamMessagesReturnValue = v.object({ ...vPaginationResult(vMessageDoc).fields, - streams: v.optional( - v.union( - v.object({ kind: v.literal("list"), messages: v.array(vStreamMessage) }), - v.object({ kind: v.literal("deltas"), deltas: v.array(vStreamDelta) }), - ), - ), + streams: vSyncStreamsReturnValue, +}); + +/** + * The `returns` validator for a streaming query that returns UIMessages, + * e.g. one whose handler returns {@link listUIMessagesWithStreams} + * (used with the `useUIMessages` React hook). + */ +export const vStreamUIMessagesReturnValue = v.object({ + ...vPaginationResult(vUIMessage).fields, + streams: vSyncStreamsReturnValue, }); +/** + * List a page of UIMessages for a thread along with any streaming message + * deltas: everything the `useUIMessages` React hook needs when passed + * `stream: true`. The return value satisfies + * {@link vStreamUIMessagesReturnValue}, so a complete streaming query is: + * + * ```ts + * export const listThreadMessages = query({ + * args: { + * threadId: v.string(), + * paginationOpts: paginationOptsValidator, + * streamArgs: vStreamArgs, + * }, + * returns: vStreamUIMessagesReturnValue, + * handler: async (ctx, args) => { + * // await authorizeThreadAccess(ctx, args.threadId); + * return listUIMessagesWithStreams(ctx, components.agent, args); + * }, + * }); + * ``` + * + * If you want to filter or enrich the results, you can instead compose + * {@link listUIMessages} and {@link syncStreams} yourself and return + * `{ ...paginated, streams }`. + * + * @param ctx A ctx object from a query, mutation, or action. + * @param component The agent component, usually `components.agent`. + * @param args.threadId The thread to list messages & streams for. + * @param args.paginationOpts Pagination options (e.g. from usePaginatedQuery). + * @param args.streamArgs The stream arguments passed from the client hook. + * @param args.includeStatuses Which stream statuses to include + * (defaults to only "streaming"). + * @returns The paginated UIMessages with a `streams` field of deltas. + */ +export async function listUIMessagesWithStreams( + ctx: QueryCtx | MutationCtx | ActionCtx, + component: AgentComponent, + args: { + threadId: string; + paginationOpts: PaginationOptions; + streamArgs?: StreamArgs | undefined; + includeStatuses?: ("streaming" | "finished" | "aborted")[]; + }, +): Promise< + PaginationResult & { streams: SyncStreamsReturnValue } +> { + const [paginated, streams] = await Promise.all([ + listUIMessages(ctx, component, args), + syncStreams(ctx, component, args), + ]); + return { ...paginated, streams }; +} + +/** + * List a page of MessageDocs for a thread along with any streaming message + * deltas: everything the `useThreadMessages` React hook needs when passed + * `stream: true`. The return value satisfies + * {@link vStreamMessagesReturnValue}. + * + * If you use the `useUIMessages` hook, use + * {@link listUIMessagesWithStreams} instead. + * + * @param ctx A ctx object from a query, mutation, or action. + * @param component The agent component, usually `components.agent`. + * @param args.threadId The thread to list messages & streams for. + * @param args.paginationOpts Pagination options (e.g. from usePaginatedQuery). + * @param args.streamArgs The stream arguments passed from the client hook. + * @param args.excludeToolMessages Whether to exclude tool messages. + * @param args.statuses What message statuses to include. All by default. + * @param args.includeStatuses Which stream statuses to include + * (defaults to only "streaming"). + * @returns The paginated MessageDocs with a `streams` field of deltas. + */ +export async function listMessagesWithStreams( + ctx: QueryCtx | MutationCtx | ActionCtx, + component: AgentComponent, + args: { + threadId: string; + paginationOpts: PaginationOptions; + streamArgs?: StreamArgs | undefined; + excludeToolMessages?: boolean; + statuses?: MessageStatus[]; + includeStatuses?: ("streaming" | "finished" | "aborted")[]; + }, +): Promise< + PaginationResult & { streams: SyncStreamsReturnValue } +> { + const [paginated, streams] = await Promise.all([ + listMessages(ctx, component, args), + syncStreams(ctx, component, args), + ]); + return { ...paginated, streams }; +} + /** * A function that handles fetching stream deltas, used with the React hooks * `useThreadMessages` or `useStreamingThreadMessages`. diff --git a/src/react/types.ts b/src/react/types.ts index 5c9ce793..b127e9d9 100644 --- a/src/react/types.ts +++ b/src/react/types.ts @@ -10,7 +10,11 @@ export type StreamQuery> = FunctionReference< threadId: string; streamArgs?: StreamArgs; // required for stream query } & Args, - { streams: SyncStreamsReturnValue } + // `streams` is optional so that queries with a `returns` validator + // (e.g. `vStreamUIMessagesReturnValue`, where `streams` is `v.optional`) + // also count as stream queries. Queries that don't return `streams` at + // all still don't match, since there are no properties in common. + { streams?: SyncStreamsReturnValue } >; export type StreamQueryArgs> = diff --git a/src/react/useUIMessages.ts b/src/react/useUIMessages.ts index 7db30488..6a85490b 100644 --- a/src/react/useUIMessages.ts +++ b/src/react/useUIMessages.ts @@ -86,15 +86,23 @@ export type UIMessagesQueryResult< * streamArgs: vStreamArgs, * ... other arguments you want * }, + * returns: vStreamUIMessagesReturnValue, // optional + * handler: async (ctx, args) => { + * // await authorizeThreadAccess(ctx, args.threadId); + * return listUIMessagesWithStreams(ctx, components.agent, args); + * }, + * }); + * ``` + * + * To filter or modify the messages & stream deltas, compose it yourself: + * ```ts * handler: async (ctx, args) => { - * // await authorizeThreadAccess(ctx, threadId); * // NOTE: listUIMessages returns UIMessages, not MessageDocs. * const paginated = await listUIMessages(ctx, components.agent, args); * const streams = await syncStreams(ctx, components.agent, args); * // Here you could filter out / modify the documents & stream deltas. * return { ...paginated, streams }; * }, - * }); * ``` * * Then the hook can be used like this: