Build Real-Time Voice Apps: OpenAI API & VideoSDK Integration

1from videosdk import (
2    VideoSDK,
3    Meeting,
4    MeetingConfig,
5    MeetingEventHandler,
6    ParticipantEventHandler,
7    Stream,
8    Participant
9)
10
11import logging
12
13logger = logging.getLogger(__name__) # use a specific logger
14
15class VideoSDKAgent:
16    """Represents an AI Agent connected to a VideoSDK meeting."""
17
18    def __init__(self, room_id: str, videosdk_token: str, agent_name: str = "AI Assistant"):
19        self.room_id = room_id
20        self.videosdk_token = videosdk_token
21        self.agent_name = agent_name
22        self.meeting = None
23        self.is_connected = False
24        self.participant_handlers = {} # Store participant handlers for cleanup
25
26        logger.info(f"[{self.agent_name}] Initializing for Room ID: {self.room_id}")
27        self._initialize_meeting()
28
29    def _initialize_meeting(self):
30        """Sets up the Meeting object using VideoSDK.init_meeting."""
31        try:
32            # Configure the agent's meeting settings
33            meeting_config = {
34                "meeting_id": self.room_id,
35                "token": self.videosdk_token,
36                "name": self.agent_name,
37                "mic_enabled": False,         # Agent doesn't speak initially
38                "webcam_enabled": False,      # Agent has no camera
39                "auto_consume": True,         # Automatically receive streams from others
40                # Add other relevant config if needed
41            }
42            logger.debug(f"[{self.agent_name}] Meeting Config: {meeting_config}")
43
44            self.meeting = VideoSDK.init_meeting(**meeting_config)
45
46            # Attach event handlers
47            meeting_event_handler = AgentMeetingEventHandler(self.agent_name, self)
48            self.meeting.add_event_listener(meeting_event_handler)
49
50            logger.info(f"[{self.agent_name}] Meeting object initialized.")
51
52        except Exception as e:
53            logger.exception(f"[{self.agent_name}] Failed to initialize VideoSDK Meeting: {e}")
54            self.meeting = None # Ensure meeting is None if init fails
55
56    async def connect(self):
57        """Connects the agent to the meeting asynchronously."""
58        if not self.meeting:
59            logger.error(f"[{self.agent_name}] Cannot connect, meeting not initialized.")
60            return
61
62        if self.is_connected:
63            logger.warning(f"[{self.agent_name}] Already connected or connecting.")
64            return
65
66        logger.info(f"[{self.agent_name}] Attempting to join meeting...")
67        try:
68            await self.meeting.async_join()
69            # Note: on_meeting_joined event confirms connection
70            self.is_connected = True # Mark as connected
71            logger.info(f"[{self.agent_name}] async_join call completed.")
72        except Exception as e:
73            logger.exception(f"[{self.agent_name}] Error during async_join: {e}")
74            self.is_connected = False
75

1class AgentMeetingEventHandler(MeetingEventHandler):
2    """Handles meeting-level events for the Agent."""
3    def __init__(self, agent_name: str, agent_instance: 'VideoSDKAgent'):
4        self.agent_name = agent_name
5        self.agent_instance = agent_instance # Reference to the agent itself
6        logger.info(f"[{self.agent_name}] MeetingEventHandler initialized.")
7
8    def on_meeting_joined(self, data) -> None:
9        logger.info(f"[{self.agent_name}] Successfully JOINED meeting.")
10        # You could potentially trigger actions here if needed
11
12    def on_meeting_left(self, data) -> None:
13        logger.warning(f"[{self.agent_name}] LEFT meeting.")
14        # Perform cleanup if needed
15        self.agent_instance.mark_disconnected() # Update agent state
16
17    def on_participant_joined(self, participant: Participant) -> None:
18        logger.info(
19            f"[{self.agent_name}] Participant JOINED: Id='{participant.id}', "
20            f"Name='{participant.display_name}', IsLocal={participant.local}"
21        )
22        # Add event listener for this specific participant's streams
23        participant_event_handler = AgentParticipantEventHandler(self.agent_name, participant)
24        participant.add_event_listener(participant_event_handler)
25        # Store handler reference
26        self.agent_instance.participant_handlers[participant.id] = participant_event_handler
27
28    def on_participant_left(self, participant: Participant) -> None:
29        logger.warning(
30            f"[{self.agent_name}] Participant LEFT: Id='{participant.id}', "
31            f"Name='{participant.display_name}'"
32        )
33        # Cleanup participant event handler
34        if participant.id in self.agent_instance.participant_handlers:
35            handler_to_remove = self.agent_instance.participant_handlers.pop(participant.id)
36            participant.remove_event_listener(handler_to_remove)
37            logger.debug(f"[{self.agent_name}] Removed event handler for participant {participant.id}")
38

