> 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. # OpenAI Realtime This guide shows how to integrate the AgentDuet SDK with the [OpenAI Agents SDK](https://openai.github.io/openai-agents-python/) and Realtime API for bidirectional real-time audio streaming, barge-in, and tool calling. ## What this integration provides A phone agent that answers incoming calls, streams 24 kHz PCM to an OpenAI Realtime session, plays model audio back, clears the outbound buffer on `audio_interrupted`, and closes the Realtime session on hangup. ## Prerequisites - Python **3.12+** - Get an API key and connector UUID at [agentduet.com](https://agentduet.com) - OpenAI API key with Realtime access ([API keys](https://platform.openai.com/api-keys)) ```bash pip install "agentduet==1.0.0" "openai-agents" python-dotenv ``` ## Step 1: Create a project folder ```bash mkdir agentduet-openai-realtime-bridge && cd $_ python3.12 -m venv .venv && source .venv/bin/activate pip install "agentduet==1.0.0" "openai-agents" python-dotenv ``` ## Step 2: Configure `.env` ```bash cat > .env << 'EOF' AGENTDUET_API_KEY=your-connector-api-key AGENTDUET_CONNECTOR_UUID=your-connector-uuid OPENAI_API_KEY=sk-... EOF ``` ## Step 3: Write `openai_realtime_bridge.py` Create the file next to `.env`: ```python import asyncio import logging import os from typing import Optional from agents.realtime import RealtimeAgent, RealtimeRunner from agents.realtime.session import RealtimeSession from dotenv import load_dotenv 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(__name__) AGENT = RealtimeAgent( name="Assistant", instructions="You are a helpful and friendly AI assistant.", ) RUNNER = RealtimeRunner( starting_agent=AGENT, config={ "model_settings": { "model_name": "gpt-realtime-1.5", "audio": { "input": { "format": "pcm16", "transcription": {"model": "gpt-4o-mini-transcribe"}, "turn_detection": { "type": "semantic_vad", "interrupt_response": True, }, }, "output": { "format": "pcm16", "voice": "ash", }, }, } }, ) class OpenAIRealtimeIntegration: def __init__(self, call: Call, session: RealtimeSession): self._call = call self._session = session self._send_task: Optional[asyncio.Task] = None self._recv_task: Optional[asyncio.Task] = None self._terminated = False async def _on_hangup(self, evt): logger.info("Call terminated - closing OpenAI Realtime session") self._terminated = True try: await self._session.close() except Exception: logger.exception("Error closing OpenAI Realtime session") tasks = [t for t in (self._send_task, self._recv_task) if t] for t in tasks: t.cancel() if tasks: await asyncio.gather(*tasks, return_exceptions=True) async def run(self): self._call.on_hangup(self._on_hangup) self._send_task = asyncio.create_task(self.stream_to_openai()) self._recv_task = asyncio.create_task(self.receive_from_openai()) await asyncio.gather(self._send_task, self._recv_task, return_exceptions=True) async def stream_to_openai(self): try: async for audio_chunk in self._call.caller.audio_stream(): await self._session.send_audio(audio_chunk) except asyncio.CancelledError: raise except CallClosedError: logger.debug("Call closed; stopping stream to OpenAI") except Exception: logger.exception("Error in stream to OpenAI") raise async def receive_from_openai(self): try: async for event in self._session: if event.type == "audio": audio_bytes = event.audio.data if audio_bytes: try: await self._call.send_audio(audio_bytes) except BufferFullError: logger.warning( "Send buffer full - drop chunk or raise buffer_size" ) elif event.type == "audio_interrupted": await self._call.clear_send_audio_buffer() logger.debug("OpenAI interrupted - cleared send buffer") elif event.type == "error": logger.error("OpenAI Realtime error: %s", event.error) except asyncio.CancelledError: raise except CallClosedError: logger.debug("Call closed; stopping receive from OpenAI") except Exception: logger.exception("Error in receive from OpenAI") raise async def main(): config = SessionManagerConfig.create( api_key=os.getenv("AGENTDUET_API_KEY"), connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"), call_audio=CallAudioConfig( sample_rate=24000, buffer_size=1024 * 1024, ), ) async with SessionManager(config) as sm: logger.info("Connected. Waiting for calls...") @sm.on_incoming_call async def on_call(noti: IncomingCallNotification): session = await sm.open_session(new_session_id(), noti.subscriber) call = await session.process_call(noti) logger.info("Incoming call %s", call.id) try: async with await RUNNER.run() as oai_session: result = await call.answer() if not result: logger.error( "Answer failed %s: %s (%s)", call.id, result.error_message, result.error_code, ) return await OpenAIRealtimeIntegration(call, oai_session).run() except Exception: logger.exception("Error in OpenAI Realtime bridge") await call.close() raise await sm.run_forever() if __name__ == "__main__": asyncio.run(main()) ``` ## Step 4: Run the bridge ```bash python openai_realtime_bridge.py ``` Keep the process running. Confirm logs show the SessionManager connected and waiting for calls. ## Step 5: Call your number Dial the connector phone number. Verify two-way audio, then interrupt mid-reply to confirm barge-in clears playback. Hang up and confirm the process remains ready for the next call. ## How the pieces fit `SessionManager` receives the inbound call → you `open_session` + `process_call` → `answer()` → two tasks move PCM between AgentDuet and the model. Hangup closes the model session and cancels those tasks. Telephony stays AgentDuet's job; the model never sees SIP. ## Notes - **Sample rate:** Keep AgentDuet and Realtime at 24 kHz `pcm16`. - **Interrupt:** On `audio_interrupted`, use `await call.clear_send_audio_buffer()`. Do not call private `_interrupt` helpers. - **Semantic VAD:** `interrupt_response: True` lets the model cancel its turn; you still must clear AgentDuet’s outbound buffer or the caller hears leftover audio. - **`BufferFullError`:** Increase `CallAudioConfig.buffer_size` or drop chunks under bursty TTS. - **Session lifecycle:** Open the Realtime session before `answer()` so the first greeting is not delayed by WebSocket setup. ## Related - [Amazon Nova Sonic](/integrations/amazon-nova-sonic) - [Gemini Live](/integrations/gemini-live) - [Grok Voice](/integrations/grok-voice) - [Audio Streaming](/concepts/audio-streaming)