Handling Webhook Events

handle_webhook_event verifies a delivery and returns the full typed event union, so one endpoint can handle inbound mail, x402 payment settlements, and email-native payment interactions. All three families arrive on the same webhook URL, distinguished only by the X-Webhook-Event header.

If your endpoint only ever needs email.received, use the narrower handle_webhook instead, it's hard-typed to that one event and skips the union-narrowing step.

Why the header, not the body, decides the event type#

The header is the only discriminator present on all three families: interaction.* bodies carry no event or type field at all. Primitive posts every webhook with the event name in the X-Webhook-Event header, and the stored payload is sent verbatim with no wrapping envelope, and each event family shapes that payload differently:

Because interaction.* bodies carry no discriminator at all, the header is the only reliable signal across every family. handle_webhook_event reads it for you and overlays a canonical event key onto the parsed result so your code always branches on one field.

Verify and dispatch in one call#

Pass the raw body, the request headers, and your webhook secret to handle_webhook_event, then narrow the returned event with the typed guards.

1. 1

   Import handle_webhook_event and the typed guards#

   from primitive import (
       handle_webhook_event,
       is_payment_settled_event,
       is_payment_failed_event,
       is_interaction_x402_event,
   )
   

2. 2

   Call it with the raw body, headers, and your webhook secret#

   body must be the exact request bytes before any JSON parsing; headers is any mapping-like object (e.g. request.headers from your framework). Get the secret from your Primitive dashboard or GET /account/webhook-secret.

   import os
   
   event = handle_webhook_event(
       body=raw_body,
       headers=request.headers,
       secret=os.environ["PRIMITIVE_WEBHOOK_SECRET"],
   )
   

   This verifies the Primitive-Signature HMAC (or a Standard Webhooks signature, if those headers are present) over the raw body first, then parses the JSON and classifies it using the X-Webhook-Event header. Verification failures raise WebhookVerificationError; a malformed body raises WebhookPayloadError.

3. 3

   Branch on the typed guards#

   if is_payment_settled_event(event):
       # flat fields; amount is in token base units (USDC has 6 decimals)
       print(event["challenge_id"], event["amount"], event["settle_tx"])
   elif is_payment_failed_event(event):
       print(event["challenge_id"], event["failure_reason"])
   elif is_interaction_x402_event(event):
       # interaction.x402.* lifecycle: challenge, payment, settled, rejected, ...
       print(event["event"], event["interaction"])
   elif event.get("event") == "email.received":
       # a normal inbound email; hand off to your ReceivedEmail flow
       ...
   else:
       # UnknownEvent: a future event type the SDK doesn't have a typed
       # shape for yet. Log it and move on rather than raising.
       print("unhandled event:", event.get("event"))
   

   Expected result: each branch fires exactly once per delivery, keyed off the header the platform sent, regardless of whether the body carries event, type, or nothing at all.

The full event catalog#

WEBHOOK_EVENT_TYPES, a tuple in primitive.events, enumerates all 19 event types the X-Webhook-Event header can currently carry: five email.*, two payment.*, and twelve interaction.*.

from primitive.events import WEBHOOK_EVENT_TYPES, is_known_webhook_event_type

print(WEBHOOK_EVENT_TYPES)
# ('email.received', 'email.bounced', 'email.tls_report', 'email.dmarc_report',
#  'email.dmarc_failure', 'payment.settled', 'payment.failed',
#  'interaction.ack.acked', 'interaction.ack.canceled', 'interaction.ack.expired',
#  'interaction.ack.received', 'interaction.ack.requested',
#  'interaction.x402.challenge', 'interaction.x402.declined',
#  'interaction.x402.expired', 'interaction.x402.payment',
#  'interaction.x402.rejected', 'interaction.x402.settled',
#  'interaction.x402.verify_timeout')

is_known_webhook_event_type("payment.settled")  # True
is_known_webhook_event_type("email.future_event")  # False

Three sub-catalogs feed into it, exported separately if you only care about one family:

• EMAIL_EVENT_TYPES, the five first-party email events (subject: an email)
• PAYMENT_EVENT_TYPES, payment.settled, payment.failed (subject: a payment)
• INTERACTION_EVENT_TYPES, the interaction.ack.* and interaction.x402.* step events (subject: an interaction)

Typed shapes for payment and interaction events#

primitive.events exports PaymentEvent, PaymentSettledEvent, PaymentFailedEvent, and InteractionEvent as TypedDicts, so type checkers and your editor know which fields are available after a guard narrows the type.

from primitive.events import (
    InteractionEvent,
    PaymentEvent,
    PaymentFailedEvent,
    PaymentSettledEvent,
)

• PaymentEvent: base shape shared by both settlement outcomes, event, type, challenge_id, network, amount (base units), asset, payer_org
• PaymentSettledEvent: PaymentEvent narrowed to event: Literal["payment.settled"], adds settle_tx
• PaymentFailedEvent: PaymentEvent narrowed to event: Literal["payment.failed"], adds failure_reason
• InteractionEvent: event, interaction (a dict), optional id, covers every interaction.ack.* and interaction.x402.* step

Use the four guard functions rather than checking event["event"] == "..." directly; they handle both dict and dataclass-shaped inputs and keep your dispatch code forward-compatible if the internal representation ever changes.

Forward compatibility: UnknownEvent#

An event type outside the current catalog comes back as an UnknownEvent, the body plus an overlaid event field, instead of raising, so a new Primitive event type never 500s your handler. A payload whose header names a type outside the current catalog is not rejected. parse_webhook_event (called internally by handle_webhook_event) returns it as an UnknownEvent, the body plus an overlaid event field, instead of raising. This is deliberate: Primitive can ship new event types before your SDK version updates, and your handler should degrade gracefully rather than 500.

import logging
import os

from primitive import handle_webhook_event
from primitive.events import is_known_webhook_event_type

logger = logging.getLogger(__name__)

event = handle_webhook_event(
    body=raw_body,
    headers=request.headers,
    secret=os.environ["PRIMITIVE_WEBHOOK_SECRET"],
)

if not is_known_webhook_event_type(event.get("event")):
    # Log for visibility, but don't crash the request.
    logger.info("unrecognized webhook event: %s", event.get("event"))

email.received still needs normalization#

handle_webhook_event hands back the raw EmailReceivedEvent, so call normalize_received_email on it when you need the normalized ReceivedEmail fields. It returns the raw, schema-validated email.received webhook payload (EmailReceivedEvent), not the normalized ReceivedEmail object. If your email.received branch needs sender, reply_target, or thread, pass the event to primitive.received_email.normalize_received_email, see Receiving and Parsing Inbound Email for the full shape.

Legacy alternative: handle_webhook#

handle_webhook runs the same signature verification but is hard-typed to email.received: it validates every body against that schema, so a payment.* or interaction.* delivery raises WebhookValidationError. Keep using it only if your integration genuinely never needs payment or interaction events; otherwise handle_webhook_event is a strict superset with no extra cost.

Next steps#