Zenn
KWCPLUSPublicationへの投稿
🀄

電話をかけサーバ側のコマンドを実行してみる

KWC sutou
に公開
Vonage
tech

はじめに

Vonageのサンプルプログラムをベースに今回は特定の電話番号からVonageへ電話をかけ1を押すとTopコマンドの実行結果をMicrosoft Teamsへ通知する環境を作ってみました。

設備の状態を確認するコマンドを電話から実行し、Teamsへ通知することでオペレーションができない状況でもコマンドの実行結果を確認でき、正常性確認や状況確認を誰でもできるようになり、オペレーションミスも減らすことが可能です。

Vonageとは

Vonage_logo

Vonage は、米国ニュージャージー州に本社を置く、CPaaS（Communication Platform as a Service）企業です。

もともとは VoIP（Voice over IP）企業としてスタートしましたが、いくつかの企業買収を行うことで、コミュニケーションサービス全般をサポートすることができる企業に発展しました。現在はスウェーデンの大手通信機器会社エリクソンの傘下に入っています。

2024年2月14日より、株式会社KDDIウェブコミュニケーションズ（以後、KWC）が Vonage の再販事業を開始することとなりました。

前提条件

  • Vonageの電話番号が取得済みであること

    →Vonageアカウント作成についてはこちらを参考にしてください。

    →番号取得についてはこちらを参考にしてください。
  • 発着信可能な電話番号があること(携帯電話など)
  • ngrokをインストール可能なインターネットに接続されたLinuxサーバがあること
  • 通知可能なMicrosoft Teamsがあること

Vonage ダッシュボードからアプリケーションを作成して番号を割り当て

  1. Vonageダッシュボードへログインし左メニューからアプリケーションを選択し新しいアプリケーションを作成するを選択
    電話でサーバ側のコマンド実行001

  2. 名前にアプリケーション名を入れ機能から音声をONにし新しいアプリケーションの作成を選択
    電話でサーバ側のコマンド実行002

  3. 番号を所有している場合は以下画像のように番号が表示されるのでリンクを選択し番号を割り当てる
    電話でサーバ側のコマンド実行003

Ubuntu Server 24.04にngrok環境を準備してサンプルプログラムで動作確認

  1. ngrokのインストールが終わり実行すると以下のようになります。(今回はport：8000で起動しています)
$ ngrok http http://localhost:8000
ngrok                                                                                             (Ctrl+C to quit)
                                                                                                                  
🛡️ Protect endpoints w/ IP Intelligence: https://ngrok.com/r/ipintel                                               
                                                                                                                  
Session Status                online                                                                              
Account                       yyyyy-yyyy@xxxxx.xxx (Plan: Free)                                                   
Version                       3.20.0                                                                              
Region                        Japan (jp)                                                                          
Web Interface                 http://127.0.0.1:4040                                                               
Forwarding                    https://1111-222-33-44-55.ngrok-free.app -> http://localhost:8000                   
                                                                                                                  
Connections                   ttl     opn     rt1     rt5     p50     p90                                         
                              0       0       0.00    0.00    0.00    0.00

  1. Vonageダッシュボードのアプリケーションから先程作成したアプリケーションを選択し編集を押下
    電話でサーバ側のコマンド実行004

  2. 音声の回答URL、イベントURLへngrokで保存したURLに/webhooks/answerを追加して入力し変更を保存を選択
    電話でサーバ側のコマンド実行005
    入力例：https://1111-222-33-44-55.ngrok-free.app/webhooks/answer

  3. ngrokを動かしているサーバでVonageのドキュメントにあるサンプルプログラムを準備

    まずはpipにて必要なパッケージをインストール

pip install vonage 'fastapi[standard]' python-dotenv

Vonageのドキュメントにあるサンプルプログラムを使うのですがそのままだと押した番号を正しく取得できなかったため少し修正(音声の日本語化と数値取得の修正)したものを使います。

handle-user-input.py
import os
from os.path import dirname, join
from dotenv import load_dotenv
from fastapi import Body, FastAPI, Request
from vonage_voice.models import Input, NccoAction, Talk


def load_env_variables():
    dotenv_path = join(dirname(__file__), "../.env")
    load_dotenv(dotenv_path)

    return {
        "VONAGE_NUMBER": os.environ.get("VONAGE_NUMBER"),
        "RECIPIENT_NUMBER": os.environ.get("RECIPIENT_NUMBER"),
    }


env_vars = load_env_variables()

app = FastAPI()


def create_ncco(request: Request):
    return [
        Talk(text="番号を選択してください。", language="ja-JP"),
        Input(
            type=["dtmf"],
            maxDigits=1,
            eventUrl=[str(request.base_url) + "webhooks/dtmf"],
        ),
    ]


@app.get("/webhooks/answer")
async def answer_call(request: Request):
    ncco = create_ncco(request)
    return [action.model_dump(by_alias=True, exclude_none=True) for action in ncco]


