Webhooks and Integrations
What You'll Learn
How to send notifications to external services and receive webhook events from them — the foundation of modern automation.
Sending Slack Messages
pip install requests
import os
import requests
def send_slack_message(text: str, channel: str = "#alerts") -> bool:
"""Send a message to Slack via incoming webhook."""
webhook_url = os.environ["SLACK_WEBHOOK_URL"]
response = requests.post(
webhook_url,
json={
"text": text,
"channel": channel,
},
timeout=10,
)
return response.status_code == 200
# Usage
send_slack_message(":white_check_mark: Backup completed: 1,423 files")
send_slack_message(":red_circle: Sync failed — check logs")
Rich Slack Messages (Blocks)
def notify_job_result(job_name: str, status: str, details: dict) -> None:
icon = ":white_check_mark:" if status == "ok" else ":red_circle:"
webhook_url = os.environ["SLACK_WEBHOOK_URL"]
requests.post(webhook_url, json={
"blocks": [
{
"type": "header",
"text": {"type": "plain_text", "text": f"{icon} {job_name}"}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Status:*\n{status}"},
{"type": "mrkdwn", "text": f"*Records:*\n{details.get('count', '–')}"},
{"type": "mrkdwn", "text": f"*Errors:*\n{details.get('errors', '–')}"},
{"type": "mrkdwn", "text": f"*Duration:*\n{details.get('elapsed', '–')}s"},
]
}
]
}, timeout=10)
Sending Email
Using Python's built-in smtplib:
import smtplib
import os
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
def send_email(
to: str,
subject: str,
body: str,
html_body: str = None,
) -> None:
smtp_host = os.environ.get("SMTP_HOST", "smtp.gmail.com")
smtp_port = int(os.environ.get("SMTP_PORT", "587"))
smtp_user = os.environ["SMTP_USER"]
smtp_pass = os.environ["SMTP_PASS"]
sender = os.environ.get("SMTP_FROM", smtp_user)
msg = MIMEMultipart("alternative")
msg["From"] = sender
msg["To"] = to
msg["Subject"] = subject
msg.attach(MIMEText(body, "plain"))
if html_body:
msg.attach(MIMEText(html_body, "html"))
with smtplib.SMTP(smtp_host, smtp_port) as server:
server.starttls()
server.login(smtp_user, smtp_pass)
server.send_message(msg)
Sending HTTP Webhooks to Other Services
import requests
import hmac
import hashlib
import json
import os
def send_webhook(url: str, payload: dict, secret: str = None) -> bool:
"""Send a signed webhook POST request."""
body = json.dumps(payload).encode("utf-8")
headers = {"Content-Type": "application/json"}
if secret:
sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
headers["X-Signature"] = f"sha256={sig}"
response = requests.post(url, data=body, headers=headers, timeout=10)
return response.ok
Receiving GitHub Webhooks
import hmac
import hashlib
import logging
from flask import Flask, request, jsonify
app = Flask(__name__)
log = logging.getLogger(__name__)
GITHUB_SECRET = os.environ["GITHUB_WEBHOOK_SECRET"]
def verify_github_signature(payload: bytes, signature: str) -> bool:
expected = "sha256=" + hmac.new(
GITHUB_SECRET.encode(), payload, hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, signature)
@app.route("/webhooks/github", methods=["POST"])
def github_webhook():
# 1. Verify signature
sig = request.headers.get("X-Hub-Signature-256", "")
if not verify_github_signature(request.data, sig):
log.warning("Invalid signature from %s", request.remote_addr)
return jsonify({"error": "Unauthorized"}), 403
event = request.headers.get("X-GitHub-Event", "unknown")
payload = request.get_json()
log.info("GitHub event: %s", event)
if event == "push":
branch = payload["ref"].split("/")[-1]
pusher = payload["pusher"]["name"]
commits = len(payload["commits"])
log.info("Push by %s to %s: %d commit(s)", pusher, branch, commits)
handle_push(branch, payload)
elif event == "pull_request":
action = payload["action"]
pr_number = payload["pull_request"]["number"]
log.info("PR #%d %s", pr_number, action)
# Always return 200 quickly — process async if needed
return jsonify({"ok": True})
Receiving Stripe Webhooks
import stripe
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
STRIPE_WEBHOOK_SECRET = os.environ["STRIPE_WEBHOOK_SECRET"]
@app.route("/webhooks/stripe", methods=["POST"])
def stripe_webhook():
payload = request.data
sig_header = request.headers.get("Stripe-Signature")
try:
event = stripe.Webhook.construct_event(
payload, sig_header, STRIPE_WEBHOOK_SECRET
)
except ValueError:
return jsonify({"error": "Invalid payload"}), 400
except stripe.error.SignatureVerificationError:
return jsonify({"error": "Invalid signature"}), 403
if event["type"] == "payment_intent.succeeded":
payment = event["data"]["object"]
amount = payment["amount"] / 100 # cents to dollars
log.info("Payment succeeded: $%.2f from %s", amount, payment["id"])
elif event["type"] == "payment_intent.payment_failed":
payment = event["data"]["object"]
log.warning("Payment failed: %s", payment["id"])
return jsonify({"received": True})
Idempotent Webhook Processing
Webhooks can be delivered multiple times. Process idempotently:
processed_events = set() # use Redis/database in production
@app.route("/webhooks/events", methods=["POST"])
def handle_event():
payload = request.get_json()
event_id = payload.get("id")
if event_id in processed_events:
log.info("Duplicate event %s — skipping", event_id)
return jsonify({"ok": True, "duplicate": True})
try:
process_event(payload)
processed_events.add(event_id)
return jsonify({"ok": True})
except Exception as e:
log.error("Event %s failed: %s", event_id, e)
return jsonify({"error": str(e)}), 500
Quick Reference
# Slack
requests.post(webhook_url, json={"text": "message"}, timeout=10)
# Email
import smtplib
with smtplib.SMTP(host, port) as s:
s.starttls()
s.login(user, password)
s.send_message(msg)
# Webhook signature verification
sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
hmac.compare_digest(f"sha256={sig}", header_signature)
# Flask webhook endpoint
@app.route("/webhooks/x", methods=["POST"])
def webhook():
verify_signature(request.data, request.headers["X-Signature"])
event = request.get_json()
process(event)
return jsonify({"ok": True}), 200
# Idempotency
if event_id in already_processed:
return {"ok": True, "duplicate": True}