Skip to main content

Streaming

For AI agents: see llms.txt for the complete documentation index. Markdown versions are available by adding .md to a page URL or requesting Accept: text/markdown.

Streaming messages is a great way to give a user feedback and keep an application feeling responsive while using LLMs.

Traditionally streaming happens via HTTP streaming, where the client sends a request and waits until the full response is streamed back. This works out of the box when using the Agent, in the same way you would with the AI SDK. See below if that is all you're looking for.

However, with the Agent component you can also stream messages asynchronously, meaning the generation doesn't have to happen in an HTTP handler (httpAction), and the response can be streamed back to one or more clients even if their network connection is interrupted.

It works by saving the streaming parts to the database in groups (deltas), and the clients subscribe to new deltas for the given thread, as they're generated. As a bonus, you don't even need to use the Agent's version of streamText to use the delta streaming approach (see below).

Example:

Streaming message deltas​

The easiest way to stream is to pass { saveStreamDeltas: true } to agent.streamText. This will save chunks of the response as deltas as they're generated, so all clients can subscribe to the stream and get live-updating text via normal Convex queries.

agent.streamText(ctx, { threadId }, { prompt }, { saveStreamDeltas: true });

This can be done in an async function, where http streaming to a client is not possible. Under the hood it will chunk up the response and debounce saving the deltas to prevent excessive bandwidth usage. You can pass more options to saveStreamDeltas to configure the chunking and debouncing.

  { saveStreamDeltas: { chunking: "line", throttleMs: 1000 } },
  • chunking can be "word", "line", a regex, or a custom function.
  • sendSources includes source parts emitted by the model. It defaults to false.
  • throttleMs is how frequently the deltas are saved. This will send multiple chunks per delta, writes sequentially, and will not write faster than the throttleMs (single-flighted ).

Retrieving streamed deltas​

For clients to stream messages, you need to expose a query that returns the stream deltas. This is very similar to retrieving messages, with a few changes:

import { paginationOptsValidator } from "convex/server";
import { vStreamArgs, listUIMessages, syncStreams } from "@convex-dev/agent";
import { components } from "./_generated/api";

export const listThreadMessages = query({
  args: {
    threadId: v.string(),
    // Pagination options for the non-streaming messages.
    paginationOpts: paginationOptsValidator,
    streamArgs: vStreamArgs,
  },
  handler: async (ctx, args) => {
    await authorizeThreadAccess(ctx, threadId);

    // Fetches the regular non-streaming messages.
    const paginated = await listUIMessages(ctx, components.agent, args);

    const streams = await syncStreams(ctx, components.agent, args);

    return { ...paginated, streams };
  },
});

Similar to with non-streaming messages, you can use the useUIMessages hook to fetch the messages, passing in stream: true to enable streaming.

const { results, status, loadMore } = useUIMessages(
  api.chat.streaming.listMessages,
  { threadId },
  { initialNumItems: 10, stream: true },
);

Text smoothing with SmoothText and useSmoothText​

The useSmoothText hook is a simple hook that smooths the text as it changes. It can work with any text, but is especially handy for streaming text.

import { useSmoothText } from "@convex-dev/agent/react";

// in the component
const [visibleText] = useSmoothText(message.text);

You can configure the initial characters per second. It will adapt over time to match the average speed of the text coming in.

By default it won't stream the first text it receives unless you pass in startStreaming: true. To start streaming immediately when you have a mix of streaming and non-streaming messages, do:

import { useSmoothText, type UIMessage } from "@convex-dev/agent/react";

function Message({ message }: { message: UIMessage }) {
  const [visibleText] = useSmoothText(message.text, {
    startStreaming: message.status === "streaming",
  });
  return <div>{visibleText}</div>;
}

If you don't want to use the hook, you can use the SmoothText component.

import { SmoothText } from "@convex-dev/agent/react";

//...
<SmoothText text={message.text} />;

Consuming the stream yourself with the Agent​

You can consume the stream in all the ways you can with the underlying AI SDK - for instance iterating over the content, or using result.toUIMessageStreamResponse().

