Validating and Downloading Raw Email

Every email.received event carries the original MIME source of the message, either inlined as base64 in the payload or available at a time-limited download URL. Validate the payload against Primitive's JSON schema, then decode or download the raw bytes with a SHA-256 check so you never process corrupted or truncated mail.

You need this when you're building your own webhook receiver (bypassing handle_webhook) and want to validate a payload independently, or when you need the original .eml bytes for archival, re-parsing, or forwarding as an attachment.

Validate a payload against the schema#

Call validate_email_received_event(payload), which checks an arbitrary dict against the canonical email.received JSON schema and returns a fully-typed EmailReceivedEvent on success.

from primitive import validate_email_received_event, WebhookValidationError

payload = {
    "id": "evt_0123...",
    "event": "email.received",
    "version": "2025-12-14",
    "email": {
        # ... full email.received shape
    },
}

try:
    event = validate_email_received_event(payload)
except WebhookValidationError as error:
    print(error.code)  # "SCHEMA_VALIDATION_FAILED"
    raise

handle_webhook and handle_webhook_event call this internally after verifying the signature, so in the normal webhook-handler path you never call it directly, see Verifying Webhook Signatures and Handling Webhook Events. Call it yourself only when you already have a parsed dict from somewhere else (a replay queue, a stored fixture, a different transport) and need the same validation and typed result handle_webhook gives you.

A malformed payload raises WebhookValidationError with code == "SCHEMA_VALIDATION_FAILED", not a bare ValueError or a pydantic.ValidationError, catch that specific exception rather than a broad Exception.

Check whether the raw MIME source is inline#

Call is_raw_included(event): it returns True when the raw email is inlined as base64 in the payload, and False when the message exceeded email.content.raw.max_inline_bytes and is only available at the download URL.

from primitive import is_raw_included

if is_raw_included(event):
    print("raw bytes are inline in email.content.raw.data")
else:
    print("must fetch email.content.download.url instead")

is_raw_included(event) reads event.email.content.raw.included and accepts either a validated EmailReceivedEvent or a plain dict with that shape.

Decode the inline raw email#

Call decode_raw_email(event) to base64-decode email.content.raw.data and verify the result against email.content.raw.sha256, returning the original MIME bytes.

1. 1

   Confirm the raw content is inline#

   Call is_raw_included(event) first. If it returns False, skip to Download and verify a non-inline raw email, decode_raw_email raises when the content isn't inline.

2. 2

   Decode the bytes#

   from primitive import decode_raw_email, RawEmailDecodeError
   
   try:
       raw_bytes = decode_raw_email(event)
   except RawEmailDecodeError as error:
       print(error.code, str(error))
       raise
   

   decode_raw_email base64-decodes email.content.raw.data and, by default, verifies the result against email.content.raw.sha256 before returning it.

3. 3

   Handle the failure modes#

   decode_raw_email raises RawEmailDecodeError with one of these codes:

   Code	Cause
   NOT_INCLUDED	The raw content isn't inline; the error's suggestion points at the download URL to fetch instead.
   INVALID_BASE64	email.content.raw.data isn't valid base64.
   HASH_MISMATCH	The decoded bytes don't match email.content.raw.sha256. Treat this as corrupted or truncated data, not something to retry blindly.

On success, raw_bytes is a bytes object holding the exact MIME source of the message, headers first, ready to hand to an .eml parser or write to disk.

Download and verify a non-inline raw email#

Fetch event.email.content.download.url, then pass the response bytes to verify_raw_email_download to confirm the SHA-256 matches. The URL expires, so check is_download_expired first; there is no automatic hash check on an HTTP response you fetched yourself.

import httpx
from primitive import is_download_expired, get_download_time_remaining, verify_raw_email_download

if is_download_expired(event):
    raise RuntimeError("download URL has expired; the webhook must be redelivered")

remaining_ms = get_download_time_remaining(event)
print(f"{remaining_ms}ms left to download")  # 0 once expired

response = httpx.get(event.email.content.download.url)
response.raise_for_status()

raw_bytes = verify_raw_email_download(response.content, event)

verify_raw_email_download(downloaded, event) hashes the bytes you give it with SHA-256 and compares against email.content.raw.sha256, raising RawEmailDecodeError with code == "HASH_MISMATCH" on any mismatch. It works on bytes, bytearray, or memoryview.

is_download_expired and get_download_time_remaining both read email.content.download.expires_at and default to comparing against the current time; pass an explicit now (milliseconds since epoch) in tests. get_download_time_remaining returns milliseconds and clamps to 0 once the URL has expired.

Next steps#