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())
