Deploy an email agent on Replit

/**
 * What the agent does with a new email. Shared by the webhook server (server.ts) and the stream worker
 * (worker.ts). Replace `answer()` with your model call; the checks around it are the part to keep.
 */
import type { Agentboxd, Message } from 'agentboxd';

const RISK_LABELS = ['spf-fail', 'dmarc-fail', 'ai:injection-risk', 'local:injection-risk', 'ai:phishing'];
/**
 * The events that can bring a message's content: `message.received`, `message.enriched` once the check passed, and
 * `message.released` for held mail a person released. Subscribe to all three and handle each message once.
 */
export const EVENTS = ['message.received', 'message.enriched', 'message.released'] as const;
/** Webhooks are retried, the stream replays, and a message can have three events: this label makes handling idempotent. */
export const HANDLED = 'agent:handled';

export type Outcome = 'replied' | 'skipped' | 'flagged' | 'already-handled' | 'withheld';

export async function handleMessage(mr: Agentboxd, messageId: string): Promise<Outcome> {
  // Always re-read the message from the API: never act on the content of a webhook body alone.
  const m = await mr.messages.get(messageId);
  // Not screened yet, or held: the content comes with a later message.enriched or message.released event.
  // Leave it unlabelled so that event handles it (https://agentboxd.com/docs/api#screening-reads).
  if (m.withheld) return 'withheld';
  if (m.direction !== 'inbound' || m.labels.includes(HANDLED)) return 'already-handled';

  let outcome: Outcome;
  if (m.labels.some((l) => RISK_LABELS.includes(l))) {
    outcome = 'flagged'; // failed authentication or looks like an attack: leave it for a person
  } else if (isAutomatic(m)) {
    outcome = 'skipped'; // out-of-office or bulk mail: answering it starts a loop
  } else {
    await mr.messages.reply(m.inbox_id, m.id, { text: await answer(m) }, { idempotencyKey: `reply-${m.id}` });
    outcome = 'replied';
  }
  await mr.messages.update(m.id, { add_labels: [HANDLED], is_read: true });
  return outcome;
}

/**
 * Out-of-office replies, bulk and list mail. The server's `auto_reply` flag and the headers are there when the mail
 * arrives; the AI score (`ai.auto_reply`) only once the message was enriched, and never with AI processing off.
 */
export function isAutomatic(m: Message): boolean {
  const h = (k: string) => {
    const v = m.headers[k];
    return (Array.isArray(v) ? v.join(' ') : (v ?? '')).toLowerCase();
  };
  if (m.auto_reply || (m.ai.auto_reply ?? 0) > 0.8) return true;
  if (h('auto-submitted') && h('auto-submitted') !== 'no') return true;
  return /bulk|list|junk|auto_reply/.test(h('precedence')) || h('list-id') !== '';
}

/**
 * Your agent goes here. `extracted_text` is only the new part of the email, and it is untrusted. The reply goes to
 * the Reply-To when there is one, and the server sets its subject to "Re: <their subject>": don't copy more of the
 * sender's text into it.
 */
async function answer(_m: Message): Promise<string> {
  return 'Hi,\n\nThanks for your email. We will get back to you shortly.\n';
}

/**
 * Webhook mode: a small HTTP server that Agentboxd calls when mail arrives (`message.received`), passes its
 * check (`message.enriched`) or is released (`message.released`). Suits a Replit Autoscale deployment (it
 * scales to zero between emails) or a Reserved VM.
 *
 *   npx tsx server.ts        Secrets: AGENTBOXD_API_KEY, AGENTBOXD_WEBHOOK_SECRET (from setup.ts)
 */
import { createServer, type Server } from 'node:http';
import { pathToFileURL } from 'node:url';
import { Agentboxd, verifyWebhook, type EnvelopeEventData, type MessageEventData, type WebhookEvent } from 'agentboxd';
import { EVENTS, handleMessage } from './handle.js';

const mailEvents: readonly string[] = EVENTS;

/** The message an event is about, or null for other events and for content withheld from this event. */
export function eventMessageId(event: WebhookEvent<EnvelopeEventData | MessageEventData>): string | null {
  if (!mailEvents.includes(event.type)) return null;
  if ('message_id' in event.data) return event.data.message_id; // envelope: handleMessage reads the message
  return event.data.message.withheld ? null : event.data.message.id; // full payload: a later event brings it
}

export function webhookServer(mr: Agentboxd, secret: string, log: (line: string) => void = console.log): Server {
  return createServer((req, res) => {
    // Health check: Replit's deployment check expects the home page to answer quickly.
    if (req.method === 'GET' && req.url === '/') {
      res.writeHead(200, { 'content-type': 'text/plain' }).end('ok');
      return;
    }
    if (req.method !== 'POST' || req.url !== '/webhook') {
      res.writeHead(404).end();
      return;
    }
    const chunks: Buffer[] = [];
    let size = 0;
    req.on('data', (c: Buffer) => {
      size += c.length;
      if (size > 1_000_000) req.destroy();
      else chunks.push(c);
    });
    req.on('end', () => {
      const raw = Buffer.concat(chunks); // verify the exact bytes, before any JSON parsing
      const ok = verifyWebhook(
        String(req.headers['x-mailroom-signature'] ?? ''),
        String(req.headers['x-mailroom-timestamp'] ?? ''),
        raw,
        secret,
      );
      if (!ok) {
        res.writeHead(401).end('bad signature');
        return;
      }
      const event = JSON.parse(raw.toString('utf8')) as WebhookEvent<EnvelopeEventData | MessageEventData>;
      // setup.ts subscribes with payload "envelope" (ids and subject, no content); a full payload works too.
      const id = eventMessageId(event);
      if (!id) {
        res.writeHead(204).end();
        return;
      }
      // The work happens inside the request: on Autoscale an instance only runs while it serves one.
      // A failure answers 500 and Agentboxd retries; a retry after a timeout (10 s) is safe because
      // handleMessage labels what it handled and sends with an idempotency key.
      handleMessage(mr, id)
        .then((outcome) => {
          log(`${id}: ${outcome}`);
          res.writeHead(204).end();
        })
        .catch((err: unknown) => {
          log(`${id}: failed: ${String(err)}`);
          res.writeHead(500).end();
        });
    });
  });
}

