DEV Community

Otieno Keith
Otieno Keith Subscriber

Posted on

Using Webhooks for Instant Automation Workflows (with Code Example)

#github #stripe #flask #webhooks

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
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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')))
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode
  • 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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

Running locally and end-to-end check

1) Start Flask:

python app.py
Enter fullscreen mode Exit fullscreen mode

2) Expose publicly:

ngrok http 5000
Enter fullscreen mode Exit fullscreen mode

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.

Top comments (0)