Twilio API: Sending SMS and Voice Messages

pip install twilio

# Set environment variables
export TWILIO_ACCOUNT_SID=ACxxxxxxxxxxxx
export TWILIO_AUTH_TOKEN=xxxxxxxxxxxx
export TWILIO_PHONE_NUMBER=+15551234567

# send_sms.py
# Send an SMS message via Twilio REST API

from twilio.rest import Client
import os

# Initialize client
account_sid = os.environ['TWILIO_ACCOUNT_SID']
auth_token = os.environ['TWILIO_AUTH_TOKEN']
client = Client(account_sid, auth_token)

def send_sms(to_number, message_body):
    """Send an SMS message."""
    message = client.messages.create(
        body=message_body,
        from_=os.environ['TWILIO_PHONE_NUMBER'],
        to=to_number,
    )

    print(f"Message sent!")
    print(f"  SID: {message.sid}")
    print(f"  From: {message.from_}")
    print(f"  To: {message.to}")
    print(f"  Status: {message.status}")
    print(f"  Price: {message.price} {message.price_unit}")

    return message

# Test
send_sms(
    to_number='+15559876543',
    message_body='Your Doda Browser scan found a threat. Check the app for details.'
)

# send_sms_callback.py
# SMS with delivery status tracking

from twilio.rest import Client
import os

client = Client(os.environ['TWILIO_ACCOUNT_SID'], os.environ['TWILIO_AUTH_TOKEN'])

def send_sms_with_tracking(to_number, message_body):
    """Send SMS and receive delivery status via webhook."""
    message = client.messages.create(
        body=message_body,
        from_=os.environ['TWILIO_PHONE_NUMBER'],
        to=to_number,
        status_callback='https://myapp.com/twilio/status',
        # Twilio POSTs to this URL when status changes
        # Statuses: queued → sent → delivered/failed/undelivered
    )

    print(f"SMS sent: {message.sid}")
    print(f"Initial status: {message.status}")
    print(f"Status callback will POST to: {message.status_callback}")

    return message

send_sms_with_tracking(
    to_number='+15559876543',
    message_body='Your verification code is: 842910'
)

# send_whatsapp.py
# Send WhatsApp message via Twilio

from twilio.rest import Client
import os

client = Client(os.environ['TWILIO_ACCOUNT_SID'], os.environ['TWILIO_AUTH_TOKEN'])

def send_whatsapp(to_number, message_body):
    """Send a WhatsApp message. Requires Twilio WhatsApp Sandbox setup."""
    message = client.messages.create(
        body=message_body,
        from_='whatsapp:+14155238886',  # Twilio WhatsApp sandbox number
        to=f'whatsapp:{to_number}',
    )

    print(f"WhatsApp message sent: {message.sid}")
    print(f"Status: {message.status}")

    # For media messages
    if message_body.startswith('http'):
        print(f"Media URL detected — sending as media message")

    return message

# Text message
send_whatsapp(
    to_number='+15559876543',
    message_body='Hello from Doda! Your weekly security report is ready.'
)

# Media message (image)
send_whatsapp(
    to_number='+15559876543',
    message_body='https://dodatech.com/charts/weekly_report.png'
)

# make_call.py
# Make a voice call with Twilio

from twilio.rest import Client
import os

client = Client(os.environ['TWILIO_ACCOUNT_SID'], os.environ['TWILIO_AUTH_TOKEN'])

def make_voice_call(to_number, message_text):
    """Make a voice call and speak a message using TwiML."""
    from twilio.twiml.voice_response import VoiceResponse

    # Build TwiML response — Twilio fetches this URL when call connects
    response = VoiceResponse()
    response.say(message_text, voice='alice', language='en-US')
    response.hangup()

    # In production, host this TwiML at a URL
    # For testing, use TwiML Bin or host it on your server
    call = client.calls.create(
        twiml=response,
        to=to_number,
        from_=os.environ['TWILIO_PHONE_NUMBER'],
        status_callback='https://myapp.com/twilio/call-status',
        status_callback_event=['initiated', 'ringing', 'answered', 'completed'],
    )

    print(f"Voice call initiated: {call.sid}")
    print(f"To: {call.to}")
    print(f"Status: {call.status}")

    return call

