💨
Azure で障害自動電話通知コールセンターを構築してみた🐛

2025/09/07に公開
 はじめに障害が発生しているのに気づかない。。

気づくのに少し遅れてしまった。。
こんなことないでしょうか。
それを解決すべく、Azure環境にて障害通知コールセンター機能を実装してみました。

 概要今回構築システムの機能概要です：
障害発生時に Logic Apps からトリガーされる（障害アラートによりトリガー）
登録された担当者リストからランダムに電話をかける
音声ガイダンスで状況を説明し、プッシュボタンでの応答を求める
対応可能であれば「1」、別の担当者に転送するには「2」を押してもらう
応答がない場合や転送を選択された場合、次の担当者へ自動発信
対応状況を Slack に通知する

 アーキテクチャ
Logic Apps: 障害検知システムからのアラートを受け取り、Azure Functions をトリガー

Azure Functions: 電話発信ロジックとコールフロー管理を担当

Azure Communication Services: 実際の電話発信と音声ガイダンス再生を実行

Azure Cognitive Services: テキストから音声への変換サービス

Slack: 対応状況の通知先

 前提条件Azure アカウント
Azure Communication Services リソース（電話番号取得済み）
Azure Cognitive Services（Speech Services）
Azure Functions（Python 対応）
Slack Webhook URL（通知用）

 実装
 1. 環境変数の設定Azure Functions に以下の環境変数を設定します：

COMMUNICATION_SERVICES_CONNECTION_STRING: Azure Communication Servicesの接続文字列

ACS_PHONE_NUMBER: 発信元の電話番号（communicationServiceから取得）

COGNITIVE_SERVICES_ENDPOINT: Cognitive Servicesのエンドポイント

CALLBACK_URL: コールバックを受け取るURL

DEVELOPER_NUMBER: 開発者リスト（JSON形式）

SLACK_WEBHOOK_URL: SlackのWebhook URL
開発者リストは以下のような JSON 形式で設定します：
[
  {"phone_number": "+818012345678", "name": "山田"},
  {"phone_number": "+818087654321", "name": "佐藤"}
]

 2. Azure Functionsの実装（主要な関数）
 初回電話呼び出し関数@app.route(methods=["GET"])
def firstCall(req: func.HttpRequest) -> func.HttpResponse:
    # 環境変数から情報を取得
    developer_list_str = os.environ.get("DEVELOPER_NUMBER")
    connection_string = os.environ.get("COMMUNICATION_SERVICES_CONNECTION_STRING")
    acs_phone_number = os.environ.get("ACS_PHONE_NUMBER")
    cognitive_services_endpoint = os.environ.get("COGNITIVE_SERVICES_ENDPOINT")
    callback_url_env = os.environ.get("CALLBACK_URL")

    # 開発者リストから電話番号を選択
    if developer_list_str:
        try:
            developer_list = json.loads(developer_list_str)
            if developer_list:
                # ランダムで電話先を選択
                selected_developer = random.choice(developer_list)
                target_phone_number = selected_developer['phone_number']
                member_name = selected_developer.get('name', '不明')
            else:
                raise ValueError("開発者リストが空です")
        except (json.JSONDecodeError, KeyError, ValueError):
            initialize_call_order()
            target_phone_number = get_next_phone_number()
            member_name = get_member_name(target_phone_number) if target_phone_number else None
    else:
        # 電話リストを３回ループ、一度のループ内で重複した人に電話をかけないようにする実装。記事内では詳細割愛。
        initialize_call_order()
        target_phone_number = get_next_phone_number()
        member_name = get_member_name(target_phone_number) if target_phone_number else None

    # 電話発信
    call_automation_client = CallAutomationClient.from_connection_string(connection_string)
    call_invite = CallInvite(
        target=PhoneNumberIdentifier(target_phone_number),
        source_caller_id_number=PhoneNumberIdentifier(acs_phone_number)
    )
    # 電話接続後、下記コールバック関数にコールバック
    create_call_result = call_automation_client.create_call(
        call_invite,
        callback_url_env,
        operation_context="emergency_call",
        cognitive_services_endpoint=cognitive_services_endpoint
    )

    return func.HttpResponse(
        f"緊急通報開始: {member_name} に発信中 (Call ID: {create_call_result.call_connection_id})",
        status_code=200
    )


 コールバック処理@app.route( methods=["POST", "GET"])
