Email infrastructure for Mastra agents

import { createTool } from '@mastra/core/tools';
import { z } from 'zod';

const MM_API = 'https://api.multimail.dev';
const headers = {
  Authorization: `Bearer ${process.env.MULTIMAIL_API_KEY}`,
  'Content-Type': 'application/json',
};

export const sendEmailTool = createTool({
  id: 'send_email',
  description: 'Send an email from a MultiMail mailbox. Respects the mailbox oversight mode — sends may be queued for approval.',
  inputSchema: z.object({
    to: z.string().email(),
    subject: z.string(),
    body: z.string(),
    from_mailbox: z.string().default('[email protected]'),
  }),
  execute: async ({ context }) => {
    const res = await fetch(`${MM_API}/send_email`, {
      method: 'POST',
      headers,
      body: JSON.stringify(context),
    });
    return res.json();
  },
});

export const checkInboxTool = createTool({
  id: 'check_inbox',
  description: 'List recent emails in a MultiMail mailbox.',
  inputSchema: z.object({
    mailbox: z.string().default('[email protected]'),
    limit: z.number().int().min(1).max(50).default(10),
  }),
  execute: async ({ context }) => {
    const params = new URLSearchParams({
      mailbox: context.mailbox,
      limit: String(context.limit),
    });
    const res = await fetch(`${MM_API}/check_inbox?${params}`, { headers });
    return res.json();
  },
});

export const readEmailTool = createTool({
  id: 'read_email',
  description: 'Fetch the full content of a single email by ID.',
  inputSchema: z.object({ email_id: z.string() }),
  execute: async ({ context }) => {
    const res = await fetch(`${MM_API}/read_email/${context.email_id}`, { headers });
    return res.json();
  },
});

import { Agent } from '@mastra/core/agent';
import { openai } from '@ai-sdk/openai';
import { sendEmailTool, checkInboxTool, readEmailTool } from './multimail-tools';

export const supportAgent = new Agent({
  name: 'SupportAgent',
  instructions: `
    You are a customer support agent. You can read the inbox, understand customer issues,
    and draft or send replies. When a reply involves a refund, billing change, or account
    action, send it — the mailbox is configured in gated_send mode so a human will
    approve before delivery. For routine status updates, send directly.
  `,
  model: openai('gpt-4o'),
  tools: {
    send_email: sendEmailTool,
    check_inbox: checkInboxTool,
    read_email: readEmailTool,
  },
});

"cm">// Run the agent
const result = await supportAgent.generate(
  'Check the support inbox and draft a reply to any unanswered tickets from the last 24 hours.'
);
console.log(result.text);

import { createWorkflow, createStep } from '@mastra/core/workflows';
import { z } from 'zod';

const MM_API = 'https://api.multimail.dev';
const headers = {
  Authorization: `Bearer ${process.env.MULTIMAIL_API_KEY}`,
  'Content-Type': 'application/json',
};

const draftOutreach = createStep({
  id: 'draft_outreach',
  inputSchema: z.object({ prospect_email: z.string().email(), context: z.string() }),
  outputSchema: z.object({ message_id: z.string(), status: z.string() }),
  execute: async ({ inputData }) => {
    "cm">// Send with oversight — mailbox is in gated_send mode
    const res = await fetch(`${MM_API}/send_email`, {
      method: 'POST',
      headers,
      body: JSON.stringify({
        to: inputData.prospect_email,
        from_mailbox: '[email protected]',
        subject: 'Following up on your request',
        body: inputData.context,
      }),
    });
    const data = await res.json();
    "cm">// status will be 'pending_approval' if oversight mode requires it
    return { message_id: data.message_id, status: data.status };
  },
});

const waitForApproval = createStep({
  id: 'wait_for_approval',
  inputSchema: z.object({ message_id: z.string(), status: z.string() }),
  outputSchema: z.object({ approved: z.boolean(), message_id: z.string() }),
  execute: async ({ inputData }) => {
    if (inputData.status !== 'pending_approval') {
      return { approved: true, message_id: inputData.message_id };
    }
    "cm">// Poll list_pending until the message is no longer queued
    for (let i = 0; i < 60; i++) {
      await new Promise(r => setTimeout(r, 5000));
      const res = await fetch(`${MM_API}/list_pending`, { headers });
      const { pending } = await res.json();
      const still_waiting = pending.some((m: { id: string }) => m.id === inputData.message_id);
      if (!still_waiting) return { approved: true, message_id: inputData.message_id };
    }
    return { approved: false, message_id: inputData.message_id };
  },
});

export const outreachWorkflow = createWorkflow({
  id: 'gated_outreach',
  inputSchema: z.object({ prospect_email: z.string().email(), context: z.string() }),
  outputSchema: z.object({ approved: z.boolean(), message_id: z.string() }),
})
  .then(draftOutreach)
  .then(waitForApproval)
  .commit();

import { Hono } from 'hono';
import { outreachWorkflow } from './workflows/outreach';
import { mastra } from './mastra';

const app = new Hono();

"cm">// MultiMail posts to this endpoint when a new email arrives
app.post('/webhooks/multimail/inbound', async (c) => {
  const event = await c.req.json();

  if (event.type !== 'email.received') {
    return c.json({ ok: true });
  }

  const { from, subject, body_text, email_id } = event.data;

  "cm">// Trigger a Mastra workflow to handle the inbound message
  const run = await mastra.getWorkflow('support_triage').createRun();
  await run.start({
    inputData: {
      from,
      subject,
      body: body_text,
      email_id,
    },
  });

  return c.json({ ok: true, run_id: run.runId });
});

export default app;