Project: Build a Multi-Agent Research System with Web Search

CREATE TABLE research_runs (
  id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  question    text NOT NULL,
  status      text NOT NULL DEFAULT 'queued',  -- queued | running | done | failed
  attempts    int  NOT NULL DEFAULT 0,
  lease_until timestamptz,
  report      text,
  error       text,
  created_at  timestamptz NOT NULL DEFAULT now()
);
 
CREATE TABLE research_steps (
  run_id     uuid NOT NULL REFERENCES research_runs(id),
  step_key   text NOT NULL,
  result     jsonb NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (run_id, step_key)
);

import pg from "pg";
 
export const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

import { pool } from "./db.ts";
 
export async function step<T>(runId: string, key: string, fn: () => Promise<T>): Promise<T> {
  const saved = await pool.query<{ result: T }>("SELECT result FROM research_steps WHERE run_id = $1 AND step_key = $2", [runId, key]);
  if (saved.rows[0]) return saved.rows[0].result;
  const result = await fn();
  await pool.query("INSERT INTO research_steps (run_id, step_key, result) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", [runId, key, JSON.stringify(result)]);
  await pool.query("UPDATE research_runs SET lease_until = now() + interval '5 minutes' WHERE id = $1", [runId]);
  return result;
}
 
export async function claimRun(): Promise<{ id: string; question: string } | null> {
  const { rows } = await pool.query<{ id: string; question: string }>(
    `UPDATE research_runs SET status = 'running', attempts = attempts + 1, lease_until = now() + interval '5 minutes'
      WHERE id = (SELECT id FROM research_runs
                   WHERE status IN ('queued', 'running') AND (lease_until IS NULL OR lease_until < now()) AND attempts < 3
                   ORDER BY created_at LIMIT 1 FOR UPDATE SKIP LOCKED)
      RETURNING id, question`,
  );
  return rows[0] ?? null;
}

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
export interface Source {
  url: string;
  title: string;
}
 
export interface Findings {
  subtask: string;
  summary: string;
  sources: Source[];
  tokens: number;
}
 
export async function researchSubtask(subtask: string, brief: string): Promise<Findings> {
  const messages: Anthropic.MessageParam[] = [{ role: "user", content: brief }];
  const sources = new Map<string, string>();
  let tokens = 0;
  let summary = "";
 
  for (let i = 0; i < 4; i++) {
    const message = await client.messages.create({
      model: "claude-sonnet-5-5",
      max_tokens: 8192,
      system:
        "You are a research subagent. Research only the brief you are given, using web search and page fetching. Prefer primary sources. Reply with at most 200 words of findings, and say clearly what you could not verify.",
      tools: [
        { type: "web_search_20260318", name: "web_search", max_uses: 5 },
        { type: "web_fetch_20260318", name: "web_fetch", max_uses: 3 },
      ],
      messages,
    });
    tokens += message.usage.input_tokens + message.usage.output_tokens;
    messages.push({ role: "assistant", content: message.content });
 
    for (const block of message.content) {
      if (block.type !== "text") continue;
      for (const c of block.citations ?? []) {
        if (c.type === "web_search_result_location") sources.set(c.url, c.title ?? c.url);
      }
    }
    summary = message.content.flatMap((b) => (b.type === "text" ? [b.text] : [])).join("");
    if (message.stop_reason !== "pause_turn") break; // pause_turn: the server paused a long turn; continue it
  }
  return { subtask, summary, sources: [...sources].map(([url, title]) => ({ url, title })), tokens };
}

import Anthropic from "@anthropic-ai/sdk";
import { zodOutputFormat } from "@anthropic-ai/sdk/helpers/zod";
import { z } from "zod";
import { step } from "./durable.ts";
import { researchSubtask, type Findings } from "./subagent.ts";
 
const client = new Anthropic();
const MAX_SUBTASKS = 4;
 
const Plan = z.object({
  subtasks: z
    .array(z.object({ name: z.string(), brief: z.string().describe("Objective, what to look for, output, and boundaries") }))
    .describe(`2 to ${MAX_SUBTASKS} independent subtasks`),
});
 