make_voice_call(
    to_number='+15559876543',
    message_text='Hello! This is Doda Antivirus calling with an important security alert. A threat was detected on your device. Please open the app for details.'
)

# verify.py
# Phone number verification with Twilio Verify

from twilio.rest import Client
import os

client = Client(os.environ['TWILIO_ACCOUNT_SID'], os.environ['TWILIO_AUTH_TOKEN'])

# Create a verification service in Twilio console first
VERIFY_SERVICE_SID = 'VAxxxxxxxxxxxx'

def send_verification_code(phone_number):
    """Send a 6-digit verification code via SMS."""
    verification = client.verify.v2.services(VERIFY_SERVICE_SID) \
        .verifications.create(
            to=phone_number,
            channel='sms',  # Also: 'call', 'email', 'whatsapp'
        )

    print(f"Verification sent to {phone_number}")
    print(f"Status: {verification.status}")
    print(f"Channel: {verification.channel}")
    return verification

def check_verification_code(phone_number, code):
    """Verify the code entered by the user."""
    verification_check = client.verify.v2.services(VERIFY_SERVICE_SID) \
        .verification_checks.create(
            to=phone_number,
            code=code,
        )

    if verification_check.status == 'approved':
        print(f"✅ Phone {phone_number} verified successfully!")
        return True
    else:
        print(f"❌ Invalid code for {phone_number}")
        return False

# Flow: send code → user enters code → verify
send_verification_code('+15559876543')

# User enters the 6-digit code they received
is_verified = check_verification_code('+15559876543', '842910')

# incoming_webhook.py
# Handle incoming SMS and WhatsApp messages

from flask import Flask, request, twiml
from twilio.twiml.messaging_response import MessagingResponse
import json

app = Flask(__name__)

# Store conversation history (use Redis in production)
conversations = {}

@app.route('/twilio/sms', methods=['POST'])
def handle_incoming_sms():
    """Handle incoming SMS and WhatsApp messages."""
    # Extract message details
    from_number = request.form['From']
    to_number = request.form['To']
    body = request.form['Body']
    message_sid = request.form['MessageSid']
    num_media = int(request.form.get('NumMedia', 0))

    print(f"Incoming message from {from_number}:")
    print(f"  Body: {body}")
    print(f"  Media count: {num_media}")

    # Handle media attachments
    if num_media > 0:
        for i in range(num_media):
            media_url = request.form.get(f'MediaUrl{i}')
            content_type = request.form.get(f'MediaContentType{i}')
            print(f"  Media {i}: {media_url} ({content_type})")

    # Build response
    response = MessagingResponse()

    # Simple auto-reply logic
    body_lower = body.lower()
    if 'help' in body_lower:
        response.message("Available commands:\n- STATUS: Check your account\n- REPORT: Get security report\n- STOP: Unsubscribe from messages")
    elif 'stop' in body_lower or 'unsubscribe' in body_lower:
        response.message("You've been unsubscribed from Doda alerts.")
        # Update user preference in database
    elif 'status' in body_lower:
        response.message("Your Doda Browser subscription is active. No threats detected in the last 24 hours.")
    else:
        response.message("Thanks for your message! A Doda support agent will respond shortly.")

    print(f"Response: {str(response)}")

    return str(response), 200, {'Content-Type': 'text/xml'}

@app.route('/twilio/status', methods=['POST'])
def handle_delivery_status():
    """Receive delivery status updates."""
    message_sid = request.form['MessageSid']
    message_status = request.form['MessageStatus']
    error_code = request.form.get('ErrorCode', 'none')
    to_number = request.form['To']

    print(f"Status update for {message_sid}:")
    print(f"  To: {to_number}")
    print(f"  Status: {message_status}")
    print(f"  Error: {error_code}")

    # Log to database for analytics
    # update_message_status(message_sid, message_status)

    return '', 200

