Webhooks: Complete Developer Guide
Webhooks are user-defined HTTP callbacks that enable real-time event-driven communication by having a server send an HTTP POST request to a registered URL when a specific event occurs, replacing the need for polling.
What You’ll Learn
By the end of this tutorial, you’ll understand how webhooks work, how to implement a webhook consumer with verification, retry logic, and idempotency, and how to use tools like Svix and Webhook.site.
Why Webhooks Matter
Polling wastes resources — your app asks “anything new?” every 30 seconds, and most responses are “no”. Webhooks flip the model: the server tells you when something happens, instantly. Doda Browser uses webhooks to receive real-time malware signature updates from threat intelligence partners.
Webhooks vs Polling
flowchart LR
subgraph "Polling (Inefficient)"
A[Your App] -- "Any new data?" --> B[Server]
B -- "No" --> A
A -- "Any new data?" --> B
B -- "No" --> A
A -- "Any new data?" --> B
B -- "Yes! Here is data" --> A
end
subgraph "Webhook (Efficient)"
C[Your App] -- "Register callback URL" --> D[Server]
D -- "Event! POST data → callback URL" --> C
end
style C fill:#22c55e,color:#fff
How Webhooks Work
- Register your callback URL with the provider (Stripe, GitHub, SendGrid)
- Subscribe to specific events (e.g.,
payment_intent.succeeded,push) - Receive HTTP POST requests with JSON payloads when events occur
- Verify the signature to confirm the request is legitimate
- Respond with 200 OK to acknowledge receipt
Building a Webhook Consumer
# webhook_consumer.py
# Flask webhook receiver with signature verification
from flask import Flask, request, jsonify
import hmac
import hashlib
import json
import time
app = Flask(__name__)
# Shared secret used to sign webhooks
WEBHOOK_SECRET = b'super-secret-key-change-in-production'
def verify_signature(payload_body, signature_header, timestamp):
"""Verify HMAC-SHA256 signature of webhook payload."""
if not signature_header:
return False
# Expected format: t=1723456789,s=abc123def456
parts = {}
for pair in signature_header.split(','):
key, value = pair.strip().split('=', 1)
parts[key] = value
received_sig = parts.get('s', '')
message = f"{parts.get('t', '')}.{payload_body}".encode()
expected_sig = hmac.new(
WEBHOOK_SECRET,
message,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(received_sig, expected_sig)
@app.route('/webhook', methods=['POST'])
def handle_webhook():
"""Receive and process incoming webhook."""
payload = request.get_data(as_text=True)
signature = request.headers.get('X-Webhook-Signature', '')
timestamp = request.headers.get('X-Webhook-Timestamp', '')
# Verify signature
if not verify_signature(payload, signature, timestamp):
return jsonify({"error": "Invalid signature"}), 401
# Parse payload
data = json.loads(payload)
event_type = data.get('event', 'unknown')
print(f"Received webhook event: {event_type}")
# Process based on event type
if event_type == 'payment.succeeded':
handle_payment_success(data)
elif event_type == 'user.created':
handle_user_created(data)
else:
print(f"Unhandled event type: {event_type}")
# Always return 200 to acknowledge
return jsonify({"status": "ok"}), 200
def handle_payment_success(data):
"""Process successful payment event."""
payment_id = data['payment']['id']
amount = data['payment']['amount']
print(f"Payment {payment_id}: ${amount} succeeded")
# Update order status in database
def handle_user_created(data):
"""Process new user creation event."""
user_email = data['user']['email']
print(f"New user registered: {user_email}")
# Send welcome email, create default resources
if __name__ == '__main__':
app.run(port=8080, debug=True)Expected output:
* Running on http://127.0.0.1:8080
Received webhook event: payment.succeeded
Payment pi_3Mqwerty: $49.99 succeededSimulating a Webhook Sender
# webhook_sender.py
# Simulates a webhook provider sending events
import requests
import hmac
import hashlib
import json
import time
WEBHOOK_SECRET = b'super-secret-key-change-in-production'
WEBHOOK_URL = 'http://localhost:8080/webhook'
def send_webhook(event_type, data):
"""Send a signed webhook to the consumer."""
payload = json.dumps({
"event": event_type,
"data": data,
"timestamp": int(time.time()),
})
# Create signature
timestamp = str(int(time.time()))
message = f"{timestamp}.{payload}".encode()
signature = hmac.new(WEBHOOK_SECRET, message, hashlib.sha256).hexdigest()
headers = {
'Content-Type': 'application/json',
'X-Webhook-Signature': f't={timestamp},s={signature}',
'X-Webhook-Timestamp': timestamp,
}
response = requests.post(WEBHOOK_URL, data=payload, headers=headers)
print(f"Sent {event_type}: HTTP {response.status_code}")
return response
# Test events
send_webhook('payment.succeeded', {
"payment": {"id": "pi_3Mqwerty", "amount": 49.99, "currency": "USD"},
"customer": {"email": "alice@example.com"}
})
send_webhook('user.created', {
"user": {"email": "bob@example.com", "name": "Bob Smith"}
})
send_webhook('invalid.event', {
"test": True
})Expected output:
Sent payment.succeeded: HTTP 200
Sent user.created: HTTP 200
Sent invalid.event: HTTP 200Retry Logic with Idempotency
# robust_webhook_consumer.py
# Webhook consumer with retry idempotency
from flask import Flask, request, jsonify
import time
app = Flask(__name__)
# In-memory store of processed webhook IDs (use Redis in production)
processed_webhooks = {}
@app.route('/webhook', methods=['POST'])
def handle_webhook_idempotent():
"""Receive webhook with idempotency guarantees."""
data = request.get_json()
webhook_id = data.get('id')
event_type = data.get('event')
# Idempotency check — ignore duplicate deliveries
if webhook_id in processed_webhooks:
print(f"Duplicate webhook {webhook_id}, already processed")
return jsonify({"status": "already_processed"}), 200
# Process the webhook
try:
print(f"Processing {event_type}: {webhook_id}")
# Business logic here
time.sleep(0.1)
# Mark as processed
processed_webhooks[webhook_id] = {
"status": "completed",
"processed_at": time.time()
}
return jsonify({"status": "ok"}), 200
except Exception as e:
print(f"Failed to process {webhook_id}: {e}")
# Return non-200 to trigger provider retry
return jsonify({"error": str(e)}), 500
# Cleanup old entries periodically (optional)Expected retry behavior for a failing webhook:
Processing payment.succeeded: wh_001
... (processing fails) ...
Processing payment.succeeded: wh_001 (retry 1 after 60s)
... (processing fails) ...
Processing payment.succeeded: wh_001 (retry 2 after 300s)Using Svix for Webhook Management
Svix is a managed webhook gateway that handles delivery, retries, and signing.
# svix_example.py
# Sending webhooks via Svix
import svix
svix_client = svix.Svix("YOUR_SVIX_API_KEY")
# Create an endpoint (register a consumer URL)
endpoint = svix_client.endpoint.create(
"app_123",
{
"url": "https://myapp.com/webhook",
"description": "Production webhook consumer",
"filter_types": ["payment.succeeded", "user.created"],
}
)
# Send a webhook
svix_client.message.create(
"app_123",
{
"eventType": "payment.succeeded",
"payload": {
"payment_id": "pi_3Mqwerty",
"amount": 4999,
"currency": "usd"
}
}
)Common Errors
1. Returning Non-200 Status Codes
If your webhook consumer returns anything other than 2xx, providers retry. Accidentally returning 500 due to a minor error causes repeated deliveries. Always return 200 after receiving, even if you queue processing for later.
2. Not Verifying Signatures
Anyone can send POST requests to your endpoint. Without HMAC verification, attackers can trigger fake events. Always verify the signature before processing.
3. Missing Idempotency Handling
Providers may deliver the same webhook multiple times (network retries, at-least-once semantics). Without idempotency checking, you might process a payment twice, send two welcome emails, or create duplicate accounts.
4. Slow Webhook Processing
Providers have timeouts (typically 5-10 seconds). If your handler takes too long, the provider times out and retries. Acknowledge immediately (200) and process asynchronously.
5. Forgetting to Log Webhook Payloads
When a webhook fails, you need to replay it. Log the full payload (or store it) so you can reprocess events manually. Skip logging sensitive data like credit card numbers — log IDs, not secrets.
6. Hard-Coded Webhook Secrets
Storing secrets in source code is a security risk. Use environment variables or a secret manager (AWS Secrets Manager, HashiCorp Vault). Rotate secrets periodically.
Practice Questions
1. What is the main advantage of webhooks over polling?
Webhooks provide instant notifications without wasting resources on periodic checks. The server pushes data when an event occurs instead of the client repeatedly asking for updates.
2. How do you verify a webhook came from a trusted source?
Use HMAC-SHA256 signing. The provider signs the payload with a shared secret. Your consumer recomputes the signature and compares it using hmac.compare_digest().
3. Why is idempotency important for webhook consumers?
Webhooks can be delivered more than once (at-least-once semantics). Idempotency ensures that processing the same event multiple times has the same effect as processing it once, preventing duplicate actions.
4. What status code should you return after processing a webhook?
200 OK. Return 200 immediately after receiving and validating the webhook. Do the heavy processing asynchronously to avoid provider timeouts.
5. Challenge: Build a webhook system where (a) the provider sends events with HMAC signing and exponential backoff retry, (b) the consumer verifies signatures, checks idempotency via webhook ID, and processes events asynchronously via a task queue.
Implement with Flask for the receiver, Redis for idempotency cache, and Celery for async processing. Store processed webhook IDs in Redis with TTL. Return 200 immediately and enqueue a Celery task for business logic.
Mini Project: Webhook Testing Tool
# webhook_test_server.py
# Interactive webhook test server with logging
from flask import Flask, request, jsonify
from datetime import datetime
app = Flask(__name__)
received_webhooks = []
@app.route('/webhook-test', methods=['POST'])
def capture_webhook():
"""Capture and display incoming webhooks for debugging."""
data = request.get_json()
headers = dict(request.headers)
webhook_entry = {
"id": len(received_webhooks) + 1,
"timestamp": datetime.now().isoformat(),
"headers": headers,
"payload": data,
"source_ip": request.remote_addr,
}
received_webhooks.append(webhook_entry)
print(f"\n{'='*60}")
print(f"Webhook #{webhook_entry['id']} received at {webhook_entry['timestamp']}")
print(f"Source IP: {webhook_entry['source_ip']}")
print(f"Content-Type: {headers.get('Content-Type', 'N/A')}")
print(f"Signature: {headers.get('X-Webhook-Signature', 'None')}")
print(f"{'='*60}")
print(f"Payload:")
import json
print(json.dumps(data, indent=2))
print(f"{'='*60}\n")
return jsonify({"status": "captured", "id": webhook_entry['id']}), 200
@app.route('/webhook-test/history', methods=['GET'])
def webhook_history():
"""View all captured webhooks."""
return jsonify(received_webhooks)
if __name__ == '__main__':
print("Webhook test server running on http://localhost:8080")
print("Use tools like Webhook.site or curl to send test webhooks")
app.run(port=8080, debug=True)Run with curl:
curl -X POST http://localhost:8080/webhook-test \
-H "Content-Type: application/json" \
-H "X-Webhook-Signature: t=1723456789,s=test123" \
-d '{"event":"test.ping","data":{"message":"hello"}}'FAQ
Related Concepts
What’s Next
You now understand webhooks! Next, learn about Celery for processing webhook events asynchronously, then explore background job patterns for handling async workloads at scale.
- Practice daily — Set up a webhook consumer with ngrok and test with Stripe or GitHub test webhooks
- Build a project — Build a webhook relay service that verifies, filters, and routes incoming webhooks to internal services
- Explore related topics — Check out Svix, Zapier webhooks, and enterprise event-driven architectures
Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro. Updated 2026-06-20.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro