🀄

Vonage で複数人へ順番に通話を転送する

2025/01/15に公開

はじめに

こんにちは。KDDI ウェブコミュニケーションズの西嵜(にしざき)です。

この記事では、Vonage を用いて、着信通話を複数人へ順番に転送する方法について、ご紹介いたします。

本記事の対象となる読者

  • Vonage に興味のある方
  • Vonage を用いて、着信通話を複数人へ順番に通話を転送する方法を学びたい方

準備

この記事でご紹介する手順では、以下のものが必要となります。

  • Vonage アカウント
    • API キー(アカウントを開設すると、自動的に付与されます)
    • Vonage 電話番号
  • Python
    • 最新の安定バージョンを推奨します。
  • Vonage CLI
    • Vonage のさまざまな要素を操作するためのコマンドライン・インターフェースです。
  • ngrok
    • ローカル PC で作成したスクリプトを外部に公開するために利用します。
  • 電話機 2 台以上
    • 今回は複数人への通話転送を実装するため、発信側 1 台と着信側に最低 1 台、合わせて 2 台以上の電話機が必要となります。3 台以上あれば理想です。

手順

Python での NCCO スクリプトの実装

複数人への通話転送を実現するためのスクリプトを、Python を用いて実装します。

以下の Python のコードをコピーしてファイルに貼り付け、rotate-inbound-calls.py というファイル名で保存します。

rotate-inbound-calls.py
import os
import json
import flask
import vonage
from vonage_voice.models import CreateCallRequest, Conversation, Input, Talk
from vonage_voice.models.input_types import Dtmf

APP_DEBUG = True
VONAGE_APPLICATION_ID = 'VONAGE_APPLICATION_ID'
VONAGE_APPLICATION_PRIVATE_KEY_PATH = 'VONAGE_APPLICATION_PRIVATE_KEY_PATH'
VONAGE_CONVERSATION = 'AwesomeConversation'
VONAGE_NUMBER = 'VONAGE_NUMBER'
DESTINATION_NUMBERS = [
    'DESTINATION_NUMBER_1',
    'DESTINATION_NUMBER_2',
]
MAX_LOOPS = 3
LOCAL_HOSTNAME = 'localhost'
LOCAL_PORT = 3000
NGROK_HOSTNAME = 'NGROK_HOSTNAME'

url = f'https://{NGROK_HOSTNAME}'
url_webhooks_event = f'{url}/webhooks/event'
url_webhooks_input = f'{url}/webhooks/input'

with open(VONAGE_APPLICATION_PRIVATE_KEY_PATH) as f:
    private_key = f.read()

vonage_client = vonage.Vonage(
    vonage.Auth(
        application_id=VONAGE_APPLICATION_ID,
        private_key=private_key,
    ),
)

app = flask.Flask(__name__)

@app.route('/ncco/index', methods=['POST'])
def route_ncco_index():
    request_body = flask.request.get_data()
    app.logger.debug(request_body)
    data = json.loads(request_body)
    parent_uuid = data['uuid']
    to_number = DESTINATION_NUMBERS[0]
    create_call(to_number, ncco_input(parent_uuid, 1), parent_uuid, 1)
    return dump_ncco(ncco_introduction(parent_uuid))

@app.route('/webhooks/default', methods=['POST'])
def route_webhooks_default():
    request_body = flask.request.get_data()
    app.logger.debug(request_body)
    return ''

@app.route('/webhooks/event/<parent_uuid>/<int:trial>', methods=['POST'])
def route_webhooks_event(parent_uuid, trial):
    request_body = flask.request.get_data()
    app.logger.debug(request_body)
    data = json.loads(request_body)
    if data.get('status') is None:
        return ''
    next_trial = trial + 1
    match data['status']:
        case 'completed':
            if next_trial > MAX_LOOPS * len(DESTINATION_NUMBERS):
                transfer_call(parent_uuid, ncco_failed())
            elif remove_token(parent_uuid):
                transfer_call(parent_uuid, ncco_succeeded())
            else:
                index = next_trial % len(DESTINATION_NUMBERS) - 1
                to_number = DESTINATION_NUMBERS[index]
                transfer_call(parent_uuid, ncco_introduction(parent_uuid))
                create_call(to_number, ncco_input(parent_uuid, next_trial), parent_uuid, next_trial)
            return ''
        case _:
            return ''

@app.route('/webhooks/input/<parent_uuid>/<int:trial>', methods=['POST'])
def route_webhooks_input(parent_uuid, trial):
    request_body = flask.request.get_data()
    app.logger.debug(request_body)
    data = json.loads(request_body)
    if retrieve_call(parent_uuid).status == 'completed':
        return dump_ncco(ncco_disconnected())
    elif data['dtmf']['digits'] == '1':
        create_token(parent_uuid)
        return dump_ncco(ncco_connecting(parent_uuid))
    else:
        return dump_ncco(ncco_input(parent_uuid, trial))

def ncco_introduction(uuid):
    return [
        Talk(
            text='複数人への通話転送を開始します',
            language='ja-JP',
        ),
        Conversation(
            name=f'{VONAGE_CONVERSATION}.{uuid}',
        ),
    ]

def ncco_succeeded():
    return [
        Talk(
            text='通話を終了します',
            language='ja-JP',
        ),
    ]

def ncco_failed():
    return [
        Talk(
            text='規定の回数、呼び出しましたので、通話を終了します',
            language='ja-JP',
        ),
    ]

def ncco_disconnected():
    return [
        Talk(
            text='発信者が通話を切断したため、通話を終了します',
            language='ja-JP',
        ),
    ]

def ncco_input(uuid, trial):
    return [
        Talk(
            text='通話するには1を入力してください',
            language='ja-JP',
            bargeIn=True,
        ),
        Input(
            type=['dtmf'],
            dtmf=Dtmf(
                maxDigits=1,
                submitOnHash=True,
                timeOut=10,
            ),
            eventUrl=[f'{url_webhooks_input}/{uuid}/{trial}'],
        ),
    ]

def ncco_connecting(uuid):
    return [
        Talk(
            text='接続しますので少々お待ちください',
            language='ja-JP',
        ),
        Conversation(
            name=f'{VONAGE_CONVERSATION}.{uuid}',
        ),
    ]

def dump_ncco(ncco):
    return list(map(lambda x: x.model_dump(), ncco))

def retrieve_call(uuid):
    response = vonage_client.voice.get_call(uuid)
    return response

def create_call(to, ncco, uuid, trial):
    if retrieve_call(uuid).status == 'completed':
        return
    call = CreateCallRequest(
        to=[{
            'type': 'phone',
            'number': to,
        }],
        from_={
            'type': 'phone',
            'number': VONAGE_NUMBER,
        },
        ncco=ncco,
        event_url=[f'{url_webhooks_event}/{uuid}/{trial}'],
    )
    response = vonage_client.voice.create_call(call)
    return response

def transfer_call(uuid, ncco):
    if retrieve_call(uuid).status == 'completed':
        return
    vonage_client.voice.transfer_call_ncco(
        uuid=uuid,
        ncco=ncco,
    )

def create_token(uuid):
    f = open(f'.token.{uuid}', 'w')
    f.close()

def remove_token(uuid):
    result = False
    try:
        os.remove(f'.token.{uuid}')
        result = True
    except Exception:
        pass
    return result

if __name__ == '__main__':
    app.run(host=LOCAL_HOSTNAME, port=LOCAL_PORT, debug=APP_DEBUG)
  • VONAGE_NUMBER の部分は、取得された Vonage 電話番号に置き換えます。この番号は、通話が転送された際、着信側の電話機に発信元番号として表示されるものです。
  • DESTINATION_NUMBER_1 および DESTINATION_NUMBER_2 の部分は、着信側の電話番号に置き換えます。ここで指定された番号に対して、転送通話を発信します。
    • 電話機が 2 台しかない場合、DESTINATION_NUMBER_1 と DESTINATION_NUMBER_2 には、同じ電話番号を入れても構いません。
    • この配列の要素数を増やすことで、転送先の電話番号を増やすことができます。
  • 上記の電話番号については、MSISDN 形式で指定します。日本の番号であれば、先頭の 0 を除去し、代わりに国番号である 81 を加えます。たとえば、090AAAABBBB であれば、MSISDN 形式は 8190AAAABBBB となります。
  • VONAGE_APPLICATION_ID および VONAGE_APPLICATION_PRIVATE_KEY_PATH の部分については後述します。

