mails-agent provides three methods for receiving inbound emails, each suited to different architectures. Pick the one that fits your setup, or combine them.

Method	Best for	Requires public URL?
Webhooks	Servers, serverless functions	Yes
SSE Streaming	Long-running agents, local dev	No
Polling	Cron jobs, simple scripts	No

Method 1: Webhooks

mails-agent sends a POST request to your URL whenever a new email arrives. This is the most common pattern for production systems.

Configure a webhook

Set a webhook URL on your mailbox using the CLI or API:

# Via CLI
mails webhook set https://your-server.com/webhook/email

# Via API
curl -X PATCH https://mails-worker.genedai.workers.dev/v1/mailbox \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"webhook_url": "https://your-server.com/webhook/email"}'

Webhook payload

When an email arrives, your endpoint receives a POST with this flat JSON body:

{
  "event": "message.received",
  "email_id": "msg_abc123",
  "mailbox": "[email protected]",
  "from": "[email protected]",
  "subject": "Hello from a human"
}

Signature verification (HMAC-SHA256)

Every webhook request includes an X-Webhook-Signature header. Always verify it to ensure the request is authentic.

The signature is computed as HMAC-SHA256(webhook_secret, raw_request_body) and sent as a hex string.

# Python - verify webhook signature
import hmac, hashlib

def verify_signature(body: bytes, signature_header: str, secret: str) -> bool:
    # Strip the "sha256=" prefix before comparing
    signature = signature_header.replace("sha256=", "")
    expected = hmac.new(
        secret.encode(), body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

# In your Flask/FastAPI handler:
@app.post("/webhook/email")
async def handle_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Webhook-Signature", "")

    if not verify_signature(body, signature, WEBHOOK_SECRET):
        return Response(status_code=401)

    # Webhook payload is flat (no "data" wrapper)
    payload = json.loads(body)
    print(f"New email from {payload['from']}: {payload['subject']}")
    return Response(status_code=200)

Method 2: SSE Streaming

Server-Sent Events let you receive emails in real time without exposing a public URL. Your agent opens a persistent connection to GET /v1/events and receives events as they happen.

curl

# Stream events (runs until you Ctrl-C)
curl -N -H "Authorization: Bearer YOUR_API_KEY" \
  https://mails-worker.genedai.workers.dev/v1/events

Each event arrives as an SSE message:

event: message.received
data: {"event":"message.received","email_id":"msg_abc123","from":"[email protected]","subject":"Hello"}

: keepalive

Python

import requests

url = "https://mails-worker.genedai.workers.dev/v1/events"
headers = {"Authorization": "Bearer YOUR_API_KEY"}

with requests.get(url, headers=headers, stream=True) as resp:
    for line in resp.iter_lines(decode_unicode=True):
        if line.startswith("data: "):
            payload = json.loads(line[6:])
            if "from" in payload:
                print(f"Email from {payload['from']}: {payload['subject']}")
                # Process or reply here

Node.js

const EventSource = require("eventsource");

const es = new EventSource(
  "https://mails-worker.genedai.workers.dev/v1/events",
  { headers: { Authorization: "Bearer YOUR_API_KEY" } }
);

es.addEventListener("message.received", (e) => {
  const email = JSON.parse(e.data);
  console.log(`Email from ${email.from}: ${email.subject}`);
});

Method 3: Polling

The simplest approach: periodically call GET /v1/inbox with a since parameter to fetch only new emails since your last check.

curl

# Fetch recent emails
curl -H "Authorization: Bearer YOUR_API_KEY" \
  "https://mails-worker.genedai.workers.dev/v1/inbox"

Python polling loop

import time, requests

API = "https://mails-worker.genedai.workers.dev"
HEADERS = {"Authorization": "Bearer YOUR_API_KEY"}

while True:
    resp = requests.get(f"{API}/v1/inbox", headers=HEADERS)
    emails = resp.json().get("emails", [])

    for email in emails:
        print(f"New: {email['from_address']} - {email['subject']}")
        # Process each email...

    time.sleep(30)  # Poll every 30 seconds

Sending Replies (Threading)

To send a reply that threads correctly in the recipient's email client, include the in_reply_to field with the original email's ID. mails-agent automatically sets the correct References and In-Reply-To headers.

curl

curl -X POST https://mails-worker.genedai.workers.dev/v1/send \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "from": "[email protected]",
    "to": ["[email protected]"],
    "subject": "Re: Hello from a human",
    "text": "Thanks for reaching out! I am an AI agent and I received your email.",
    "in_reply_to": "msg_abc123"
  }'

Python - full receive-and-reply flow

import requests, json

API = "https://mails-worker.genedai.workers.dev"
HEADERS = {
    "Authorization": "Bearer YOUR_API_KEY",
    "Content-Type": "application/json"
}

# 1. Fetch recent emails
inbox = requests.get(f"{API}/v1/inbox", headers=HEADERS).json()

for email in inbox.get("emails", []):
    # 2. Process the email (your logic here)
    reply_text = f"Got your message about: {email['subject']}"

    # 3. Send a threaded reply
    requests.post(f"{API}/v1/send", headers=HEADERS, json={
        "from": "[email protected]",
        "to": [email["from_address"]],
        "subject": f"Re: {email['subject']}",
        "text": reply_text,
        "in_reply_to": email["id"]
    })
    print(f"Replied to {email['from_address']}")

CLI

# Use mails send with the original sender as recipient
mails send --to [email protected] --subject "Re: Hello" --body "Thanks, I got your email!"

Choosing the right method

Consideration	Webhooks	SSE	Polling
Latency	Near-instant	Near-instant	Depends on interval
Complexity	Medium (need public URL + signature verification)	Low	Lowest
Reliability	High (retries built in)	Medium (reconnect on drop)	High
Firewall-friendly	No (inbound POST)	Yes (outbound GET)	Yes (outbound GET)
Resource usage	Low (event-driven)	Low (one connection)	Higher (repeated requests)

For most AI agent use cases, SSE streaming is the recommended approach -- it gives you real-time delivery without needing to expose a public endpoint or deal with webhook signatures.