How to let an AI agent send email with MCP, with a person approving each send

{
  "permissions": {
    "ask": [
      "mcp__rasket__send_email",
      "mcp__rasket__send_broadcast",
      "mcp__rasket__delete_contact",
      "mcp__rasket__delete_webhook"
    ],
    "deny": [
      "mcp__rasket__list_received_emails",
      "mcp__rasket__get_received_email"
    ]
  }
}

// gate.mjs: a human approval gate around an MCP client's callTool.
// npm i @modelcontextprotocol/sdk@1.31.0   (Node 22+)
import { createHash } from "node:crypto";
import { readFile, rename, writeFile } from "node:fs/promises";
import { createInterface } from "node:readline/promises";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

const API = process.env.RASKET_API ?? "https://api.rasket.com";
// Written by sign-in.mjs. One process per file: two refreshing the same token revoke it.
const TOKENS_FILE = process.env.RASKET_TOKENS ?? "tokens.json";
const GATED = new Set(["send_email", "send_broadcast", "delete_contact", "delete_webhook"]);
const list = (value) =>
  value
    .split(",")
    .map((item) => item.trim().toLowerCase())
    .filter(Boolean);
const ALLOWED_TO = list(process.env.ALLOWED_RECIPIENT_DOMAINS ?? "example.com");
const ALLOWED_FROM = list(process.env.ALLOWED_SENDERS ?? "orders@mail.acme.example");

let tokens = JSON.parse(await readFile(TOKENS_FILE, "utf8"));
if (!tokens.access_token || !tokens.refresh_token) {
  throw new Error(`No tokens in ${TOKENS_FILE}: run sign-in.mjs`);
}

// Access tokens last an hour. Refresh a minute early, once even when the model calls tools in
// parallel, and save the new refresh token before using it: the old one is now dead, and
// presenting it again revokes the whole connection.
let refreshing = null;
function accessToken() {
  if (Date.now() < tokens.expires_at - 60_000) return Promise.resolve(tokens.access_token);
  refreshing ??= refresh().finally(() => (refreshing = null));
  return refreshing;
}

async function refresh() {
  const res = await fetch(`${API}/oauth/token`, {
    method: "POST",
    headers: { "content-type": "application/x-www-form-urlencoded" },
    body: new URLSearchParams({
      grant_type: "refresh_token",
      client_id: tokens.client_id,
      refresh_token: tokens.refresh_token,
      resource: `${API}/mcp`,
    }),
  });
  if (!res.ok) throw new Error(`Refresh refused (${res.status}): run sign-in.mjs again`);
  const next = await res.json();
  tokens = { ...tokens, ...next, expires_at: Date.now() + next.expires_in * 1000 };
  await writeFile(`${TOKENS_FILE}.tmp`, JSON.stringify(tokens, null, 2), { mode: 0o600 });
  await rename(`${TOKENS_FILE}.tmp`, TOKENS_FILE); // whole file or old file, never half of one
  return tokens.access_token;
}

// One connection per access token, shared by parallel calls, and replaced after a refresh (the old
// one is left to finish its calls: the server keeps no session to close). The SDK
// runs initialize, sends the protocol-version header and reads JSON or SSE replies.
let connection = null;
async function connected() {
  const token = await accessToken();
  if (connection?.token !== token) {
    const client = new Client({ name: "acme-agent", version: "1.0.0" });
    const transport = new StreamableHTTPClientTransport(new URL(`${API}/mcp`), {
      requestInit: { headers: { authorization: `Bearer ${token}` } },
    });
    const entry = { token };
    entry.ready = client.connect(transport).then(
      () => client,
      (error) => {
        if (connection === entry) connection = null; // a failed connect is retried next call
        throw error;
      },
    );
    connection = entry;
  }
  return connection.ready;
}

