feat(python-sdk): add webhook verification and event handling (#344)

* feat(python-sdk): add webhook verification and event handling

Add webhook support to the Python SDK matching the JS SDK implementation:
- Add Webhooks class with verify() and construct_event() methods
- Implement HMAC-SHA256 signature verification with timing-safe comparison
- Add timestamp validation with configurable tolerance (default 5 minutes)
- Add comprehensive webhook event types (18 events: email, contact, domain, test)
- Add WebhookVerificationError with typed error codes
- Export webhook constants (headers) and types

* fix(python-sdk): harden webhook parsing and typing

Normalize invalid UTF-8 webhook payloads to INVALID_BODY errors so verify() safely returns false, and narrow base email webhook event types to avoid discriminated-union overlap. Add regression tests for both paths.

* chore(python-sdk): bump package version to 0.2.9

* feat(python-sdk): add local webhook test example project

Add a runnable Flask receiver and signed webhook sender under packages/python-sdk/example, and link it from the Python SDK README for local verification.

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: KMKoushik

packages/python-sdk/README.md

@@ -84,6 +84,12 @@ else:
     print("ok:", raw)
 ```
 
+## Webhook Local Example
+
+For a runnable webhook verification demo project, see:
+
+- `example/webhook-test-project/README.md`
+
 ## Development
 
 This package is managed with Poetry. Models are maintained in-repo under

packages/python-sdk/example/webhook-test-project/README.md

@@ -0,0 +1,43 @@
+# Webhook Test Project (Flask)
+
+This example project helps you validate Python SDK webhook signature verification locally.
+
+## What it includes
+
+- `receiver.py`: local webhook endpoint that verifies and parses events
+- `send_test_webhook.py`: sends a signed test webhook request to your local endpoint
+
+## Setup
+
+```bash
+cd packages/python-sdk/example/webhook-test-project
+python -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+```
+
+## Run
+
+Terminal 1:
+
+```bash
+python receiver.py
+```
+
+Terminal 2:
+
+```bash
+python send_test_webhook.py
+```
+
+You should see:
+
+- `200` response from the receiver
+- parsed webhook event output in the receiver terminal
+
+## Environment variables
+
+- `USESEND_WEBHOOK_SECRET` (default: `whsec_test`)
+- `WEBHOOK_URL` (default: `http://127.0.0.1:8000/webhook`)
+
+Use the same `USESEND_WEBHOOK_SECRET` for both scripts.

packages/python-sdk/example/webhook-test-project/receiver.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import os
+
+from flask import Flask, jsonify, request
+from flask.typing import ResponseReturnValue
+
+from usesend import UseSend, WebhookVerificationError  # type: ignore[import-not-found]
+
+
+WEBHOOK_SECRET = os.getenv("USESEND_WEBHOOK_SECRET", "whsec_test")
+
+app = Flask(__name__)
+usesend = UseSend("us_test")
+webhooks = usesend.webhooks(WEBHOOK_SECRET)
+
+
+@app.post("/webhook")
+def webhook() -> ResponseReturnValue:
+    raw_body = request.get_data()
+
+    try:
+        event = webhooks.construct_event(raw_body, headers=request.headers)
+    except WebhookVerificationError as exc:
+        return jsonify({"ok": False, "code": exc.code, "message": str(exc)}), 400
+
+    print(f"Received event: {event['type']}")
+
+    if event["type"] == "email.bounced":
+        bounce = event["data"].get("bounce", {})
+        print("Bounce details:", bounce)
+
+    return jsonify({"ok": True, "type": event["type"]}), 200
+
+
+if __name__ == "__main__":
+    app.run(host="127.0.0.1", port=8000)

packages/python-sdk/example/webhook-test-project/requirements.txt

@@ -0,0 +1,2 @@
+-e ../..
+flask>=3.0,<4.0

packages/python-sdk/example/webhook-test-project/send_test_webhook.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+import time
+import uuid
+
+import requests
+
+
+WEBHOOK_SECRET = os.getenv("USESEND_WEBHOOK_SECRET", "whsec_test")
+WEBHOOK_URL = os.getenv("WEBHOOK_URL", "http://127.0.0.1:8000/webhook")
+
+
+def _signature(secret: str, timestamp_ms: str, body: str) -> str:
+    digest = hmac.new(
+        secret.encode("utf-8"),
+        f"{timestamp_ms}.{body}".encode("utf-8"),
+        hashlib.sha256,
+    ).hexdigest()
+    return f"v1={digest}"
+
+
+def main() -> None:
+    payload = {
+        "id": f"evt_{uuid.uuid4().hex[:8]}",
+        "type": "email.bounced",
+        "createdAt": "2026-02-08T10:00:00.000Z",
+        "data": {
+            "id": "email_123",
+            "status": "BOUNCED",
+            "from": "sender@example.com",
+            "to": ["recipient@example.com"],
+            "occurredAt": "2026-02-08T10:00:00.000Z",
+            "bounce": {
+                "type": "Permanent",
+                "subType": "General",
+                "message": "Mailbox unavailable",
+            },
+        },
+    }
+
+    body = json.dumps(payload)
+    timestamp = str(int(time.time() * 1000))
+    signature = _signature(WEBHOOK_SECRET, timestamp, body)
+
+    headers = {
+        "Content-Type": "application/json",
+        "X-UseSend-Signature": signature,
+        "X-UseSend-Timestamp": timestamp,
+        "X-UseSend-Event": payload["type"],
+        "X-UseSend-Call": f"call_{uuid.uuid4().hex[:10]}",
+    }
+
+    response = requests.post(WEBHOOK_URL, data=body, headers=headers, timeout=10)
+    print("Status:", response.status_code)
+    print("Body:", response.text)
+
+
+if __name__ == "__main__":
+    main()

packages/python-sdk/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "usesend"
-version = "0.2.8"
+version = "0.2.9"
 description = "Python SDK for the UseSend API"
 authors = ["UseSend"]
 license = "MIT"

packages/python-sdk/tests/test_webhooks.py

@@ -0,0 +1,103 @@
+import hashlib
+import hmac
+import json
+import time
+from typing import get_args, get_type_hints
+
+import pytest
+
+from usesend import types
+from usesend.webhooks import (
+    WEBHOOK_SIGNATURE_HEADER,
+    WEBHOOK_TIMESTAMP_HEADER,
+    WebhookVerificationError,
+    Webhooks,
+)
+
+
+def _sign(secret: str, timestamp: str, body: str) -> str:
+    digest = hmac.new(
+        secret.encode("utf-8"),
+        f"{timestamp}.{body}".encode("utf-8"),
+        hashlib.sha256,
+    ).hexdigest()
+    return f"v1={digest}"
+
+
+def test_verify_returns_false_for_non_utf8_bytes_body() -> None:
+    webhooks = Webhooks("whsec_test")
+    timestamp = str(int(time.time() * 1000))
+
+    is_valid = webhooks.verify(
+        b"\xff",
+        headers={
+            WEBHOOK_SIGNATURE_HEADER: "v1=deadbeef",
+            WEBHOOK_TIMESTAMP_HEADER: timestamp,
+        },
+    )
+
+    assert is_valid is False
+
+
+def test_construct_event_raises_invalid_body_for_non_utf8_bytes() -> None:
+    webhooks = Webhooks("whsec_test")
+    timestamp = str(int(time.time() * 1000))
+
+    with pytest.raises(WebhookVerificationError) as exc:
+        webhooks.construct_event(
+            b"\xff",
+            headers={
+                WEBHOOK_SIGNATURE_HEADER: "v1=deadbeef",
+                WEBHOOK_TIMESTAMP_HEADER: timestamp,
+            },
+        )
+
+    assert exc.value.code == "INVALID_BODY"
+
+
+def test_email_webhook_event_type_excludes_specialized_events() -> None:
+    email_event_type = get_type_hints(types.EmailWebhookEvent)["type"]
+    supported = set(get_args(email_event_type))
+
+    assert "email.delivered" in supported
+    assert "email.bounced" not in supported
+    assert "email.failed" not in supported
+    assert "email.suppressed" not in supported
+    assert "email.opened" not in supported
+    assert "email.clicked" not in supported
+
+
+def test_construct_event_parses_bounced_event_with_valid_signature() -> None:
+    secret = "whsec_test"
+    webhooks = Webhooks(secret)
+    timestamp = str(int(time.time() * 1000))
+
+    payload = {
+        "id": "evt_123",
+        "type": "email.bounced",
+        "createdAt": "2026-02-08T10:00:00.000Z",
+        "data": {
+            "id": "email_123",
+            "status": "BOUNCED",
+            "from": "from@example.com",
+            "to": ["to@example.com"],
+            "occurredAt": "2026-02-08T10:00:00.000Z",
+            "bounce": {
+                "type": "Permanent",
+                "subType": "General",
+            },
+        },
+    }
+    body = json.dumps(payload)
+    signature = _sign(secret, timestamp, body)
+
+    event = webhooks.construct_event(
+        body,
+        headers={
+            WEBHOOK_SIGNATURE_HEADER: signature,
+            WEBHOOK_TIMESTAMP_HEADER: timestamp,
+        },
+    )
+
+    assert event["type"] == "email.bounced"
+    assert event["data"]["bounce"]["type"] == "Permanent"

packages/python-sdk/usesend/__init__.py

@@ -3,6 +3,26 @@
 from .usesend import UseSend, UseSendHTTPError
 from .domains import Domains  # type: ignore
 from .campaigns import Campaigns  # type: ignore
+from .webhooks import (
+    Webhooks,
+    WebhookVerificationError,
+    WEBHOOK_SIGNATURE_HEADER,
+    WEBHOOK_TIMESTAMP_HEADER,
+    WEBHOOK_EVENT_HEADER,
+    WEBHOOK_CALL_HEADER,
+)
 from . import types
 
-__all__ = ["UseSend", "UseSendHTTPError", "types", "Domains", "Campaigns"]
+__all__ = [
+    "UseSend",
+    "UseSendHTTPError",
+    "types",
+    "Domains",
+    "Campaigns",
+    "Webhooks",
+    "WebhookVerificationError",
+    "WEBHOOK_SIGNATURE_HEADER",
+    "WEBHOOK_TIMESTAMP_HEADER",
+    "WEBHOOK_EVENT_HEADER",
+    "WEBHOOK_CALL_HEADER",
+]