OpenAI との接続ガイド

お申し込みと、接続に使う情報のやりとり

お申し込み後に、クラコールから接続に使う情報をお渡しします。お客様からは、OpenAI 側の情報をお知らせください。

お問い合わせの本文に「BYOC 接続」とお書きのうえ、お送りください。BYOC では、着信・発信とも、INVITE の宛先（Request-URI）・To・From の電話番号を +81 から始まる E.164 形式にしてください（例：03-1234-5678 → +81312345678）。

Webhook を登録する

OpenAI は、着信があると realtime.call.incoming というイベントを Webhook で知らせます。OpenAI の settings > Project > Webhooks（platform.openai.com）を開き、Create（作成）のボタンで Webhook を登録します。入れるのは、次の3つです。

• 名前（ご自身が見分けるためのものです）
• お客様のサーバーの公開 URL（手順3で動かすサーバーのアドレスです）
• 受け取るイベントの種類：realtime.call.incoming

Webhook を作ると、signing secret（署名用のシークレット）が表示されます。この値はあとからは見られないので、必ず保存してください。手順3で、この値を環境変数 OPENAI_WEBHOOK_SECRET に入れます。なくしたとき、うっかり公開してしまったときは、rotating the signing secret（シークレットの再発行）で新しいものを作れます。

• OpenAI から届く知らせは、本文の type が realtime.call.incoming のイベントで、通話の ID は data の中の call_id です。ヘッダーには webhook-id・webhook-timestamp・webhook-signature が付きます。
• 知らせを受けたお客様のサーバーは、すぐに 2xx（成功）の応答を返します。2xx でないときや応答が遅いときは、OpenAI は同じ知らせを再送します（最長 72 時間）。
• 同じ知らせが重複して届くことがあります。webhook-id を重複を見分ける印（idempotency key）に使えます。

着信を受け入れるサーバーを動かす（webhook.py）

次の webhook.py は、Webhook を受けて署名を確かめ、realtime.call.incoming なら accept で着信を受け入れます。Python の標準ライブラリだけで動くので、追加のインストールは要りません（Python 3.9 以上）。

webhook.py

import base64
import hashlib
import hmac
import json
import os
import time
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

OPENAI_BASE = os.environ.get("OPENAI_BASE", "https://api.openai.com")
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
WEBHOOK_SECRET = os.environ["OPENAI_WEBHOOK_SECRET"]
PORT = int(os.environ.get("PORT", "8000"))
TOLERANCE_SECONDS = 300

# シークレットが whsec_ で始まる場合は、その後ろを base64 として復号します（起動時に1回だけ）
if WEBHOOK_SECRET.startswith("whsec_"):
    try:
        KEY = base64.b64decode(WEBHOOK_SECRET[6:], validate=True)
    except ValueError:
        raise SystemExit("OPENAI_WEBHOOK_SECRET が正しくありません。OpenAI の画面に表示された signing secret をそのまま設定してください。")
else:
    KEY = WEBHOOK_SECRET.encode()

# 着信を受け入れるときに送る内容（値は OpenAI 公式の例です。お客様の用途に合わせて変えてください）
ACCEPT_BODY = {
    "type": "realtime",
    "model": "gpt-realtime-2.1",
    "instructions": "You are Alex, a friendly concierge for Example Corp.",
}


def signature_is_valid(headers, body):
    # Standard Webhooks の署名検証：msg_id.timestamp.payload を HMAC-SHA256 で署名し、base64 にしたものを比べます
    msg_id = headers.get("webhook-id", "")
    timestamp = headers.get("webhook-timestamp", "")
    signatures = headers.get("webhook-signature", "")
    try:
        if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
            return False
    except ValueError:
        return False
    signed = msg_id.encode() + b"." + timestamp.encode() + b"." + body
    expected = base64.b64encode(hmac.new(KEY, signed, hashlib.sha256).digest()).decode()
    for item in signatures.split():
        version, _, value = item.partition(",")
        if version == "v1" and hmac.compare_digest(value.encode(), expected.encode()):
            return True
    return False


