Deploy an agent with the Claude Agent SDK

import type { SessionStore } from "@anthropic-ai/claude-agent-sdk";
import pg from "pg";

export const pool = new pg.Pool({ connectionString: process.env.DB_URL });

// Session transcripts live in PostgreSQL, so any instance can resume any session.
export const sessionStore: SessionStore = {
  async append(key, entries) {
    for (const entry of entries) {
      // The SDK can send an entry twice when it retries, so the uuid deduplicates it.
      await pool.query(
        `INSERT INTO session_entries (project_key, session_id, subpath, uuid, entry)
         VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING`,
        [key.projectKey, key.sessionId, key.subpath ?? "", entry.uuid ?? null, entry],
      );
    }
  },
  async load(key) {
    const { rows } = await pool.query(
      `SELECT entry FROM session_entries
       WHERE project_key = $1 AND session_id = $2 AND subpath = $3 ORDER BY id`,
      [key.projectKey, key.sessionId, key.subpath ?? ""],
    );
    return rows.length > 0 ? rows.map((row) => row.entry) : null;
  },
};

import { pool } from "./db.ts";

await pool.query(`
  CREATE TABLE IF NOT EXISTS orders (
    id integer PRIMARY KEY,
    status text NOT NULL,
    total numeric(10, 2) NOT NULL,
    delivered_on date
  );
  INSERT INTO orders VALUES
    (1001, 'delivered', 59.90, current_date - 10),
    (1002, 'shipped', 120.00, NULL),
    (1003, 'delivered', 18.50, current_date - 90)
  ON CONFLICT DO NOTHING;

  CREATE TABLE IF NOT EXISTS tasks (
    id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
    prompt text NOT NULL,
    session_id text,
    status text NOT NULL DEFAULT 'queued',
    result text,
    cost_usd numeric,
    created_at timestamptz NOT NULL DEFAULT now(),
    updated_at timestamptz NOT NULL DEFAULT now()
  );
  CREATE TABLE IF NOT EXISTS task_events (
    id bigserial PRIMARY KEY,
    task_id uuid NOT NULL REFERENCES tasks ON DELETE CASCADE,
    data jsonb NOT NULL
  );
  CREATE TABLE IF NOT EXISTS session_entries (
    id bigserial PRIMARY KEY,
    project_key text NOT NULL,
    session_id text NOT NULL,
    subpath text NOT NULL,
    uuid text,
    entry jsonb NOT NULL,
    UNIQUE (project_key, session_id, subpath, uuid)
  );
`);
await pool.end();
console.log("Database is ready");

import { join } from "node:path";
import { createSdkMcpServer, query, tool } from "@anthropic-ai/claude-agent-sdk";
import * as z from "zod";
import { pool, sessionStore } from "./db.ts";

// Read-only tools that run in this process. The agent sees them as mcp__store__<name>.
const store = createSdkMcpServer({
  name: "store",
  version: "1.0.0",
  tools: [
    tool(
      "get_order",
      "Get the status, total and delivery date of one order by its number, such as 1001.",
      { order_id: z.number().int().positive() },
      async ({ order_id }) => {
        const { rows } = await pool.query(
          "SELECT id, status, total, delivered_on::text FROM orders WHERE id = $1",
          [order_id],
        );
        if (rows.length === 0) {
          return { content: [{ type: "text", text: `Order ${order_id} not found` }], isError: true };
        }
        return { content: [{ type: "text", text: JSON.stringify(rows[0]) }] };
      },
      { annotations: { readOnlyHint: true } },
    ),
  ],
});

export function runAgent(prompt: string, sessionId: string | null, abortController: AbortController) {
  return query({
    prompt,
    options: {
      model: process.env.AGENT_MODEL ?? "claude-haiku-5-5",
      systemPrompt:
        "You draft replies to customer support tickets for an online store. " +
        "Look up orders with get_order and read the store policies in your working directory before you answer. " +
        "Never promise anything the policies don't allow. Reply with the draft only.",
      // The agent can read the policies in knowledge/ and call get_order, nothing else.
      cwd: join(import.meta.dirname, "knowledge"),
      tools: ["Read", "Grep", "Glob"],
      mcpServers: { store },
      allowedTools: ["mcp__store__get_order"],
      permissionMode: "dontAsk",
      settingSources: [],
      maxTurns: 10,
      maxBudgetUsd: 0.5,
      resume: sessionId ?? undefined,
      sessionStore,
      abortController,
    },
  });
}

