SendGrid Event Webhook: Real-Time Email Delivery Tracking & Monitoring
In this tutorial, you will learn about SendGrid Event Webhook: Real. We cover key concepts, practical examples, and best practices to help you master this topic.
SendGrid Event Webhook sends real-time HTTP POST requests to your server when email events occur — delivery, open, click, bounce, spam report, unsubscribe, and more.
What You'll Learn
How to configure the Event Webhook, receive and process event payloads, verify webhook authenticity, handle different event types, store events for analytics, and prevent replay attacks.
Why It Matters
Without Webhooks, you don't know if emails were delivered or bounced. DodaTech processes 50K+ daily events to track deliverability, detect bounces in real-time, and update user preferences.
Real-World Use
A user clicks a link in a security report email. SendGrid sends a "click" event webhook. The app logs the click, updates the user's engagement score, and records the click-through rate for analytics.
flowchart LR
A["SendGrid\nProcesses Email"] --> B["Event Generated\nDelivered, Open, Click"]
B --> C["HTTP POST\nto Your Webhook URL"]
C --> D["Verify\nSignature"]
D --> E["Process Event\nPer Type"]
E --> F["Update\nDatabase"]
E --> G["Trigger\nActions"]
style A fill:#1a82e2,color:#fff
style C fill:#dbeafe,stroke:#2563eb
style D fill:#fef3c7,stroke:#d97706
Setting Up Webhook
# In SendGrid Dashboard:
# Settings > Mail Settings > Event Webhook
# URL: https://api.dodatech.com/sendgrid/events
# Select events: delivered, open, click, bounce, dropped, spam_report, unsubscribe
# Enable: POST, URL, and signature verification
Receiving Events
from flask import Flask, request, jsonify
import hashlib
import hmac
import os
app = Flask(__name__)
SENDGRID_WEBHOOK_SECRET = os.environ["SENDGRID_WEBHOOK_SECRET"]
@app.route("/sendgrid/events", methods=["POST"])
def handle_events():
events = request.get_json()
if not events:
return jsonify({"error": "No payload"}), 400
for event in events:
event_type = event.get("event")
email = event.get("email")
timestamp = event.get("timestamp")
sg_event_id = event.get("sg_event_id")
print(f"[{event_type}] {email} at {timestamp}")
if event_type == "delivered":
handle_delivered(event)
elif event_type == "open":
handle_open(event)
elif event_type == "click":
handle_click(event)
elif event_type == "bounce":
handle_bounce(event)
elif event_type == "spam_report":
handle_spam(event)
elif event_type == "unsubscribe":
handle_unsubscribe(event)
elif event_type == "group_unsubscribe":
handle_group_unsubscribe(event)
return jsonify({"status": "ok"}), 200
def handle_delivered(event):
email = event["email"]
print(f" Delivered: {email}")
def handle_open(event):
email = event["email"]
user_agent = event.get("useragent", "unknown")
ip = event.get("ip", "unknown")
print(f" Opened by {email} from {ip}")
def handle_click(event):
email = event["email"]
url = event.get("url", "unknown")
print(f" Click: {email} -> {url}")
log_click(email, url)
def handle_bounce(event):
email = event["email"]
bounce_type = event.get("bounce_type", "unknown")
reason = event.get("reason", "No reason")
print(f" Bounce ({bounce_type}): {email} — {reason[:80]}")
remove_from_active_list(email)
def handle_spam(event):
email = event["email"]
print(f" Spam report: {email}")
flag_for_review(email)
def handle_unsubscribe(event):
email = event["email"]
print(f" Unsubscribed: {email}")
mark_as_unsubscribed(email)
def handle_group_unsubscribe(event):
email = event["email"]
group_id = event.get("asm_group_id")
print(f" Group unsubscribe: {email} from group {group_id}")
update_preferences(email, group_id, False)
Verifying Webhook Signatures
def verify_signature(request):
signature = request.headers.get("X-Twilio-Email-Event-Webhook-Signature")
timestamp = request.headers.get("X-Twilio-Email-Event-Webhook-Timestamp")
if not signature or not timestamp:
print("Missing signature headers")
return False
# Build expected signature
payload = timestamp + request.get_data(as_text=True)
expected = hmac.new(
SENDGRID_WEBHOOK_SECRET.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
# Constant-time comparison
is_valid = hmac.compare_digest(signature, expected)
if not is_valid:
print(f"Invalid signature: got {signature[:20]}..., expected {expected[:20]}...")
else:
print("Signature verified")
return is_valid
@app.route("/sendgrid/events", methods=["POST"])
def secure_handle_events():
if not verify_signature(request):
return jsonify({"error": "Invalid signature"}), 403
return handle_events()
Storing Events for Analytics
def log_click(email, url):
# Store in database for analytics
click_event = {
"email": email,
"url": url,
"timestamp": datetime.utcnow().isoformat(),
"campaign": extract_campaign(url)
}
# db.clicks.insert_one(click_event)
print(f" Logged click: {email} -> {url[:50]}...")
def extract_campaign(url):
# Extract campaign from URL
if "utm_campaign" in url:
from urllib.parse import parse_qs, urlparse
params = parse_qs(urlparse(url).query)
return params.get("utm_campaign", ["unknown"])[0]
return "direct"
Common Mistakes
1. Not Verifying Webhook Signatures
Without signature verification, anyone can POST fake events to your endpoint. Always verify using the shared secret.
2. Responding Slowly (Timeout)
SendGrid expects a 200 response within 30 seconds. Slow handlers cause retries. Acknowledge immediately with 200, process asynchronously.
3. Not Handling Duplicate Events
SendGrid may deliver the same event multiple times. Use sg_event_id for deduplication. Process each event ID only once.
4. Processing Events Synchronously in the Handler
Processing events inline blocks the response. Acknowledge immediately, push to a queue (Redis, SQS), and process asynchronously.
5. Ignoring sg_message_id
The sg_message_id links events to the original send. Store it to correlate delivery, opens, and clicks for each message.
Practice Questions
- How do you verify the authenticity of a webhook request?
- How do you deduplicate event webhooks?
- What should you return from the webhook endpoint?
- How do you correlate open events with the original send?
Answers:
- Use the
X-Twilio-Email-Event-Webhook-Signatureheader and verify with HMAC-SHA256 using your webhook secret. - Store processed
sg_event_idvalues and check for duplicates before processing. Use an idempotent database operation (upsert). - Return HTTP 200 immediately. Process events asynchronously to avoid timeout. Non-200 responses cause retries.
- The
sg_message_idfield in the event payload matches the message ID from the original send response headers.
Challenge: Build an event processing pipeline: verify webhook signatures, process 6 event types (delivered, open, click, bounce, spam, unsubscribe), deduplicate by sg_event_id, store in database, and generate a daily analytics report showing delivery rates, open rates, and click-through rates per campaign.
FAQ
Mini Project
Build a complete event tracking system: configure Event Webhook for all event types, implement signature verification, process events into a database (MongoDB/PostgreSQL), deduplicate by sg_event_id, and create a real-time dashboard showing delivery rates, open/click rates, and bounce alerts.
What's Next
Inbound Parse Webhook — receive and process incoming email replies.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro