KaiMail

How to Test KaiMail Webhooks Using Python

KaiMail Webhook Testing for Developers

KaiMail Webhook Testing for Developers

KaiMail can deliver incoming emails to your application as HTTP POST requests. Instead of polling a mailbox, your code gets called the moment an email arrives. This is useful if you are building something that reacts to email: a support ticket system, an invoice processor or a notification pipeline.

The problem is testing it. Your laptop is not on the public internet, so KaiMail cannot reach it. This tutorial walks through the whole setup: a Python webhook receiver, ngrok to punch a hole through your firewall, and a real email to prove it works.

We will not use and 3rd party dependencies. Everything here uses the Python standard library.

Prerequisites

You need three things:

  1. A KaiMail account with at least one custom domain configured and a mailbox (e.g. [email protected]) that supports webhooks.
  2. Python 3.10 or later
  3. An ngrok account (free tier is fine)

Step 1: Write the Webhook Receiver

We will build a small HTTP server that accepts POST requests and prints whatever KaiMail sends.

The minimal version

Start with the simplest thing that works:

from http.server import BaseHTTPRequestHandler, HTTPServer
import json

class WebhookHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length)

        payload = json.loads(body)
        print(json.dumps(payload, indent=2, ensure_ascii=False))

        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"OK\n")

server = HTTPServer(("0.0.0.0", 8000), WebhookHandler)
server.serve_forever()

This is enough to receive a webhook. Save it, run it with python receiver.py, and it will listen on port 8000. But we want to verify that the request actually came from KaiMail, not from some random bot.

Add signature verification

KaiMail signs every webhook payload with HMAC-SHA256 using the route's webhook secret. The signature is sent in the X-KAI-Webhook-Signature header in the format sha256=<hex_digest>.

To verify it:

  1. Compute the HMAC-SHA256 of the raw request body using the shared secret
  2. Strip the sha256= prefix from the header value
  3. Compare the two hex digests in constant time
import hashlib
import hmac

def verify_signature(payload, secret, received_signature):
    expected = hmac.new(
        secret.encode("utf-8"),
        payload,
        hashlib.sha256,
    ).hexdigest()
    received_hex = received_signature.removeprefix("sha256=")
    return hmac.compare_digest(expected, received_hex)

The removeprefix call is important. The header value is sha256=abc123..., but hmac.new().hexdigest() returns just abc123.... If you compare them without stripping the prefix, verification will always fail. (I know because that is exactly the bug I found in our own sample code.)

We use hmac.compare_digest instead of == to prevent timing attacks.

The complete script

Here is the full receiver with signature verification and CLI arguments. Save this as webhook_test_receiver.py:

#!/usr/bin/env python

import argparse
import hashlib
import hmac
import json
import sys
from datetime import UTC, datetime
from http.server import BaseHTTPRequestHandler, HTTPServer

KAI_HEADERS = [
    "X-KAI-Webhook-Signature",
    "X-KAI-Tracking-ID",
    "X-KAI-Event",
]


def verify_signature(payload, secret, received_signature):
    expected = hmac.new(
        secret.encode("utf-8"),
        payload,
        hashlib.sha256,
    ).hexdigest()
    received_hex = received_signature.removeprefix("sha256=")
    return hmac.compare_digest(expected, received_hex)


class WebhookHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length) if content_length else b""

        now = datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S UTC")
        print(f"\n{'=' * 70}")
        print(f"  POST {self.path}  [{now}]")
        print(f"{'=' * 70}")

        for header in KAI_HEADERS:
            value = self.headers.get(header)
            if value:
                print(f"  {header}: {value}")

        secret = self.server.webhook_secret
        signature = self.headers.get("X-KAI-Webhook-Signature", "")
        if secret:
            if signature:
                valid = verify_signature(body, secret, signature)
                status = "VALID" if valid else "INVALID"
            else:
                status = "MISSING"
            print(f"  Signature: {status}")

        print()

        if body:
            try:
                payload = json.loads(body)
                print(json.dumps(payload, indent=2, ensure_ascii=False))
            except (json.JSONDecodeError, UnicodeDecodeError):
                print(body.decode("utf-8", errors="replace"))

        print()

        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.end_headers()
        self.wfile.write(b"OK\n")
        sys.stdout.flush()

    def log_message(self, format, *args):
        pass


