Receive Email in Python with a Signed Webhook

To receive email in Python, enable receiving on a verified Samva domain and point it at an HTTPS endpoint that subscribes to message.received. Samva accepts the mail, parses it, and POSTs a signed JSON event to your app. You verify the signature over the raw request bytes with the standard library, then act on the event. Samva has no Python SDK, so this guide uses Flask for the endpoint and requests for the REST calls, with a FastAPI variant at the end.

Before you start

You need three things from Samva:

• A verified domain with receiving enabled, or a subdomain such as reply.example.com when your team's mailboxes already use the root domain.
• A webhook endpoint subscribed to message.received. Creating it returns the signing secret once; keep it in SAMVA_WEBHOOK_SECRET.
• An API key in SAMVA_API_KEY, sent as X-API-Key, for reading the full email after the event arrives.

pip install flask requests

Verify the signature

Samva follows the Standard Webhooks scheme. The signature is an HMAC-SHA256 over {webhook-id}.{webhook-timestamp}.{body}, keyed with the base64 part of the signing secret. The check below needs only the standard library. It rejects a missing header and a timestamp more than five minutes old, then compares in constant time.

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

TOLERANCE_SECONDS = 300


def verify_webhook(body: bytes, headers) -> tuple[str, dict]:
    """Return (event id, 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 event_id, json.loads(body)

Pass it the exact bytes of the request body, before any JSON parsing. A signature header can hold more than one v1 entry while a secret is rotating, which is why the check accepts any match. The event catalog lists the tolerance and the retry schedule; keep TOLERANCE_SECONDS in step with it.

Accept the event in Flask

from flask import Flask, request

from handle_received import handle_received
from verify_webhook import verify_webhook

app = Flask(__name__)
seen: set[str] = set()  # use a database unique key in production


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

    if event["type"] == "message.received" and event_id not in seen:
        handle_received(event["data"])  # an exception answers 500, so Samva retries
        seen.add(event_id)  # mark only after success
    return "", 204

Three behaviors matter here:

• A request that fails verification gets 400 and never reaches your code.
• Delivery is at least once. webhook-id is stable across retries of the same event, so record it and skip repeats. The in-memory set is for the example only; use a unique key in a database so a restart does not forget what you handled.
• Any non-2xx answer makes Samva retry, up to eight automatic attempts on the published schedule. Let a failure in your own processing raise so the event is delivered again rather than lost.

Read the event, then the full email

This is a real message.received payload; the field list is in the event catalog.

{
  "type": "message.received",
  "timestamp": "2026-01-15T10:05:33.900Z",
  "data": {
    "messageId": "msg_01j8k4n6p8q2r4s6t9vw",
    "organizationId": "org_01j8k0a2b4c6d8e0f2gh",
    "channel": "email",
    "direction": "inbound",
    "from": "ada@example.com",
    "to": "support@your-domain.com",
    "toEmails": ["support@your-domain.com"],
    "subject": "Re: Welcome to Samva",
    "conversationId": "conv_01j8k5p7q9r2s4t6v8wx",
    "hasAttachments": false,
    "isAutoReply": false,
    "body": "Thanks for the welcome. How do I invite a teammate?"
  }
}

body is the HTML when the email has one and the text otherwise. When you need more, GET /v1/messages/{id}/email returns the addresses beyond from and to, the parsed threading headers, each attachment, and, with includeDownloadUrls=true, signed links to the HTML, the text, and the raw MIME.

import os
from pathlib import Path

import requests

API_URL = os.environ.get("SAMVA_API_URL", "https://api.samva.dev")
HEADERS = {"X-API-Key": os.environ["SAMVA_API_KEY"]}


def handle_received(event: dict) -> None:
    # Out-of-office and delivery-status mail is machine mail: record it, never answer it.
    if event.get("isAutoReply"):
        return

    response = requests.get(
        f"{API_URL}/v1/messages/{event['messageId']}/email",
        params={"includeDownloadUrls": "true"},
        headers=HEADERS,
        timeout=10,
    )
    response.raise_for_status()
    email = response.json()
    print(email["fromEmail"], email["subject"], email["conversationId"])

    if not event.get("hasAttachments"):
        return
    directory = Path("attachments")
    directory.mkdir(exist_ok=True)
    for attachment in email["attachments"]:
        if attachment["isInline"]:
            continue  # images embedded in the HTML body
        link = requests.get(
            f"{API_URL}/v1/attachments/{attachment['id']}/download-url",
            headers=HEADERS,
            timeout=10,
        )
        link.raise_for_status()
        file = requests.get(link.json()["url"], timeout=30)
        file.raise_for_status()
        # Name the file by its id: the sender chose `filename`, so never build a path from it.
        (directory / attachment["id"]).write_bytes(file.content)

The download link expires, so fetch the file when you need it and keep your own copy if you must retain it. Check the content type and size before you open a file, and treat the sender's filename as display text. Samva returns the body as the sender wrote it, including any history their mail client quoted below the reply. To answer on the same conversation, follow how a reply finds its conversation.

Use FastAPI

FastAPI parses JSON bodies when you declare a model, which loses the exact bytes. Take the Request and read the body yourself:

import asyncio

from fastapi import FastAPI, HTTPException, Request, Response

from handle_received import handle_received
from verify_webhook import verify_webhook

app = FastAPI()
seen: set[str] = set()  # use a database unique key in production


@app.post("/webhooks/samva")
async def samva_webhook(request: Request) -> Response:
    try:
        event_id, event = verify_webhook(await request.body(), request.headers)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid webhook")

    if event["type"] == "message.received" and event_id not in seen:
        # handle_received makes blocking requests calls, so keep it off the event loop
        await asyncio.to_thread(handle_received, event["data"])
        seen.add(event_id)
    return Response(status_code=204)

handle_received makes blocking requests calls, so the route hands it to a worker thread with asyncio.to_thread. Declaring the route with plain def works too, because FastAPI then runs it in its thread pool.

Test it without sending an email

Sign a fixture the way Samva does and POST it. The script below uses only the standard library, so it works with either server, and it also sends a tampered copy to prove the handler rejects it.

import base64
import hashlib
import hmac
import json
import os
import time
import urllib.error
import urllib.request
import uuid

URL = os.environ.get("TARGET_URL", "http://localhost:3000/webhooks/samva")

body = json.dumps(
    {
        "type": "message.received",
        "timestamp": "2026-01-15T10:05:33.900Z",
        "data": {
            "messageId": "msg_test_1",
            "organizationId": "org_test_1",
            "channel": "email",
            "direction": "inbound",
            "from": "ada@example.com",
            "to": "support@your-domain.com",
            "toEmails": ["support@your-domain.com"],
            "subject": "Re: Welcome",
            "conversationId": "conv_test_1",
            "hasAttachments": False,
            "isAutoReply": True,  # the handler returns before it calls the API
            "body": "How do I invite a teammate?",
        },
    }
).encode()

event_id = f"whevt_{uuid.uuid4()}"
timestamp = str(int(time.time()))
secret = base64.b64decode(os.environ["SAMVA_WEBHOOK_SECRET"].removeprefix("whsec_"))
digest = hmac.new(secret, f"{event_id}.{timestamp}.".encode() + body, hashlib.sha256).digest()
signature = "v1," + base64.b64encode(digest).decode()


def post(payload: bytes) -> int:
    request = urllib.request.Request(
        URL,
        data=payload,
        method="POST",
        headers={
            "content-type": "application/json",
            "webhook-id": event_id,
            "webhook-timestamp": timestamp,
            "webhook-signature": signature,
        },
    )
    try:
        return urllib.request.urlopen(request).status
    except urllib.error.HTTPError as error:
        return error.code


print("signed:", post(body))  # 204
print("tampered:", post(body.replace(b"invite", b"delete")))  # 400

The fixture's messageId does not exist in Samva, so it is marked isAutoReply: the handler returns before it reads the email and the test needs no account. To exercise the read and the attachment download, send a messageId from mail you received and set isAutoReply back to False.

Run the server with flask --app app run --port 3000, then python send_test.py in another terminal. To check the real path from Samva, expose the port with a tunnel and send a webhook.test event; the handler verifies it and answers 204 because it is not a message.received.

What to read next

• Receive, thread, and reply to configure receiving and send an answer on the same conversation.
• Receive email in Node.js for the same endpoint with the TypeScript SDK.
• Send email with Python for the send side over the same REST API.
• Email webhooks for the retry schedule, secret rotation, and redelivery.