ConversationRelayとPythonを使用してGoogle GeminiとTwilio Voiceを連携

Time to read:

April 20, 2026

執筆者

Rishab Kumar

Twilion

レビュー担当者

Dylan Frankcom

Twilion

AI生成による要約

FastAPIサーバーを構築し、ConversationRelayを活用してGoogle GeminiとTwilio Voiceを連携させます。
安全なコミュニケーションを行うために、必要な環境変数と依存関係を設定します。
Twilioに発信し、AIとのリアルタイムな会話を通じて連携をテストします。

この要約は、AIにより生成されました。

ConversationRelayとPythonを使用してGoogle GeminiとTwilio Voiceを連携

電話越しに、AIモデルとリアルタイムで自然な会話ができる場面を想像してみてください。Twilio Conversation Relayを使用すれば、そんな体験を実現できます。高速なイベント駆動型WebSocketを介して、音声通話と各種LLMを連携させ、シームレスなコミュニケーションを実現します。

この記事では、ConversationRelayを使用してTwilio VoiceでGoogle Geminiを設定する方法について説明します。最終的には、シンプルなPythonサーバーが稼働し、Twilioの電話番号にかけるだけでGeminiと会話できるようになります。ここから、さらにインタラクティブで高機能な音声アプリケーションへと拡張していくことができます。

前提条件

このチュートリアルを進めるには、以下が必要です。

Python 3.10以降がインストールされていること
Twilioの電話番号（Twilioのサインアップはこちら）
任意のIDE（Visual Studio Codeなど）
Ngrokなどのトンネリングサービス
Google AI StudioアカウントとAPIキー
Twilioへの発信に使用する電話

プロジェクトのセットアップ

まず、プロジェクトの新しいディレクトリを作成します。

mkdir twilio-cr-gemini-python
cd twilio-cr-gemini-python

次に、仮想環境を作成し、有効化します。

python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

その後、pipを使用して必要な依存関係をインストールします。

pip install google-genai python-dotenv uvicorn fastapi websockets

FastAPIをフレームワークとして使用し、WebSocketとTwilioのルートの両方を備えたサーバーをすばやく構築します。

このクイックスタートのコードはすべて、GitHub上で公開されています。

環境変数の設定

Google GeminiのAPIキーは、安全な場所に保存してください。プロジェクトフォルダで.envファイルを作成し、以下を追加します。

GOOGLE_API_KEY="YOUR_GEMINI_API_KEY"
NGROK_URL="your-ngrok-subdomain.ngrok-free.app"

プレースホルダー部分を、実際のAPIキーとNgrokの転送URLに差し替えます（手順については、後ほど説明します）。

サーバーの作成

次に、「main.py」という名前で新しいPythonファイルを作成します。このファイルに、アプリケーションのメインのロジックを記述します。

1. 必要なライブラリをインポートし、環境をセットアップする

サーバー構築の最初のステップとして、必要なライブラリをインポートします。これらのライブラリは、サーバーの構築、WebSocket接続の処理、Google Geminiとの連携、環境変数の読み込みに使用します。

import os
import json
import uvicorn
from google import genai
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import Response
from dotenv import load_dotenv

FastAPI: サーバーの構築に使用するメインのWebフレームワーク。簡単にルートを簡単に定義して、HTTPリクエストを処理できるほか、サーバーとTwilio ConversationRelay間のWebSocket通信にも対応しています。
uvicorn: FastAPIアプリケーションを実行する高速ASGIサーバー。
genai: GoogleのGen AI SDK。GoogleのGeminiモデルと連携して応答を生成するために使用します。
dotenv: .envファイルから環境変数（Google APIキーやNgrok URLなど）を読み込むためのモジュール。

ライブラリをインポートした後、以下の行を追加して.envファイルから環境変数を読み込みます。

load_dotenv()

これで、APIキーなどの機密情報がコードにハードコード化されるのを防ぐことができます。

2. 定数と設定 :

コード全体で共有する定数を定義し、アプリケーションの動作をカスタマイズします。

PORT = int(os.getenv("PORT", "8080"))
DOMAIN = os.getenv("NGROK_URL")
if not DOMAIN:
    raise ValueError("NGROK_URL environment variable not set.")
WS_URL = f"wss://{DOMAIN}/ws"
WELCOME_GREETING = "Hi! I am a voice assistant powered by Twilio and Google Gemini. Ask me anything!"
SYSTEM_PROMPT = """You are a helpful and friendly voice assistant. This conversation is happening over a phone call, so your responses will be spoken aloud. 
Please adhere to the following rules:
1. Provide clear, concise, and direct answers.
2. Spell out all numbers (e.g., say 'one thousand two hundred' instead of 1200).
3. Do not use any special characters like asterisks, bullet points, or emojis.
4. Keep the conversation natural and engaging."""
sessions = {}