# Returns and refunds

- Customers can return unworn items within 30 days of delivery for a full refund.
- Refunds go back to the original payment method within 5 business days of the return arriving.
- Final-sale items can't be returned.

# Shipping

- Orders ship within 2 business days.
- Standard delivery takes 3 to 5 business days after an order ships.

import { setTimeout as sleep } from "node:timers/promises";
import { runAgent } from "./agent.ts";
import { pool } from "./db.ts";

let stopping = false;
let running: AbortController | null = null;

// Sevalla stops an instance 30 seconds after SIGTERM, too soon to finish a run:
// abort it, and the task goes back to the queue for the next worker.
process.on("SIGTERM", () => {
  console.log("SIGTERM received, stopping the worker");
  stopping = true;
  running?.abort();
});

// Claim the oldest queued task. SKIP LOCKED lets more than one worker share the queue.
async function claimTask() {
  const { rows } = await pool.query(`
    UPDATE tasks SET status = 'running', updated_at = now()
    WHERE id = (SELECT id FROM tasks WHERE status = 'queued'
                ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1)
    RETURNING id, prompt, session_id`);
  return rows[0];
}

async function addEvent(taskId: string, data: object) {
  await pool.query("INSERT INTO task_events (task_id, data) VALUES ($1, $2)", [taskId, data]);
}

async function runTask(task: { id: string; prompt: string; session_id: string | null }) {
  running = new AbortController();
  let finished = false;
  let failure = "";
  try {
    for await (const message of runAgent(task.prompt, task.session_id, running)) {
      if (message.type === "assistant") {
        for (const block of message.message.content) {
          if (block.type === "text") await addEvent(task.id, { type: "text", text: block.text });
          if (block.type === "tool_use") await addEvent(task.id, { type: "tool", name: block.name, input: block.input });
        }
      }
      if (message.type === "result") {
        const status = message.subtype === "success" && !message.is_error ? "succeeded" : "failed";
        const result = message.subtype === "success" ? message.result : message.subtype;
        await pool.query(
          `UPDATE tasks SET status = $2, result = $3, session_id = $4, cost_usd = $5, updated_at = now()
           WHERE id = $1`,
          [task.id, status, result, message.session_id, message.total_cost_usd],
        );
        await addEvent(task.id, { type: "done", status, session_id: message.session_id, cost_usd: message.total_cost_usd });
        console.log(
          `Task ${task.id} ${status}: ${message.num_turns} turns, $${message.total_cost_usd}, ` +
            `${message.permission_denials.length} tool calls denied`,
        );
        finished = true;
      }
    }
  } catch (error) {
    // query() also throws after an error result, which is already saved.
    failure = String(error);
  } finally {
    running = null;
  }
  if (finished) return;
  if (stopping) {
    await pool.query("UPDATE tasks SET status = 'queued', updated_at = now() WHERE id = $1", [task.id]);
    await addEvent(task.id, { type: "requeued" });
    console.log(`Task ${task.id} returned to the queue`);
  } else {
    await pool.query("UPDATE tasks SET status = 'failed', result = $2, updated_at = now() WHERE id = $1", [task.id, failure]);
    await addEvent(task.id, { type: "done", status: "failed" });
    console.error(`Task ${task.id} failed: ${failure}`);
  }
}

console.log("Worker started");
while (!stopping) {
  const task = await claimTask();
  if (task) await runTask(task);
  else await sleep(1000);
}
await pool.end();
console.log("Worker stopped");

import { createHash, timingSafeEqual } from "node:crypto";
import { setTimeout as sleep } from "node:timers/promises";
import express from "express";
import type { NextFunction, Request, Response } from "express";
import { pool } from "./db.ts";

