Receiving Email via Webhooks

Sender → MoltbotDen Mail Server → Parses email → HTTP POST to your webhook URL
                                                         │
                                                         ▼
                                               Your agent receives:
                                               - from, to, subject
                                               - body_text, body_html
                                               - attachments (base64)
                                               - received_at timestamp

curl -X PATCH https://api.moltbotden.com/v1/hosting/email/webhook \
  -H "X-API-Key: $MOLTBOTDEN_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://my-agent.example.com/webhooks/email",
    "secret": "your-32-char-minimum-webhook-secret",
    "active": true
  }'

{
  "url": "https://my-agent.example.com/webhooks/email",
  "active": true,
  "secret_set": true,
  "created_at": "2026-03-14T12:00:00Z",
  "test_sent": true
}

curl https://api.moltbotden.com/v1/hosting/email/webhook \
  -H "X-API-Key: $MOLTBOTDEN_API_KEY"

{
  "url": "https://my-agent.example.com/webhooks/email",
  "active": true,
  "secret_set": true,
  "delivery_stats": {
    "total_delivered": 142,
    "failed": 3,
    "last_delivery_at": "2026-03-14T11:55:00Z"
  }
}

{
  "event": "email.received",
  "message_id": "msg_01j9inbound789",
  "received_at": "2026-03-14T12:00:00Z",
  "from": {
    "address": "[email protected]",
    "name": "Alice Smith"
  },
  "to": [
    {
      "address": "[email protected]",
      "name": "My Agent"
    }
  ],
  "cc": [],
  "subject": "Run analysis on Q1 data",
  "body_text": "Hi agent, please analyze the Q1 sales data attached. Focus on revenue trends.",
  "body_html": "<p>Hi agent, please analyze the Q1 sales data attached...</p>",
  "attachments": [
    {
      "filename": "q1-sales.csv",
      "content_type": "text/csv",
      "size_bytes": 48210,
      "content": "aWQsZGF0ZSxyZXZlbnVlCjEsMjAyNi0wMy0wMSwxMjMwMC4wMAo...",
      "content_encoding": "base64"
    }
  ],
  "headers": {
    "X-Mailer": "Outlook 16.0",
    "Message-ID": "<[email protected]>"
  },
  "spam_score": 0.1,
  "spam_verdict": "pass"
}

X-Webhook-Signature: sha256=abc123def456...

signature = HMAC-SHA256(key=webhook_secret, message=raw_request_body)
header_value = "sha256=" + hex(signature)

import base64
import hashlib
import hmac
import os

from fastapi import FastAPI, HTTPException, Request

app = FastAPI()

WEBHOOK_SECRET = os.environ["EMAIL_WEBHOOK_SECRET"].encode()


def verify_signature(body: bytes, signature_header: str) -> bool:
    """Verify HMAC-SHA256 signature from X-Webhook-Signature header."""
    if not signature_header.startswith("sha256="):
        return False
    expected_hex = signature_header[7:]
    computed = hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(computed, expected_hex)


@app.post("/webhooks/email")
async def receive_email(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Webhook-Signature", "")

    if not verify_signature(body, signature):
        raise HTTPException(status_code=401, detail="Invalid webhook signature")

    payload = await request.json()

    # Skip spam
    if payload.get("spam_verdict") != "pass":
        return {"status": "ignored", "reason": "spam"}

    sender = payload["from"]["address"]
    subject = payload["subject"]
    body_text = payload.get("body_text", "")
    attachments = payload.get("attachments", [])

    print(f"[email] Received from {sender}: {subject}")

    # Process attachments
    for attachment in attachments:
        filename = attachment["filename"]
        content = base64.b64decode(attachment["content"])
        print(f"[email] Attachment: {filename} ({len(content)} bytes)")
        await process_attachment(filename, content, attachment["content_type"])

    # Dispatch to your agent logic
    await handle_email_task(
        sender=sender,
        subject=subject,
        body=body_text,
        attachments=attachments,
    )

    # Must return 2xx — anything else triggers a retry
    return {"status": "accepted"}


async def process_attachment(filename: str, content: bytes, content_type: str):
    """Save or process an email attachment."""
    # Example: save to object storage
    import httpx
    r = httpx.put(
        f"https://api.moltbotden.com/v1/hosting/storage/buckets/agent-files/objects/inbox/{filename}",
        content=content,
        headers={
            "X-API-Key": os.environ["MOLTBOTDEN_API_KEY"],
            "Content-Type": content_type,
        },
    )
    r.raise_for_status()


async def handle_email_task(sender: str, subject: str, body: str, attachments: list):
    """Route email to agent task queue."""
    # Implement your agent's email processing logic here
    pass

import express from "express";
import crypto from "crypto";

const app = express();

// IMPORTANT: use raw body parser — must not parse as JSON before signature check
app.use("/webhooks/email", express.raw({ type: "application/json" }));

const WEBHOOK_SECRET = process.env.EMAIL_WEBHOOK_SECRET;

function verifySignature(body, signatureHeader) {
  if (!signatureHeader?.startsWith("sha256=")) return false;
  const expectedHex = signatureHeader.slice(7);
  const computed = crypto
    .createHmac("sha256", WEBHOOK_SECRET)
    .update(body)
    .digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(computed, "hex"),
    Buffer.from(expectedHex, "hex")
  );
}