def callback(req: func.HttpRequest) -> func.HttpResponse:
    if req.method == "GET":
        return func.HttpResponse("OK", status_code=200)

    # イベントデータを解析
    request_body = req.get_body()
    event_payload = json.loads(request_body.decode('utf-8'))
    event_data = event_payload[0] if isinstance(event_payload, list) else event_payload
    event_type = event_data.get("type")
    data_section = event_data.get("data", {})
    call_connection_id = data_section.get("callConnectionId")

    # イベントタイプに応じた処理（communication serviceからイベント受信）
    if event_type == "Microsoft.Communication.CallConnected":
        # 電話がつながったらガイダンスを再生
        play_guidance(connection_string, call_connection_id)

    elif event_type == "Microsoft.Communication.PlayCompleted":
        # ガイダンス再生完了後、DTMF入力待ち
        start_dtmf_recognition(connection_string, call_connection_id)

    elif event_type == "Microsoft.Communication.RecognizeCompleted":
        # DTMF入力を処理
        pressed_tone = get_pressed_tone_from_data(data_section)
        normalized_tone = normalize_tone(pressed_tone)

        if normalized_tone == "1":
            # 1が押された場合：対応開始
            member_name = get_member_name(get_current_phone_number())
            send_slack_notification(f"🚨🚨障害通知、{member_name}さんが対応を開始されました🚨🚨")
            hang_up_call(connection_string, call_connection_id)

        elif normalized_tone == "2":
            # 2が押された場合：次の担当者へ転送
            play_transfer_message(connection_string, call_connection_id)

    # その他のイベント処理...

    return func.HttpResponse("OK", status_code=200)

 音声ガイダンス再生def play_guidance(connection_string, call_connection_id):
    try:
        client = CallAutomationClient.from_connection_string(connection_string)
        call_connection = client.get_call_connection(call_connection_id)

        # テキスト音声合成の設定
        text_source = TextSource(
            text="障害通知システムです。1番で対応開始、2番で次の担当者に転送します。",
            source_locale="ja-JP",
            voice_name="ja-JP-NanamiNeural"
        )

        # 音声再生
        call_connection.play_media_to_all(
            play_source=text_source,
            loop=False,
            operation_context="emergency_guidance"
        )
        return True
    except Exception:
        return False


 DTMF（プッシュボタン）認識def start_dtmf_recognition(connection_string, call_connection_id):
    try:
        client = CallAutomationClient.from_connection_string(connection_string)
        call_connection = client.get_call_connection(call_connection_id)
        call_properties = call_connection.get_call_properties()
        participants = getattr(call_properties, 'targets', [])

        if not participants:
            return False

        target_participant = participants[0]
        call_connection.start_recognizing_media(
            dtmf_max_tones_to_collect=1,
            input_type=RecognizeInputType.DTMF,
            target_participant=target_participant,
            end_silence_timeout=15,
            initial_silence_timeout=120,
            operation_context="emergency_dtmf"
        )
        return True
    except Exception:
        return False


 その他諸々
 1. Azure Functions のデプロイVS Code の Azure Functions 拡張機能や、Azure CLI を使ってデプロイ：

 2. Logic Apps の設定Logic Apps では、障害検知トリガーから以下のような HTTP アクションを設定：
HTTP Action

Method: GET

URI: https://your-function-app.azurewebsites.net/api/pomameACS

(初回電話呼び出し関数を指定)

 3. 電話で案内を鳴らすサービス電話はなるが、音声が出ない問題にかなり頭を悩ませました。

解決策は以下でした。
＊解決策

AI-Multi-Account（Classic）を使用してCognitiveServiceと連携で解決。

ソース記事：

https://learn.microsoft.com/en-us/answers/questions/2278843/action-failed-due-to-a-bad-request-to-cognitive-se
最新のAIでテキストを音声再生するサービスは「AiSpeechService」。

今回のCognitiveServiceとの接続においては「SpeechService」がCognitiveServiceと互換性がないため電話がかかっても音声が再生されなかったのが原因でした。

 まとめAzure Communication Services と Azure Functions を組み合わせることで、シンプルながらも効果的な自動電話通知システムを構築できました。このシステムにより、障害発生時の連絡プロセスを自動化し、迅速な対応につなげることが可能です。
サーバーレスアーキテクチャを採用しているため、実行時のみコストが発生し、運用負担も少なくなっています。障害対応の効率化に興味がある方は、ぜひ試してみてください。
はじめに

概要

アーキテクチャ

前提条件

実装

1. 環境変数の設定

2. Azure Functionsの実装（主要な関数）

初回電話呼び出し関数

コールバック処理

音声ガイダンス再生

DTMF（プッシュボタン）認識

その他諸々

1. Azure Functions のデプロイ

2. Logic Apps の設定

3. 電話で案内を鳴らすサービス

まとめ

Discussion