1class AgentParticipantEventHandler(ParticipantEventHandler):
2    """Handles events specific to a participant within the meeting for the Agent."""
3    def __init__(self, agent_name: str, participant: Participant):
4        self.agent_name = agent_name
5        self.participant = participant
6        logger.info(f"[{self.agent_name}] ParticipantEventHandler initialized.")
7
8    def on_stream_enabled(self, stream: Stream) -> None:
9        """Handle Participant stream enabled event."""
10        logger.info(
11            f"[{self.agent_name}] Stream ENABLED: Kind='{stream.kind}', "
12            f"StreamId='{stream.id}', ParticipantId='{self.participant.id}', "
13            f"ParticipantName='{self.participant.display_name}'"
14        )
15        
16        # Process audio stream
17        if stream.kind == "audio" and not self.participant.local: # Process remote audio
18            logger.info(f"[{self.agent_name}] ===> Received AUDIO stream from {self.participant.display_name}")
19            
20            # Here you would implement audio processing with OpenAI Realtime Voice API
21            # Example:
22            # asyncio.create_task(self.process_audio_with_openai(stream))
23            
24    def on_stream_disabled(self, stream: Stream) -> None:
25        """Handle Participant stream disabled event."""
26        logger.info(
27            f"[{self.agent_name}] Stream DISABLED: Kind='{stream.kind}', "
28            f"StreamId='{stream.id}', ParticipantId='{self.participant.id}'"
29        )
30        if stream.kind == "audio" and not self.participant.local:
31            logger.info(f"[{self.agent_name}] ===> Stopped receiving AUDIO stream")
32            # Stop any corresponding audio processing task for this stream
33

1from fastapi.middleware.cors import CORSMiddleware
2import os
3import logging
4import httpx
5import asyncio
6from fastapi import Request, Response, HTTPException, FastAPI
7from twilio.twiml.voice_response import VoiceResponse
8from twilio.request_validator import RequestValidator
9from dotenv import load_dotenv
10
11# VideoSDKAgent
12from agent import VideoSDKAgent
13
14# Load environment variables
15load_dotenv()
16
17logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
18logger = logging.getLogger(__name__)
19
20# Environment variables
21VIDEOSDK_SIP_USERNAME = os.getenv("VIDEOSDK_SIP_USERNAME")
22VIDEOSDK_SIP_PASSWORD = os.getenv("VIDEOSDK_SIP_PASSWORD")
23VIDEOSDK_AUTH_TOKEN = os.getenv("VIDEOSDK_AUTH_TOKEN")
24TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN")
25
26validator = RequestValidator(TWILIO_AUTH_TOKEN)
27app = FastAPI()
28
29app.add_middleware(
30    CORSMiddleware,
31    allow_origins=["*"],
32    allow_credentials=True,
33    allow_methods=["*"],
34    allow_headers=["*"],
35)
36
37# Store active agents
38active_agents = {}
39
40# Create VideoSDK Room
41async def create_videosdk_room() -> str:
42    """Creates a new VideoSDK room and returns the roomId."""
43    if not VIDEOSDK_AUTH_TOKEN:
44         logging.error("VideoSDK Auth Token is not configured.")
45         raise HTTPException(status_code=500, detail="Server configuration error")
46
47    headers = {"Authorization": VIDEOSDK_AUTH_TOKEN}
48
49    async with httpx.AsyncClient() as client:
50        try:
51            logging.info("Attempting to create VideoSDK room")
52            response = await client.post("https://api.videosdk.live/v2/rooms", headers=headers, timeout=10.0)
53            response.raise_for_status()
54
55            response_data = response.json()
56            room_id = response_data.get("roomId")
57
58            if not room_id:
59                logging.error(f"VideoSDK response missing 'roomId'. Response: {response_data}")
60                raise HTTPException(status_code=500, detail="Failed to get roomId from VideoSDK")
61
62            logging.info(f"Successfully created VideoSDK room: {room_id}")
63            return room_id
64
65        except Exception as exc:
66            logging.error(f"Error creating VideoSDK room: {exc}")
67            raise HTTPException(status_code=500, detail=f"Error creating room: {exc}")
68