If you are not also saving the deltas, it might look like this:

const result = await agent.streamText(ctx, { threadId }, { prompt });

for await (const textPart of result.textStream) {
  console.log(textPart);
}

agent.streamObject saves the object from the AI SDK's onFinish, which only runs once the stream has been read to the end. Read result.partialObjectStream to the end before the action returns: awaiting result.object alone does not read the stream.

If you want to both iterate as the stream is happening, as well as save the deltas, you can pass { saveStreamDeltas: { returnImmediately: true } } to streamText. This will return immediately, and you can then iterate over the stream live, or return the stream in an HTTP Response.

const result = await agent.streamText(
  ctx,
  { threadId },
  { prompt },
  { saveStreamDeltas: { returnImmediately: true } },
);

return result.toUIMessageStreamResponse();

If you don't want to have the Agent involved at all, the next section will show you how to save the deltas yourself.

Using useChat from @ai-sdk/react​

toUIMessageStreamResponse() returns the response shape useChat consumes, so the AI SDK's React hook and the tooling built around it work against an httpAction. The hook posts its messages array. Validate the last user message, use its text as the prompt, and let the agent load the rest of the history from the thread:

// convex/chat.ts
import { z } from "zod";
import { getThreadMetadata } from "@convex-dev/agent";
import { components } from "./_generated/api";
import { httpAction } from "./_generated/server";
import { storyAgent } from "./agents/story";
import { getAuthUserId } from "./utils";

const chatRequest = z.object({
  threadId: z.string().min(1),
  messages: z.array(
    z.object({ role: z.string(), parts: z.array(z.unknown()) }),
  ),
});
const textPart = z.object({ type: z.literal("text"), text: z.string() });

export const chat = httpAction(async (ctx, request) => {
  const parsed = chatRequest.safeParse(await request.json().catch(() => null));
  if (!parsed.success) return new Response("Invalid request", { status: 400 });
  const { messages, threadId } = parsed.data;
  const last = messages.at(-1);
  if (last?.role !== "user") {
    return new Response("Expected a user message", { status: 400 });
  }
  const prompt = last.parts
    .flatMap((part) => {
      const text = textPart.safeParse(part);
      return text.success ? [text.data.text] : [];
    })
    .join("");
  if (!prompt) return new Response("No prompt text", { status: 400 });
  const userId = await getAuthUserId(ctx);
  if (!userId) return new Response("Unauthorized", { status: 401 });
  const thread = await getThreadMetadata(ctx, components.agent, { threadId });
  if (thread.userId !== userId) {
    return new Response("Forbidden", { status: 403 });
  }
  const result = await storyAgent.streamText(
    ctx,
    { threadId },
    { prompt },
    { saveStreamDeltas: { returnImmediately: true } },
  );
  return result.toUIMessageStreamResponse();
});

Route it in convex/http.ts with CORS, as in http.ts:

// convex/http.ts
import { httpRouter } from "convex/server";
import { corsRouter } from "convex-helpers/server/cors";
import { chat } from "./chat";

const http = httpRouter();
const cors = corsRouter(http, {
  allowedHeaders: ["Authorization", "Content-Type"],
});
cors.route({ path: "/chat", method: "POST", handler: chat });
export default http;

Then point the transport at it. VITE_CONVEX_SITE_URL is your deployment's .convex.site URL, which npx convex dev writes for Vite apps:

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

const { messages, sendMessage } = useChat({
  transport: new DefaultChatTransport({
    api: `${import.meta.env.VITE_CONVEX_SITE_URL}/chat`,
    body: { threadId },
    headers: async () => ({ Authorization: `Bearer ${await getToken()}` }),
  }),
});

This example handles text-only user sends. Add handling for attachments or regeneration if your chat UI supports them.

The access check returns 401 or 403 itself, so useChat sees the status instead of an opaque CORS failure, while unexpected errors still surface as server errors. getAuthUserId is the example app's demo stub (example/convex/utils.ts), which ignores the request, so in your app derive the user from ctx.auth.getUserIdentity() with Convex auth configured. DefaultChatTransport does not automatically send the Convex client's auth token to an HTTP action, so the transport passes it as an Authorization: Bearer <token> header, with getToken standing in for your auth provider's getter for the Convex JWT. See HTTP action auth.

