Webhooks

Webhooks are the other half of driving Envless from code: the SDK writes changes, and webhooks tell your systems that a change happened. The client manages the endpoints, and one exported function verifies what arrives.

Create an endpoint#

from envless import envless

endpoint = envless.webhooks.create(
    {
        'name': 'deploy-trigger',
        'url': 'https://acme.com/hooks/envless',
        'eventTypes': ['variable_version.created', 'variable.updated'],
    }
)

print(endpoint['secret'])

Webhooks are workspace-wide, so a key locked to specific projects cannot reach them. Every webhook route, list_event_types included, answers 403 with auth.key_project_restricted for such a key.

Discover the valid event types rather than hard-coding them. An unknown name is refused with a 400 whose message names the offending values.

from envless import envless

catalogue = envless.webhooks.list_event_types()

print(catalogue['totalCount'], catalogue['eventTypes'])

Verify every incoming request#

Verification is mandatory in production. verify_webhook_signature checks the HMAC-SHA256 signature over id.timestamp.body in constant time, checks the timestamp drift, and returns the parsed event. It raises WebhookVerificationError, a ValueError, on anything it cannot prove.

import os

from flask import Flask, Response, request

from envless import WebhookVerificationError, verify_webhook_signature

app = Flask(__name__)


@app.post('/hooks/envless')
def envless_webhook() -> Response:
    try:
        event = verify_webhook_signature(
            payload=request.get_data(),
            headers=dict(request.headers),
            secret=os.environ['ENVLESS_WEBHOOK_SECRET'],
        )
    except WebhookVerificationError:
        return Response(status=401)

    print(event['type'], event['id'])

    return Response(status=204)

import os

from fastapi import FastAPI, Request, Response

from envless import WebhookVerificationError, verify_webhook_signature

app = FastAPI()


@app.post('/hooks/envless')
async def envless_webhook(request: Request) -> Response:
    try:
        event = verify_webhook_signature(
            payload=await request.body(),
            headers=request.headers,
            secret=os.environ['ENVLESS_WEBHOOK_SECRET'],
        )
    except WebhookVerificationError:
        return Response(status_code=401)

    print(event['type'], event['id'])

    return Response(status_code=204)

headers is any mapping from header names to values, and names are matched case-insensitively, so Starlette's request.headers, Django's request.headers and a plain dict all work unchanged. A value may also be a list of strings, in which case the first one is used.

All four are keyword only. A webhook-signature header carrying several space-separated v1, signatures passes when any one of them matches. Verification is CPU only and never touches the network, so it runs the same in sync and async code.

The event shape#

from typing import Any, TypedDict


class ActorKey(TypedDict):
    id: str | None
    name: str | None
    prefix: str | None


class Actor(TypedDict):
    userId: str | None
    name: str | None
    email: str | None
    key: ActorKey | None


class WebhookEvent(TypedDict):
    id: str
    type: str
    apiVersion: str
    createdAt: str
    workspaceId: str
    actor: Actor
    data: dict[str, Any]

That is the shape envless.types.WebhookEvent declares, generic over data. apiVersion is 2026-05-31 today, and actor['key'] names the API key behind the change, or is None when a person made it in the dashboard or the CLI.

verify_webhook_signature returns WebhookEvent[dict[str, Any]]. Once you have switched on type, narrow data with a TypedDict of your own:

from typing import TypedDict, cast

from envless import verify_webhook_signature


class Version(TypedDict):
    id: str
    label: str
    note: str | None
    variableCount: int
    environmentId: str
    createdAt: str


class VersionCreated(TypedDict):
    version: Version


def on_delivery(payload: bytes, headers: dict[str, str], secret: str) -> None:
    event = verify_webhook_signature(payload=payload, headers=headers, secret=secret)

    if event['type'] == 'variable_version.created':
        data = cast('VersionCreated', event['data'])

        print(data['version']['label'])

Idempotency#

Delivery is at-least-once. Track event['id'], which is the webhook-id header, in a store with a TTL and skip duplicates. A delivery is attempted at most 5 times, spaced 30s, 2m, 10m and 30m apart, so the last attempt lands about 43 minutes after the first and 24 hours is plenty.

Inspecting deliveries#

from envless import envless

endpoint_id = '0194f2b1-9c3a-7c21-8f0e-2b6d4a1c5e90'

for delivery in envless.webhooks.list_all_deliveries(endpoint_id, status='failed'):
    detail = envless.webhooks.get_delivery(endpoint_id, delivery['id'])

    print(detail['responseStatus'], detail['responseBody'])

responseBody is only returned by the single-delivery route, truncated at 4000 characters. The list omits it.

Replaying#

from envless import envless

replay = envless.webhooks.redeliver(
    '0194f2b1-9c3a-7c21-8f0e-2b6d4a1c5e90', '0194f2c8-11ad-7a03-b6c2-9d70f4e8a112'
)

print(replay['id'], replay['sourceDeliveryId'])

It only replays finished work. A delivery still pending or processing is refused with a 409, and so is a replay onto a disabled endpoint. status on the response is always pending, and the route answers 201.