@app.post("/webhooks/dtmf")
async def answer_dtmf(data: dict = Body(...)):
    dtmf_input = data.get("dtmf", {}).get("digits")
    response_text = f"あなたが押した番号は {dtmf_input}です"

    return [
        Talk(text=response_text, language="ja-JP").model_dump(
            by_alias=True, exclude_none=True
        )
    ]

上記プログラムを保存したら以下でプログラム(handle-user-input.py)を実行する。

uvicorn handle-user-input:app --reload

以下のような実行結果になりエラーなどが出ていなければ起動完了

$ uvicorn handle-user-input:app --reload
INFO:     Will watch for changes in these directories: ['/home/hogehoge']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [31249] using WatchFiles
INFO:     Started server process [31251]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
  1. 動作確認
  • ngrokが起動
  • Vonageアプリケーションへngrok URL + /webhooks/answerの設定
  • サンプルプログラムの実行
    上記3つが完了して正常にすべて動作していれば動作確認をしてみましょう。

    Vonage番号へ発信→番号を聞かれる→キーパッドで番号を選択→押した番号を読み上げ→通話終了

    となればサンプルプログラムの動作環境構築は完了となります。

サーバのプログラムを実行してみる

  1. Microsoft Teamsの通知するチャネルにIncoming Webhookのコネクタを作成しURLをコピー

  2. topコマンドの実行結果をMicrosoft Teamsへ通知するスクリプトを用意
    今回はpost_ms-teams.pyとして保存(次項のhandle-user-input.pyでファイル名をハードコーディングしているため)

post_ms-teams.py
import subprocess

import requests

# TeamsのIncoming WebhookのURLを設定
webhook_url = "コピーしたTeamsのIncoming WebhookのURL"

# Topコマンドを実行し出力を取得
result = subprocess.run(["top", "-b", "-n", "1"], capture_output=True, text=True)
top_output = result.stdout

# 改行を保持するためにMarkdown形式に変換
formatted_output = top_output.replace('\n', '\n\n')

# Teamsに送信するペイロードを作成
payload = {"text": formatted_output}

# HTTP POSTリクエストを送信して、Teamsに通知する
response = requests.post(webhook_url, json=payload)

# レスポンスを確認
if response.status_code == 200:
    print("通知が成功しました")
else:
    print(f"通知に失敗しました: {response.status_code}, {response.text}")

手動で実行し、Teamsへ通知されれば成功

  1. 先ほどのhandle-user-input.pyを参考に特定番号以外は処理せず、1を押されたら上記の通知スクリプトを実行する内容へ改修
handle-user-input.py
import os
import sys
from os.path import join, dirname
from dotenv import load_dotenv
from fastapi import FastAPI, Body, Request
from vonage_voice.models import Input, NccoAction, Talk
import runpy


def load_env_variables():
    dotenv_path = join(dirname(__file__), "../.env")
    load_dotenv(dotenv_path)

    return {
        "VONAGE_NUMBER": os.environ.get("VONAGE_NUMBER"),
        "RECIPIENT_NUMBER": os.environ.get("RECIPIENT_NUMBER"),
    }


env_vars = load_env_variables()
script_path = "./post_ms-teams.py"

app = FastAPI()


def create_ncco(request: Request):
    return [
        Talk(text="番号を入力してください", language="ja-JP"),
        Input(
            type=["dtmf"],
            maxDigits=1,
            eventUrl=[str(request.base_url) + "webhooks/dtmf"],
        ),
    ]


@app.get("/webhooks/answer")
async def answer_call(request: Request):
    from_number = request.query_params.get("from")

    # 発信する番号を指定
    if from_number != "8190YYYYZZZZ":
        sys.exit(1)

    ncco = create_ncco(request)
    return [action.model_dump(by_alias=True, exclude_none=True) for action in ncco]


@app.post("/webhooks/dtmf")
async def answer_dtmf(request: Request, data: dict = Body(...)):
    dtmf_input = data.get("dtmf", {}).get("digits")

    if dtmf_input == "1":
        runpy.run_path(script_path)
        response_text = (
            f"{dtmf_input}が入力されました。top コマンドの結果をTeamsへ通知します。"
        )
    else:
        response_text = f"こんにちは、あなたは {dtmf_input} を押しました"

    return [
        Talk(text=response_text, language="ja-JP").model_dump(
            by_alias=True, exclude_none=True
        )
    ]

これで090YYYYZZZZからの着信で1を押された場合のみTopコマンドを実行し結果をTeamsへ通知する仕組みが完成。

Vonageの電話番号へ発信し動作確認。

終わりに

今回は電話にてサーバ側のコマンドを実行する仕組みを紹介しました。

インターネットから直接アクセス可能なサーバであればngrokを使わずに実施することも可能です。

参考資料

KWCPLUS
KWCPLUSPublication

株式会社KDDIウェブコミュニケーションズ Tech Blog

Discussion