def accept_call(call_id):
    accept_endpoint = f"{OPENAI_BASE}/v1/realtime/calls/{call_id}/accept"
    request = urllib.request.Request(
        accept_endpoint,
        data=json.dumps(ACCEPT_BODY).encode(),
        headers={
            "Authorization": f"Bearer {OPENAI_API_KEY}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        return response.status


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
        if not signature_is_valid(self.headers, body):
            self.send_response(400)
            self.end_headers()
            return
        event = json.loads(body)
        if event.get("type") == "realtime.call.incoming":
            call_id = event["data"]["call_id"]
            try:
                accept_call(call_id)
                print("accepted", call_id)
            except urllib.error.HTTPError as error:
                if error.code == 429 or error.code >= 500:
                    # 一時的なエラー：502 を返し、OpenAI に再送してもらいます
                    print("accept failed", call_id, error.code)
                    self.send_response(502)
                    self.end_headers()
                    return
                # それ以外で断られた場合（恒久的）：ログに残し、Webhook には 200 を返します（再送を避けるため）
                print("accept refused", call_id, error.code)
            except OSError as error:
                # ネットワークの失敗・タイムアウト：502 を返し、OpenAI に再送してもらいます
                print("accept failed", call_id, error)
                self.send_response(502)
                self.end_headers()
                return
        self.send_response(200)
        self.end_headers()


ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()

ターミナル

export OPENAI_API_KEY="（OpenAI の API キー）"
export OPENAI_WEBHOOK_SECRET="（Webhook の署名用のシークレット）"
python3 webhook.py

• OPENAI_API_KEY：OpenAI の API キーです。accept の呼び出しの Authorization: Bearer に使います。コードには書かず、環境変数で渡します。
• OPENAI_WEBHOOK_SECRET：手順2で保存した signing secret です。コードは、値が whsec_ で始まる場合は、その後ろを base64 として復号して鍵にし、そうでない場合は、値をそのまま鍵にします。
• 署名の検証は、Standard Webhooks の仕様のとおり、webhook-id・webhook-timestamp・本文を . でつないだものを HMAC-SHA256 で署名し、webhook-signature の v1, のあとの値と比べます。時刻が許容差（例では5分）を超えるものと、署名が合わないものは、400 で断ります。
• accept の本文の type は realtime です。model と instructions の値は、OpenAI の公式の例のものです。お客様の用途に合わせて変えてください。
• accept が成功すると、OpenAI は 200 OK を返し、通話の準備が始まります。お客様のサーバーも 200 を返します。
• accept が一時的なエラー（429・5xx）や通信エラーの場合は 502 を返して再送してもらい、それ以外で断られた場合は 200 を返します。
• 例の OPENAI_BASE は、試験用に宛先を切り替えるためのもので、通常は設定しません（標準は https://api.openai.com です）。

SIP の宛先をクラコールにお知らせする

クラコールの SBC が着信を送る先は、OpenAI の SIP の宛先（SIP URI）です。次の形で、$PROJECT_ID をお客様のプロジェクト ID に置き換えたものをクラコールへお知らせください。

OpenAI の SIP の宛先（クラコールの SBC が着信を送る先）

sip:[email protected];transport=tls

• $PROJECT_ID：OpenAI のプロジェクト ID（proj_ の接頭辞が付いた文字列）に置き換えます。settings > Project > General で確認できます。
• ポートは TCP/TLS の 5061 です。TLS を使うため、暗号化の有料オプションが必要です。
• 着信の電話番号は、+81 から始まる E.164 形式で届きます。

クラコールとの接続では、Realtime API の SIP の音声の形式は G.711 μ-law（PCMU）です。

つながることを確認する

• 着信：クラコールの電話番号へ、携帯電話などから電話をかけ、お客様のサーバー（webhook.py）に、OpenAI から realtime.call.incoming が届くことを確認します（例のコードは、受け入れると accepted と通話の ID を表示します）。
• 着信を受け入れたあと、相手の声が聞こえることを確認します。