const token = process.env.AGENT_TOKEN;
if (!token) throw new Error("AGENT_TOKEN is not set");

// Hashing both sides gives equal lengths, which timingSafeEqual needs.
const expected = createHash("sha256").update(`Bearer ${token}`).digest();
function requireToken(req: Request, res: Response, next: NextFunction) {
  const given = createHash("sha256").update(req.headers.authorization ?? "").digest();
  if (timingSafeEqual(given, expected)) return next();
  res.status(401).set("WWW-Authenticate", "Bearer").json({ error: "invalid_token" });
}

const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
let stopping = false;
const app = express();
app.use(express.json());

app.get("/health", (req, res) => {
  res.send("ok");
});

app.use(requireToken);

// Queue a task. Pass the session_id of an earlier task to continue its conversation.
app.post("/tasks", async (req, res) => {
  const { prompt, session_id } = req.body ?? {};
  if (typeof prompt !== "string" || prompt.length < 1 || prompt.length > 4000) {
    return res.status(400).json({ error: "prompt must be 1 to 4000 characters" });
  }
  if (session_id !== undefined && !uuid.test(session_id)) {
    return res.status(400).json({ error: "session_id must be a UUID" });
  }
  const { rows } = await pool.query(
    "INSERT INTO tasks (prompt, session_id) VALUES ($1, $2) RETURNING id, status",
    [prompt, session_id ?? null],
  );
  res.status(202).json(rows[0]);
});

app.get("/tasks/:id", async (req, res) => {
  const { rows } = uuid.test(req.params.id)
    ? await pool.query("SELECT id, status, result, session_id, cost_usd FROM tasks WHERE id = $1", [req.params.id])
    : { rows: [] };
  if (rows.length === 0) return res.status(404).json({ error: "not_found" });
  res.json(rows[0]);
});

// Stream a task's progress as Server-Sent Events. The events come from the
// database, so any instance can serve them and a client can reconnect with Last-Event-ID.
app.get("/tasks/:id/events", async (req, res) => {
  const { rowCount } = uuid.test(req.params.id)
    ? await pool.query("SELECT 1 FROM tasks WHERE id = $1", [req.params.id])
    : { rowCount: 0 };
  if (!rowCount) return res.status(404).json({ error: "not_found" });

  res.set({ "Content-Type": "text/event-stream", "Cache-Control": "no-cache" });
  res.flushHeaders();
  let lastId = Number(req.get("Last-Event-ID")) || 0;
  let lastWrite = Date.now();
  while (!res.closed && !stopping) {
    const { rows } = await pool.query(
      "SELECT id, data FROM task_events WHERE task_id = $1 AND id > $2 ORDER BY id",
      [req.params.id, lastId],
    );
    for (const row of rows) {
      res.write(`id: ${row.id}\ndata: ${JSON.stringify(row.data)}\n\n`);
      lastId = row.id;
      lastWrite = Date.now();
      if (row.data.type === "done") return res.end();
    }
    // A comment every 15 seconds keeps the connection open while the agent works.
    if (Date.now() - lastWrite >= 15_000) {
      res.write(": keep-alive\n\n");
      lastWrite = Date.now();
    }
    await sleep(1000);
  }
  res.end();
});

const port = Number(process.env.PORT ?? 3000);
const server = app.listen(port, () => {
  console.log(`Agent API listening on port ${port}`);
});

// Open streams end within a second, and clients reconnect to the new instance.
process.on("SIGTERM", () => {
  console.log("SIGTERM received, closing the server");
  stopping = true;
  server.close(() => pool.end());
});

node_modules/

SIGTERM received, stopping the worker
Task d559b1f6-69c2-4c4e-9be2-bb7232a60c62 returned to the queue
Worker stopped

FROM node:24-trixie-slim

WORKDIR /app
ENV NODE_ENV=production PORT=8080

COPY package.json package-lock.json ./
RUN npm ci --omit=dev && npm cache clean --force

COPY . .

USER 1000:1000
EXPOSE 8080
CMD ["node", "server.ts"]

.git
node_modules
npm-debug.log*
.env*
Dockerfile
.dockerignore