PORT: サーバーが接続を待ち受けるポート番号（環境変数から取得するか、デフォルト値の8080を使用）
DOMAIN: TwilioとのWebSocket通信の確立に使用されるNgrok URL。このコードは設定値を即座に検証し、未設定の場合はエラーを発生させます。
WS_URL: Twilioの接続先であるエンドポイントの完全なWebSocket URL。
WELCOME_GREETING: 通話接続時に流れる挨拶メッセージ。
SYSTEM_PROMPT: Geminiモデルに動作の指示を与える特殊なメッセージ。AIの応答のトーンやフォーマット（例：数字を文字で表記する、絵文字を使用しないなど）を定義する重要な役割を果たします。今回は音声会話であるため、これらのルールを設定することで、クリアで聞き取りやすい音声応答を出力できるようになります。
sessions: 各通話のアクティブなチャットセッションオブジェクトを保持する辞書（キーとしてcallSidを使用）。注記: この実装では、正常な切断時のみクリーンアップ処理が行われます。本番環境で運用する場合は、セッションのタイムアウト処理の追加を検討してください。

3. FastAPIとGeminiクライアントの初期化

FastAPIアプリケーションを初期化し、.envファイルに保存されているAPIキーを使用してGeminiクライアントを設定します。

GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
if not GOOGLE_API_KEY:
    raise ValueError("GOOGLE_API_KEY environment variable not set.")
client = genai.Client(api_key=GOOGLE_API_KEY)
app = FastAPI()

GOOGLE_API_KEY: 環境変数からAPIキーを読み込み、処理を実行する前にAPIキーが正しく設定されているか検証するコード。
client: genai.Client()を使用して、GoogleのGen AIクライアントを初期化します。これは新しいSDKのパターンです。以前のgoogle-generativeaiパッケージを使用していた場合、これは新しいgoogle-genai SDKであること注意してください。
FastAPIアプリ: HTTPリクエストとWebSocket接続の両方を処理するWebサーバーが生成されます。

4. AI応答関数

次に、Gemini APIと連携して、AIからの応答を取得する関数を作成します。

def gemini_response(chat_session, user_prompt):
    """Get a response from the Gemini API."""
    response = chat_session.send_message(user_prompt)
    return response.text

gemini_response(): Geminiのチャットセッションオブジェクトとユーザーのプロンプトを取り、メッセージを送信してテキストの応答を返す関数。チャットセッションオブジェクトが会話履歴を自動的に維持するため、Geminiはメッセージごとに文脈を把握した上で応答できます。

5. TwilioのTwiML応答

/twimlエンドポイントは、着信した音声通話をWebSocketアプリケーションへ接続するようTwilioに指示します。Twilioは、TwiML（Twilio Markup Language）を使って、音声対話の処理方法を判断します。

@app.post("/twiml")
async def twiml_endpoint():
    """Endpoint that returns TwiML for Twilio to connect to the WebSocket"""
    xml_response = f"""<?xml version="1.0" encoding="UTF-8"?>
    <Response>
    <Connect>
    <ConversationRelay url="{WS_URL}" welcomeGreeting="{WELCOME_GREETING}" ttsProvider="ElevenLabs" voice="FGY2WhTYpPnrIDTdsKH5" />
    </Connect>
    </Response>"""
    return Response(content=xml_response, media_type="text/xml")

TwiMLの応答によって、Twilioは指定されたWebSocketサーバー（WS_URL）に接続し、発信者に対してWELCOME_GREETINGの挨拶メッセージを発信します。
この例では、音声合成プロバイダーとしてElevenLabsを使用し、特定の音声ID（FGY2WhTYpPnrIDTdsKH5）を設定しています。これはサンプル音声IDです。実際に利用する際は、ご自身の音声IDに置き換えてください。このカスタマイズはオプションです。ttsProvider属性とvoice属性を完全に削除し、TwilioのデフォルトのTTSを使用するか、「ElevenLabs」を「Amazon」や「Google」に置き換えることができます。
言語、ttsProvider、音声など、サポートされている属性の詳細については、ConversationRelayドキュメントを参照してください。
注記: このエンドポイントは、Twilioリクエスト署名を検証しません。本番運用については、Twilioのリクエスト検証ドキュメントを参照してください。

6. WebSocket接続処理