The trade-off is where state lives. useChat holds the conversation in the browser tab that made the request; the agent still persists every message, but a new useChat instance does not automatically load or subscribe to that history. On reload, seed its messages from a query over the thread. useUIMessages reads the same persisted messages and deltas reactively from the database, so every subscriber sees the stream. Use useChat when you need its ecosystem, and useUIMessages when you need the conversation to be shared.

Advanced: Streaming deltas asynchronously without using an Agent​

To stream messages without using the Agent's wrapper of streamText, you can use the streamText function from the AI SDK directly.

It consists of using the DeltaStreamer class to save the deltas to the database, and then using the above approach to retrieve the messages, though you can use a more direct useStreamingUIMessages hook that doesn't involve reading any non-streaming messages from the database.

The requirements for reading and writing the streams are just that they use a threadId from the Agent component, and that each stream is saved with a distinct order, for ordering on the client side.

The example uses the Convex AI Gateway; see Getting Started for installation and requirements.

import { components } from "./_generated/api";
import { type ActionCtx } from "./_generated/server";
import { DeltaStreamer, compressUIMessageChunks } from "@convex-dev/agent";
import { streamText } from "ai";
import { convexGateway } from "@convex-dev/ai-sdk-provider";

async function stream(ctx: ActionCtx, threadId: string, order: number) {
  const streamer = new DeltaStreamer(
    components.agent,
    ctx,
    {
      throttleMs: 100,
      onAsyncAbort: async () => console.error("Aborted asynchronously"),
      // This will collapse multiple tiny deltas into one if they're being sent
      // in quick succession.
      compress: compressUIMessageChunks,
      abortSignal: undefined,
    },
    {
      threadId,
      format: "UIMessageChunk",
      order,
      stepOrder: 0,
      userId: undefined,
    },
  );
  // Do the normal streaming with the AI SDK
  const response = streamText({
    model: convexGateway("openai/gpt-5-mini"),
    prompt: "Tell me a joke",
    abortSignal: streamer.abortController.signal,
    onError: (error) => {
      console.error(error);
      streamer.fail(errorToString(error.error));
    },
  });

  // We could await here if we wanted to wait for the stream to finish,
  // but instead we have it process asynchronously so we can return a streaming
  // http Response.
  void streamer.consumeStream(response.toUIMessageStream());

  return {
    // e.g. to do `response.toTextStreamResponse()` for HTTP streaming.
    response,
    // We don't need this on the client, but with it we can have some clients
    // selectively not stream down deltas when they're using HTTP streaming
    // already.
    streamId: await streamer.getStreamId(),
  };
}

To fetch the deltas for the client, you can use the syncStreams function, as you would with normal Agent streaming. If you don't want to fetch the non-streaming messages, it can be simplified to:

import { v } from "convex/values";
import { vStreamArgs, syncStreams } from "@convex-dev/agent";
import { query } from "./_generated/server";
import { components } from "./_generated/api";

export const listStreams = query({
  args: {
    threadId: v.string(),
    streamArgs: vStreamArgs,
  },
  handler: async (ctx, args) => {
    // await authorizeThreadAccess(ctx, args.threadId);
    const streams = await syncStreams(ctx, components.agent, {
      ...args,
      // By default syncStreams only returns streaming messages. However, if
      // your messages aren't saved in the same transaction as the streaming
      // ends, you might want to include them here to avoid UI flashes.
      includeStatuses: ["streaming", "aborted", "finished"],
    });
    return { streams };
  },
});

On the client side, you can use the useStreamingUIMessages hook to fetch the messages. If you defined more arguments than just threadId, they'll get passed along with threadId here.

const messages = useStreamingUIMessages(api.example.listStreams, { threadId });

You can pass in another parameter to either skip certain streamIds or to start at some order to ignore previous streams.