> 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. # Qwen Omni This guide shows how to integrate the AgentDuet SDK with [Alibaba Cloud's Qwen-Omni Realtime](https://www.alibabacloud.com/help/en/model-studio/realtime) API for bidirectional real-time audio streaming, including sample-rate bridging between AgentDuet and Qwen. ## What this integration provides A phone agent that answers calls, downsamples AgentDuet 24 kHz PCM to Qwen’s 16 kHz input, plays Qwen’s 24 kHz output back on the call, cancels the model turn and clears the outbound buffer on barge-in, and closes the WebSocket on hangup. ## Prerequisites - Python **3.12+** - Get an API key and connector UUID at [agentduet.com](https://agentduet.com) - DashScope API key for Qwen Omni Realtime ([get an API key](https://www.alibabacloud.com/help/en/model-studio/get-api-key)) ```bash pip install "agentduet==1.0.0" websockets soxr numpy python-dotenv ``` ## Step 1: Create a project folder ```bash mkdir agentduet-qwen-omni-bridge && cd $_ python3.12 -m venv .venv && source .venv/bin/activate pip install "agentduet==1.0.0" websockets soxr numpy python-dotenv ``` ## Step 2: Configure `.env` ```bash cat > .env << 'EOF' AGENTDUET_API_KEY=your-connector-api-key AGENTDUET_CONNECTOR_UUID=your-connector-uuid DASHSCOPE_API_KEY= DASHSCOPE_REGION=intl EOF ``` ## Step 3: Write `qwen_omni_bridge.py` Create the file next to `.env`: ```python import asyncio import base64 import json import logging import os import time from typing import Any, Callable, Dict, Optional import numpy as np import soxr import websockets 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__) REGION = os.getenv("DASHSCOPE_REGION", "intl") BASE_DOMAIN = ( "dashscope-intl.aliyuncs.com" if REGION == "intl" else "dashscope.aliyuncs.com" ) QWEN_WS_URL = f"wss://{BASE_DOMAIN}/api-ws/v1/realtime" MODEL = "qwen3.5-omni-flash-realtime" class QwenRealtimeClient: """Minimal Qwen-Omni Realtime WebSocket client.""" def __init__( self, url: str, api_key: str, model: str, voice: str = "Jennifer", instructions: str = "You are a helpful and respectful AI assistant.", on_audio_delta: Optional[Callable[[bytes], None]] = None, on_interruption: Optional[Callable[[], Any]] = None, ): self.url = f"{url}?model={model}" self.api_key = api_key self.voice = voice self.instructions = instructions self.on_audio_delta = on_audio_delta self.on_interruption = on_interruption self.ws = None self._is_responding = False async def connect(self): headers = {"Authorization": f"Bearer {self.api_key}"} self.ws = await websockets.connect(self.url, additional_headers=headers) logger.info("Connected to Qwen-Omni WebSocket") await self.send_event( { "type": "session.update", "session": { "modalities": ["text", "audio"], "voice": self.voice, "instructions": self.instructions, "input_audio_format": "pcm16", # 16-bit 16 kHz mono "output_audio_format": "pcm24", # 16-bit 24 kHz mono "turn_detection": { "type": "server_vad", "threshold": 0.5, "prefix_padding_ms": 300, "silence_duration_ms": 500, }, "input_audio_transcription": {"model": "gummy-realtime-v1"}, }, } ) async def send_event(self, event: Dict[str, Any]): if "event_id" not in event: event["event_id"] = f"evt_{int(time.time() * 1000)}" await self.ws.send(json.dumps(event)) async def stream_audio(self, audio_chunk: bytes): await self.send_event( { "type": "input_audio_buffer.append", "audio": base64.b64encode(audio_chunk).decode(), } ) async def cancel_response(self): if self._is_responding: logger.info("Sending response.cancel to Qwen") await self.send_event({"type": "response.cancel"}) self._is_responding = False async def receive_loop(self): try: async for message in self.ws: event = json.loads(message) event_type = event.get("type") if event_type == "response.audio.delta": audio_bytes = base64.b64decode(event["delta"]) if self.on_audio_delta: self.on_audio_delta(audio_bytes) elif event_type == "input_audio_buffer.speech_started": logger.debug("Speech start - possible interruption") if self.on_interruption: await self.on_interruption() elif event_type == "response.created": self._is_responding = True elif event_type == "response.done": self._is_responding = False elif event_type == "error": logger.error("Qwen error: %s", event.get("error")) elif event_type == "conversation.item.input_audio_transcription.completed": logger.info("User: %s", event.get("transcript")) elif event_type == "response.audio_transcript.done": logger.info("AI: %s", event.get("transcript")) except websockets.exceptions.ConnectionClosed: logger.info("Qwen WebSocket closed") except Exception: logger.exception("Error in Qwen receive loop") async def close(self): if self.ws: await self.ws.close() class QwenRealtimeIntegration: def __init__(self, call: Call, qwen_client: QwenRealtimeClient): self._call = call self._qwen_client = qwen_client self._stream_task: Optional[asyncio.Task] = None self._receive_task: Optional[asyncio.Task] = None self._terminated = False async def _on_hangup(self, evt): logger.info("Call terminated - cleaning up Qwen") self._terminated = True await self._qwen_client.close() if self._stream_task: self._stream_task.cancel() if self._receive_task: self._receive_task.cancel() async def handle_interruption(self): logger.info("Interruption - cancel Qwen response and clear send buffer") await self._qwen_client.cancel_response() await self._call.clear_send_audio_buffer() async def run(self): self._call.on_hangup(self._on_hangup) await self._qwen_client.connect() self._stream_task = asyncio.create_task(self.stream_to_qwen()) self._receive_task = asyncio.create_task(self._qwen_client.receive_loop()) await asyncio.gather( self._stream_task, self._receive_task, return_exceptions=True ) async def stream_to_qwen(self): try: async for audio_chunk in self._call.caller.audio_stream(): if self._terminated: break resampled = self.downsample_24to16(audio_chunk) await self._qwen_client.stream_audio(resampled) except CallClosedError: logger.debug("Call closed; stopping stream to Qwen") except Exception: logger.exception("Error in stream to Qwen") def downsample_24to16(self, audio_data: bytes) -> bytes: if not audio_data: return b"" samples = np.frombuffer(audio_data, dtype=np.int16) resampled = soxr.resample(samples, 24000, 16000) return resampled.astype(np.int16).tobytes() def on_qwen_audio(self, audio_bytes: bytes): if self._terminated: return asyncio.create_task(self._send_audio_to_call(audio_bytes)) async def _send_audio_to_call(self, audio_bytes: bytes): try: await self._call.send_audio(audio_bytes) except BufferFullError: logger.warning("Call audio buffer full, dropping chunk") except CallClosedError: logger.debug("Call closed; dropping Qwen audio") except Exception: logger.exception("Error sending audio to call") 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=8 * 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: integration: Optional[QwenRealtimeIntegration] = None async def on_interruption_handler(): if integration: await integration.handle_interruption() qwen_client = QwenRealtimeClient( url=QWEN_WS_URL, api_key=os.getenv("DASHSCOPE_API_KEY"), model=MODEL, on_audio_delta=lambda data: ( integration.on_qwen_audio(data) if integration else None ), on_interruption=on_interruption_handler, ) integration = QwenRealtimeIntegration(call, qwen_client) result = await call.answer() if not result: logger.error( "Answer failed %s: %s (%s)", call.id, result.error_message, result.error_code, ) return await integration.run() except Exception: logger.exception("Error in Qwen bridge") await call.close() await sm.run_forever() if __name__ == "__main__": asyncio.run(main()) ``` ## Step 4: Run the bridge ```bash python qwen_omni_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 - **Rate mismatch:** AgentDuet and Qwen *output* are 24 kHz; Qwen *input* is 16 kHz. Always downsample uplink with soxr (or equivalent). Do not change AgentDuet to 16 kHz unless you also resample downlink. - **Interrupt:** On speech start, cancel the Qwen response *and* `await call.clear_send_audio_buffer()`. - **Buffer size:** Qwen can burst audio; a larger `buffer_size` (e.g. 8 MB) reduces `BufferFullError` drops. - **Region:** Use `DASHSCOPE_REGION=cn` for the China endpoint; default is international. - **Callbacks:** Wire `on_audio_delta` / `on_interruption` after constructing the integration so closures see a live instance. ## Related - [Gemini Live](/integrations/gemini-live) - [Grok Voice](/integrations/grok-voice) - [Audio Streaming](/concepts/audio-streaming)