> For clean Markdown of any page, append .md to the page URL. > For a complete documentation index, see https://docs.agentduet.com/llms.txt. > For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.agentduet.com/_mcp/server. # Vapi This guide shows how to integrate the AgentDuet SDK with [Vapi](https://vapi.ai) conversational AI agents over Vapi's custom WebSocket transport (`provider: "vapi.websocket"`). ## What This Integration Provides AgentDuet provides telephony carrier connectivity and real-time audio streaming. Vapi orchestrates conversation logic, speech-to-text, LLM intelligence, and low-latency voice synthesis over its WebSocket transport. * **Telephony Carrier Control**: AgentDuet manages inbound phone numbers, SIP trunks, call answering, and hangup signals. * **Low-Latency Audio Streaming**: 16 kHz 16-bit Mono Linear PCM streamed bidirectionally between the telephone caller and Vapi. * **Instant Barge-In**: User speech triggers instant clearing of AgentDuet audio buffers (`call.clear_send_audio_buffer()`). * **Farewell Audio Protection**: Ensures the assistant's final goodbye audio fully plays to the caller before terminating the carrier session. * **Graceful Lifecycle Management**: Bidirectional termination handles caller hangup (`call.on_hangup`) and agent-initiated hangup (`hang` or `end-call`). ## Architecture ```mermaid flowchart TD Caller[Caller: PSTN Phone] -->|Inbound Telephony Call| AgentDuet[AgentDuet: Carrier & Media Engine] AgentDuet -->|16 kHz 16-bit Linear PCM| Bridge[Vapi Integration Bridge: Python] Bridge -->|wss://api.vapi.ai/call_id/transport| Vapi[Vapi Conversational AI Engine] Vapi -->|Speech Synthesis PCM Chunks| Bridge Bridge -->|Audio Playback Stream| AgentDuet AgentDuet -->|Ear Receiver Audio| Caller ``` | Layer | Responsibilities | | :--- | :--- | | **AgentDuet** | Telephone number provisioning, SIP carrier management, answering, packetization, and sub-millisecond audio buffer clearing | | **Vapi Integration Bridge** | Translates audio and control frames between AgentDuet and Vapi WebSocket protocol | | **Vapi Platform** | STT, conversational LLM reasoning, TTS synthesis, function calling, and conversation state | ## Prerequisites * Python **3.12+** * An AgentDuet API key and connector UUID from [AgentDuet Console](https://agentduet.com) * A Vapi Private API Key and Assistant ID from [Vapi Dashboard](https://dashboard.vapi.ai) ```bash pip install "agentduet==1.0.0" httpx websockets python-dotenv ``` ## Step 1: Create an Assistant in Vapi 1. Navigate to the [Vapi Dashboard](https://dashboard.vapi.ai). 2. Go to **Assistants** and click **Create Assistant**. 3. Choose your desired LLM (for example, GPT-4o, Claude 3.5 Sonnet, or Groq) and voice provider (such as ElevenLabs, PlayHT, or Deepgram). 4. Configure your system prompt and opening greeting. 5. Copy your **Assistant ID** from the top of the assistant settings page. 6. Copy your **Private API Key** from **Settings** -> **API Keys**. ## Step 2: Configure Environment Variables Create a `.env` file with your credentials: ```ini # AgentDuet Credentials AGENTDUET_API_KEY=your_agentduet_api_key AGENTDUET_CONNECTOR_UUID=your_connector_uuid # Vapi Credentials VAPI_API_KEY=your_vapi_private_api_key VAPI_ASSISTANT_ID=your_vapi_assistant_id ``` ## Step 3: Implement the Integration Bridge The bridge listens for incoming phone calls via AgentDuet's `SessionManager`, answers the call, creates a Vapi WebSocket transport session, and starts bidirectional streaming. Create `main.py`: ```python 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()) ``` ## Step 4: Run and Test 1. Start the integration bridge: ```bash python main.py ``` 2. Dial your AgentDuet phone number from your phone. 3. Your Vapi assistant answers immediately with its configured opening greeting. 4. Speak normally to interact with the assistant. If you interrupt the assistant while it is speaking, `call.clear_send_audio_buffer()` cancels the speech immediately, providing a natural conversational rhythm. ## Next Step Trigger visual n8n workflows and custom automation from AgentDuet voice agent events.