Send and receive email with Python

Send email and receive replies from Python — one file, requests and Flask.

Create an API key (a test key never sends) and a webhook endpoint in Developers.
Set IOTABOT_API_KEY and IOTABOT_WEBHOOK_SECRET (the endpoint's whsec_… secret) in your environment.
Run the program below. Point the endpoint at /iotabot/webhook on your server.

app.py

# pip install requests flask
import hashlib, hmac, os, time, uuid
import requests
from flask import Flask, abort, jsonify, request

API = "https://api.iotabot.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['IOTABOT_API_KEY']}"}
SECRET = os.environ["IOTABOT_WEBHOOK_SECRET"]   # the endpoint's signing secret (whsec_…)


def send(body: dict) -> dict:
    # One Idempotency-Key per message: reuse it if you retry, so it is never sent twice.
    res = requests.post(f"{API}/messages", json=body,
                        headers={**HEADERS, "Idempotency-Key": str(uuid.uuid4())}, timeout=15)
    data = res.json()
    if not res.ok:
        raise RuntimeError(f"{data['error']['code']}: {data['error']['message']}")
    return data["data"]


# 1. Start a new email thread from one of your verified addresses (GET /v1/email/senders).
try:
    sent = send({
        "channel": "email",
        "to": "priya@example.com",
        "from": "support@acme.com",
        "subject": "Your order #1042",
        "html": "<p>Hi Priya, your order shipped yesterday.</p>",
    })
    print("queued", sent["id"], "in conversation", sent["conversation_id"])
except RuntimeError as err:                        # e.g. template_not_approved — use your own
    print("send refused:", err)


# 2. Receive inbound email as JSON and reply in the same thread.
app = Flask(__name__)


def verified(raw: bytes, header: str) -> bool:
    parts = [p.split("=", 1) for p in header.split(",") if "=" in p]
    t = int(next((v for k, v in parts if k == "t"), "0"))
    if abs(time.time() - t) > 300:                     # older than 5 minutes
        return False
    expected = hmac.new(SECRET.encode(), f"{t}.".encode() + raw, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(v, expected) for k, v in parts if k == "v1")


@app.post("/iotabot/webhook")
def webhook():
    raw = request.get_data()
    event = request.get_json()
    if event["type"] == "endpoint.verification":       # sent once, when you add the endpoint
        return jsonify(challenge=event["data"]["challenge"])
    if not verified(raw, request.headers.get("IotaBot-Signature", "")):
        abort(401)
    if event["type"] == "message.received":
        conversation = event["data"]["conversation"]
        email = event["data"]["message"]["email"]
        print(email["from"], "wrote:", email["subject"])
        send({"conversation_id": conversation["id"], "text": "Thanks — we are on it!"})   # threads onto their email
    return "", 200


if __name__ == "__main__":
    app.run(port=3000)