Twilio Communications

Twilioで通信機能を構築します：SMSメッセージング、音声通話、WhatsApp Business API、ユーザー認証（2FA）。シンプルな通知から複雑なIVRシステム、マルチチャネル認証まで、幅広い機能をカバーしています。コンプライアンス、レート制限、エラーハンドリングが重要です。

パターン

SMSメッセージ送信パターン

Twilioでメッセージを送信する基本パターンです。 電話番号のフォーマット、メッセージ配信、配信ステータスコールバックを扱います。

主な注意点：

• 電話番号はE.164形式（+1234567890）である必要があります
• デフォルトレート制限：1秒あたり80メッセージ（MPS）
• 160文字を超えるメッセージは分割されます（料金も増加します）
• キャリアフィルタリングはメッセージをブロックできます（特に米国の番号の場合）

使用時機：ユーザーへの通知送信、トランザクションメッセージ（注文確認、配送）、アラートとリマインダー

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

class TwilioSMS:
    """
    適切なエラーハンドリングと検証を使用したSMS送信。
    """

    def __init__(self):
        self.client = Client(
            os.environ["TWILIO_ACCOUNT_SID"],
            os.environ["TWILIO_AUTH_TOKEN"]
        )
        self.from_number = os.environ["TWILIO_PHONE_NUMBER"]

    def validate_e164(self, phone: str) -> bool:
        """電話番号がE.164形式であることを検証します。"""
        pattern = r'^\+[1-9]\d{1,14}$'
        return bool(re.match(pattern, phone))

    def send_sms(
        self,
        to: str,
        body: str,
        status_callback: str = None
    ) -> dict:
        """
        SMSメッセージを送信します。

        Args:
            to: E.164形式の受信者電話番号
            body: メッセージテキスト（160文字 = 1セグメント）
            status_callback: 配信ステータスウェブフックのURL

        Returns:
            メッセージSIDとステータス
        """
        # 電話番号形式を検証
        if not self.validate_e164(to):
            return {
                "success": False,
                "error": "Phone number must be in E.164 format (+1234567890)"
            }

        # メッセージ長をチェック（分割について警告）
        segment_count = (len(body) + 159) // 160
        if segment_count > 1:
            print(f"Warning: Message will be sent as {segment_count} segments")

        try:
            message = self.client.messages.create(
                to=to,
                from_=self.from_number,
                body=body,
                status_callback=status_callback
            )

            return {
                "success": True,
                "message_sid": message.sid,
                "status": message.status,
                "segments": segment_count
            }

        except TwilioRestException as e:
            return self._handle_error(e)

    def _handle_error(self, error: TwilioRestException) -> dict:
        """Twilio固有のエラーを処理します。"""
        error_handlers = {
            21610: "Recipient has opted out. They must reply START.",
            21614: "Invalid 'To' phone number format.",
            21211: "'From' phone number is not valid.",
            30003: "Phone is unreachable (off, airplane mode, no signal).",
            30005: "Unknown destination (invalid number or landline).",
            30006: "Landline or unreachable carrier.",
            30429: "Rate limit exceeded. Implement exponential backoff.",
        }

        return {
            "success": False,
            "error_code": error.code,
            "error": error_handlers.get(error.code, error.msg),
            "details": str(error)
        }

# 使用例
sms = TwilioSMS()
result = sms.send_sms(
    to="+14155551234",
    body="Your order #1234 has shipped!",
    status_callback="https://your-app.com/webhooks/twilio/status"
)

アンチパターン

• E.164形式を検証してから送信していない
• Twilioの認証情報をコードにハードコーディングしている
• 配信ステータスコールバックを無視している
• オプトアウト（21610）エラーを処理していない

Twilio Verifyパターン（2FA/OTP）

電話番号の検証と2FAにTwilio Verifyを使用します。 コード生成、配信、レート制限、詐欺防止を処理します。

DIY OTPに比べた主な利点：

• Twilioがコード生成と有効期限を管理
• 組み込みの詐欺防止（顧客が7億4700万件の試行をブロックし、8200万ドル削減）
• 自動的なレート制限を処理
• マルチチャネル：SMS、音声、メール、プッシュ、WhatsApp

Googleの調査では、SMS 2FAは「自動ボットの100%、大量フィッシング攻撃の96%、標的型攻撃の76%」をブロックしています。

使用時機：サインアップ時の電話番号検証、二要素認証（2FA）、パスワードリセット検証、高額取引の確認

from twilio.rest import Client
from twilio.base.exceptions import TwilioRestException
import os
from enum import Enum
from typing import Optional

class VerifyChannel(Enum):
    SMS = "sms"
    CALL = "call"
    EMAIL = "email"
    WHATSAPP = "whatsapp"

