import asyncio
import json
import logging
import os
from typing import Optional
from dotenv import load_dotenv
import httpx
import websockets
from agentduet import (
BufferFullError,
Call,
CallAudioConfig,
CallClosedError,
IncomingCallNotification,
SessionManager,
SessionManagerConfig,
new_session_id,
)
load_dotenv()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger("agentduet-vapi")
SAMPLE_RATE = 16000 # Vapi native telephony sample rate
class VapiIntegration:
"""Bridges AgentDuet inbound phone calls with Vapi conversational AI agents."""
def __init__(
self,
call: Call,
assistant_id: str,
api_key: str,
vapi_base_url: str = "https://api.vapi.ai",
):
self._call = call
self._assistant_id = assistant_id
self._api_key = api_key
self._vapi_base_url = vapi_base_url.rstrip("/")
self._ws: Optional[websockets.WebSocketClientProtocol] = None
self._inbound_task: Optional[asyncio.Task] = None
self._outbound_task: Optional[asyncio.Task] = None
self._vapi_call_id: Optional[str] = None
self._closed = False
self._hangup_received = False
self._farewell_drain_seconds = 2.5
async def start(self) -> None:
"""Creates the Vapi transport call and connects WebSocket streaming."""
# 1. Create a call session with custom websocket transport
create_url = f"{self._vapi_base_url}/call"
payload = {
"assistantId": self._assistant_id,
"transport": {
"provider": "vapi.websocket",
"format": {
"format": "pcm_s16le",
"sampleRate": SAMPLE_RATE,
},
},
"customer": {
"number": getattr(self._call, "caller", "unknown"),
},
}
headers = {
"Authorization": f"Bearer {self._api_key}",
"Content-Type": "application/json",
}
async with httpx.AsyncClient(timeout=10.0) as client:
resp = await client.post(create_url, json=payload, headers=headers)
resp.raise_for_status()
call_data = resp.json()
self._vapi_call_id = call_data.get("id")
ws_url = (
call_data.get("transport", {}).get("websocketUrl")
or f"wss://api.vapi.ai/{self._vapi_call_id}/transport"
)
logger.info("Connecting to Vapi WebSocket transport: %s", ws_url)
# 2. Connect WebSocket
self._ws = await websockets.connect(
ws_url,
additional_headers=headers,
ping_interval=20,
ping_timeout=20,
)
# 3. Launch bidirectional streaming tasks
self._inbound_task = asyncio.create_task(
self._stream_mic_to_vapi(), name="vapi_inbound_stream"
)
self._outbound_task = asyncio.create_task(
self._stream_vapi_to_call(), name="vapi_outbound_stream"
)
async def _stream_mic_to_vapi(self) -> None:
"""Reads microphone audio from telephone caller and sends binary PCM to Vapi."""
try:
async for chunk in self._call.audio_stream():
if self._closed or not self._ws or self._ws.closed:
break
await self._ws.send(chunk)
except (CallClosedError, asyncio.CancelledError):
pass
except Exception as e:
logger.warning("Error in microphone stream to Vapi: %s", e)
async def _stream_vapi_to_call(self) -> None:
"""Receives audio and control messages from Vapi and plays audio to caller."""
try:
async for message in self._ws:
if self._closed:
break
# Binary frame: synthesized voice PCM from Vapi
if isinstance(message, bytes):
try:
await self._call.send_audio(message)
except (CallClosedError, BufferFullError):
break
# Text frame: JSON control messages
elif isinstance(message, str):
await self._handle_control_message(message)
except (websockets.ConnectionClosed, asyncio.CancelledError):
pass
except Exception as e:
logger.warning("Error in Vapi audio receiver: %s", e)
async def _handle_control_message(self, message: str) -> None:
"""Handles barge-in interruption and hangup signals from Vapi."""
try:
data = json.loads(message)
msg_type = data.get("type", "")
# User barge-in: clear telephone playback buffer instantly
if msg_type in ("user-interrupted", "conversation-update"):
if data.get("interrupted"):
logger.info("Caller interrupted: clearing outbound audio buffer")
self._call.clear_send_audio_buffer()
# End of call initiated by assistant
elif msg_type in ("hang", "end-of-call-report"):
logger.info("Vapi hangup signal received (%s)", msg_type)
self._hangup_received = True
# Allow final goodbye audio to drain before closing
await asyncio.sleep(self._farewell_drain_seconds)
await self._call.close()
except json.JSONDecodeError:
pass
async def stop(self) -> None:
"""Closes all streaming resources cleanly."""
self._closed = True
for task in (self._inbound_task, self._outbound_task):
if task and not task.done():
task.cancel()
if self._ws and not self._ws.closed:
await self._ws.close()
async def handle_call(sm: SessionManager, noti: IncomingCallNotification) -> None:
"""Answers the incoming phone call and attaches the Vapi bridge."""
session = await sm.open_session(new_session_id(), noti.subscriber)
call: Call = await session.process_call(noti)
if not await call.answer():
logger.error("Failed to answer incoming call %s", call.id)
return
assistant_id = os.environ["VAPI_ASSISTANT_ID"]
api_key = os.environ["VAPI_API_KEY"]
bridge = VapiIntegration(call, assistant_id, api_key)
async def on_hangup(_evt):
logger.info("Remote caller hung up call %s", call.id)
await bridge.stop()
call.on_hangup(on_hangup)
try:
await bridge.start()
# Keep bridge active while call remains open
while not call.is_closed:
await asyncio.sleep(0.5)
except Exception as e:
logger.exception("Error in Vapi call session: %s", e)
finally:
await bridge.stop()
try:
await call.close()
except Exception:
pass
async def main() -> None:
api_key = os.environ["AGENTDUET_API_KEY"]
connector_uuid = os.environ["AGENTDUET_CONNECTOR_UUID"]
config = SessionManagerConfig.create(
api_key=api_key,
connector_uuid=connector_uuid,
call_audio=CallAudioConfig(
sample_rate=SAMPLE_RATE,
buffer_size=1024 * 1024,
),
)
async with SessionManager(config) as sm:
logger.info("AgentDuet SessionManager connected: %s", sm.id)
@sm.on_incoming_call
async def on_incoming_call(noti: IncomingCallNotification):
logger.info("Incoming phone call received from %s", noti.subscriber)
asyncio.create_task(handle_call(sm, noti))
logger.info("Vapi Telephony Gateway active: waiting for calls...")
await sm.run_forever()
if __name__ == "__main__":
asyncio.run(main())