def main():
    parser = argparse.ArgumentParser(
        description="Webhook receiver for testing KaiMail webhooks."
    )
    parser.add_argument(
        "--port", type=int, default=8000, help="Port (default: 8000)"
    )
    parser.add_argument(
        "--secret", type=str, default="", help="Webhook signing secret"
    )
    args = parser.parse_args()

    server = HTTPServer(("0.0.0.0", args.port), WebhookHandler)
    server.webhook_secret = args.secret

    print(f"Listening on http://0.0.0.0:{args.port}")
    if args.secret:
        print("Signature verification: ENABLED")
    else:
        print("Signature verification: DISABLED (use --secret to enable)")
    print("Waiting for POST requests... (Ctrl+C to stop)\n")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down.")
        server.server_close()
        sys.exit(0)


if __name__ == "__main__":
    main()

Step 2: Install and Configure ngrok

ngrok creates a public HTTPS URL that tunnels traffic to your local machine. Install it:

# Ubuntu/Debian
sudo snap install ngrok

# macOS
brew install ngrok

# Or download from https://ngrok.com/download

Sign up for a free account at ngrok.com and add your auth token:

ngrok config add-authtoken YOUR_AUTH_TOKEN

Start the tunnel pointing at your receiver's port:

ngrok http 8000

ngrok will print something like:

Forwarding  https://a1b2c3d4.ngrok-free.app -> http://localhost:8000

Copy that ngrok-free.app HTTPS URL. That is what KaiMail will POST to.

Note: the free tier gives you a random URL that changes every time you restart ngrok. That is fine for testing.

Step 3: Configure KaiMail

Go to the Routes page in KaiMail, edit the mailbox you want to test (e.g. [email protected]), and set the delivery method to Webhook. Paste the ngrok HTTPS URL into the webhook URL field.

Take note of the Signing Secret shown on the route. You will need it to verify signatures.

Step 4: Send a Test Email

Start the receiver with the signing secret:

python webhook_test_receiver.py --port 8000 --secret YOUR_SIGNING_SECRET

Send an email to [email protected]that you registered on KaiMail from any email client, and the email should be sent out to your receiver throught the webhook.

Watch the receiver terminal. Within a few seconds, you should see the full JSON payload printed, with Signature: VALID.

Understanding the Payload

Here is what a typical webhook payload looks like:

{
  "version": "1.0",
  "timestamp": "2025-01-31T10:30:00Z",
  "headers": {
    "From": "[email protected]",
    "To": "[email protected]",
    "Subject": "Quick question",
    "Date": "Fri, 31 Jan 2025 10:30:00 +0000",
    "Message-ID": "<[email protected]>"
  },
  "body_text": "Hi,\n\nDo you offer annual discounts?\n\nThanks,\nAlice",
  "body_html": null,
  "attachments": [],
  "envelope": {
    "sender": "[email protected]",
    "recipient": "[email protected]"
  },
  "account": {
    "email": "[email protected]",
    "domain": "yourdomain.com",
    "mailbox": "[email protected]"
  },
  "metadata": {
    "authentication_results": {}
  }
}

The fields:

  • version: Payload schema version. Currently "1.0".
  • timestamp: When KaiMail processed the email (ISO 8601, UTC).
  • headers: All original email headers. Multi-value headers (like Received) are stored as arrays.
  • body_text: The plain text body, if present. null if the email is HTML-only.
  • body_html: The HTML body, if present. null if the email is text-only. Multipart emails can have both.
  • attachments: Array of attachment objects. Each has filename, content_type, size (bytes), and url (a pre-signed URL that expires after 24 hours). Attachments are stored externally, not embedded in the JSON, so payloads stay small.
  • envelope: The SMTP envelope sender and recipient (these can differ from the From/To headers).
  • account: Your KaiMail account email, the domain, and the specific mailbox that received the email.
  • metadata: Additional data, including email authentication results (SPF, DKIM, ARC).

Responding to Webhooks

Your server must return a 2xx HTTP status code for the delivery to be considered successful.

If your server returns a 5xx error, 408 (Request Timeout), or 429 (Too Many Requests), KaiMail will retry up to 3 times with exponential backoff (5 seconds, 10 seconds, 20 seconds).

Any other 4xx response is treated as a permanent failure. KaiMail will not retry.

If your server does not respond within the timeout window, that is treated the same as a 5xx.

Next Steps

This setup is for development and testing. For production:

  1. Use HTTPS. Your production webhook endpoint should be behind TLS. KaiMail verifies SSL certificates by default.
  2. Persist to a database before processing. Accept the webhook, store it, return 200 immediately, then process asynchronously. This avoids timeouts on slow operations.
  3. Deduplicate using the tracking ID. The X-KAI-Tracking-ID header is unique per delivery. If a retry arrives after your server already processed the first attempt, check the tracking ID before processing again.
  4. Always verify the signature. In production, reject any request with an invalid or missing X-KAI-Webhook-Signature. The test receiver prints the result; a production server should return 403.
← Back to Integration