class TwilioVerify:
    """
    Twilio Verifyを使用した電話検証。
    OTPコードを保存しないでください - Twilioが処理します。
    """

    def __init__(self, verify_service_sid: str = None):
        self.client = Client(
            os.environ["TWILIO_ACCOUNT_SID"],
            os.environ["TWILIO_AUTH_TOKEN"]
        )
        # Twilio Consoleでまず検証サービスを作成
        self.service_sid = verify_service_sid or os.environ["TWILIO_VERIFY_SID"]

    def send_verification(
        self,
        to: str,
        channel: VerifyChannel = VerifyChannel.SMS,
        locale: str = "en"
    ) -> dict:
        """
        検証コードを電話またはメールに送信します。

        Args:
            to: 電話番号（E.164）またはメール
            channel: SMS、通話、メール、またはWhatsApp
            locale: メッセージの言語コード

        Returns:
            検証ステータス
        """
        try:
            verification = self.client.verify \
                .v2 \
                .services(self.service_sid) \
                .verifications \
                .create(
                    to=to,
                    channel=channel.value,
                    locale=locale
                )

            return {
                "success": True,
                "status": verification.status,  # "pending"
                "channel": channel.value,
                "valid": verification.valid
            }

        except TwilioRestException as e:
            return self._handle_verify_error(e)

    def check_verification(self, to: str, code: str) -> dict:
        """
        検証コードが正しいかチェックします。

        Args:
            to: コードを受け取った電話番号またはメール
            code: ユーザーが入力したコード

        Returns:
            検証結果
        """
        try:
            check = self.client.verify \
                .v2 \
                .services(self.service_sid) \
                .verification_checks \
                .create(
                    to=to,
                    code=code
                )

            return {
                "success": True,
                "valid": check.status == "approved",
                "status": check.status  # "approved" or "pending"
            }

        except TwilioRestException as e:
            # コードが間違っているか有効期限切れ
            return {
                "success": False,
                "valid": False,
                "error": str(e)
            }

    def _handle_verify_error(self, error: TwilioRestException) -> dict:
        """検証固有のエラーを処理します。"""
        error_handlers = {
            60200: "Invalid phone number format",
            60203: "Max send attempts reached for this number",
            60205: "Service not found - check VERIFY_SID",
            60223: "Failed to create verification - carrier rejected",
        }

        return {
            "success": False,
            "error_code": error.code,
            "error": error_handlers.get(error.code, error.msg)
        }

# 使用例 - サインアップフロー
verify = TwilioVerify()

# ステップ1：ユーザーが電話番号を入力
result = verify.send_verification("+14155551234", VerifyChannel.SMS)
if result["success"]:
    print("Code sent! Check your phone.")

# ステップ2：ユーザーが受け取ったコードを入力
code = "123456"  # ユーザー入力から
check = verify.check_verification("+14155551234", code)

if check["valid"]:
    print("Phone verified! Create account.")
else:
    print("Invalid code. Try again.")

# ベストプラクティス：音声フォールバックを提供
async def verify_with_fallback(phone: str, max_attempts: int = 3):
    """音声フォールバックを使用した検証。"""
    for attempt in range(max_attempts):
        channel = VerifyChannel.SMS if attempt == 0 else VerifyChannel.CALL
        result = verify.send_verification(phone, channel)

        if result["success"]:
            return result

        # SMS失敗時、待機して音声を試行
        if channel == VerifyChannel.SMS:
            await asyncio.sleep(30)
            continue

    return {"success": False, "error": "All verification attempts failed"}

アンチパターン

• データベースにOTPコードを保存している（Twilioが処理します）
• verifyエンドポイントにレート制限を実装していない
• 同じコードで再試行している（Verifyに新しいコードを生成させます）
• SMS失敗時のフォールバックチャネルがない

TwiML IVRパターン

TwiMLを使用したインタラクティブボイスレスポンス（IVR）システムを構築します。 TwiML（Twilio Markup Language）はTwilioに何をするかを指示するXMLです。

コアTwiML動詞：

• <Say>：テキスト音声変換
• <Play>：オーディオファイルの再生
• <Gather>：キーパッドまたは音声入力の収集
• <Dial>：別の番号に接続
• <Record>：発信者の音声を記録
• <Redirect>：別のTwiMLエンドポイントに移動

重要な概念：TwilioはあなたのウェブフックにHTTPリクエストを送信し、あなたがTwiMLを返し、Twilioがそれを実行します。ステートレスなので、URLパラメータやセッションを使用します。

使用時機：電話メニューシステム（営業は1を押す...）、自動カスタマーサポート、予約リマインダーと確認、ボイスメールシステム

from flask import Flask, request, Response
from twilio.twiml.voice_response import VoiceResponse, Gather
from twilio.request_validator import RequestValidator
import os

app = Flask(__name__)

def validate_twilio_request(f):
    """Twilioからのリクエストを検証するデコレータ。"""
    def wrapper(*args, **kwargs):
        validator = RequestValidator(os.environ["TWILIO_AUTH_TOKEN"])

        # リクエスト詳細を取得
        url = request.url
        params = request.form.to_dict()
        signature = request.headers.get("X-Twilio-Signature", "")

        if not validator.validate(url, params, signature):
            return "Invalid request", 403

        return f(*args, **kwargs)
    wrapper.__name__ = f.__name__
    return wrapper

@app.route("/voice/incoming", methods=["POST"])
@validate_twilio_request
def incoming_call():
    """IVRメニュー付きで着信通話を処理します。"""
    response = VoiceResponse()

    # タイムアウト付きで数字を収集
    gather = Gather(
        num_digits=1,
        action="/voice/menu-selection",
        method="POST",
        timeout=5
    )
    gather.say(
        "Welcome to Acme Corp. "
        "Press 1 for sales. "
        "Press 2 for support. "
        "Press 3 to leave a message."
    )
    response.append(gather)

    # 入力がない場合は繰り返す
    response.redirect("/voice/incoming")

    return Response(str(response), mimetype="text/xml")

