Complete Twilio Project: Build a Multi-Channel Notification System
In this tutorial, you will learn about Complete Twilio Project: Build a Multi. We cover key concepts, practical examples, and best practices to help you master this topic.
Build a production-ready multi-channel notification system integrating SMS, WhatsApp, voice calls, email alerts, message templates, delivery tracking, error handling, credential security, and deployment.
What You'll Learn
How to architect and implement a complete Twilio notification system combining all channels: Programmable SMS, WhatsApp Business API, Programmable Voice, Messaging Service pools, Conversations API, Twilio Functions, error handling, and credential management.
Why It Matters
A well-designed notification system is critical for user engagement, security alerts, and operational communications. DodaTech sends 50,000+ notifications daily across SMS, WhatsApp, and voice, with 99.5% delivery rate.
Real-World Use
A customer's order ships. The system checks their preferred channel (WhatsApp), sends a shipping confirmation with photo and tracking link, logs delivery status, and if unread after 2 hours, falls back to SMS, then voice call for critical updates.
flowchart LR
subgraph Sender
A["Notification\nTrigger"] --> B{"Preferred\nChannel"}
end
subgraph Channels
B -->|WhatsApp| C["Send\nTemplate"]
B -->|SMS| D["Messaging\nService"]
B -->|Voice| E["Text-to-Speech\nCall"]
end
subgraph Tracking
C --> F["Status\nCallback"]
D --> F
E --> F
F --> G{"Delivered?"}
G -->|Yes| H["Success\nLog"]
G -->|No| I["Fallback\nto Next Channel"]
I --> B
end
style A fill:#f22f46,color:#fff
style G fill:#fef3c7,stroke:#d97706
style H fill:#bbf7d0,stroke:#16a34a
Project Architecture
"""
notification_system/
config.py # Credentials, environment setup
channels/
sms.py # SMS sender with Messaging Service
whatsapp.py # WhatsApp template sender
voice.py # Voice call sender
core/
router.py # Channel preference logic
templates.py # Message template management
fallback.py # Channel fallback logic
monitoring/
delivery_tracker.py # Status callback handler
error_handler.py # Retry and error categorization
metrics.py # Delivery statistics
security/
credential_manager.py # Key rotation and subaccounts
tests/
test_all_channels.py # Comprehensive test suite
"""
1. Configuration and Multi-Channel Router
# config.py
import os
from twilio.rest import Client
class TwilioConfig:
def __init__(self, environment="production"):
self.environment = environment
sid = os.environ["TWILIO_ACCOUNT_SID"]
if environment == "production":
token = os.environ["TWILIO_AUTH_TOKEN"]
self.messaging_service_sid = os.environ["TWILIO_MESSAGING_SID"]
self.whatsapp_number = os.environ["TWILIO_WHATSAPP_NUMBER"]
self.voice_number = os.environ["TWILIO_VOICE_NUMBER"]
else:
token = os.environ["TWILIO_TEST_AUTH_TOKEN"]
self.messaging_service_sid = os.environ["TWILIO_TEST_MESSAGING_SID"]
self.whatsapp_number = "whatsapp:+14155238886" # Sandbox
self.voice_number = os.environ["TWILIO_TEST_VOICE_NUMBER"]
self.client = Client(sid, token)
self.status_callback = f"https://api.dodatech.com/notifications/status"
def verify(self):
"""Verify configuration is valid."""
account = self.client.api.accounts(sid).fetch()
print(f"Environment: {self.environment}")
print(f"Account: {account.friendly_name}")
print(f"Status: {account.status}")
print(f"Messaging Service: {self.messaging_service_sid[:20]}...")
return account.status == "active"
config = TwilioConfig("test")
config.verify()
# Expected output:
# Environment: test
# Account: DodaTech Notifications
# Status: active
# Messaging Service: MGxxxxxxxxxxxxxxxx...
2. Multi-Channel Sender
# core/router.py
from enum import Enum
class Channel(Enum):
SMS = "sms"
WHATSAPP = "whatsapp"
VOICE = "voice"
def send_notification(recipient, message, preferred_channel, config):
"""Route notification through the preferred channel with fallback."""
channels_ordered = [preferred_channel, Channel.SMS, Channel.VOICE]
attempted = []
for channel in channels_ordered:
try:
if channel == Channel.WHATSAPP and config.whatsapp_number:
result = send_whatsapp(recipient, message, config)
elif channel == Channel.SMS:
result = send_sms(recipient, message, config)
elif channel == Channel.VOICE:
result = send_voice(recipient, message, config)
else:
continue
print(f"Sent via {channel.value}: {result.sid}")
return result
except Exception as e:
attempted.append(channel.value)
print(f"{channel.value} failed: {e}")
continue
raise RuntimeError(f"All channels failed for {recipient}: {attempted}")
# channels/sms.py
def send_sms(recipient, body, config):
return config.client.messages.create(
messaging_service_sid=config.messaging_service_sid,
to=recipient,
body=body,
status_callback=config.status_callback
)
# channels/whatsapp.py
def send_whatsapp(recipient, body, config):
return config.client.messages.create(
from_=config.whatsapp_number,
to=f"whatsapp:{recipient}",
body=body,
status_callback=config.status_callback
)
3. Delivery Tracking and Status Callbacks
# monitoring/delivery_tracker.py
from flask import Flask, request, jsonify
import sqlite3
import json
app = Flask(__name__)
def init_db():
conn = sqlite3.connect("delivery.db")
conn.execute("""
CREATE TABLE IF NOT EXISTS deliveries (
message_sid TEXT PRIMARY KEY,
recipient TEXT,
channel TEXT,
status TEXT,
error_code TEXT,
error_message TEXT,
timestamp TEXT,
callback_count INTEGER DEFAULT 1
)
""")
conn.close()
@app.route("/notifications/status", methods=["POST"])
def delivery_status():
message_sid = request.form["MessageSid"]
status = request.form["MessageStatus"]
to_number = request.form["To"]
channel = "whatsapp" if "whatsapp" in to_number else "sms"
error_code = request.form.get("ErrorCode", "")
error_message = request.form.get("ErrorMessage", "")
conn = sqlite3.connect("delivery.db")
existing = conn.execute(
"SELECT status FROM deliveries WHERE message_sid = ?",
(message_sid,)
).fetchone()
if existing:
conn.execute("""
UPDATE deliveries SET status = ?, error_code = ?,
error_message = ?, callback_count = callback_count + 1
WHERE message_sid = ?
""", (status, error_code, error_message, message_sid))
else:
conn.execute("""
INSERT INTO deliveries (message_sid, recipient, channel, status,
error_code, error_message, timestamp)
VALUES (?, ?, ?, ?, ?, ?, datetime('now'))
""", (message_sid, to_number, channel, status, error_code, error_message))
conn.commit()
conn.close()
return jsonify({"status": "logged"}), 200
4. Error Handler with Fallback
# monitoring/error_handler.py
import time
class NotificationErrorHandler:
def __init__(self, config, max_retries=3):
self.config = config
self.max_retries = max_retries
self.retryable_codes = {20429, 21610, 21611, 30001, 30002, 30003}
self.fatal_codes = {20003, 21211, 21212, 21401}
def send_with_retry(self, recipient, body, channel_preference):
for attempt in range(self.max_retries):
try:
return send_notification(
recipient, body, channel_preference, self.config
)
except TwilioRestException as e:
if e.code in self.fatal_codes:
raise # No point retrying
if e.code in self.retryable_codes:
wait = (2 ** attempt) + (time.time() % 1) # Backoff + jitter
print(f"Retrying in {wait:.1f}s...")
time.sleep(wait)
raise RuntimeError(f"Failed after {self.max_retries} retries")
def send_with_fallback(self, recipient, body, channels):
"""Try each channel in order until one succeeds."""
for channel in channels:
for attempt in range(2): # 2 attempts per channel
try:
return self.send_with_retry(recipient, body, channel)
except Exception:
continue
return None
5. Test Suite
# tests/test_all_channels.py
def run_notification_tests(config):
test_number = os.environ.get("TEST_PHONE_NUMBER", "+14155551234")
results = []
# Test SMS
try:
msg = send_sms(test_number, "Test notification from DodaTech", config)
results.append(("SMS", "PASS", msg.sid))
except Exception as e:
results.append(("SMS", "FAIL", str(e)))
# Test WhatsApp (if sandbox is active)
try:
msg = send_whatsapp(test_number,
"Test WhatsApp from DodaTech", config)
results.append(("WhatsApp", "PASS", msg.sid))
except Exception as e:
results.append(("WhatsApp", "FAIL", str(e)))
# Test channel fallback
try:
result = send_notification(
test_number, "Test fallback",
Channel.VOICE, config # Voice fails, falls to SMS
)
results.append(("Fallback", "PASS", result.sid))
except Exception as e:
results.append(("Fallback", "FAIL", str(e)))
print("Notification System Test Results:")
print(f"{'Channel':<15} {'Result':<8} {'SID/Error':<40}")
print("-" * 63)
for channel, result, detail in results:
print(f"{channel:<15} {result:<8} {detail:<40}")
# run_notification_tests(config)
# Expected output:
# Notification System Test Results:
# Channel Result SID/Error
# ---------------------------------------------------------------
# SMS PASS SMxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
# WhatsApp PASS SMxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
# Fallback PASS SMxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
Common Mistakes
1. Not Implementing Channel Fallback
If SMS fails, the user never gets notified. Always implement fallback: WhatsApp -> SMS -> Voice. Each channel has different failure modes and fallback covers them.
2. Ignoring Status Callbacks
Without tracking, you don't know delivery rates. Status callbacks are essential for monitoring, troubleshooting, and trigger retries. Always set status_callback on every message.
3. Not Testing All Channels
Each channel has unique failure modes. Test SMS with long messages (segmentation), WhatsApp without active sandbox session, voice with unanswered calls. Don't assume all channels work the same.
4. Hardcoding Environment Configuration
Development and production use different keys, numbers, and services. Use environment-specific configuration (test vs production) to prevent accidentally sending test messages to real customers.
5. Forgetting Rate Limits
Each channel and number has rate limits. Sending 1000 SMS instantly causes rate limit errors. Implement a rate limiter per channel with configurable max per second.
Practice Questions
- What is the architecture of a multi-channel notification system?
- How does channel fallback work?
- Why are status callbacks critical for a notification system?
- How do you test all notification channels?
Answers:
- Three layers: Channel layer (SMS, WhatsApp, Voice senders), Core layer (router, templates, fallback), Monitoring layer (delivery tracking, error handling, metrics).
- Channel fallback tries the user's preferred channel first. If it fails, the system automatically tries the next channel in priority order (WhatsApp -> SMS -> Voice for non-critical, SMS -> Voice for critical).
- Status callbacks provide delivery confirmation — without them, you don't know if messages arrived. They enable retry logic, delivery rate monitoring, and a proof of delivery for Compliance.
- Create a test suite that sends via each channel, verifies delivery through status callbacks, tests fallback by forcing channel failures, checks error handling with invalid numbers, and validates Rate Limiting behavior.
Challenge: Build and deploy the complete notification system as described. Implement all 5 components, test all channels, verify fallback behavior, set up delivery tracking with a status callback endpoint, configure credential security with restricted API keys and subaccounts, create a monitoring dashboard, and write a production deployment checklist.
FAQ
Mini Project
The mini project is this complete notification system. Implement all components, test with SMS and WhatsApp sandbox, implement channel fallback with retry, set up delivery tracking with status callbacks, configure credential security, create a monitoring dashboard, and document the architecture.
What's Next
API Automated Testing — apply automated testing techniques to your Twilio integration.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro