Durable Agents

Copy codenpm install convex-durable-agents ai zod

Copy code// convex/convex.config.ts
import { defineApp } from "convex/server";
import durableAgents from "convex-durable-agents/convex.config.js";

const app = defineApp();
app.use(durableAgents);

export default app;

Copy code// convex/chat.ts
import { z } from "zod";
import { components, internal } from "./_generated/api";
import { createActionTool, defineAgentApi, streamHandlerAction } from "convex-durable-agents";

// Define the stream handler with your model and tools
export const chatAgentHandler = streamHandlerAction(components.durableAgents, {
  model: "anthropic/claude-haiku-4.5",
  system: "You are a helpful AI assistant.",
  tools: {
    get_weather: createActionTool({
      description: "Get weather for a location",
      args: z.object({ location: z.string() }),
      handler: internal.tools.weather.getWeather,
      retry: {
        enabled: true,
        maxAttempts: 5,
      },
    }),
  },
  saveStreamDeltas: true, // Enable real-time streaming
});

// Export the agent API (public - callable from clients)
export const {
  createThread,
  sendMessage,
  getThread,
  listMessages,
  streamUpdates,
  listThreads,
  deleteThread,
  resumeThread,
  stopThread,
  addToolResult,
  addToolError,
} = defineAgentApi(components.durableAgents, internal.chat.chatAgentHandler, {
  // Optional: Add authorization to protect thread access
  authorizationCallback: async (ctx, threadId) => {
    // Example: verify the user owns this thread
    // const identity = await ctx.auth.getUserIdentity();
    // if (!identity) throw new Error("Unauthorized");
  },
});

Copy code// convex/chat.ts
import { defineInternalAgentApi } from "convex-durable-agents";

// Export internal agent API (only callable from other Convex functions)
export const {
  createThread,
  sendMessage,
  getThread,
  // ... other functions
} = defineInternalAgentApi(components.durableAgents, internal.chat.chatAgentHandler);

Copy code// convex/myChat.ts
import { mutation, query } from "./_generated/server";
import { internal } from "./_generated/api";
import { v } from "convex/values";

// Public wrapper with auth check
export const sendMessage = mutation({
  args: { threadId: v.string(), prompt: v.string() },
  handler: async (ctx, args) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Not authenticated");

    // Call the internal agent API
    return ctx.runMutation(internal.chat.sendMessage, args);
  },
});

Copy code// convex/tools/weather.ts
import { v } from "convex/values";
import { internalAction } from "../_generated/server";

export const getWeather = internalAction({
  args: { location: v.string() },
  returns: v.object({ weather: v.string(), temperature: v.number() }),
  handler: async (_ctx, args) => {
    // Call your weather API here
    return { weather: "sunny", temperature: 72 };
  },
});

Copy codeimport { useAgentChat } from "convex-durable-agents/react";
import { api } from "../convex/_generated/api";

function ChatView({ threadId }: { threadId: string }) {
  const { messages, status, isRunning, sendMessage, stop } = useAgentChat({
    listMessages: api.chat.listMessages,
    streamUpdates: api.chat.streamUpdates,
    getThread: api.chat.getThread,
    sendMessage: api.chat.sendMessage,
    stopThread: api.chat.stopThread,
    resumeThread: api.chat.resumeThread,
    threadId,
  });

  return (
    <div>
      {messages.map((msg) => (
        <div key={msg.metadata?.key}>
          <strong>{msg.role}:</strong> {msg.parts.map((p) => (p.type === "text" ? p.text : null))}
        </div>
      ))}

      {isRunning && <button onClick={() => stop()}>Stop</button>}

      <input
        onKeyPress={(e) => {
          if (e.key === "Enter" && !isRunning) {
            sendMessage(e.currentTarget.value);
            e.currentTarget.value = "";
          }
        }}
        disabled={isRunning}
      />
    </div>
  );
}

Copy codetype AgentApiOptions = {
  authorizationCallback?: (ctx: QueryCtx | MutationCtx | ActionCtx, threadId: string) => Promise<void> | void;
  workpoolEnqueueAction?: FunctionReference<"mutation", "internal">;
  toolExecutionWorkpoolEnqueueAction?: FunctionReference<"mutation", "internal">;
  onStatusChange?: FunctionReference<"mutation", "internal">;
  excludeSystemMessages?: boolean;
};

Copy codedefineAgentApi(components.durableAgents, internal.chat.chatAgentHandler, {
  authorizationCallback: async (ctx, threadId) => {
    const identity = await ctx.auth.getUserIdentity();
    if (!identity) throw new Error("Unauthorized");

    // Query your threads table to verify ownership
    const thread = await ctx.runQuery(api.threads.getOwner, { threadId });
    if (thread?.userId !== identity.subject) {
      throw new Error("Access denied");
    }
  },
});

Copy codestreamHandlerAction(component, {
  model: languageModel,        // AI SDK language model
  system?: string,             // System prompt
  tools?: Record<string, DurableTool>,
  saveStreamDeltas?: boolean | StreamingOptions,
  transformMessages?: (messages) => messages,
  // ... other AI SDK streamText options
});

Copy codecreateActionTool({
  description: string,
  args: ZodSchema,
  handler: FunctionReference<"action">,
  retry?: true | {
    enabled: true,
    maxAttempts?: number,        // default 3
    backoff?: {
      strategy?: "fixed",
      delayMs: number,
      jitter?: boolean,
    } | {
      strategy: "exponential",
      initialDelayMs: number,
      multiplier?: number,       // default 2
      maxDelayMs?: number,
      jitter?: boolean,
    },
    shouldRetryError?: FunctionReference<"action">,
  },
});

Copy codecreateAsyncTool({
  description: string,
  args: ZodSchema,
  callback: FunctionReference<"action">,
});

Copy codeconst {
  messages, // UIMessageWithConvexMetadata[]
  thread, // ThreadDoc | null
  status, // ThreadStatus
  isLoading, // boolean
  isRunning, // boolean
  isComplete, // boolean
  isFailed, // boolean
  isStopped, // boolean
  sendMessage, // (prompt: string) => Promise<null>
  stop, // () => Promise<null>
  resume, // (prompt?: string) => Promise<null>
} = useAgentChat({
  listMessages: api.chat.listMessages,
  streamUpdates: api.chat.streamUpdates,
  getThread: api.chat.getThread,
  sendMessage: api.chat.sendMessage,
  stopThread: api.chat.stopThread,
  resumeThread: api.chat.resumeThread,
  threadId,
});

// Send a message (threadId is automatically included)
await sendMessage("Hello!");

// Stop the agent
await stop();

// Resume after stopping or failure
await resume();

Copy codeconst {
  messages, // UIMessageWithConvexMetadata[]
  thread, // ThreadDoc | null
  status, // ThreadStatus
  isLoading, // boolean
  isRunning, // boolean
  isComplete, // boolean
  isFailed, // boolean
  isStopped, // boolean
} = useThreadMessages({
  messagesQuery: api.chat.listMessages,
  streamingMessageUpdatesQuery: api.chat.streamUpdates

[README truncated for prompt length]