@app.route("/voice/menu-selection", methods=["POST"])
@validate_twilio_request
def menu_selection():
    """メニュー選択に基づいてルーティング。"""
    response = VoiceResponse()
    digit = request.form.get("Digits", "")

    if digit == "1":
        # 営業に転送
        response.say("Connecting you to sales.")
        response.dial(os.environ["SALES_PHONE"])

    elif digit == "2":
        # サポートに転送
        response.say("Connecting you to support.")
        response.dial(os.environ["SUPPORT_PHONE"])

    elif digit == "3":
        # ボイスメール
        response.say("Please leave a message after the beep.")
        response.record(
            action="/voice/voicemail-saved",
            max_length=120,
            transcribe=True,
            transcribe_callback="/voice/transcription"
        )

    else:
        response.say("Invalid selection.")
        response.redirect("/voice/incoming")

    return Response(str(response), mimetype="text/xml")

@app.route("/voice/voicemail-saved", methods=["POST"])
@validate_twilio_request
def voicemail_saved():
    """保存されたボイスメールを処理します。"""
    response = VoiceResponse()

    recording_url = request.form.get("RecordingUrl")
    recording_sid = request.form.get("RecordingSid")

    # データベースに保存、チームに通知など
    print(f"Voicemail saved: {recording_url}")

    response.say("Thank you. Goodbye.")
    response.hangup()

    return Response(str(response), mimetype="text/xml")

@app.route("/voice/transcription", methods=["POST"])
@validate_twilio_request
def transcription_callback():
    """ボイスメール文字起こしを処理します。"""
    transcription = request.form.get("TranscriptionText")
    recording_sid = request.form.get("RecordingSid")

    # 文字起こしを保存、Slackに送信など
    print(f"Transcription: {transcription}")

    return "", 200

# 発信通話の例
from twilio.rest import Client

def make_outbound_call(to: str, message: str):
    """カスタムTwiML付きで発信通話を作成します。"""
    client = Client(
        os.environ["TWILIO_ACCOUNT_SID"],
        os.environ["TWILIO_AUTH_TOKEN"]
    )

    # TwiML Bin URLまたはあなたのエンドポイント
    call = client.calls.create(
        to=to,
        from_=os.environ["TWILIO_PHONE_NUMBER"],
        url="https://your-app.com/voice/outbound-message",
        status_callback="https://your-app.com/voice/status"
    )

    return call.sid

if __name__ == "__main__":
    app.run(debug=True)

アンチパターン

• X-Twilio-Signatureを検証していない（セキュリティリスク）
• Twilioへの非XMLレスポンスを返している
• タイムアウト/入力なしケースを処理していない
• TwiMLに電話番号をハードコーディングしている

WhatsApp Business APIパターン

Twilio APIを使用してWhatsAppメッセージを送受信します。 SMSと同じTwilio Messages APIを使用しますが、小さな変更があります。

WhatsAppの重要なルール：

• 24時間セッションウィンドウ：ユーザーメッセージから24時間以内にのみ返信可能
• テンプレートメッセージ：セッションウィンドウ外での事前承認済みテンプレート
• オプトイン必須：ユーザーがメッセージの受信に明確に同意する必要があります
• レート制限：デフォルト80 MPS（承認で最大400）
• 文字制限：非テンプレート1024文字、テンプレート約550文字

使用時機：リッチメディア付きカスタマーサポート、ボタン付き注文通知、マーケティングメッセージ（テンプレート）、インタラクティブフロー（予約、アンケート）

from twilio.rest import Client
from twilio.base.exceptions import TwilioRestException
import os
from datetime import datetime, timedelta
from typing import Optional

class TwilioWhatsApp:
    """
    Twilio経由のWhatsApp Business API。
    セッションウィンドウとテンプレートメッセージを処理します。
    """

    def __init__(self):
        self.client = Client(
            os.environ["TWILIO_ACCOUNT_SID"],
            os.environ["TWILIO_AUTH_TOKEN"]
        )
        # WhatsApp番号形式：whatsapp:+14155551234
        self.from_number = os.environ["TWILIO_WHATSAPP_NUMBER"]

    def send_message(
        self,
        to: str,
        body: str,
        media_url: Optional[str] = None
    ) -> dict:
        """
        24時間セッション内でWhatsAppメッセージを送信します。

        Args:
            to: 受信者番号（E.164、whatsappプレフィックスなし）
            body: メッセージテキスト（非テンプレート最大1024文字）
            media_url: オプションの画像/ドキュメントURL

        Returns:
            メッセージ結果
        """
        # WhatsAppでフォーマット
        to_whatsapp = f"whatsapp:{to}"
        from_whatsapp = f"whatsapp:{self.from_number}"

        try:
            message_params = {
                "to": to_whatsapp,
                "from_": from_whatsapp,
                "body": body
            }

            if media_url:
                message_params["media_url"] = [media_url]

            message = self.client.messages.create(**message_params)

            return {
                "success": True,
                "message_sid": message.sid,
                "status": message.status
            }

        except TwilioRestException as e:
            return self._handle_whatsapp_error(e)

    def send_template_message(
        self,
        to: str,
        content_sid: str,
        content_variables: dict
    ) -> dict:
        """
        事前承認済みテンプレートメッセージを送信します。
        24時間ウィンドウ外のメッセージに使用します。

        コンテンツテンプレートはWhatsAppに最初に承認される必要があります。
        Twilio Console > Content Template Builderで作成できます。
        """
        to_whatsapp = f"whatsapp:{to}"
        from_whatsapp = f"whatsapp:{self.from_number}"

        try:
            message = self.client.messages.create(
                to=to_whatsapp,
                from_=from_whatsapp,
                content_sid=content_sid,
                content_variables=content_variables
            )

            return {
                "success": True,
                "message_sid": message.sid,
                "template": True
            }

        except TwilioRestException as e:
            return self._handle_whatsapp_error(e)

    def _handle_whatsapp_error(self, error: TwilioRestException) -> dict:
        """WhatsApp固有のエラーを処理します。"""
        error_handlers = {
            63016: "Outside 24-hour window. Use template message.",
            63018: "Template not approved or doesn't exist.",
            63025: "Too many template messages sent to this user.",
            63038: "Rate limit exceeded for WhatsApp.",
        }

        return {
            "success": False,
            "error_code": error.code,
            "error": error_handlers.get(error.code, error.msg)
        }

