利用規約運営会社
EnglishMiiTel へログイン

チュートリアル - FastAPI Outgoing Webhook を受信する

Suggest Edits

基本設定

基本的な利用方法は、以下のサポートページを確認してください。

実装ガイド

Python の FastAPI フレームワーク、および、Serverless Framework を使った例を使用して、応対履歴の Outgoing Webhook を受信する API の実装方法を紹介します。

事前準備

この手順を実施するには、以下の事前準備が必要です。

  • Python3 をインストールする
  • Serverless Framework V3 をインストールする
  • AWS アカウント、および、AWS プロファイルを用意する

FastAPI 開発を準備する

Fast API および Mangum をインストールする

pip install uvicorn fastapi mangum

app フォルダと __init__.py を作成する

フォルダと空ファイルを作成します。

以下は Shell のコマンドを使用しています。(別の方法で作成した場合でも利用できます)

mkdir app
touch app/__init__.py

app/main.py を作成する

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from mangum import Mangum

app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/hello")
def read_root():
    return {"Hello": "World"}


handler = Mangum(app, "off")

アプリを起動する

uvicorn app.main:app

起動後、http://127.0.0.1:8000/hello にアクセスすると、アプリ画面が表示されます。

{"Hello":"World"} と表示されたら設定完了です。

Serverless Framework を準備する

Outgoing Webhook を利用するには、インターネット上への API 公開が必要です。

ここでは、Serverless Framework を使って、 AWS 環境上に Outgoing Webhook を受信する API の公開手順を例に説明します。

requirements.txt を作成する

aws-lambda-powertools を使います。AWS Lambda にログを出力する時に便利なツールです。

uvicorn
fastapi
mangum
aws-lambda-powertools

serverless.yaml を作成する

service: ow-fastapi-sample

plugins:
  - serverless-python-requirements
  - serverless-add-api-key

custom:
  pythonRequirements:
    dockerizePip: true
  apiKeys:
    - name: OwFastApiSampleAPIKey
      value: <API_KEY> # 適宜２０文字以上の英数字で置き換えてください。

provider:
  name: aws
  runtime: python3.10
  region: ap-northeast-1
  stage: ${opt:stage, 'dev'}
  timeout: 30


functions:
  application:
    handler: app.main.handler
    events:
      - http:
          path: /{path+}
          method: ANY
          cors: true
          private: true

package.json と npm install を作成する

{
  "devDependencies": {
    "serverless-add-api-key": "^4.2.1",
    "serverless-python-requirements": "^6.0.1",
    "serverless": "^3.36.0"
  },
  "dependencies": {}
}

npm install --include=dev

デプロイする

以下のコマンドで、Serverless Framework を使って、AWS 上に API をデプロイします。

npx sls deploy --aws-profile <your-aws-profile>

以下のようなメッセージが表示されることを確認します。

✔ Service deployed to stack ow-fastapi-sample-dev (44s)

endpoint: ANY - https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/dev/{path+}
functions:
  application: ow-fastapi-sample-dev-application (200 kB)

上記メッセージの endpoint がこの API のエンドポイントとなり、後に Outgoing Webhook の送信先として設定します。

Postman で APIKey を設定後、リクエストの送信を試します。メッセージを取得できれば成功です。

Outgoing Webhook のリクエストを受け付ける API を開発する

ここまでできたら、次は Outgoing Webhook のリクエストを受け付ける API を開発します。

Challenge-Response 対応を実装する

Outgoing Webhook 用の API には、以下の処理を作成する必要があります。

  • 実際に動作するための処理
  • Challenge-Response 方式の確認リクエストに対応したレスポンスを返す処理

Challenge-Response 方式とは、Outgoing Webhook が意図しない API に対してリクエストを送信し続けないことを確認するために、決まったレスポンスを返すか確認する方式のことです。

まずは、 新しくapp/schema.pyファイルを作成して、応対履歴の Outgoing Webhook にリクエストのスキーマを定義します。

※ 以下の定義は 2023 年 11 月 17 日現在のものです。最新の定義は「Outgoing Webhook 応対履歴」から Payload (Request Body) のサンプルを元に、型定義を生成してください。

※ 一部項目は、nullとなるケースや、 項目そのものがない場合もあります。 下記の JSON スキーマを参考にしてください。