Python 依存パッケージのインストール

今回は、Python 用の Vonage Server SDK と、ウェブアプリケーションフレームワークの Flask をインストールします。

mkdir site-packages
pip install flask -t site-packages 
pip install vonage -t site-packages

ngrok による URL の公開

上記のスクリプトに Vonage からアクセスできるよう、ngrok を用いて URL を与えます。

ngrok http 3000

実行した際、「Forwarding」の項目に表示される URL が、ngrok によって外部に公開された URL なので、これをメモしておきます。またこの後は、Control-C などは入力せず、新しいターミナル画面を開きます。

Vonage CLI によるアプリケーションの構成

  • Vonage CLI をインストールします。
npm install -g @vonage/cli
  • Vonage CLI を用いて、Vonage アプリケーションを新たに作成します。
    • URL のドメイン部分(NGROK_HOSTNAME)は、先ほどコピーしておいたものを指定します。
      • NGROK_HOSTNAME は、先ほどの Python スクリプト中にもあるため、ここでも合わせて指定します。
    • 実行した際に表示される「Application ID」および「Private Key File」は、後の手順で必要となるため、コピーしておいてください。
vonage apps:create "Vonage Tutorial" \
--voice_answer_url="https://NGROK_HOSTNAME/ncco/index" \
--voice_event_url="https://NGROK_HOSTNAME/webhooks/default"
  • Vonage CLI を用いて、作成した Vonage アプリケーションに Vonage 電話番号(MSISDN 形式)を割り当てます。
vonage apps:link (上記でコピーしたApplication ID) --number="(Vonage電話番号)"
  • 最初に作成した Python スクリプトのうち、以下の部分を置き換えます。
    • VONAGE_APPLICATION_ID: (上記でコピーしたApplication ID)
    • VONAGE_APPLICATION_PRIVATE_KEY_PATH: (上記でコピーしたPrivate Key File)

Python スクリプトの実行

作成した Python スクリプトを実行します。

PYTHONPATH="${PYTHONPATH}:./site-packages" python rotate-inbound-calls.py

すると、以下の 4 つの URL で HTTP リクエストを待機します(そして、すでに起動済みの ngrok により、外部からのリクエストを受け付けます)。この後は、Control-C などは入力せず、ターミナルを開いたままにします。

通話の実行

  • 発信側の電話機から、Vonage 電話番号に対して通話を発信し、以下のように動作すれば成功です。
    • 発信側の電話機で「複数人への通話転送を開始します」という音声が流れ、続いて待ち受け音声が始まる。
    • DESTINATION_NUMBER_1 に指定した電話番号に、その Vonage 電話番号から着信する。
    • DESTINATION_NUMBER_1 が応答し、キー 1 を入力すると、発信側の電話との間で通話が開始される。
    • DESTINATION_NUMBER_1 がキーを入力せず通話を切断すると、次は DESTINATION_NUMBER_2 に Vonage 電話番号から着信する。
    • 以下、DESTINATION_NUMBER_1 と DESTINATION_NUMBER_2 に対して、順番に着信する。この着信のループは、MAX_LOOPS に指定された値だけ繰り返される。
  • 通話転送の実行の流れ以外にも、rotate-inbound-calls.py を実行しているターミナルに流れる JSON 文字列にも注目してください。これは、Vonage による Event webhook のリクエストで、通話において何らかのイベント(たとえば、着信側が応答した、など)に反応して HTTP リクエストを送信するものです。今回、転送先への発信も、この Event webhook からの情報を利用しています。

まとめ

  • Vonage を利用すると、Vonage 電話番号に対して着信があった時に、複数の外部の電話番号に対して通話を順番に発信し、応答した通話を着信通話と接続することができます。
  • 上記を実現するには、複数人での電話会議を実現する NCCO の Conversation action と、Voice API による通話発信を併用します。
    • Vonage 電話番号に着信した際、発信側を Conversation に入室させます。
    • 次に、Voice API で外部の電話番号に対して通話を発信します。
    • 通話発信に応答があったら、その着信側を、発信側が待つ Conversation に入室させ、通話を接続します。
  • Vonage での通話中、何かイベントが発生すると、Vonage から HTTP リクエストが実行されます。これを Event webhooks と呼びます。

参考ウェブサイト

KWCPLUS

Discussion