const refuse = (text) => ({ content: [{ type: "text", text }], isError: true });
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const asArray = (value) => (value === undefined ? [] : [value].flat());
// One address, bare or as "Name <address>". Anything else (two addresses, a comma list, a second
// "<") is unreadable, and an unreadable address is refused rather than guessed at.
const addressOf = (value) =>
  /^(?:[^<>,@]*<)?([^\s<>,@]+@[^\s<>,@]+)>?$/.exec(String(value).trim())?.[1].toLowerCase() ?? null;

export async function askInTerminal(name, args) {
  console.log(`The agent wants to call ${name}:\n${JSON.stringify(args, null, 2)}`);
  const prompt = createInterface({ input: process.stdin, output: process.stdout });
  const answer = await prompt.question("Type yes to approve: ");
  prompt.close();
  return answer.trim().toLowerCase() === "yes";
}

export const listTools = async () => (await connected()).listTools();
export const close = async () => (await connection?.ready)?.close();

export async function callTool(name, args, approve = askInTerminal) {
  if (name === "send_email") {
    // from is required here: left out, a template's sender would skip this check.
    if (!ALLOWED_FROM.includes(addressOf(args.from ?? ""))) {
      return refuse(`Sender not allowed: ${args.from ?? "(none given)"}`);
    }
    const outside = [args.to, args.cc, args.bcc, args.reply_to]
      .flatMap(asArray)
      .filter((value) => !ALLOWED_TO.includes(addressOf(value)?.split("@")[1]));
    if (outside.length > 0) return refuse(`Not on the recipient allowlist: ${outside.join(", ")}`);
    // A key the model left out: the same arguments within 24 hours then send once, even on a retry.
    const key = createHash("sha256").update(JSON.stringify(args)).digest("hex").slice(0, 32);
    args = { idempotency_key: key, ...args };
  }
  if (GATED.has(name) && !(await approve(name, args))) {
    return refuse("The person declined this call.");
  }

  for (let attempt = 1; ; attempt += 1) {
    try {
      return await (await connected()).callTool({ name, arguments: args });
    } catch (error) {
      // A budget refusal is JSON-RPC -32000 with data.retryAfter: wait that long, retry once.
      const wait = error.code === -32000 ? error.data?.retryAfter : undefined;
      if (attempt === 1 && wait !== undefined) {
        await sleep(wait * 1000);
        continue;
      }
      // It may still have run: resending the same arguments replays it rather than sending twice.
      return refuse(`${name} failed: ${error.message}`);
    }
  }
}

The agent wants to call send_email:
{
  "idempotency_key": "d715b13318399a6dd0448cd632227eed",
  "from": "Acme <orders@mail.acme.example>",
  "to": [
    "customer@example.com"
  ],
  "subject": "Your order #1042 has shipped",
  "text": "Your order is on its way."
}
Type yes to approve: no
{ "content": [{ "type": "text", "text": "The person declined this call." }], "isError": true }

// approve-webhook.mjs: approval from Slack, a web page or a phone, through a service you run.
// It takes POST {name, args} and answers {id}, shows the call to a person, then answers
// GET /<id> with {decision: "pending" | "approved" | "declined"}.
const APPROVALS = process.env.APPROVAL_URL ?? "https://approvals.acme.example/requests";
const headers = {
  authorization: `Bearer ${process.env.APPROVAL_SECRET}`,
  "content-type": "application/json",
};
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

export async function approveByWebhook(
  name,
  args,
  { everyMs = 5_000, giveUpMs = 15 * 60_000 } = {},
) {
  try {
    const created = await fetch(APPROVALS, {
      method: "POST",
      headers,
      body: JSON.stringify({ name, args }),
    });
    if (!created.ok) return false;
    const { id } = await created.json();
    for (const deadline = Date.now() + giveUpMs; Date.now() < deadline;) {
      await sleep(everyMs);
      const res = await fetch(`${APPROVALS}/${encodeURIComponent(id)}`, { headers });
      if (!res.ok) return false;
      const { decision } = await res.json();
      if (decision !== "pending") return decision === "approved";
    }
    return false; // nobody answered in time
  } catch {
    return false; // unreachable service or a malformed reply: never a yes
  }
}

// In the agent loop: await callTool("send_email", args, approveByWebhook);