An email monitoring bot watches an inbox and takes action when certain emails arrive. Common use cases: alerts when a payment fails, notifications when a customer replies, or triggers when a server sends an error email.
Approach 1: Long-Poll Loop
import httpx, time
BASE = "https://api.agentmailr.com"
KEY = "ak_your_key"
INBOX_ID = "inb_your_inbox"
HEADERS = {"X-API-Key": KEY}
def watch_inbox():
seen = set()
while True:
try:
r = httpx.get(
f"{BASE}/v1/inboxes/{INBOX_ID}/wait",
headers=HEADERS,
params={"timeout": 60},
timeout=65
)
if r.status_code == 200:
email = r.json()
if email["id"] not in seen:
seen.add(email["id"])
handle_email(email)
except httpx.TimeoutException:
continue # No email arrived, poll again
def handle_email(email):
subject = email.get("subject", "")
sender = email["from"]["address"]
print(f"New email from {sender}: {subject}")
# Route by category
category = email["parsed"]["category"]
if category == "verification":
handle_verification(email)
elif category == "security":
handle_security_alert(email)
else:
print("Unhandled category:", category)
watch_inbox()
Approach 2: Webhooks (Production)
For production use, register a webhook instead of polling:
POST /v1/webhooks
{
"url": "https://yourserver.com/email-webhook",
"events": ["email.received"]
}
Then in your Flask or FastAPI handler:
from fastapi import FastAPI, Request
import hmac, hashlib
app = FastAPI()
@app.post("/email-webhook")
async def handle_webhook(req: Request):
body = await req.body()
sig = req.headers.get("X-AgentMailr-Signature")
expected = hmac.new(b"your_webhook_secret", body, hashlib.sha256).hexdigest()
if not hmac.compare_digest(sig, expected):
return {"error": "Invalid signature"}, 401
payload = await req.json()
process_email(payload["data"])
return {"ok": True}
Which to Use
Use long-poll for scripts and simple bots. Use webhooks for production servers that handle email at scale.