Webhooks: receiving and sending events

Start with a plain webhook endpoint

A webhook is an HTTP request sent when an event happens. This is the smallest inbound webhook: it accepts any JSON event and returns it, with no signature or authentication check. LenAPI does not require HMAC; you can start with this route and add a verification option when your integration needs one.

from typing import Any
from lenapi import Body, LENAPI

app = LENAPI()

@app.post("/webhooks/events")
async def receive_event(payload: Any = Body()):
    # Inspect or validate the event before processing it.
    return {"received": True, "event": payload}

Choose how to verify the sender when needed

Providers retry webhook requests when delivery times out or receives an error. Verify the sender using the scheme it supports, validate each event, acknowledge promptly, and make processing idempotent.

• Use a provider's native signature verification when it publishes a signing secret and signature format.
• Use LenAPI hmac_auth when both systems can share a secret and agree to LenAPI's timestamp, method, path, body-digest, and nonce format.
• If the provider cannot sign requests, require another credential such as a bearer token or a dedicated secret header, and add network restrictions and event deduplication where available.
• If a sender supports no signature or credential, keep the route behind a trusted gateway or limit it to non-sensitive actions.

Verify a provider's raw-body HMAC signature

Use webhook_auth when the sender signs the raw request body in the format supported by this helper. The dependency verifies the signature before your route acts. Configure the provider's header names exactly. If event_id_header and security_store are enabled, LenAPI records accepted event IDs and acknowledges duplicate deliveries with HTTP 200. A timestamp header is optional; when configured, the timestamp and raw body are signed together and checked for freshness.

from typing import Any
from lenapi import Body, Depends, LENAPI
from lenapi.security import webhook_auth
import settings

app = LENAPI()
provider_hook = webhook_auth(
    settings.WEBHOOK_SECRET,  # configure this secret securely; use 32+ bytes
    security_store=app.security_store,
    provider="billing",
    signature_header="x-webhook-signature",
    event_id_header="x-webhook-id",
    timestamp_header="x-webhook-timestamp",
    tolerance_seconds=300,
)

@app.post("/webhooks/billing", auth=provider_hook)
async def receive_billing_event(
    payload: Any = Body(),
    verified=Depends(provider_hook),
):
    # Validate event type and required fields before changing data.
    return {"received": True, "event_id": verified["event_id"]}

Use hmac_auth for a partner-defined request signature

Use hmac_auth when your partner can sign each request using LenAPI's canonical format: timestamp, HTTP method, path and query, SHA-256 digest of the exact body, and unique nonce. This format binds the signature to the route and rejects stale timestamps and replayed nonces. The sender must use the same canonicalization and header names.

from lenapi import Depends, LENAPI, Request
from lenapi.security import hmac_auth
import settings

app = LENAPI()
partner_auth = hmac_auth(
    settings.PARTNER_HMAC_SECRET,
    tolerance_seconds=300,
    nonce_store=app.security_store,
    header_name="X-Partner-Signature",
)

@app.post("/webhooks/partner", auth=partner_auth)
async def receive_partner_event(
    request: Request,
    identity=Depends(partner_auth),
):
    payload = await request.json()  # can be any JSON shape
    return {"accepted": True, "payload": payload}

Accept a sender that does not support HMAC

Some integrations provide only a static bearer token or secret header. Require and compare that credential in constant time, use HTTPS, and rotate the secret if it is exposed. A static key does not sign the payload, prove its freshness, or stop replay by itself. Use the provider's API-key or bearer option when it offers one; never treat an unguessable callback URL as authentication.

import hmac
from typing import Any
from lenapi import Body, Depends, HTTPException, LENAPI, Request
import settings

app = LENAPI()
async def verify_shared_secret(request: Request):
    supplied = request.headers.get("x-provider-key", "")
    expected = settings.PROVIDER_WEBHOOK_KEY
    if not supplied or not hmac.compare_digest(supplied, expected):
        raise HTTPException(401, "Invalid webhook credential")
    return {"provider": "partner"}

@app.post("/webhooks/unsigned-provider", auth=verify_shared_secret)
async def receive_unsigned_event(
    payload: Any = Body(),
    identity=Depends(verify_shared_secret),
):
    # Validate the schema and reject event IDs already processed.
    return {"received": True, "provider": identity["provider"]}

Send an event to another service

For an outbound webhook, POST to the subscriber's configured endpoint and use a timeout. Include a stable event ID so the receiver can deduplicate retries. If the receiving service requires HMAC or an API key, send the credentials and signature format it documents; LenAPI does not force one scheme on outbound requests.

import httpx

async def send_outbound_event(target_url: str, event_id: str, payload: dict):
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.post(
            target_url,
            json=payload,
            headers={"Idempotency-Key": event_id},
        )
        response.raise_for_status()

Acknowledge quickly and handle retries safely

Return a success response after the event is verified and safely accepted. For slow processing, queue an idempotent task instead of holding the provider connection open. Store provider event IDs with a unique constraint or use webhook_auth's security-store deduplication; retries must not repeat charges, emails, or other side effects.

from lenapi import BackgroundTasks, Body, Depends

@app.post("/webhooks/billing", auth=provider_hook)
async def billing_event(
    background_tasks: BackgroundTasks,
    payload: dict = Body(),
    verified=Depends(provider_hook),
):
    background_tasks.add_task(
        process_billing_event, verified["event_id"], payload
    )
    return {"received": True}