Building AI Features

This page shows the extension points you use most often when turning RED into your own AI product.

Start with the smallest tool that matches the job:

Direct sync tool — fast work, no task row.
Sync internal task tool — fast work, but tracked as a task.
Async internal task — Convex-backed background work with progress.
Async external task — remote worker or third-party service calls back over HTTP.
Agent registration — instructions, model, tools, and optional per-turn context.

Recipe 1: Direct Sync Tool

Use defineSyncTool() for work that should finish quickly and does not need task history.

import { z } from "zod/v4"
import type { Id } from "../../../_generated/dataModel"
import type { ToolContext } from "../core/types"
import { defineSyncTool } from "../core/defineSyncTool"
import type { ProjectAgentContext } from "../agents/project.agent"

const inputSchema = z.object({
  projectId: z.string(),
})

export const readProject = defineSyncTool({
  description: "Read one project in the current organization.",
  inputSchema,
  toolName: "readProject",
  execute: async (input, options) => {
    const toolContext = options.experimental_context as ToolContext<ProjectAgentContext> | undefined
    if (!toolContext) throw new Error("readProject: missing tool context")

    const project = await toolContext.ctx.db.get(input.projectId as Id<"projects">)
    if (!project || project.organizationId !== toolContext.organizationId) {
      throw new Error("Project not found")
    }

    return {
      id: project._id,
      name: project.name,
      status: project.status,
    }
  },
})

Key points:

Read RED context through options.experimental_context.
Verify organization ownership after loading any org-scoped document.
Keep sync tools fast. If it can take several seconds, make it a task.

Recipe 2: Sync Tool Backed by an Internal Task

Use defineSyncInternalTaskTool() when you want the model to receive the result in the same turn, but you also want task history.

First, create an internal node that supports inline execution:

import { z } from "zod/v4"
import type { Id } from "../../../_generated/dataModel"
import type { InternalNodeRunOptions, TaskMutationCtx } from "../tasks_engine"

export const projectSummaryInputSchema = z.object({
  projectId: z.string(),
})

export const projectSummaryOutputSchema = z.object({
  summary: z.string(),
})

export async function runProjectSummary(
  ctx: TaskMutationCtx,
  input: z.infer<typeof projectSummaryInputSchema>,
  _options: InternalNodeRunOptions,
) {
  const project = await ctx.db.get(input.projectId as Id<"projects">)
  if (!project) throw new Error("Project not found")

  return {
    summary: `${project.name}: ${project.status}`,
  }
}

export const projectSummaryNodeDefinition = {
  key: "tasks.projects.summary",
  name: "Projects - Summary",
  description: "Builds a short project summary.",
  kind: "internal" as const,
  inlineExecution: "sync" as const,
  scope: "member" as const,
  inputSchema: projectSummaryInputSchema,
  outputSchema: projectSummaryOutputSchema,
  run: runProjectSummary,
}

Then expose it as a tool:

import type { z } from "zod/v4"
import {
  projectSummaryInputSchema,
  projectSummaryOutputSchema,
} from "../../tasks/node_types/project_summary_node"
import { defineSyncInternalTaskTool } from "../core/defineSyncInternalTaskTool"

export const projectSummary = defineSyncInternalTaskTool<
  z.infer<typeof projectSummaryInputSchema>,
  z.infer<typeof projectSummaryOutputSchema>
>({
  description: "Summarize the current project.",
  inputSchema: projectSummaryInputSchema,
  nodeTypeKey: "tasks.projects.summary",
  toolName: "projectSummary",
})

Use this when you want short task-backed work. The node must be internal and set inlineExecution: "sync".

Recipe 3: Async Internal Task

Use defineAsyncTool() with an internal node when work runs in Convex but should happen in the background or emit progress.

This mirrors the Brief export pattern. The real Brief node is tasks.briefs.export, kind: "internal", and schedules progress plus a final artifact write.

import { z } from "zod/v4"
import { internal } from "../../../_generated/api"
import type { ProjectAgentContext } from "../agents/project.agent"
import { defineAsyncTool } from "../core/defineAsyncTool"
import type { ToolContext } from "../core/types"

const inputSchema = z.object({
  format: z.literal("md").default("md"),
})

export const exportProject = defineAsyncTool<z.infer<typeof inputSchema>>({
  description: "Start a background Markdown export for the current project.",
  inputSchema,
  startTask: async (input, baseContext) => {
    const toolContext = baseContext as ToolContext<ProjectAgentContext>
    if (!toolContext.extras?.projectId) {
      throw new Error("exportProject: this thread is not bound to a project")
    }

    const { taskId } = await toolContext.ctx.runMutation(
      internal.modules.tasks.tasks_internal.startTaskForAi,
      {
        organizationId: toolContext.organizationId,
        runBy: toolContext.startedBy,
        nodeTypeKey: "tasks.projects.export",
        input: {
          ...input,
          projectId: toolContext.extras.projectId,
          durationMs: 30_000,
        },
        threadId: toolContext.threadId,
        pollIntervalMs: 1_000,
      },
    )

    return {
      taskIds: [taskId],
      summary: "Exporting project",
      pollIntervalMs: 1_000,
      blocking: false,
    }
  },
})

Use blocking: false for exports and reports where the user can keep working. Use blocking: true when the agent needs the result before answering.

The internal node itself uses run, not start:

