Python Examples

There is no official SDK. These examples need Python 3.9 or later and requests (pip install requests); the webhook receiver uses only the standard library. Put the setup block first; every later snippet uses its api helper.

Setup

The helper returns the payload together with meta (which carries the pagination cursor), raises RouteonApiError on any non-2xx answer, and retries only when the request-rate limiter answered 429 with meta.details.code = "rate_limited", waiting for that response’s Retry-After.

import time
import uuid
from dataclasses import dataclass
from typing import Any, Iterator, Optional

import requests

BASE_URL = "https://emails.routeon.io/api/v1"
API_KEY = "<API_KEY>"  # load it from your secret store; never commit it

session = requests.Session()
session.headers["Authorization"] = f"Bearer {API_KEY}"

@dataclass
class ApiResult:
    status: int
    payload: Any
    meta: dict
    replayed: bool  # True when the response came from the idempotency cache

class RouteonApiError(Exception):
    def __init__(self, status, reason, code, validation_errors, details, retry_after):
        self.status = status
        self.reason = reason
        self.code = code
        self.validation_errors = validation_errors
        self.details = details
        self.retry_after = retry_after
        super().__init__(f"Routeon API {status} {reason}" + (f" ({code})" if code else ""))

def api(method: str, path: str, *, json_body=None, params=None,
        idempotency_key: Optional[str] = None, max_attempts: int = 5) -> ApiResult:
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    attempt = 1
    while True:
        resp = session.request(method, BASE_URL + path, json=json_body, params=params,
                               headers=headers, timeout=30)
        try:
            envelope = resp.json() if resp.content else {}
        except ValueError:
            envelope = {}

        if resp.ok:
            return ApiResult(resp.status_code, envelope.get("payload"), envelope.get("meta") or {},
                             resp.headers.get("Idempotency-Replayed") == "true")

        meta = envelope.get("meta") or {}
        details = meta.get("details") or {}
        code = details.get("code")
        header = resp.headers.get("Retry-After", "")
        retry_after = int(header) if header.isdigit() else None

        # Only the request-rate limiter clears in seconds. Daily caps also answer 429
        # (otp_recipient_daily_cap, transactional_daily_limit_exceeded) and are not retried.
        if resp.status_code == 429 and code == "rate_limited" and attempt < max_attempts:
            time.sleep(retry_after or 1)
            attempt += 1
            continue  # same Idempotency-Key: a rate-limited request never ran

        raise RouteonApiError(resp.status_code, meta.get("reason", f"HTTP_{resp.status_code}"), code,
                              meta.get("validation_error") or [], details, retry_after)

Check your API key

me = api("GET", "/me").payload
print(me["auth_type"], me["scopes"])

A call outside the key’s scopes fails with 403 FORBIDDEN and no code.

Send a transactional email

The template must be active and of class transactional. Derive the idempotency key from your own business event so a retry after a timeout cannot send twice.

sent = api("POST", "/transactional/send", idempotency_key="order-1042-confirmation", json_body={
    "domain_id": "<DOMAIN_ID>",
    "to": "jane.doe@example.com",
    "template_id": "<TRANSACTIONAL_TEMPLATE_ID>",
    "variables": {"first_name": "Jane", "order_number": "1042"},
    "tags": ["order-confirmation", "shop:eu"],
    "metadata": {"order_id": "1042", "customer_tier": "gold"},
})
message_id = sent.payload["message_id"]

To schedule it, add "send_at": "<FUTURE_ISO8601_TIMESTAMP>" (explicit UTC offset, at most 30 days ahead). Cancel a scheduled message that has not been picked up yet:

api("DELETE", f"/transactional/{message_id}")

Send a batch of transactional emails

recipients = [
    {"to": "jane.doe@example.com", "variables": {"first_name": "Jane"}},
    {"to": "john.smith@example.com", "variables": {"first_name": "John"}},
]
batch = api("POST", "/transactional/send-batch", idempotency_key=f"shipping-notices-{uuid.uuid4()}", json_body={
    "domain_id": "<DOMAIN_ID>",
    "template_id": "<TRANSACTIONAL_TEMPLATE_ID>",
    "tags": ["shipping-notice"],
    "items": recipients,
})

# Results are matched by index. Resend failed recipients under a NEW key;
# replaying the old key returns the same failures.
for item in batch.payload["results"]:
    if item["status"] == "failed":
        print(recipients[item["index"]]["to"], item["code"], item.get("errors", []))

Look up messages

message = api("GET", f"/messages/{message_id}").payload
print(message["status"], [e["type"] for e in message["events"]], message["tags"], message["metadata"])

tagged = api("GET", "/messages", params={"tag": "order-confirmation", "per_page": 50})
print(len(tagged.payload), "messages tagged order-confirmation")

template_id is None for a message that carries its own content instead of a template. Event types (delivery, bounce, …) are spelled differently from message statuses (delivered, bounced, …).

Request and verify an OTP code

OTP needs a template of class critical.