# 着信WhatsAppメッセージ用Flaskウェブフック
from flask import Flask, request

app = Flask(__name__)

@app.route("/webhooks/whatsapp", methods=["POST"])
def whatsapp_webhook():
    """着信WhatsAppメッセージを処理します。"""
    from_number = request.form.get("From", "").replace("whatsapp:", "")
    body = request.form.get("Body", "")
    media_url = request.form.get("MediaUrl0")  # 最初の添付ファイル

    # セッション開始を追跡（24時間ウィンドウが始まります）
    session_start = datetime.now()
    session_expires = session_start + timedelta(hours=24)

    # セッション追跡のためにデータベースに保存
    # user_sessions[from_number] = session_expires

    # メッセージを処理して応答
    response = process_whatsapp_message(from_number, body, media_url)

    # セッション内で返信
    whatsapp = TwilioWhatsApp()
    whatsapp.send_message(from_number, response)

    return "", 200

def process_whatsapp_message(phone: str, text: str, media: str) -> str:
    """着信メッセージを処理し応答を生成します。"""
    text_lower = text.lower()

    if "order status" in text_lower:
        return "Your order #1234 is out for delivery!"
    elif "support" in text_lower:
        return "A support agent will contact you shortly."
    else:
        return "Thanks for your message! Reply with 'order status' or 'support'."

# タイピングインジケータを送信（2025年機能）
def send_typing_indicator(to: str):
    """ユーザーに入力中であることを知らせます。"""
    # Senders API設定が必要
    pass

アンチパターン

• 24時間ウィンドウ外で非テンプレートメッセージを送信している
• ユーザーごとのセッションウィンドウを追跡していない
• セッションメッセージの1024文字制限を超えている
• テンプレート拒否エラーを処理していない

ウェブフックハンドラーパターン

配信ステータス、着信メッセージ、通話イベント用のTwilioウェブフックを処理します。 重要：常にX-Twilio-Signatureを検証してください。

Twilioは以下のウェブフックを送信します：

• メッセージステータス更新（キューイング → 送信 → 配信/失敗）
• 着信SMS/WhatsAppメッセージ
• 通話イベント（開始、着信音、応答、完了）
• 録音/文字起こし準備完了

使用時機：メッセージ配信ステータスの追跡、着信メッセージの受信、通話分析とログ、ボイスメール文字起こし処理

from flask import Flask, request, abort
from twilio.request_validator import RequestValidator
from functools import wraps
import os
import logging

app = Flask(__name__)
logger = logging.getLogger(__name__)

def validate_twilio_signature(f):
    """
    リクエストがTwilioから来たことを検証します。
    重要：ウェブフックエンドポイントに常にこれを使用してください。
    """
    @wraps(f)
    def wrapper(*args, **kwargs):
        validator = RequestValidator(os.environ["TWILIO_AUTH_TOKEN"])

        # フルURLを構築（クエリパラメータを含む）
        url = request.url

        # POSTボディを辞書として取得
        params = request.form.to_dict()

        # ヘッダーから署名を取得
        signature = request.headers.get("X-Twilio-Signature", "")

        if not validator.validate(url, params, signature):
            logger.warning(f"Invalid Twilio signature from {request.remote_addr}")
            abort(403)

        return f(*args, **kwargs)
    return wrapper

@app.route("/webhooks/twilio/sms/status", methods=["POST"])
@validate_twilio_signature
def sms_status_callback():
    """
    SMS配信ステータス更新を処理します。

    ステータスの進行：キューイング → 送信 → 送信済み → 配信完了
    または：キューイング → 送信 → 未配信/失敗
    """
    message_sid = request.form.get("MessageSid")
    status = request.form.get("MessageStatus")
    error_code = request.form.get("ErrorCode")
    error_message = request.form.get("ErrorMessage")

    logger.info(f"SMS {message_sid}: {status}")

    if status == "delivered":
        # メッセージが正常に配信された
        update_message_status(message_sid, "delivered")

    elif status == "undelivered":
        # キャリアが拒否またはその他の失敗
        logger.error(f"SMS failed: {error_code} - {error_message}")
        handle_failed_message(message_sid, error_code, error_message)

    elif status == "failed":
        # Twilioが送信できなかった
        logger.error(f"SMS send failed: {error_code}")
        handle_failed_message(message_sid, error_code, error_message)

    return "", 200

@app.route("/webhooks/twilio/sms/incoming", methods=["POST"])
@validate_twilio_signature
def incoming_sms():
    """
    着信SMSメッセージを処理します。
    """
    from_number = request.form.get("From")
    to_number = request.form.get("To")
    body = request.form.get("Body")
    num_media = int(request.form.get("NumMedia", 0))

    # メディア添付ファイルを処理
    media_urls = []
    for i in range(num_media):
        media_urls.append(request.form.get(f"MediaUrl{i}"))

    # オプトアウトキーワードをチェック
    if body.strip().upper() in ["STOP", "UNSUBSCRIBE", "CANCEL"]:
        handle_opt_out(from_number)
        return "", 200

    # オプトインキーワードをチェック
    if body.strip().upper() in ["START", "SUBSCRIBE"]:
        handle_opt_in(from_number)
        return "", 200

    # メッセージを処理
    process_incoming_sms(from_number, body, media_urls)

    return "", 200