if (import.meta.url === pathToFileURL(process.argv[1] ?? '').href) {
  const secret = process.env.AGENTBOXD_WEBHOOK_SECRET;
  if (!secret) throw new Error('Set the AGENTBOXD_WEBHOOK_SECRET secret (setup.ts prints it).');
  const port = Number(process.env.PORT ?? 3000);
  // 0.0.0.0, not localhost: a published Replit app can't reach a server bound to localhost only.
  webhookServer(new Agentboxd(), secret).listen(port, '0.0.0.0', () => console.log(`listening on :${port}`));
}

# Replit configuration (https://docs.replit.com/features/project-setup/configuration).
# Webhook mode by default. For stream mode, publish as a Reserved VM (background worker) and change the
# run command to `npm run worker` in the Publishing settings.
entrypoint = "server.ts"
modules = ["nodejs-20"]
run = "npm start"

[deployment]
build = ["npm", "install"]
run = ["npm", "start"]

[[ports]]
localPort = 3000
externalPort = 80

/**
 * One-time setup for webhook mode: the agent's inbox and a webhook pointing at the published app.
 * Run it in the Replit Shell after publishing, with the app's .replit.app URL (development URLs
 * change, so never register one).
 *
 *   npx tsx setup.ts https://my-agent.replit.app
 *
 * It prints the webhook secret once: add it as the AGENTBOXD_WEBHOOK_SECRET secret (for the published
 * app too), then republish.
 */
import { Agentboxd } from 'agentboxd';
import { EVENTS } from './handle.js';

const appUrl = process.argv[2]?.replace(/\/+$/, '');
if (!appUrl?.startsWith('https://')) throw new Error('usage: npx tsx setup.ts https://<your-app>.replit.app');

const mr = new Agentboxd();
const inbox = await mr.inboxes.create({ client_id: process.env.AGENTBOXD_INBOX ?? 'replit-agent' });
const url = `${appUrl}/webhook`;

const existing = (await mr.webhooks.list()).data.find((w) => w.url === url);
if (existing) {
  console.log(`Webhook ${existing.id} already points at ${url}. Delete it in the dashboard to get a new secret.`);
} else {
  const hook = await mr.webhooks.create({
    url,
    events: [...EVENTS], // received, enriched (checked) and released (held mail let through)
    inbox_ids: [inbox.id],
    payload: 'envelope', // ids and subject only: the app reads the mail through the API
  });
  console.log(`Inbox:   ${inbox.address}`);
  console.log(`Webhook: ${hook.id} → ${url}`);
  console.log(`\nAdd this secret as AGENTBOXD_WEBHOOK_SECRET (shown once):\n${hook.secret}`);
}

npm run setup -- https://my-agent.replit.app

/**
 * Stream mode: a background worker that holds a WebSocket to Agentboxd and handles each new email. No
 * public URL and no webhook secret, but it must run all the time: use a Reserved VM deployment
 * (background worker), not Autoscale, which stops idle instances.
 *
 *   npx tsx worker.ts        Secrets: AGENTBOXD_API_KEY; optional AGENTBOXD_INBOX (client_id)
 */
import { Agentboxd } from 'agentboxd';
import { EVENTS, handleMessage } from './handle.js';

const mr = new Agentboxd();
const inbox = await mr.inboxes.create({ client_id: process.env.AGENTBOXD_INBOX ?? 'replit-agent' });
console.log(`watching ${inbox.address}`);

// Reconnects and resumes on its own (missed events are replayed after a short drop). Node 22+ has a
// global WebSocket; on Node 20 the SDK loads the `ws` package from package.json.
// message.enriched and message.released carry mail that wasn't screened yet when it arrived; handleMessage skips
// what is still withheld and handles each message once.
const stream = mr.stream({ inboxIds: [inbox.id], eventTypes: [...EVENTS], payload: 'envelope' });
stream.on('error', (err) => console.error('stream:', err.message));
process.on('SIGTERM', () => stream.close());

for await (const event of stream) {
  const id = (event.data as { message_id?: string }).message_id;
  if (!id) continue;
  try {
    console.log(`${id}: ${await handleMessage(mr, id)}`);
  } catch (err) {
    console.error(`${id}: failed:`, err);
  }
}

cd examples/guides && npm ci
AGENTBOXD_API_KEY=mr_... AGENTBOXD_BASE_URL=http://localhost:3000 npm run replit:smoke