otp = api("POST", "/otp/request", idempotency_key=f"otp-{uuid.uuid4()}", json_body={
    "domain_id": "<DOMAIN_ID>",
    "to": "jane.doe@example.com",
    "template_id": "<OTP_TEMPLATE_ID>",
    "purpose": "login",
    "code_length": 6,
    "ttl_seconds": 300,
}).payload

def verify_otp(otp_id: str, code: str) -> str:
    try:
        result = api("POST", "/otp/verify", json_body={"otp_id": otp_id, "code": code}).payload
        return "ok" if result["verified"] else "wrong"
    except RouteonApiError as err:
        # 410: otp_not_found | otp_expired | otp_already_used | otp_attempts_exceeded
        if err.status == 410:
            return "request_new_code"
        raise

print(verify_otp(otp["otp_id"], "<CODE_FROM_USER>"))

Create and send a campaign

The template must be active and of class marketing. sender_profile_id is optional; without it the sender profile routed for campaigns, or your default one, is used.

campaign = api("POST", "/campaigns", idempotency_key="campaign-create-product-update", json_body={
    "name": "Product update",
    "type": "blast",
    "domain_id": "<DOMAIN_ID>",
    "template_id": "<MARKETING_TEMPLATE_ID>",
    "sender_profile_id": "<SENDER_PROFILE_ID>",
    "audience": {"include": [{"kind": "list", "id": "<LIST_ID>"}]},
}).payload

def launch(campaign_id: str) -> dict:
    try:
        return api("POST", f"/campaigns/{campaign_id}/send").payload
    except RouteonApiError as err:
        # 409 without a code: some recipients would get more than one email that day.
        if err.status == 409 and err.code is None:
            print("conflicts:", err.details.get("conflict_count"))
            return api("POST", f"/campaigns/{campaign_id}/send",
                       json_body={"conflict_resolution": "exclude_conflicts"}).payload
        raise  # 409 with a code: invalid_state, campaign_needs_content_review, ...

print(launch(campaign["id"])["status"])  # "sending"

Create a webhook and verify its signature

webhook = api("POST", "/webhooks", json_body={
    "url": "https://hooks.example.com/routeon",
    "events": ["email.delivered", "email.bounced", "email.complained", "contact.unsubscribed"],
    "description": "Delivery status sync",
}).payload
# webhook["secret"] is returned only here. Store it in your secret store.

Each delivery is a POST with X-Timestamp (Unix seconds) and X-Signature, the hex HMAC-SHA256 of "<timestamp>.<raw body>" keyed with the secret. Verify the raw bytes before parsing them. A minimal receiver, as a separate program:

import hashlib
import hmac
import json
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

WEBHOOK_SECRET = "<WEBHOOK_SECRET>"
TOLERANCE_SECONDS = 300

def verify_signature(raw_body: bytes, signature: str, timestamp: str) -> bool:
    if not signature or not timestamp or not timestamp.isdigit():
        return False
    if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
        return False
    expected = hmac.new(WEBHOOK_SECRET.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        raw_body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        if not verify_signature(raw_body, self.headers.get("X-Signature", ""), self.headers.get("X-Timestamp", "")):
            self.send_response(401)
            self.end_headers()
            return
        event = json.loads(raw_body)
        # A delivery that timed out is retried, so deduplicate on event["id"].
        print(event["type"], event["id"])
        self.send_response(204)  # answer 2xx within 10 seconds
        self.end_headers()

HTTPServer(("", 8080), Handler).serve_forever()

See Webhooks for the data of each event type.

Handle errors

The api helper already waits out rate_limited. Everything else raises RouteonApiError. Branch on code (from meta.details.code) first, then on reason, and read validation_errors for VALIDATION_ERROR: some codes, such as send_at_too_far or template_not_active, arrive only as a validation_errors[].message.

try:
    api("POST", "/otp/request", idempotency_key=f"otp-{uuid.uuid4()}", json_body={
        "domain_id": "<DOMAIN_ID>", "to": "jane.doe@example.com",
        "template_id": "<OTP_TEMPLATE_ID>", "purpose": "login",
    })
except RouteonApiError as err:
    if err.code == "otp_resend_cooldown":
        print(f"resend allowed in {err.details.get('retry_after_seconds')} s")
    elif err.code in ("otp_recipient_daily_cap", "transactional_daily_limit_exceeded"):
        print("daily cap reached; do not retry today")
    elif err.code == "rate_limited":
        print("still rate limited after all retries; slow down")
    elif err.reason == "VALIDATION_ERROR":
        for e in err.validation_errors:
            print(f"{e.get('field') or '(request)'}: {e.get('message')}")
    else:
        print(err.status, err.reason, err.code)

See Errors for every code and where it appears.

Page through a list

Cursor-paginated lists return meta.cursor while more pages exist. It is absent on the last page.

def list_all(path: str, **params) -> Iterator[dict]:
    while True:
        page = api("GET", path, params=params)
        yield from page.payload
        cursor = page.meta.get("cursor")
        if not cursor:
            return
        params["cursor"] = cursor

for m in list_all("/messages", tag="order-confirmation", per_page=200):
    print(m["id"], m["status"])