@app.route("/webhooks/twilio/voice/status", methods=["POST"])
@validate_twilio_signature
def voice_status_callback():
    """通話ステータス更新を処理します。"""
    call_sid = request.form.get("CallSid")
    status = request.form.get("CallStatus")
    duration = request.form.get("CallDuration")
    direction = request.form.get("Direction")

    # 通話ステータス：開始、着信音、通話中、完了、ビジー、応答なし、キャンセル、失敗

    logger.info(f"Call {call_sid}: {status} ({duration}s)")

    if status == "completed":
        # 通話が正常に終了
        log_call_completion(call_sid, duration)

    elif status in ["busy", "no-answer", "canceled", "failed"]:
        # 通話が接続されなかった
        handle_failed_call(call_sid, status)

    return "", 200

# ヘルパー関数
def update_message_status(message_sid: str, status: str):
    """データベースのメッセージステータスを更新します。"""
    pass

def handle_failed_message(message_sid: str, error_code: str, error_msg: str):
    """失敗したメッセージ配信を処理します。"""
    # チームに通知、再試行ロジックなど
    pass

def handle_opt_out(phone: str):
    """ユーザーのオプトアウトを処理します。"""
    # ユーザーをデータベースでオプトアウト済みにマーク
    # 重要：これを尊重する必要があります！
    pass

def handle_opt_in(phone: str):
    """ユーザーの再オプトインを処理します。"""
    pass

def process_incoming_sms(from_phone: str, body: str, media: list):
    """着信SMSメッセージを処理します。"""
    pass

def log_call_completion(call_sid: str, duration: str):
    """完了した通話をログします。"""
    pass

def handle_failed_call(call_sid: str, status: str):
    """接続されなかった通話を処理します。"""
    pass

アンチパターン

• X-Twilio-Signatureを検証していない
• 認証なしでウェブフックURLを公開している
• オプトアウトキーワード（STOP）を処理していない
• ウェブフックレスポンスをブロック（高速である必要があります）

レート制限と再試行パターン

Twilioのレート制限を処理し、適切な再試行ロジックを実装します。

デフォルト制限：

• SMS：1秒あたり80メッセージ（MPS）
• 音声：地域と番号タイプに応じて異なります
• APIコール：1秒あたり100リクエスト

エラーコード：

• 20429：音声API レート制限
• 30429：メッセージングAPI レート制限

使用時機：大量メッセージングアプリケーション、一括SMS キャンペーン、自動通話システム

import time
import random
from functools import wraps
from twilio.base.exceptions import TwilioRestException
import logging

logger = logging.getLogger(__name__)