export const projectExportNodeDefinition = {
  key: "tasks.projects.export",
  name: "Projects - Export Markdown",
  description: "Exports a project as Markdown.",
  kind: "internal" as const,
  scope: "member" as const,
  inputSchema: projectExportInputSchema,
  outputSchema: projectExportOutputSchema,
  run: runProjectExport,
}

Recipe 4: Async External Task

Use an external task when a remote worker does the real work.

import { z } from "zod/v4"
import type { ExternalHandle, TaskActionCtx } from "../tasks_engine"
import type { Doc } from "../../../_generated/dataModel"

export const remoteRenderInputSchema = z.object({
  projectId: z.string(),
  format: z.literal("pdf"),
})

export const remoteRenderOutputSchema = z.object({
  downloadUrl: z.string().url(),
  size: z.number().int().nonnegative(),
})

async function triggerRemoteRender(
  _ctx: TaskActionCtx,
  task: Doc<"tasks">,
  handle: ExternalHandle,
) {
  if (!handle.handleUrl) throw new Error("Missing external callback URL")

  await fetch("https://worker.example.com/render", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      taskId: task._id,
      input: task.input,
      callbackUrl: handle.handleUrl,
    }),
  })
}

export const remoteRenderNodeDefinition = {
  key: "tasks.projects.remote_render",
  name: "Projects - Remote Render",
  description: "Renders a project through an external worker.",
  kind: "external" as const,
  scope: "member" as const,
  inputSchema: remoteRenderInputSchema,
  outputSchema: remoteRenderOutputSchema,
  pollIntervalMs: 5_000,
  timeoutMs: 120_000,
  trigger: triggerRemoteRender,
}

The remote worker posts events back to RED:

POST /api/tasks/{handle}/event
Content-Type: application/json

{
  "type": "success",
  "payload": {
    "output": {
      "downloadUrl": "https://storage.example.com/render.pdf",
      "size": 204800
    }
  }
}

Supported event types are started, progress, heartbeat, success, error, cancelled, and custom. The {handle} is a capability token. Do not expose it except to the service that should report task progress.

Recipe 5: Register Task Nodes

Task nodes are registered from the built-in node registry:

import type { NodeDefinition } from "../tasks_engine"
import { briefExportNodeDefinition } from "./brief_export_node"
import { projectExportNodeDefinition } from "./project_export_node"
import { projectSummaryNodeDefinition } from "./project_summary_node"
import { remoteRenderNodeDefinition } from "./remote_render_node"

export const builtInNodeDefinitions: NodeDefinition[] = [
  briefExportNodeDefinition,
  projectSummaryNodeDefinition,
  projectExportNodeDefinition,
  remoteRenderNodeDefinition,
]

Do not add product nodes by editing the engine loop. The engine reads node definitions from the registry and stays generic.

Recipe 6: Register a New Agent

Create an agent file:

import { DEFAULT_MODEL } from "@red/backend-contract/ai/models"
import type { Id } from "../../../_generated/dataModel"
import { defineAgent } from "../core/types"
import { exportProject } from "../tools/exportProject.tool"
import { projectSummary } from "../tools/projectSummary.tool"
import { readProject } from "../tools/readProject.tool"

export type ProjectAgentContext = {
  projectId?: Id<"projects">
}

const tools = {
  readProject,
  projectSummary,
  exportProject,
}

export const projectAgent = defineAgent<typeof tools, ProjectAgentContext>({
  key: "project",
  resolveContext: async ({ ctx, threadId }) => {
    const project = await ctx.db
      .query("projects")
      .withIndex("by_thread", (q) => q.eq("threadId", threadId))
      .first()

    return { projectId: project?._id }
  },
  instructions: ({ boundaryMemory }) => `You help users manage projects.
${boundaryMemory ? `\nWorkspace memory:\n${boundaryMemory}` : ""}`,
  modelDefault: DEFAULT_MODEL,
  tools,
})

Add the key to the frontend-safe contract:

export const AGENT_KEYS = ["assistant", "project"] as const

import { projectAgent } from "./project.agent"

export const AGENTS = {
  assistant: assistantAgent,
  project: projectAgent,
} as const satisfies Record<AgentKey, unknown>

Create threads with that agent key from your product UI. Keep the key stable because it is persisted on threads and runs.

Recipe 7: Attachments

Agents can translate attachments into model content with composeUserAttachments. Tools can also read uploaded files through ToolContext.attachments.

execute: async (_input, options) => {
  const toolContext = options.experimental_context as ToolContext<ProjectAgentContext> | undefined
  if (!toolContext) throw new Error("Missing tool context")

  for (const attachment of toolContext.attachments) {
    const buffer = await attachment.loadBuffer({ maxBytes: 10 * 1024 * 1024 })
    const text = new TextDecoder().decode(buffer)
    // Process text here.
  }

  return { processed: toolContext.attachments.length }
}

The Brief assistant uses composeUserAttachments to inline supported text, image, and PDF content when the selected model can handle it.

Checklist

When adding an AI feature:

Decide whether the work is direct sync, sync task-backed, async internal, or async external.
Put product-specific IDs in agent resolveContext, then read them through ToolContext.extras.
Register task nodes in node_types/registry.ts.
Add the agent key to backend-contract/src/ai/agents.ts.
Register the agent in agents.registry.ts.
Run bun run lint.
If backend Convex code changed, run bunx convex dev --once.