Using Webhooks for Instant Automation Workflows (with Code Example) – ██FR█████ █INTELL███████████

This content originally appeared on DEV Community and was authored by keith mark

Intro – webhook vs API polling

APIs let you fetch data on demand; polling is when your system repeatedly asks, “Anything new yet?” at a schedule. Polling is simple but inefficient: you waste requests when there’s nothing to fetch, and you discover changes only after your next poll. Webhooks invert the flow. Instead of you polling, external systems call your endpoint immediately when something happens e.g., Stripe charges succeeded, GitHub push events, or a CRM contact update. The result is:

Faster reactions (near real-time).
Lower infrastructure and API costs (no wasteful polling).
More scalable automation (events drive workflows).

A good webhook receiver must be reliable, secure, and fast: accept the request, verify authenticity, enqueue work, and return 2xx quickly.

Explanation diagram in words

Think of a five-step pipeline:

1) External service (Stripe/GitHub) detects an event (charge succeeded, push committed).

2) The service sends an HTTPS POST to your public URL /webhook with a JSON payload and security headers.

3) Your reverse proxy/load balancer forwards the request to your app.

4) Your Flask route receives raw bytes, verifies the signature, parses the JSON, and enqueues a job (e.g., Celery/queue) for downstream processing.

5) Your worker updates databases, triggers emails/Slack, or fans out to other services. The HTTP handler returns 200 quickly to avoid retries.

Code Example: How to build a webhook receiver using Python Flask

Start from a minimal Flask receiver and then harden it.

Install dependencies:

pip install flask requests

Minimal example (works for local testing):

from flask import Flask, request
app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def receive():
    data = request.json
    print(data)
    return '', 200

if __name__ == '__main__':
    app.run(port=5000)

Production-ready version with structured logging, raw-body access, and quick returns:

import json
import logging
import os
from datetime import datetime
from flask import Flask, request, abort

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("webhook")

@app.route('/health', methods=['GET'])
def health():
    return {'status': 'ok', 'time': datetime.utcnow().isoformat()}, 200

@app.route('/webhook', methods=['POST'])
def webhook():
    # Always read raw bytes first; some signature checks require exact bytes
    raw_body = request.get_data(cache=False, as_text=False)
    headers = {k.lower(): v for k, v in request.headers.items()}

    # Parse JSON safely
    try:
        payload = json.loads(raw_body.decode('utf-8') or '{}')
    except json.JSONDecodeError:
        logger.warning("Invalid JSON payload")
        abort(400, description="Invalid JSON")

    # Optionally route by provider (via path, header, or secret separation)
    provider = headers.get('user-agent', 'unknown')

    # TODO: Verify signature here (see dedicated Security section below)
    # verify_provider_signature_or_abort(raw_body, headers)

    # At this point, return 2xx quickly to prevent retries
    logger.info("Accepted webhook from %s: %s", provider, payload.get('type') or 'unknown')
    # Enqueue or trigger async processing (mocked here)
    process_event_async(payload)

    return '', 200

def process_event_async(payload):
    # Replace with Celery/RQ/ThreadPool this is just a placeholder
    logger.info("Queued processing for event: %s", payload.get('type'))

if __name__ == '__main__':
    # Use host='0.0.0.0' in containers; add SSL termination upstream
    app.run(port=int(os.getenv('PORT', '5000')))

Notes:

Read request.get_data() for exact bytes needed in signature verification.
Return fast; do heavy work asynchronously.
Separate endpoints per provider or verify per-request which provider sent the event.

Test sending a request using Python requests.post(…)

You can simulate a webhook locally to validate your route.

import requests

url = "http://localhost:5000/webhook"
payload = {
    "type": "test.event",
    "data": {"message": "Hello, webhook!"}
}
headers = {
    "Content-Type": "application/json",
    # Simulate provider headers if needed
    "User-Agent": "local-tester/1.0"
}

resp = requests.post(url, json=payload, headers=headers, timeout=5)
print(resp.status_code, resp.text)

Tip: Use ngrok or cloudflared to expose your local server to the public internet for real provider callbacks.

Install and run:

pip install pyngrok
ngrok http 5000

You’ll get a URL like https://abcd1234.ngrok.io. Configure providers to target https://abcd1234.ngrok.io/webhook.

Connect to Stripe or GitHub webhooks

Stripe

In the Stripe Dashboard, create an endpoint pointing to your URL /webhook.
Choose events (e.g., payment_intent.succeeded, customer.subscription.updated).
Copy the “Signing secret” for that endpoint.
For local dev, the Stripe CLI can forward events:

# Install: https://stripe.com/docs/stripe-cli
stripe listen --forward-to localhost:5000/webhook

Handle events in code (after signature verification):

def handle_stripe_event(event):
    event_type = event.get('type')
    obj = event.get('data', {}).get('object', {})

    if event_type == 'payment_intent.succeeded':
        payment_intent_id = obj.get('id')
        amount = obj.get('amount_received')
        # update DB, send emails, etc.
    elif event_type == 'charge.refunded':
        pass
    else:
        pass

GitHub

In your repo’s Settings → Webhooks → Add webhook:
- Payload URL: https://yourdomain.com/webhook
- Content type: application/json
- Secret: set a strong secret and store it as GITHUB_WEBHOOK_SECRET.
- Select events (e.g., push, pull_request).
GitHub retries on non-2xx; ensure you return quickly.

Handle GitHub events in code:

def handle_github_event(event, headers):
    event_name = headers.get('x-github-event', 'unknown')
    delivery_id = headers.get('x-github-delivery', 'unknown')

    if event_name == 'push':
        commits = event.get('commits', [])
        # react to new commits
    elif event_name == 'pull_request':
        action = event.get('action')
        # react to PR opened/synchronize/closed
    else:
        pass

You can route in /webhook by checking headers like Stripe-Signature or X-GitHub-Event.

Security: checking signature in code

Stripe signature verification (recommended: official SDK)

Install:

pip install stripe

Code:

import os
import stripe
from flask import abort, request

stripe.api_key = os.getenv('STRIPE_API_KEY')  # for API calls if needed
STRIPE_SIGNING_SECRET = os.getenv('STRIPE_WEBHOOK_SECRET')  # endpoint secret

@app.route('/webhook/stripe', methods=['POST'])
def stripe_webhook():
    payload = request.get_data(cache=False, as_text=False)
    sig_header = request.headers.get('Stripe-Signature', '')

    try:
        event = stripe.Webhook.construct_event(
            payload=payload, sig_header=sig_header, secret=STRIPE_SIGNING_SECRET
        )
    except stripe.error.SignatureVerificationError:
        abort(400, description="Invalid Stripe signature")
    except ValueError:
        abort(400, description="Invalid Stripe payload")

    # Process event securely
    handle_stripe_event(event)
    return '', 200

Why this works:

Stripe signs each payload with your endpoint secret.
The library validates HMAC and timestamp tolerance to prevent replay attacks.

GitHub signature verification (HMAC SHA-256)

GitHub sends X-Hub-Signature-256: sha256=<hex>.

import os
import hmac
import hashlib
from flask import abort, request

GITHUB_WEBHOOK_SECRET = os.getenv('GITHUB_WEBHOOK_SECRET', '')

def verify_github_signature_or_abort(raw_body: bytes, headers: dict):
    signature_header = headers.get('X-Hub-Signature-256') or headers.get('x-hub-signature-256')
    if not signature_header or not signature_header.startswith('sha256='):
        abort(400, description="Missing GitHub signature")

    expected = hmac.new(
        key=GITHUB_WEBHOOK_SECRET.encode('utf-8'),
        msg=raw_body,
        digestmod=hashlib.sha256
    ).hexdigest()

    provided = signature_header.split('=', 1)[1].strip()
    if not hmac.compare_digest(expected, provided):
        abort(400, description="Invalid GitHub signature")

@app.route('/webhook/github', methods=['POST'])
def github_webhook():
    raw_body = request.get_data(cache=False, as_text=False)
    verify_github_signature_or_abort(raw_body, request.headers)
    event = request.get_json(silent=True) or {}
    handle_github_event(event, request.headers)
    return '', 200

Additional hardening:

Rate-limit by IP or provider ASN if feasible.
Enforce HTTPS only; terminate TLS before Flask or use a proper reverse proxy.
Validate content types and required headers.
Idempotency: store processed id (Stripe event.id, GitHub X-GitHub-Delivery) to prevent duplicate handling.
Respond with 2xx only after enqueuing work; if verification fails, return 4xx immediately.

Putting it together: a single route that detects provider

You can keep separate endpoints (cleanest) or branch on headers:

@app.route('/webhook', methods=['POST'])
def unified_webhook():
    raw_body = request.get_data(cache=False, as_text=False)
    headers = request.headers

    if 'Stripe-Signature' in headers:
        # Verify and handle Stripe
        try:
            event = stripe.Webhook.construct_event(
                payload=raw_body,
                sig_header=headers['Stripe-Signature'],
                secret=STRIPE_SIGNING_SECRET
            )
        except Exception:
            abort(400)
        handle_stripe_event(event)
    elif 'X-GitHub-Event' in headers or 'x-github-event' in {k.lower(): v for k, v in headers.items()}:
        verify_github_signature_or_abort(raw_body, headers)
        event = request.get_json(silent=True) or {}
        handle_github_event(event, headers)
    else:
        abort(400, description="Unknown provider")

    return '', 200

Running locally and end-to-end check

1) Start Flask:

python app.py

2) Expose publicly:

ngrok http 5000

3) Configure a test provider:

Stripe: stripe listen --forward-to localhost:5000/webhook/stripe and trigger events from the Dashboard or CLI.
GitHub: Add a webhook to your test repo pointing to the ngrok URL /webhook/github, choose push, set your secret.

4) Watch logs; confirm 200s and downstream processing.

Conclusion

Webhooks turn external events into instant triggers for your automation: faster than polling, cheaper, and more scalable. A reliable receiver does four things well:

Accepts and returns 2xx quickly.
Verifies authenticity with provider signatures.
Routes and enqueues work for asynchronous processing.
Implements idempotency and observability so you can replay or debug safely.

Use the minimal Flask receiver to get started, then adopt signature verification for Stripe and GitHub, separate endpoints where helpful, and queue heavy tasks. With a simple public tunnel for local testing and production-grade security (HTTPS, secrets, idempotency, and rate limiting), you’ll have a robust webhook backbone for everything from payments to CI to CRM automation.

Keep the handler tiny and fast; push work to workers.
Verify signatures using provider-recommended methods.
Store and deduplicate event IDs.
Prefer separate routes per provider for clarity.

Summary:

Built a Flask webhook receiver with both minimal and hardened versions.
Showed local testing via requests.post(...) and tunneling with ngrok.
Integrated with Stripe and GitHub and verified signatures securely.
Outlined best practices: fast returns, async processing, idempotency, HTTPS, and observability.

This content originally appeared on DEV Community and was authored by keith mark