def exponential_backoff_retry(
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    rate_limit_codes: list = [20429, 30429]
):
    """
    レート制限での指数バックオフ再試行用デコレータ。

    ジッターを使用して雷群効果を防ぎます。
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)

                except TwilioRestException as e:
                    last_exception = e

                    # レート制限エラーのみ再試行
                    if e.code not in rate_limit_codes:
                        raise

                    if attempt == max_retries:
                        logger.error(f"Max retries exceeded: {e}")
                        raise

                    # ジッター付き遅延を計算
                    delay = min(
                        base_delay * (2 ** attempt) + random.uniform(0, 1),
                        max_delay
                    )

                    logger.warning(
                        f"Rate limited (attempt {attempt + 1}/{max_retries}). "
                        f"Retrying in {delay:.1f}s"
                    )
                    time.sleep(delay)

            raise last_exception

        return wrapper
    return decorator

# 使用例
from twilio.rest import Client

client = Client(account_sid, auth_token)

@exponential_backoff_retry(max_retries=5)
def send_sms(to: str, body: str):
    return client.messages.create(
        to=to,
        from_=from_number,
        body=body
    )

# レート制限付きで一括送信
import asyncio
from asyncio import Semaphore

class RateLimitedSender:
    """
    組み込みレート制限付きでメッセージを送信します。
    Twilioの80 MPS制限以下に保ちます。
    """

    def __init__(self, client, from_number: str, mps: int = 50):
        self.client = client
        self.from_number = from_number
        self.mps = mps
        self.semaphore = Semaphore(mps)

    async def send_bulk(self, messages: list[dict]) -> list[dict]:
        """
        レート制限付きでメッセージを送信します。

        Args:
            messages: リスト [{"to": "+1...", "body": "..."}]

        Returns:
            各メッセージの結果
        """
        tasks = [
            self._send_with_limit(msg["to"], msg["body"])
            for msg in messages
        ]

        return await asyncio.gather(*tasks, return_exceptions=True)

    async def _send_with_limit(self, to: str, body: str):
        """セマフォベースのレート制限付きで単一メッセージを送信します。"""
        async with self.semaphore:
            try:
                # スレッドプール内で同期クライアントを使用
                loop = asyncio.get_event_loop()
                result = await loop.run_in_executor(
                    None,
                    lambda: self.client.messages.create(
                        to=to,
                        from_=self.from_number,
                        body=body
                    )
                )
                return {"success": True, "sid": result.sid, "to": to}

            except TwilioRestException as e:
                return {"success": False, "error": str(e), "to": to}

            finally:
                # レート制限を維持するために遅延
                await asyncio.sleep(1 / self.mps)

# 使用例
async def send_campaign():
    sender = RateLimitedSender(client, from_number, mps=50)

    messages = [
        {"to": "+14155551234", "body": "Hello!"},
        {"to": "+14155555678", "body": "Hello!"},
        # ... 数千のメッセージ
    ]

    results = await sender.send_bulk(messages)

    successful = sum(1 for r in results if r.get("success"))
    print(f"Sent {successful}/{len(messages)} messages")

アンチパターン

• バックオフなしで即座に再試行している
• ジッターなし（雷群効果を引き起こします）
• 非レート制限エラーを再試行している
• Twilioの MPS制限を超えている

危険な注意点

オプトアウトしたユーザーに送信（エラー21610）

重大度：高

状況：ユーザーにSMSを送信

症状： メッセージがエラーコード21610で失敗します。Twilioはメッセージを拒否します。 ユーザーはSMSを受け取りません。同じ番号は前に機能していました。

理由： 受取人が前のメッセージに「STOP」（またはUNSUBSCRIBE、CANCEL等）で返信しました。Twilioは自動的にオプトアウトを尊重し、あなたのアカウントからその番号への追加メッセージをブロックします。

これは米国メッセージング（TCPA、CTIAガイドライン）で法的に必須です。 これをオーバーライドすることはできません - ユーザーは「START」に返信してオプトインしなおす必要があります。

推奨される修正：

データベースでオプトアウトステータスを追跡

# ウェブフックハンドラ内
@app.route("/webhooks/sms/incoming", methods=["POST"])
def incoming_sms():
    from_number = request.form.get("From")
    body = request.form.get("Body", "").strip().upper()

    # 標準的なオプトアウトキーワード
    if body in ["STOP", "UNSUBSCRIBE", "CANCEL", "END", "QUIT"]:
        mark_user_opted_out(from_number)
        return "", 200

    # 標準的なオプトインキーワード
    if body in ["START", "SUBSCRIBE", "YES", "UNSTOP"]:
        mark_user_opted_in(from_number)
        return "", 200

    # 他のメッセージを処理...

# 送信前に
def send_sms_safe(to: str, body: str):
    if is_user_opted_out(to):
        return {"success": False, "error": "User has opted out"}

    try:
        return send_sms(to, body)
    except TwilioRestException as e:
        if e.code == 21610:
            # キャリア経由でオプトアウトしたことをデータベースで更新
            mark_user_opted_out(to)
        raise

オプトアウト指示を含める

マーケティングメッセージに「返信STOPで登録解除」を追加します。

電話到達不可だが有効（エラー30003）

重大度：中

状況：モバイル番号にSMSを送信

症状： メッセージがエラー30003で失敗します。番号は有効で前に機能していました。 間欠的 - 時々は機能し、時々は失敗します。

理由： エラー30003は「到達不可な宛先ハンドセット」を意味します。電話は存在しますが、現在メッセージを受け取ることができません。一般的な原因：

• 電源がオフ
• 飛行機モード
• 信号範囲外
• キャリアネットワークの問題
• 電話ストレージが満杯

30006（永続的に到達不可）と異なり、30003は通常一時的です。

推奨される修正：

一時的な失敗に対して再試行ロジックを実装

TRANSIENT_ERRORS = [30003, 30008, 30009]  # 再試行可能なエラー

async def send_with_retry(to: str, body: str, max_retries: int = 3):
    for attempt in range(max_retries):
        result = send_sms(to, body)

        if result["success"]:
            return result

        if result.get("error_code") not in TRANSIENT_ERRORS:
            # 永続的な失敗は再試行しない
            return result

        # 指数バックオフ：5分、15分、45分
        delay = 300 * (3 ** attempt)
        await asyncio.sleep(delay)

    return {"success": False, "error": "Max retries exceeded"}

フォールバックチャネルを提供

async def notify_user(user, message):
    # 最初にSMSを試行
    result = await send_sms(user.phone, message)

    if result.get("error_code") == 30003:
        # 電話到達不可 - メール試行
        await send_email(user.email, message)
        return {"channel": "email", "status": "sent"}

    return {"channel": "sms", "status": result["status"]}

キャリアフィルタリングでメッセージがブロック

重大度：高

状況：米国の電話番号にSMSを送信

症状： メッセージは「送信済み」と表示されますが「配信」にはなりません。Twilioからエラーはありません。 ユーザーはメッセージを受け取ったと言いません。特定のキャリアまたはメッセージコンテンツのパターンがあります。

理由： 米国キャリア（Verizon、AT&T、T-Mobile）はSMSをスパムに対して積極的にフィルタリングします。 メッセージは以下の場合ブロックされる可能性があります：

• URLが含まれている（特に短URL または未知のドメイン）
• フィッシングのように見える（緊急、アカウント、検証、今すぐクリック）
• 同じ番号からの大量
• 登録されたA2P 10DLCを使用していない
• 低い送信者評判

キャリアはTwilioにメッセージが配信される理由を伝えません - 黙ってドロップします。

推奨される修正：

A2P 10DLCに登録（米国要件）

1. Twilio Console > Messaging > Trust Hub へ移動
2. ビジネスブランドを登録
3. メッセージングキャンペーン作成（ユースケースを説明）
4. 承認を待つ（数日かかる可能性があります）
5. キャンペーンに電話番号を関連付ける

メッセージコンテンツのベストプラクティス

def sanitize_message(text: str) -> str:
    """メッセージがフィルタリングされにくくするようにします。"""
    # URL短縮機を使用しない - フルドメインを使用
    # スパムトリガーワードを使用しない
    # 会話的に、販促的ではなく

    # 例：代わりにこれはしないでください
    bad = "URGENT: Verify your account now! Click: bit.ly/abc"

    # これをしてください
    good = "Hi! Your order #1234 is ready. Questions? Reply here."

    return text

# 高い容量にはフリーダイヤルまたはショートコードを使用
# 10DLCは<10K msg/day向け
# フリーダイヤル：最大10K msg/day
# ショートコード：100K+ msg/day

配信率を監視

def track_delivery_rate():
    sent = get_messages_with_status("sent")
    delivered = get_messages_with_status("delivered")

    rate = len(delivered) / len(sent) * 100

    if rate < 95:
        alert_team(f"Delivery rate dropped to {rate}%")

ウェブフック署名を検証していない

重大度：クリティカル

状況：Twilioウェブフックコールバックを受け取る

症状： 攻撃者がウェブフックエンドポイントに偽のウェブフックを送信します。詐欺的トランザクションが処理されます。なりすまされた着信メッセージがアクション引きトリガーします。

理由： Twilioはすべてのウェブフックリクエストに X-Twilio-Signatureヘッダーで署名しています。 これを検証しない場合、ウェブフックURLを知っている誰もがTwilioであるふりをして偽のリクエストを送信できます。

これは以下につながる可能性があります：

• 偽のメッセージ配信確認
• なりすまされた着信メッセージ
• 詐欺的な検証承認

推奨される修正：

常に署名を検証

from twilio.request_validator import RequestValidator
from flask import Flask, request, abort
from functools import wraps
import os

def require_twilio_signature(f):
    """Twilioウェブフックリクエストを検証するデコレータ。"""
    @wraps(f)
    def wrapper(*args, **kwargs):
        validator = RequestValidator(os.environ["TWILIO_AUTH_TOKEN"])

        # クエリ文字列を含むフルURL
        url = request.url

        # POSTボディを辞書として
        params = request.form.to_dict()

        # 署名ヘッダー
        signature = request.headers.get("X-Twilio-Signature", "")

        if not validator.validate(url, params, signature):
            abort(403)

        return f(*args, **kwargs)
    return wrapper

@app.route("/webhooks/twilio", methods=["POST"])
@require_twilio_signature  # 常にこれを使用
def twilio_webhook():
    # 処理するのは安全
    pass

一般的な検証の落とし穴

# URLは Twilioが呼び出したものと正確に一致する必要があります
# プロキシの後ろにある場合、以下が必要かもしれません：
url = request.headers.get("X-Forwarded-Proto", "http") + "://" + \
      request.headers.get("X-Forwarded-Host", request.host) + \
      request.path

# ngrokを使用している場合、再起動するたびにURLが変わります
# 本番環境では一貫したURLを使用してください

WhatsAppメッセージが24時間ウィンドウ外（エラー63016）

重大度：高

状況：ユーザーにWhatsAppメッセージを送信

症状： メッセージがエラー63016で失敗します。「メッセージが許可されたウィンドウの外」です。 テンプレートメッセージは機能していますが、通常のメッセージは失敗します。

理由： WhatsAppには迷惑メール以外のメッセージについて厳密なルールがあります：

• ユーザーは最初にあなたにメッセージする必要があります
• 彼らの最後のメッセージから24時間以内にのみ返信できます
• 24時間後、事前承認済みのテンプレートメッセージを使用する必要があります

これはスパムを防ぎ、プラットフォームとしてのWhatsAppの信頼を維持します。

推奨される修正：

ユーザーごとにセッションウィンドウを追跡

from datetime import datetime, timedelta

class WhatsAppSession:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.window_hours = 24

    def start_session(self, phone: str):
        """着信メッセージで24時間セッションを開始/更新。"""
        key = f"wa_session:{phone}"
        expires = datetime.now() + timedelta(hours=self.window_hours)
        self.redis.set(key, expires.isoformat(), ex=self.window_hours * 3600)

    def can_send_freeform(self, phone: str) -> bool:
        """非テンプレートメッセージを送信できるかをチェック。"""
        key = f"wa_session:{phone}"
        expires_str = self.redis.get(key)

        if not expires_str:
            return False

        expires = datetime.fromisoformat(expires_str)
        return datetime.now() < expires

    def send_message(self, phone: str, body: str, template_sid: str = None):
        """メッセージを送信、ウィンドウ外の場合テンプレートを使用。"""
        if self.can_send_freeform(phone):
            return send_whatsapp_message(phone, body)
        elif template_sid:
            return send_whatsapp_template(phone, template_sid)
        else:
            return {
                "success": False,
                "error": "Outside session window, template required"
            }

着信メッセージウェブフック

@app.route("/webhooks/whatsapp", methods=["POST"])
def whatsapp_incoming():
    from_phone = request.form.get("From").replace("whatsapp:", "")

    # セッションを開始/更新
    session.start_session(from_phone)

    # メッセージを処理...

一般的なメッセージに対して承認されたテンプレートを作成

1. Twilio Console > Content Template Builder
2. {{1}}プレースホルダーでテンプレートを作成
3. WhatsApp承認用に提出（24～48時間かかります）
4. content_sidを使用して送信

Account SIDまたはAuth Tokenが公開

重大度：クリティカル

状況：Twilio統合をデプロイ

症状： Twilioアカウント上で権限のない請求。送信していないメッセージが送信されました。 承認なしで購入した電話番号。

理由： 攻撃者が Account SID + Auth Tokenを入手した場合、あなたの Twilioアカウントへの完全なアクセス権があります。以下が可能です：

• メッセージ送信（アカウントに課金）
• 電話番号を購入
• 通話録音にアクセス
• 設定を変更

一般的な公開ポイント：

• ソースコード内にハードコーディング（GitHubにプッシュ）
• クライアント側の JavaScript
• Dockerイメージ
• ログ内

推奨される修正：

認証情報を絶対にハードコーディングしない

# 悪い - これはしないでください
client = Client("AC1234...", "abc123...")

# 良い - 環境変数
client = Client(
    os.environ["TWILIO_ACCOUNT_SID"],
    os.environ["TWILIO_AUTH_TOKEN"]
)

# 良い - シークレットマネージャー
from aws_secretsmanager import get_secret
creds = get_secret("twilio-credentials")
client = Client(creds["sid"], creds["token"])

Auth Tokenの代わりに API Key を使用

# Auth Tokenは完全なアカウントアクセス権を持ちます
# API Keysはスコープ化でき、取り消すことができます

# Twilio Console で API Keyを作成
client = Client(
    os.environ["TWILIO_API_KEY_SID"],
    os.environ["TWILIO_API_KEY_SECRET"],
    os.environ["TWILIO_ACCOUNT_SID"]
)

# 侵害された場合、そのキーだけを取り消す

トークンが公開された場合は直ちに取り直す

1. Twilio Console > Account > API credentials
2. Auth Tokenを取り直す
3. 新しいトークンですべてのデプロイを更新
4. 権限のない使用についてアカウントアクティビティをレビュー

検証レート制限超過（エラー60203）

重大度：中

状況：検証コードを送信

症状： 検証リクエストがエラー60203で失敗します。 「この電話番号の最大送信試行回数に達しました」。

理由： Twilio Verifyには濫用防止のための組み込みレート制限があります：

• サービスあたりの電話番号あたり10分あたり5回の検証試行
• SMSポンピング詐欺を防止
• ブルートフォース攻撃から保護

ユーザーが合法的に追加の試行が必要な場合、UXに問題があるかもしれません。

推奨される修正：

アプリケーションレベルのレート制限も実装

from datetime import datetime, timedelta
import redis

class VerifyRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
        # Twilioの制限より厳しい
        self.max_attempts = 3
        self.window_minutes = 10

    def can_request(self, phone: str) -> bool:
        key = f"verify_rate:{phone}"
        attempts = self.redis.get(key)

        if attempts and int(attempts) >= self.max_attempts:
            return False

        return True

    def record_attempt(self, phone: str):
        key = f"verify_rate:{phone}"
        pipe = self.redis.pipeline()
        pipe.incr(key)
        pipe.expire(key, self.window_minutes * 60)
        pipe.execute()

    def get_wait_time(self, phone: str) -> int:
        """ユーザーが再度リクエストできるまでの秒数を返す。"""
        key = f"verify_rate:{phone}"
        ttl = self.redis.ttl(key)
        return max(0, ttl)

# 使用例
limiter = VerifyRateLimiter(redis_client)

@app.route("/verify/send", methods=["POST"])
def send_verification():
    phone = request.json["phone"]

    if not limiter.can_request(phone):
        wait = limiter.get_wait_time(phone)
        return {
            "error": f"Too many attempts. Try again in {wait} seconds."
        }, 429

    result = twilio_verify.send_verification(phone)

    if result["success"]:
        limiter.record_attempt(phone)

    return result

明確なユーザーフィードバックを提供

# 残りの試行回数を表示
# カウントダウンタイマーを表示
# 代替案を提供（音声通話、メール）

検証チェック

ハードコーディングされた Twilio 認証情報

重大度：エラー

Twilio認証情報は絶対にハードコーディングしてはいけません

メッセージ：ハードコーディングされたTwilio SIDが検出されました。環境変数を使用してください。

ソースコード内のAuth Token

重大度：エラー

Auth トークンは環境変数に含まれるべきです

メッセージ：ハードコーディングされたAuth Token。 os.environ['TWILIO_AUTH_TOKEN']を使用してください。

署名検証なしのウェブフック

重大度：エラー

Twilioウェブフックは X-Twilio-Signatureを検証する必要があります

メッセージ：署名検証なしのウェブフック。RequestValidator チェックを追加してください。

クライアント側コード内のTwilio認証情報

重大度：エラー

Twilioの認証情報をブラウザに公開しないでください

メッセージ：Twilio認証情報がクライアント側に公開されています。サーバー側のみ使用してください。

E.164 電話番号検証なし

重大度：警告

電話番号は送信前に検証する必要があります

メッセージ：E.164検証なしで電話に送信しています。

ハードコーディングされた電話番号

重大度：警告

電話番号は設定またはデータベースから来る必要があります

メッセージ：ハードコーディングされた電話番号。設定または環境変数を使用してください。

Twilio例外処理なし

重大度：警告

TwilioコールはTwilioRestExceptionを処理する必要があります

メッセージ：エラーハンドリングなしのTwilio APIコール。TwilioRestExceptionをキャッチしてください。

特定のエラーコードを処理していない

重大度：情報

一般的なTwilioエラーコード を特に処理することを検討してください

メッセージ：特定のエラーコード（21610、30003等）の処理を検討してください。

オプトアウトキーワード処理なし

重大度：警告

SMSシステムはSTOP/UNSUBSCRIBEキーワードを処理する必要があります

メッセージ：オプトアウト処理があり