py-webhook

Simple webhook sender and receiver for Python/FastAPI. Send webhooks with retry logic, receive with HMAC-SHA256 signature verification.

Features

• HMAC-SHA256 Signatures -- every webhook is signed and verified
• Retry Logic -- configurable retries with exponential backoff
• Async -- built on httpx for non-blocking I/O
• FastAPI Integration -- drop-in Depends() for automatic verification
• Lightweight -- minimal dependencies, no magic

Installation

pip install py-webhook

Quick Start

Sending Webhooks

import asyncio
from pywebhook import WebhookSender

sender = WebhookSender(max_retries=3, timeout=10.0)

async def main():
    response = await sender.send(
        url="https://example.com/webhook",
        event="order.created",
        data={"order_id": 42, "total": 99.99},
        secret="your-shared-secret",
    )
    print(f"Delivered: {response.status_code}")

asyncio.run(main())

The sender automatically:

• Generates a WebhookPayload with a UUID, timestamp, event name, and your data
• Signs the JSON body with HMAC-SHA256
• Sends headers: X-Webhook-Signature, X-Webhook-Event, X-Webhook-Id
• Retries failed requests with exponential backoff

Receiving Webhooks (FastAPI)

from fastapi import Depends, FastAPI
from pywebhook import WebhookReceiver

app = FastAPI()
receiver = WebhookReceiver()

@app.post("/webhook")
async def handle_webhook(payload: dict = Depends(receiver.handler("your-shared-secret"))):
    print(f"Received event: {payload['event']}")
    print(f"Data: {payload['data']}")
    return {"status": "ok"}

The handler dependency automatically:

• Reads the raw request body
• Verifies the X-Webhook-Signature header against the shared secret
• Returns 401 if the signature is invalid
• Parses and returns the JSON payload

Manual Verification

from pywebhook import WebhookReceiver

receiver = WebhookReceiver()

body = b'{"event":"ping","data":{}}'
signature = "abc123..."
is_valid = receiver.verify(body, signature, secret="your-shared-secret")

Payload Model

from pywebhook import WebhookPayload

payload = WebhookPayload(event="user.signup", data={"user_id": 7})
print(payload.id)         # auto-generated UUID
print(payload.timestamp)  # auto-generated ISO 8601
print(payload.event)      # "user.signup"
print(payload.data)       # {"user_id": 7}

Configuration

License

MIT -- see LICENSE for details.