Twilio Python Sdk Example (#121)

* Twilio Python Sdk Example

* Add Readme + requirements

examples/conversational-ai/twilio/python-sdk/README.md

@@ -0,0 +1,61 @@
+# Conversational AI Twilio Python SDK
+
+## Getting Started
+
+### Setting Up the Environment
+
+1. **Create a Virtual Environment**  
+   Navigate to your project directory and create a virtual environment:
+
+   ```bash
+   python -m venv venv
+   ```
+
+2. **Activate the Virtual Environment**  
+   - On Windows:
+     ```bash
+     venv\Scripts\activate
+     ```
+   - On macOS/Linux:
+     ```bash
+     source venv/bin/activate
+     ```
+
+3. **Install Requirements**  
+   Install the necessary packages using pip:
+
+   ```bash
+   pip install -r requirements.txt
+   ```
+
+### Running the Application
+
+To start the application, run the following command:
+
+```bash
+uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
+```
+
+In your code, ensure that the `requires_auth` parameter is set correctly in the `Conversation` instantiation. It should have the same value as the settings of your agent.
+
+### Setting Environment Variables
+
+You need to export the following environment variables:
+
+- `ELEVENLABS_API_KEY`
+- `AGENT_ID`
+
+You can do this in your terminal:
+
+```bash
+export ELEVENLABS_API_KEY='your_api_key_here'
+export AGENT_ID='your_agent_id_here'
+```
+
+Alternatively, you can use a `.env` file with the `python-dotenv` package. Create a `.env` file in your project root and add:
+
+```
+ELEVENLABS_API_KEY=your_api_key_here
+AGENT_ID=your_agent_id_here
+```
+

examples/conversational-ai/twilio/python-sdk/main.py

@@ -0,0 +1,83 @@
+import os
+import json
+import traceback
+from dotenv import load_dotenv
+from fastapi import FastAPI, Request, WebSocket
+from fastapi.responses import HTMLResponse
+from twilio.twiml.voice_response import VoiceResponse, Connect
+from elevenlabs import ElevenLabs
+from elevenlabs.conversational_ai.conversation import Conversation
+from twilio_audio_interface import TwilioAudioInterface
+from starlette.websockets import WebSocketDisconnect
+
+load_dotenv()
+
+ELEVEN_LABS_AGENT_ID = os.getenv("AGENT_ID")
+ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY")
+
+app = FastAPI()
+
+
+@app.get("/")
+async def root():
+    return {"message": "Twilio-ElevenLabs Integration Server"}
+
+
+@app.post("/twilio/inbound_call")
+async def handle_incoming_call(request: Request):
+    form_data = await request.form()
+    call_sid = form_data.get("CallSid", "Unknown")
+    from_number = form_data.get("From", "Unknown")
+    print(f"Incoming call: CallSid={call_sid}, From={from_number}")
+
+    response = VoiceResponse()
+    connect = Connect()
+    connect.stream(url=f"wss://{request.url.hostname}/media-stream-eleven")
+    response.append(connect)
+    return HTMLResponse(content=str(response), media_type="application/xml")
+
+
+@app.websocket("/media-stream-eleven")
+async def handle_media_stream(websocket: WebSocket):
+    await websocket.accept()
+    print("WebSocket connection opened")
+
+    audio_interface = TwilioAudioInterface(websocket)
+    eleven_labs_client = ElevenLabs(api_key=ELEVENLABS_API_KEY)
+
+    try:
+        conversation = Conversation(
+            client=eleven_labs_client,
+            agent_id=ELEVEN_LABS_AGENT_ID,
+            requires_auth=True, # Security > Enable authentication
+            audio_interface=audio_interface,
+            callback_agent_response=lambda text: print(f"Agent: {text}"),
+            callback_user_transcript=lambda text: print(f"User: {text}"),
+        )
+
+        conversation.start_session()
+        print("Conversation started")
+
+        async for message in websocket.iter_text():
+            if not message:
+                continue
+            await audio_interface.handle_twilio_message(json.loads(message))
+
+    except WebSocketDisconnect:
+        print("WebSocket disconnected")
+    except Exception:
+        print("Error occurred in WebSocket handler:")
+        traceback.print_exc()
+    finally:
+        try:
+            conversation.end_session()
+            conversation.wait_for_session_end()
+            print("Conversation ended")
+        except Exception:
+            print("Error ending conversation session:")
+            traceback.print_exc()
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

examples/conversational-ai/twilio/python-sdk/requirements.txt

@@ -0,0 +1,7 @@
+fastapi
+uvicorn
+twilio
+python-dotenv
+elevenlabs
+starlette
+python-multipart

examples/conversational-ai/twilio/python-sdk/twilio_audio_interface.py

@@ -0,0 +1,61 @@
+import asyncio
+import base64
+import json
+from fastapi import WebSocket
+from elevenlabs.conversational_ai.conversation import AudioInterface
+from starlette.websockets import WebSocketDisconnect, WebSocketState
+
+
+class TwilioAudioInterface(AudioInterface):
+    def __init__(self, websocket: WebSocket):
+        self.websocket = websocket
+        self.input_callback = None
+        self.stream_sid = None
+        self.loop = asyncio.get_event_loop()
+
+    def start(self, input_callback):
+        self.input_callback = input_callback
+
+    def stop(self):
+        self.input_callback = None
+        self.stream_sid = None
+
+    def output(self, audio: bytes):
+        """
+        This method should return quickly and not block the calling thread.
+        """
+        asyncio.run_coroutine_threadsafe(self.send_audio_to_twilio(audio), self.loop)
+
+    def interrupt(self):
+        asyncio.run_coroutine_threadsafe(self.send_clear_message_to_twilio(), self.loop)
+
+    async def send_audio_to_twilio(self, audio: bytes):
+        if self.stream_sid:
+            audio_payload = base64.b64encode(audio).decode("utf-8")
+            audio_delta = {
+                "event": "media",
+                "streamSid": self.stream_sid,
+                "media": {"payload": audio_payload},
+            }
+            try:
+                if self.websocket.application_state == WebSocketState.CONNECTED:
+                    await self.websocket.send_text(json.dumps(audio_delta))
+            except (WebSocketDisconnect, RuntimeError):
+                pass
+
+    async def send_clear_message_to_twilio(self):
+        if self.stream_sid:
+            clear_message = {"event": "clear", "streamSid": self.stream_sid}
+            try:
+                if self.websocket.application_state == WebSocketState.CONNECTED:
+                    await self.websocket.send_text(json.dumps(clear_message))
+            except (WebSocketDisconnect, RuntimeError):
+                pass
+
+    async def handle_twilio_message(self, data):
+        event_type = data.get("event")
+        if event_type == "start":
+            self.stream_sid = data["start"]["streamSid"]
+        elif event_type == "media" and self.input_callback:
+            audio_data = base64.b64decode(data["media"]["payload"])
+            self.input_callback(audio_data)