export async function runResearch(runId: string, question: string): Promise<string> {
  const plan = await step(runId, "plan", async () => {
    const res = await client.messages.parse({
      model: "claude-opus-5-5",
      max_tokens: 4096,
      output_config: { effort: "medium", format: zodOutputFormat(Plan) },
      system: "You plan research. Split the question into independent subtasks for parallel researchers. Each brief must stand alone.",
      messages: [{ role: "user", content: question }],
    });
    if (!res.parsed_output) throw new Error(`No plan (stop_reason: ${res.stop_reason})`);
    return res.parsed_output.subtasks.slice(0, MAX_SUBTASKS);
  });
 
  // Each subtask is its own durable step, so finished ones are never repeated.
  const findings: Findings[] = await Promise.all(
    plan.map((t, i) =>
      step(runId, `subtask-${i}`, async () => {
        try {
          return await researchSubtask(t.name, t.brief);
        } catch (error) {
          return { subtask: t.name, summary: `FAILED: ${error instanceof Error ? error.message : String(error)}`, sources: [], tokens: 0 };
        }
      }),
    ),
  );
 
  return step(runId, "report", async () => {
    // Number sources globally so the report can cite them as [1], [2], …
    const all = [...new Map(findings.flatMap((f) => f.sources).map((s) => [s.url, s])).values()];
    const numbered = all.map((s, i) => `[${i + 1}] ${s.title} (${s.url})`).join("\n");
    const res = await client.messages.create({
      model: "claude-opus-5-5",
      max_tokens: 8192,
      system:
        "You write research reports from your researchers' findings. Cite sources as [n] using the numbered list. Do not add facts that are not in the findings. Call out disagreements and gaps.",
      messages: [
        {
          role: "user",
          content: `Question: ${question}\n\n${findings.map((f) => `<findings subtask="${f.subtask}">\n${f.summary}\n</findings>`).join("\n")}\n\n<sources>\n${numbered}\n</sources>`,
        },
      ],
    });
    const body = res.content.flatMap((b) => (b.type === "text" ? [b.text] : [])).join("");
    return `${body}\n\n## Sources\n${numbered}`;
  });
}

import { claimRun } from "./lib/durable.ts";
import { pool } from "./lib/db.ts";
import { runResearch } from "./lib/research.ts";
 
// One pass: claim and run every runnable job, then exit. Schedule it, or wrap it in a loop.
for (let run = await claimRun(); run; run = await claimRun()) {
  console.log(`run ${run.id}: ${run.question}`);
  try {
    const report = await runResearch(run.id, run.question);
    await pool.query("UPDATE research_runs SET status = 'done', report = $2, lease_until = NULL WHERE id = $1", [run.id, report]);
    console.log("  done");
  } catch (error) {
    // Leave it claimable for a retry after the lease; after 3 attempts claimRun skips it.
    await pool.query("UPDATE research_runs SET error = $2 WHERE id = $1", [run.id, error instanceof Error ? error.message : String(error)]);
    console.error("  failed:", error);
  }
}
await pool.end();

import { z } from "zod";
import { pool } from "@/lib/db";
 
const Body = z.object({ question: z.string().trim().min(10).max(500) });
 
export async function POST(request: Request) {
  const parsed = Body.safeParse(await request.json().catch(() => null));
  if (!parsed.success) return Response.json({ error: "Ask a question of 10 to 500 characters." }, { status: 400 });
  const { rows } = await pool.query<{ id: string }>("INSERT INTO research_runs (question) VALUES ($1) RETURNING id", [parsed.data.question]);
  return Response.json({ id: rows[0]!.id }, { status: 202 });
}

import { pool } from "@/lib/db";
 
export async function GET(_request: Request, { params }: { params: Promise<{ id: string }> }) {
  const { id } = await params;
  const run = await pool.query("SELECT id, question, status, report, error FROM research_runs WHERE id = $1", [id]);
  if (!run.rows[0]) return Response.json({ error: "Not found" }, { status: 404 });
  // Progress for the UI: which steps are finished.
  const steps = await pool.query<{ step_key: string }>("SELECT step_key FROM research_steps WHERE run_id = $1 ORDER BY created_at", [id]);
  return Response.json({ ...run.rows[0], completedSteps: steps.rows.map((s) => s.step_key) });
}