1@app.post("/join-agent", response_class=Response)
2async def handle_twilio_call(request: Request):
3    """
4    Handles incoming Twilio call webhook.
5    Creates a VideoSDK room and returns TwiML to connect the call via SIP.
6    """
7    # Validate Twilio request
8    form_data = await request.form()
9    twilio_signature = request.headers.get('X-Twilio-Signature', None)
10    url = str(request.url)
11
12    if not TWILIO_AUTH_TOKEN or not validator.validate(url, form_data, twilio_signature):
13        logging.warning("Twilio request validation failed.")
14        raise HTTPException(status_code=403, detail="Invalid Twilio Signature")
15    
16    room_id = None
17    agent_token = VIDEOSDK_AUTH_TOKEN
18    
19    logging.info("Received valid request on /join-agent")
20    try:
21        # Create VideoSDK Room
22        room_id = await create_videosdk_room()
23    except HTTPException as http_exc:
24        # If room creation fails, inform the caller via TwiML and hang up
25        response = VoiceResponse()
26        response.say("Sorry, we encountered an error connecting you. Please try again later.")
27        response.hangup()
28        logging.warning(f"Returning error TwiML due to room creation failure: {http_exc.detail}")
29        return Response(content=str(response), media_type="application/xml")
30    
31    # Initialize and connect AI agent (in background)
32    if room_id and agent_token:
33        logger.info(f"Initializing and connecting agent for room {room_id}")
34        agent = VideoSDKAgent(room_id=room_id, videosdk_token=agent_token, agent_name="AI Assistant")
35        active_agents[room_id] = agent # Store the agent instance
36
37        # Run the agent's connect method in the background
38        asyncio.create_task(agent.connect())
39        logger.info(f"Agent connection task created for room {room_id}")
40    else:
41        logger.error("Cannot proceed with agent connection, room_id or agent_token missing.")
42         
43    # Generate TwiML to connect to VideoSDK SIP
44    sip_uri = f"sip:{room_id}@sip.videosdk.live"
45    logging.info(f"Connecting caller to SIP URI: {sip_uri}")
46
47    response = VoiceResponse()
48    # Optional: Announce connecting
49    response.say("Thank you for calling. Connecting you to the meeting now.")
50
51    # Create the <Dial> verb with the <Sip> noun
52    dial = response.dial(caller_id=None)
53    dial.sip(sip_uri, username=VIDEOSDK_SIP_USERNAME, password=VIDEOSDK_SIP_PASSWORD)
54
55    logging.info("Generated TwiML for SIP connection.")
56    # Return the TwiML response to Twilio
57    return Response(content=str(response), media_type="application/xml")
58

1import base64
2import asyncio
3import json
4import websockets
5
6class OpenAIVoiceProcessor:
7    def __init__(self, api_key):
8        self.api_key = api_key
9        self.ws = None
10        self.audio_queue = asyncio.Queue()
11        
12    async def connect(self):
13        """Establish connection to OpenAI Realtime Voice API."""
14        ws_url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
15        
16        headers = {
17            "Authorization": f"Bearer {self.api_key}",
18            "OpenAI-Beta": "realtime=v1"
19        }
20        
21        self.ws = await websockets.connect(ws_url, extra_headers=headers)
22        
23        # Configure the session
24        await self.ws.send(json.dumps({
25            "type": "session.update",
26            "event_id": "session-update-1",
27            "session": {
28                "model": "gpt-4o-realtime-preview",
29                "instructions": "You are a helpful AI assistant.",
30                "voice": "alloy",
31                "modalities": ["text", "audio"]
32            }
33        }))
34        
35        # Start the response handler
36        asyncio.create_task(self.handle_responses())
37        
38    async def process_audio(self, audio_data):
39        """Send audio data to OpenAI for processing."""
40        if not self.ws:
41            print("WebSocket not connected")
42            return
43            
44        # Convert audio to the required format (base64-encoded PCM16)
45        base64_audio = base64.b64encode(audio_data).decode("utf-8")
46        
47        # Send the audio data
48        await self.ws.send(json.dumps({
49            "type": "input_audio_buffer.append",
50            "audio": base64_audio
51        }))
52        
53    async def handle_responses(self):
54        """Handle responses from the OpenAI Realtime Voice API."""
55        while True:
56            if not self.ws:
57                await asyncio.sleep(1)
58                continue
59                
60            try:
61                response = await self.ws.recv()
62                response_data = json.loads(response)
63                
64                if response_data["type"] == "response.audio_delta":
65                    # Handle audio response (base64-encoded PCM16)
66                    audio_bytes = base64.b64decode(response_data["delta"])
67                    await self.audio_queue.put(audio_bytes)
68                    
69                elif response_data["type"] == "response.text_delta":
70                    # Handle text response
71                    print(f"AI: {response_data['delta']}")
72                    
73            except Exception as e:
74                print(f"Error handling response: {e}")
75                await asyncio.sleep(1)
76

