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_KEYenvironment variable. - A verified sending domain, or a verified sender address. Samva sends only from addresses you own, so
hello@yourdomain.combelow 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:
channelisemail.fromis optional, with anemailand an optionalname.tois a list of recipients, each{ "email": "..." }.ccandbcctake the same shape.emailholds the content: asubjectwithhtml,text, or both. To send a template instead, replace them withtemplateSlugandtemplateData.
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.
What to read next
- Send an email for attachments, templates, and scheduling.
- The REST API conventions for rate limits and the error format.
- The OpenAPI document at api.samva.dev/v1/openapi.json, if you would rather generate a typed Python client than write requests by hand.