if __name__ == '__main__':
    # twilio phone-numbers:update +15551234567 --sms-url http://localhost:5000/twilio/sms
    app.run(port=5000, debug=True)

# error_handling.py
# Robust Twilio error handling

from twilio.rest import Client
from twilio.base.exceptions import TwilioRestException
import os
import time

client = Client(os.environ['TWILIO_ACCOUNT_SID'], os.environ['TWILIO_AUTH_TOKEN'])

def send_sms_robust(to_number, message_body, max_retries=3):
    """Send SMS with comprehensive error handling."""
    for attempt in range(max_retries):
        try:
            message = client.messages.create(
                body=message_body,
                from_=os.environ['TWILIO_PHONE_NUMBER'],
                to=to_number,
            )
            print(f"SMS sent: {message.sid}")
            return message

        except TwilioRestException as e:
            print(f"Attempt {attempt + 1}/{max_retries} failed:")

            if e.code == 21211:  # Invalid 'To' number
                print(f"  Invalid phone number: {to_number}")
                return None  # Don't retry

            elif e.code == 21610:  # Carrier blocked
                print(f"  Carrier blocked this message type")
                return None  # Don't retry

            elif e.code == 21608:  # Not verified for trial
                print(f"  Trial account: verify {to_number} first")
                return None

            elif e.code == 30001:  # Queue overflow
                print(f"  Queue full, retrying...")
                time.sleep(2 ** attempt)  # Exponential backoff

            elif e.code == 14101:  # Unreachable
                print(f"  Number unreachable at this time")
                time.sleep(60)

            else:
                print(f"  Error {e.code}: {e.msg}")
                time.sleep(2 ** attempt)

        except Exception as e:
            print(f"Unexpected error: {e}")
            time.sleep(2 ** attempt)

    print(f"Failed to send SMS after {max_retries} attempts")
    return None

# auto_responder.py
# SMS keyword auto-responder for common inquiries

from flask import Flask, request
from twilio.twiml.messaging_response import MessagingResponse
import json

app = Flask(__name__)

# Keyword → response mapping
KEYWORD_RESPONSES = {
    'BALANCE': 'Your current balance is $49.99. Next payment due: July 15, 2026.',
    'THREATS': 'No threats detected in the last 24 hours. Last scan: June 19, 2026 at 3:42 PM.',
    'HELP': 'Available keywords: BALANCE, THREATS, SUBSCRIPTION, RENEW, SUPPORT',
    'SUBSCRIPTION': 'You are on the Pro plan ($9.99/month). Next renewal: July 1, 2026.',
    'RENEW': 'To renew your subscription, visit: https://dodatech.com/account/billing',
    'SUPPORT': 'A support agent will contact you within 1 hour. Your ticket ID is TKT-789.',
    'STOP': 'You have been unsubscribed from SMS alerts. Reply START to resume.',
    'START': 'Welcome back! SMS alerts have been resumed.',
}

@app.route('/sms/auto-respond', methods=['POST'])
def auto_respond():
    """Auto-respond to SMS based on keyword."""
    from_number = request.form['From']
    body = request.form['Body'].strip().upper()

    print(f"Received from {from_number}: {body}")

    response = MessagingResponse()

    # Find matching keyword
    keyword = body.split()[0] if body else ''
    reply = KEYWORD_RESPONSES.get(keyword)

    if reply:
        response.message(reply)
        print(f"Auto-response: {reply[:50]}...")
    else:
        response.message(
            f"Unknown command: '{keyword}'. "
            f"Reply HELP for available keywords."
        )

    return str(response), 200, {'Content-Type': 'text/xml'}

if __name__ == '__main__':
    print("SMS Auto-Responder running on http://localhost:5000")
    print(f"Configured keywords: {', '.join(KEYWORD_RESPONSES.keys())}")
    app.run(port=5000, debug=True)