はじめに
先日OpenAIのYoutubeを見ていたらRealtime APIのアップデートが色々流れていました。
その中で、これまではWebSocketでしか対応していなかったところから、SIP接続に対応したということで、早速マニュアルをみながらセットアップしました。実施した手順をご紹介したいと思います。
大まかな流れ
- Open AIのコンソール
- API Keyの取得
- Project IDの取得
- Webhookの設定とWebhook Secretの取得
- TwilioのSIP Trunkingの設定
- Webhookサーバの起動
- サンプルコードの微修正
- 通話開始
セットアップ
前提
- Python 3.8以上
- Open AIのアカウント
- OpenAI APIキー
- OpenAI Webhook Secret
- Twilioアカウント
- SIP Trunkで使用する電話番号は購入済みであること
- ngrok
OpenAIコンソールでの設定
-
まずOpen AIの設定画面にアクセスしてください
-
API Keysのメニューに遷移してCreate new secret keyボタンからシークレットキーを作成してください。作成したSecret keyは必ずどこかにメモしておいてください
-
次にWebhookメニュ−からWebhookのエンドポイントを作成します。
-
Webhookのメニュー画面で、Createボタンを押してください。
-
エンドポイント情報を入力します。
- Name : 任意の名称でOK
- URL : Webhookサーバーのドメインを入れてください。後ほどNgrokを起動したときに表示されるドメインを指定するので、現時点では https://exapmles.jp.ngrok.io など適当なものを入れておいて良いです。
- Event Type : realtime.call.incomingを指定してください。
-
WebhookのSecretが作成されるので、ここのキーを何処かにメモしておいてください
-
最後にProject IDを取得します。Project -> Generalのメニューに遷移して表示されたProject IDを何処かにメモしておいてください。
ここまでの作業で下記の3点の情報が取得できていればOKです。
- Project ID
- API Keyのシークレット
- Webhookのシークレット
Twilio SIP Trunkingの設定
-
Twilioコンソールにログイン
-
Elastic SIP TrunkingのメニューからTrunkに移動し新規にTrunkを生成します。
-
FRIENDRY NAMEに任意の名前を入れて、Createボタンを押します
-
Trunkが作成されますので、Generalの設定で下記の項目をEnabledにしてSaveボタンをおしてください。
- Secure Trunking
- Call Transfer(SIP REFER)
-
Originationの設定をおこないます。メニューでOriginationを選択してください
-
OpenAIが提供しているエンドポイントを設定します。Add new Origination URIのボタンをおしてください。
-
Open AIのDocsに指定のあるSIP URIを設定してください。マニュアルには
sip:$PROJECT_ID@sip.api.openai.com;transport=tlsと記載があるので、PROJECT_IDの部分を先程書き留めた、OPEN AIのProjectIDに置き換えて登録してください。その後、Addボタンを押してください。
-
Numbersメニューに遷移してください。Numbers画面が開いたら、Add a numberボタンを押します。既に利用可能な電話番号があれば、Add an Exsisting Numberを選択してください。なければ、Buy a numberを選択して新規に電話番号を購入してください。今回は購入済みである前提で進めます。
-
表示された電話番号の中から今回利用する番号のチェックボックスをオンにして、Add Selectedボタンを押してください
-
登録が完了すると下記の画面のように一覧表示されます
ここまでで音声が開通しているかを確認したいところですが、登録した番号に電話をしてもOpenAI側の認証が出来ていないので通話がエラーになります。このあとのWebhookサーバで認証を行うので、Webhookサーバの設定完了後に通話検証を行います。
Webhookサーバの構築
WebhookサーバについてはOpen AIのDocsにサンプルコードがあるのでそちらを利用します。
そのまま利用したら少しエラーがでていたので、修正しました。完全なコードサンプルなどはこちらにおいておきました。
サンプルコード
from flask import Flask, request, Response, jsonify, make_response
from openai import OpenAI, InvalidWebhookSignatureError
import asyncio
import json
import os
import requests
import time
import threading
import websockets
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
OPENAI_WEBHOOK_SECRET = os.getenv('OPENAI_WEBHOOK_SECRET')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
app = Flask(__name__)
client = OpenAI(
webhook_secret=OPENAI_WEBHOOK_SECRET
)
AUTH_HEADER = {
"Authorization": "Bearer " + OPENAI_API_KEY
}
call_accept = {
"type": "realtime",
"instructions": "You are a support agent for Japanese. Please speak Japanese only.",
"model": "gpt-4o-realtime-preview-2024-12-17",
}
response_create = {
"type": "response.create",
"response": {
"instructions": (
"もしもし、本日のご要件はなんですか？って最初に話してください。"
)
},
}
async def websocket_task(call_id):
try:
async with websockets.connect(
"wss://api.openai.com/v1/realtime?call_id=" + call_id,
extra_headers=AUTH_HEADER,
) as websocket:
await websocket.send(json.dumps(response_create))
while True:
response = await websocket.recv()
print(f"Received from WebSocket: {response}")
except Exception as e:
print(f"WebSocket error: {e}")
@app.route("/", methods=["POST"])
def webhook():
try:
event = client.webhooks.unwrap(request.data, request.headers)
if event.type == "realtime.call.incoming":
requests.post(
"https://api.openai.com/v1/realtime/calls/"
+ event.data.call_id
+ "/accept",
headers={**AUTH_HEADER, "Content-Type": "application/json"},
json=call_accept,
)
threading.Thread(
target=lambda: asyncio.run(
websocket_task(event.data.call_id)
),
daemon=True,
).start()
return Response(status=200)
except InvalidWebhookSignatureError as e:
print("Invalid signature", e)
return Response("Invalid signature", status=400)
if __name__ == "__main__":
app.run(port=8000)
-
リポジトリのクローン
git clone https://github.com/MitsuharuNakamura/realtimeapi-sip-webhook-server.git cd realtimeapi-sip-webhook-server -
仮想環境の作成とアクティベート
# 仮想環境を作成 python3 -m venv venv # 仮想環境をアクティベート（macOS/Linux） source venv/bin/activate -
依存パッケージのインストール
pip install -r requirements.txt -
環境変数の設定です。.evn.exampleを.envにリネームして、それぞれ値を設定してください。
OPENAI_WEBHOOK_SECRET=your_webhook_secret_here OPENAI_API_KEY=your_api_key_here -
サーバーの起動(サーバーはデフォルトでポート8000で起動します。)
python main_with_env.py -
下記のようにURLが表示されていれば成功
* Serving Flask app 'main_with_env' * Debug mode: off WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead. * Running on http://127.0.0.1:8000 Press CTRL+C to quit -
別のターミナルを起動しNgrokを起動します
ngrok http 8000 -
下記のように表示されればOKです
ngrok 🐋 Create instant endpoints for local containers within Docker Desktop → https://ngrok.com/r/docker Session Status online Account Twilio Developer Network (Plan: Enterprise) Version 3.23.3 Region Japan (jp) Latency 169ms Web Interface http://127.0.0.1:4040 Forwarding https://xxxx.jp.ngrok.io -> http://localhost:8000 Connections ttl opn rt1 rt5 p50 p90 0 0 0.00 0.00 0.00 0.00 -
OpenAIの設定画面をブラウザで開き、Webhooksの設定画面に遷移します。先程作ったWebhookの設定を編集します。
-
Ngorkで表示されているURLをこちらに指定してSaveしたら完了です。
通話検証
- 自分の電話機から、TwilioのSIP Trunkで設定した電話番号に架電します
- **「もしもし、本日のご要件はなんですか？」**と聞こえたら成功です
最後に
今回はとにかく急いでつなげたかったので、接続がうまくできるのかだけを検証しました。
Webhookサーバを使ってもっと色々と指示を出して制御することができそうです。
特にSIP Referを使った通話の転送などはこのあと検証したいと思いますが、今日はここまでに。
参考情報
Open AI Realtime APIのマニュアル
Twilio SIP Trunkingのマニュアル
サンプルコード