/ws WebSocketエンドポイントは、Twilioとサーバー間のリアルタイム通信を処理します。Twilioからメッセージを受信し、処理し、返信します。

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket endpoint for real-time communication"""
    await websocket.accept()
    call_sid = None
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            if message["type"] == "setup":
                call_sid = message["callSid"]
                print(f"Setup for call: {call_sid}")
                sessions[call_sid] = client.chats.create(
                    model="gemini-2.5-flash",
                    config={"system_instruction": SYSTEM_PROMPT}
                )
            elif message["type"] == "prompt":
                if not call_sid or call_sid not in sessions:
                    print(f"Error: Received prompt for unknown call_sid {call_sid}")
                    continue
                user_prompt = message["voicePrompt"]
                print(f"Processing prompt: {user_prompt}")
                chat_session = sessions[call_sid]
                response_text = gemini_response(chat_session, user_prompt)
                await websocket.send_text(
                    json.dumps({
                        "type": "text",
                        "token": response_text,
                        "last": True
                    })
                )
                print(f"Sent response: {response_text}")
            elif message["type"] == "interrupt":
                print(f"Handling interruption for call {call_sid}.")
            else:
                print(f"Unknown message type received: {message['type']}")
    except WebSocketDisconnect:
        print(f"WebSocket connection closed for call {call_sid}")
        if call_sid in sessions:
            sessions.pop(call_sid)
            print(f"Cleared session for call {call_sid}")

whileループは、Twilioからの着信メッセージを待ち受けます。サーバーはさまざまなメッセージタイプを処理します。
setup: この通話用の新しいGeminiチャットセッションを初期化します。このコードは、client.chats.create()を使用し、gemini-2.5-flashモデルを設定して、configパラメータでシステムプロンプトを渡します。この処理により、会話の履歴を自動的に保持するステートフルなチャットオブジェクトが生成されます。
prompt: Twilioでテキスト化されたユーザーの音声入力を処理し、gemini_response()関数でGeminiに送信します。応答テキストはTwilioに送信され、音声に変換されます。
interrupt: 会話中の割り込みに対応します（例えば、AIの応答中に発信者が話し始めた場合など）。
Unknown message: 認識されていないメッセージタイプを記録します。
sessions: アクティブなGeminiチャットセッションは、call_sidでインデックス化されてsessions辞書に保存されます。WebSocket接続が切断されると、セッションはクリーンアップされます。

7. FastAPIサーバーの実行

最後に、Uvicornを使用してFastAPIサーバーを実行します。

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=PORT)

uvicorn.run(): サーバーが起動し、指定されたポート（デフォルトは8080）で待ち受けます。これでサーバーはHTTPリクエスト（TwiML用）とWebSocket接続（リアルタイムの会話用）の両方を処理できるようになります。

サーバーの実行

サーバーを実行するには、まずターミナルを開き、Ngrokトンネルを起動します。

ngrok http 8080

まず、ngrokの接続ソケットを開いておいてください。ngrokのURLは、Twilio Consoleと.envファイルの2か所で必要になります。

Ngrok URL（例: https://1234abcd.ngrok.app）を書き留め、ドメイン部分を.envファイルに追加します。

NGROK_URL="1234abcd.ngrok-free.app"

これで、以下のコマンドでサーバーを起動できるようになりました。

python main.py

Twilioの設定

Twilio Consoleにアクセスし、自分のTwilio電話番号を選択します。［Configure（設定）］タブの［Voice Configuration］（音声設定）で、［Webhook URL］を自分のNgrok URL（/twimlを含む）に設定します。

https://1234abcd.ngrok-free.app/twiml

連携のテスト

設定がすべて完了したら、携帯電話からTwilio電話番号に発信します。挨拶メッセージが流れると、AIへの質問を始めることができます。

このPythonアプリは、Twilioの音声機能をConversationRelayに接続するFastAPIサーバーを構築し、Google Geminiを搭載したAIアシスタントとのリアルタイムな対話を可能にします。サーバーはHTTPリクエスト（Twilioへの応答指示）とWebSocket接続（AIとのライブ通信）の両方を処理します。ConversationRelayは、着信時に音声の文字起こし、LLMとの通信、LLMの応答に基づく音声合成を一連の流れとして制御し、Geminiを搭載したAIアシスタントとのスムーズな会話を実現します。

リアルタイム音声アシスタントが動作するようになったので、TTSプロバイダーの変更、通話中にGeminiが天気予報を確認したり予約を入れたりできる機能呼び出しの追加、さまざまなGeminiモデルの検証など、さらなるカスタマイズを試すことができます。

その他のConversationRelayの事例やデモをお探しの場合は、Twilionのブログ記事をご覧ください。

動画もあります。TwilioとGemini Flashの統合を解説した動画をご覧ください。

Rishab Kumar は、Twilioのデベロッパーエバンジェリストであり、熱心なクラウド愛好家です。Rishabへのご連絡は、X（@rishabincloud）にお願いします。また、クラウド、DevOps、DevRelなどについては、youtube.com/@rishabincloudで個人ブログをフォローいただけます。

ConversationRelayとPythonを使用してGoogle GeminiとTwilio Voiceを連携

ConversationRelayとPythonを使用してGoogle GeminiとTwilio Voiceを連携

前提条件

プロジェクトのセットアップ

環境変数の設定

サーバーの作成

1. 必要なライブラリをインポートし、環境をセットアップする

2. 定数と設定 :

3. FastAPIとGeminiクライアントの初期化

4. AI応答関数

5. TwilioのTwiML応答

6. WebSocket接続処理

7. FastAPIサーバーの実行

サーバーの実行

Twilioの設定

連携のテスト

次のステップ

関連投稿

関連リソース