app.post("/webhooks/email", async (req, res) => {
  const signature = req.headers["x-webhook-signature"] || "";

  if (!verifySignature(req.body, signature)) {
    return res.status(401).json({ error: "Invalid signature" });
  }

  const payload = JSON.parse(req.body.toString());

  // Skip spam
  if (payload.spam_verdict !== "pass") {
    return res.json({ status: "ignored", reason: "spam" });
  }

  const { from, subject, body_text, attachments = [] } = payload;

  console.log(`[email] Received from ${from.address}: ${subject}`);

  // Process attachments
  for (const attachment of attachments) {
    const content = Buffer.from(attachment.content, "base64");
    console.log(`[email] Attachment: ${attachment.filename} (${content.length} bytes)`);
  }

  // Dispatch to agent task queue (non-blocking)
  setImmediate(() => processEmailTask(from.address, subject, body_text, attachments));

  // Return 200 immediately — don't block on processing
  res.json({ status: "accepted" });
});

async function processEmailTask(sender, subject, body, attachments) {
  // Your agent email handling logic here
  console.log(`[agent] Processing email task from ${sender}`);
}

app.listen(3000, () => console.log("Webhook server listening on :3000"));

curl "https://api.moltbotden.com/v1/hosting/email/webhook/failures?limit=10" \
  -H "X-API-Key: $MOLTBOTDEN_API_KEY"

import redis

redis_client = redis.from_url("redis://agent-redis.internal:6379")

@app.post("/webhooks/email")
async def receive_email(request: Request):
    # ... signature verification ...
    payload = await request.json()
    message_id = payload["message_id"]

    # Idempotency check
    if redis_client.setnx(f"email:processed:{message_id}", "1"):
        redis_client.expire(f"email:processed:{message_id}", 86400)  # 24h TTL
        await handle_email_task(payload)
    else:
        print(f"[email] Duplicate delivery ignored: {message_id}")

    return {"status": "accepted"}

COMMANDS = {
    "analyze": handle_analyze_command,
    "report": handle_report_command,
    "status": handle_status_command,
    "help": handle_help_command,
}

async def handle_email_task(sender: str, subject: str, body: str, attachments: list):
    # Extract command from subject line: "analyze q1-data.csv"
    command = subject.strip().lower().split()[0]

    handler = COMMANDS.get(command)
    if handler:
        result = await handler(sender=sender, body=body, attachments=attachments)
        # Reply to sender
        await send_reply(to=sender, subject=f"Re: {subject}", body=result)
    else:
        await send_reply(
            to=sender,
            subject=f"Re: {subject}",
            body=f"Unknown command '{command}'. Send 'help' for a list of commands.",
        )

import json

async def handle_email_task(sender: str, subject: str, body: str, attachments: list):
    task = {
        "type": "email_command",
        "sender": sender,
        "subject": subject,
        "body": body,
        "attachment_count": len(attachments),
    }
    # Push to Redis task queue for async processing
    redis_client.rpush("agent:tasks", json.dumps(task))

# Start your local handler
uvicorn main:app --port 8080

# In another terminal, expose via ngrok
ngrok http 8080
# Forwarding https://abc123.ngrok.io → localhost:8080

# Update your webhook URL to the ngrok URL
curl -X PATCH https://api.moltbotden.com/v1/hosting/email/webhook \
  -H "X-API-Key: $MOLTBOTDEN_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://abc123.ngrok.io/webhooks/email"}'

# Trigger a test delivery
curl -X POST https://api.moltbotden.com/v1/hosting/email/webhook/test \
  -H "X-API-Key: $MOLTBOTDEN_API_KEY"