Skip to content
Python

Send Email in Python with the REST API and requests

Send an email from Python with requests or httpx: one POST to the Samva REST API, safe retries with an idempotency key, and webhook checks. No SDK needed.

Published

To send an email from Python, POST JSON to https://api.samva.dev/v1/messages with your API key in the X-API-Key header. A 201 response returns the message id, and the email is on its way. Samva has no Python SDK, so this guide uses requests and httpx against the REST API directly. The SDK's email.send is a thin wrapper over this same request, so nothing is lost by calling it yourself.

Before you send

You need two things from Samva:

  • An API key from the dashboard, kept in the SAMVA_API_KEY environment variable.
  • A verified sending domain, or a verified sender address. Samva sends only from addresses you own, so hello@yourdomain.com below must be one of them.

Install requests (or httpx):

pip install requests

Send an email with requests

import os

import requests

API_URL = "https://api.samva.dev/v1/messages"

response = requests.post(
    API_URL,
    headers={"X-API-Key": os.environ["SAMVA_API_KEY"]},
    json={
        "channel": "email",
        "from": {"email": "hello@yourdomain.com", "name": "Acme"},
        "to": [{"email": "ada@example.com"}],
        "email": {
            "subject": "Welcome to Acme",
            "html": "<h1>Welcome!</h1><p>Thanks for joining.</p>",
            "text": "Welcome! Thanks for joining.",
        },
    },
    timeout=10,
)
response.raise_for_status()

message = response.json()
print(message["id"], message["status"])

Run it with python send.py. The body has four parts:

  • channel is email.
  • from is optional, with an email and an optional name.
  • to is a list of recipients, each { "email": "..." }. cc and bcc take the same shape.
  • email holds the content: a subject with html, text, or both. To send a template instead, replace them with templateSlug and templateData.

A successful call returns 201 with the message as it was accepted:

{ "id": "msg_7q2xk9mvt4znw8rh", "status": "pending" }

pending means Samva accepted the email. Delivery is reported separately, as sent, delivered, bounced, or failed. The full request, including attachments, scheduling, and tracking options, is on Send a message.

Retry without sending twice

Networks fail halfway. If a request times out, you cannot tell whether Samva accepted it, and sending again could deliver the email twice. An Idempotency-Key header makes the retry safe: repeating the same request with the same key returns the original message instead of sending another.

Pick one key per logical send, such as the order the receipt belongs to, and reuse it on every retry. Reusing a key with a different body returns 409 Conflict.

import os
import time

import requests

API_URL = "https://api.samva.dev/v1/messages"
RETRYABLE = {429, 500, 502, 503, 504}


class SamvaError(Exception):
    def __init__(self, status: int, body: dict):
        super().__init__(f"{status} {body.get('_tag', 'error')}: {body.get('message', '')}")
        self.status = status
        self.body = body


def send_email(payload: dict, idempotency_key: str, attempts: int = 4) -> dict:
    headers = {
        "X-API-Key": os.environ["SAMVA_API_KEY"],
        "Idempotency-Key": idempotency_key,
    }
    for attempt in range(attempts):
        last = attempt == attempts - 1
        try:
            response = requests.post(API_URL, headers=headers, json=payload, timeout=10)
        except requests.RequestException:
            if last:
                raise
            time.sleep(2**attempt)
            continue

        if response.ok:
            return response.json()

        if response.status_code in RETRYABLE and not last:
            time.sleep(float(response.headers.get("Retry-After", 2**attempt)))
            continue

        try:
            body = response.json()
        except ValueError:
            body = {"message": response.text}
        raise SamvaError(response.status_code, body)

    raise RuntimeError("unreachable")


if __name__ == "__main__":
    message = send_email(
        {
            "channel": "email",
            "from": {"email": "hello@yourdomain.com", "name": "Acme"},
            "to": [{"email": "ada@example.com"}],
            "email": {
                "subject": "Your receipt",
                "html": "<p>Thanks for your order.</p>",
            },
        },
        idempotency_key="receipt-order-1042",
    )
    print(message["id"])

Which statuses to retry, and how long to wait, comes from the error reference. A 429 carries a Retry-After header in seconds, and the function waits that long. Other errors, such as 422 for a body that fails validation, are the caller's to fix, so the function raises them with the error's _tag and message.

The same send with httpx

httpx has the same shape and adds an async client, which suits a FastAPI app:

import asyncio
import os

import httpx


async def main() -> None:
    async with httpx.AsyncClient(
        base_url="https://api.samva.dev/v1",
        headers={"X-API-Key": os.environ["SAMVA_API_KEY"]},
        timeout=10,
    ) as client:
        response = await client.post(
            "/messages",
            json={
                "channel": "email",
                "from": {"email": "hello@yourdomain.com", "name": "Acme"},
                "to": [{"email": "ada@example.com"}],
                "email": {"subject": "Welcome to Acme", "text": "Thanks for joining."},
            },
        )
        response.raise_for_status()
        print(response.json()["id"])


asyncio.run(main())

Know what happened to the email

A 201 says Samva accepted the email, not that the recipient's server did. Two ways to follow it:

  • Poll. GET /v1/messages/{id} returns the message with its current status and deliveries.
  • Receive webhooks. Samva posts a signed event for each outcome. Register an endpoint with POST /v1/webhooks, then check each request before you trust it. The email webhooks guide covers the events and retries.

Samva follows the Standard Webhooks scheme, so the check needs only the standard library. It reads three headers, rejects a stale timestamp, and compares the signature in constant time. Pass it the exact bytes of the request body, before any JSON parsing, or the signature will not match:

import base64
import hashlib
import hmac
import json
import os
import time

TOLERANCE_SECONDS = 300


def verify_webhook(body: bytes, headers) -> dict:
    """Return the parsed event, or raise ValueError when the request is not from Samva."""
    try:
        event_id = headers["webhook-id"]
        timestamp = headers["webhook-timestamp"]
        signature_header = headers["webhook-signature"]
    except KeyError as missing:
        raise ValueError(f"missing header {missing}") from None

    if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
        raise ValueError("timestamp outside the tolerance")

    secret = base64.b64decode(os.environ["SAMVA_WEBHOOK_SECRET"].removeprefix("whsec_"))
    signed = f"{event_id}.{timestamp}.".encode() + body
    expected = base64.b64encode(hmac.new(secret, signed, hashlib.sha256).digest()).decode()

    candidates = [
        entry.removeprefix("v1,")
        for entry in signature_header.split()
        if entry.startswith("v1,")
    ]
    if not any(hmac.compare_digest(expected, candidate) for candidate in candidates):
        raise ValueError("signature does not match")

    return json.loads(body)

Call it from your framework's handler with the raw body. In Flask:

from flask import Flask, request

from verify_webhook import verify_webhook

app = Flask(__name__)


@app.post("/webhooks/samva")
def samva_webhook():
    try:
        event = verify_webhook(request.get_data(), request.headers)
    except ValueError:
        return "Invalid webhook", 400

    if event["type"] == "message.bounced":
        print("bounced:", event["data"]["toEmails"])
    return "", 204

Delivery is at least once, so the same event can arrive twice. Record webhook-id before you act on an event and skip ids you have seen. The event catalog lists the retry schedule and the timestamp tolerance, so keep TOLERANCE_SECONDS in step with it.

Frequently Asked Questions

Related Resources

Send

Ship your first email today.

Transactional and product email through one typed API, with signed events, conversation threading, and deliverability handled.