1class AgentParticipantEventHandler(ParticipantEventHandler):
2    def __init__(self, agent_name: str, participant: Participant, openai_processor: OpenAIVoiceProcessor):
3        self.agent_name = agent_name
4        self.participant = participant
5        self.openai_processor = openai_processor
6        self.audio_task = None
7        logger.info(f"[{self.agent_name}] ParticipantEventHandler initialized.")
8
9    def on_stream_enabled(self, stream: Stream) -> None:
10        # Process audio stream
11        if stream.kind == "audio" and not self.participant.local:
12            logger.info(f"[{self.agent_name}] Received AUDIO stream from {self.participant.display_name}")
13            
14            # Start processing audio with OpenAI
15            self.audio_task = asyncio.create_task(self.process_audio_stream(stream))
16            
17    def on_stream_disabled(self, stream: Stream) -> None:
18        if stream.kind == "audio" and not self.participant.local and self.audio_task:
19            logger.info(f"[{self.agent_name}] Stopped receiving AUDIO stream")
20            self.audio_task.cancel()
21            self.audio_task = None
22            
23    async def process_audio_stream(self, stream: Stream):
24        """Process audio stream with OpenAI Realtime Voice API."""
25        try:
26            # Process audio frames
27            while True:
28                # Get audio frame from stream
29                frame = await stream.track.recv()
30                
31                # Convert to appropriate format and send to OpenAI
32                audio_data = frame.to_ndarray()[0]
33                pcm_data = self.prepare_audio(audio_data)
34                
35                await self.openai_processor.process_audio(pcm_data)
36                
37                # Check for responses from OpenAI
38                while not self.openai_processor.audio_queue.empty():
39                    response_audio = await self.openai_processor.audio_queue.get()
40                    await self.play_audio_response(response_audio)
41                    
42        except asyncio.CancelledError:
43            logger.info(f"[{self.agent_name}] Audio processing task cancelled")
44        except Exception as e:
45            logger.error(f"[{self.agent_name}] Error processing audio: {e}")
46            
47    def prepare_audio(self, audio_data):
48        """Convert audio data to the format expected by OpenAI."""
49        # Example processing (would depend on specific audio format details)
50        import numpy as np
51        import librosa
52        
53        # Convert to float
54        audio_float = audio_data.astype(np.float32) / np.iinfo(np.int16).max
55        
56        # Convert to mono if needed
57        audio_mono = librosa.to_mono(audio_float.T) if audio_float.ndim > 1 else audio_float
58        
59        # Resample to 16kHz (OpenAI's expected sample rate)
60        audio_resampled = librosa.resample(audio_mono, orig_sr=48000, target_sr=16000)
61        
62        # Convert back to PCM16
63        pcm_data = (audio_resampled * np.iinfo(np.int16).max).astype(np.int16).tobytes()
64        
65        return pcm_data
66        
67    async def play_audio_response(self, audio_bytes):
68        """Play audio response in the meeting."""
69        # This would depend on VideoSDK's API for sending audio
70        # For example, you might use a custom audio track to play the response
71        if hasattr(self.participant, 'meeting') and hasattr(self.participant.meeting, 'custom_microphone_audio_track'):
72            custom_track = self.participant.meeting.custom_microphone_audio_track
73            await custom_track.add_new_bytes(iter([audio_bytes]))
74

1# When setting up the session
2await self.ws.send(json.dumps({
3    "type": "session.update",
4    "event_id": "session-update-1",
5    "session": {
6        "model": "gpt-4o-realtime-preview",
7        "instructions": "You are a helpful AI assistant.",
8        "voice": "alloy",
9        "modalities": ["text", "audio"],
10        "tools": [
11            {
12                "type": "function",
13                "function": {
14                    "name": "get_weather",
15                    "description": "Get the current weather for a location",
16                    "parameters": {
17                        "type": "object",
18                        "properties": {
19                            "location": {
20                                "type": "string",
21                                "description": "The city and state or country"
22                            }
23                        },
24                        "required": ["location"]
25                    }
26                }
27            }
28        ]
29    }
30}))
31
32# In the response handler
33async def handle_responses(self):
34    while True:
35        if not self.ws:
36            await asyncio.sleep(1)
37            continue
38            
39        try:
40            response = await self.ws.recv()
41            response_data = json.loads(response)
42            
43            if response_data["type"] == "response.function_call":
44                # Handle function call
45                function_name = response_data["function"]["name"]
46                arguments = json.loads(response_data["function"]["arguments"])
47                
48                # Execute the function
49                if function_name == "get_weather":
50                    location = arguments["location"]
51                    weather_data = await self.get_weather(location)
52                    
53                    # Send the function result back
54                    await self.ws.send(json.dumps({
55                        "type": "function_call.completed",
56                        "call_id": response_data["call_id"],
57                        "output": json.dumps(weather_data)
58                    }))
59                    
60        except Exception as e:
61            print(f"Error handling response: {e}")
62            await asyncio.sleep(1)
63