from __future__ import annotations

from typing import Any, List, Optional

from pydantic import BaseModel, Extra


class RawItem(BaseModel, extra=Extra.allow):
    order: int = None
    phrase: str = None
    phrase_nofiller: str = None
    start_at: float = None
    end_at: float = None
    filler_num: int = None


class Phrase(BaseModel, extra=Extra.allow):
    participant_id: str = None
    raw: List[RawItem] = None


class Detail(BaseModel, extra=Extra.allow):
    id: str = None
    phrases: List[Phrase] = []


class Call(BaseModel, extra=Extra.allow):
    id: str = None
    tenant_code: str = None
    details: List[Detail] = []


class OutgoingWebhookPayload(BaseModel, extra=Extra.allow):
    call: Call = None
    params_key: str = None
    challenge: str = None

次に、上で定義したスキーマを元にリクエストを受け付けるエンドポイントとして、app/main.pyを以下のように修正します。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse
from mangum import Mangum

from app.schema import OutgoingWebhookPayload

app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/hello")
def read_root():
    return {"Hello": "World"}


@app.post("/outgoing-webhook")
def outgoing_webhook(data: OutgoingWebhookPayload):
    if data.challenge:
        # ヘッダーを {'Content-Type':'text/plain'} とした 200 レスポンスで 'hex_token' 値を返す
        return PlainTextResponse(data.challenge, status_code=200)
    return {"Hello": "World"}


handler = Mangum(app, "off")

これで、 Outgoing Webhook の Challenge リクエストに対してレスポンスを返せるようになりました。

再度 Postman でダミーの Challenge リクエストを送信し、送信した Challenge リクエストの Payload (Request Body) が以下のように返ってきたら成功です。

文字起こし結果のログを出力する機能を実装する

次に、実際の Outgoing Webhook の処理を受け付けた場合の処理を記述します。

ここでは、シンプルに音声の文字起こし結果をログに出力する処理を記述します。

以下のように、app/main.py を編集します。

import logging

from aws_lambda_powertools import Logger
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import PlainTextResponse
from mangum import Mangum

from app.schema import OutgoingWebhookPayload

logger = Logger(service="mangum")
logger.setLevel(logging.INFO)

app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/hello")
def read_root():
    return {"Hello": "World"}


@app.post("/outgoing-webhook")
def outgoing_webhook(data: OutgoingWebhookPayload):
    logger.info(data)
    if data.challenge:
        # ヘッダーを {'Content-Type':'text/plain'} とした 200 レスポンスで 'hex_token' 値を返す
        return PlainTextResponse(data.challenge, status_code=200)

    logger.info("---------")
    logger.info(data.call.id)
    # 文字起こしデータをログに出力する
    for call_detail in data.call.details:
        for phrase in call_detail.phrases:
            for raw_item in phrase.raw:
                logger.info(raw_item.phrase)

    return {"Hello": "World"}


handler = Mangum(app, "off")

Outgoing Webhook を設定する

最後に、MiiTel Admin で Outgoing Webhook の設定します。

設定の詳細手順は、「Outgoing Webhook 応対履歴」を確認してください。以下は、設定内容です。

  • 名前: 任意の名前を設定する
  • URL: https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/dev/outgoing-webhook ( xxxxxxxxxx は Serverless Framework のデプロイ時に出力された値) を入力する
  • 追加ヘッダー: {"X-API-KEY": "<serverless.yml の apiKeys に設定した値>"}を入力する

設定後、MiiTel で通話テストします。テスト時、以下のようなログが CloudWatch に出力されれば成功です。

リソースを削除する

リソースが不要になった場合は、 Serverless Framework の機能で削除します。

npx sls remove --aws-profile <your-aws-profile>

終わりに

今回は、ログを吐き出すだけのベーシックな API でしたが、ここから応用して、MiiTel の文字起こし結果や解析結果などを様々なシステムに連携できます。

もし、うまく動作しない場合はフォーラム からご質問ください。

利用中のシステムに連携することで、業務をより円滑に進めることができます。更に、データを蓄積することで、より高度な分析や新たな AI 開発などにも活用できます。

ぜひ、様々なチャレンジをしてみてください。

Updated 5 months ago