チュートリアル - FastAPI Outgoing Webhook を受信する
基本設定
基本的な利用方法は、以下のサポートページを確認してください。
実装ガイド
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
を作成する
__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> # 適宜20文字以上の英数字で置き換えてください。
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 7 months ago