> 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. # Amazon Nova Sonic This guide shows how to integrate the AgentDuet SDK with [Amazon Nova 2 Sonic](https://docs.aws.amazon.com/nova/latest/userguide/speech.html) on Bedrock for bidirectional real-time speech-to-speech streaming on phone calls. ## What this integration provides A phone agent that answers calls, opens a Bedrock bidirectional stream to Nova Sonic, streams 24 kHz LPCM uplink, plays Nova audio back, clears the outbound buffer when Nova signals barge-in (`interrupted`), and ends the Nova session on hangup. ## Prerequisites - Python **3.12+** - Get an API key and connector UUID at [agentduet.com](https://agentduet.com) - AWS credentials with Bedrock access to Nova 2 Sonic ([AWS console](https://console.aws.amazon.com/), [Bedrock model access](https://docs.aws.amazon.com/bedrock/latest/userguide/model-access.html)) ```bash pip install "agentduet==1.0.0" aws-sdk-bedrock-runtime smithy-aws-core python-dotenv ``` ## Step 1: Create a project folder ```bash mkdir agentduet-nova-sonic-bridge && cd $_ python3.12 -m venv .venv && source .venv/bin/activate pip install "agentduet==1.0.0" aws-sdk-bedrock-runtime smithy-aws-core python-dotenv ``` ## Step 2: Configure `.env` ```bash cat > .env << 'EOF' AGENTDUET_API_KEY=your-connector-api-key AGENTDUET_CONNECTOR_UUID=your-connector-uuid AWS_ACCESS_KEY_ID= AWS_SECRET_ACCESS_KEY= AWS_REGION=us-east-1 EOF ``` ## Step 3: Write `nova_sonic_bridge.py` Create the file next to `.env`: ```python import asyncio import base64 import json import logging import os import uuid from typing import Optional from aws_sdk_bedrock_runtime.client import ( BedrockRuntimeClient, InvokeModelWithBidirectionalStreamOperationInput, ) from aws_sdk_bedrock_runtime.config import Config from aws_sdk_bedrock_runtime.models import ( BidirectionalInputPayloadPart, InvokeModelWithBidirectionalStreamInputChunk, ) from dotenv import load_dotenv from smithy_aws_core.identity.environment import EnvironmentCredentialsResolver from websockets.exceptions import ConnectionClosed 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__) MODEL_ID = "amazon.nova-2-sonic-v1:0" REGION = os.getenv("AWS_REGION", "us-east-1") SYSTEM_PROMPT = ( "You are a warm, professional AI assistant on a phone call. Give accurate answers " "that sound natural and direct. Answer clearly in 1-2 sentences, then expand only " "enough to stay understandable (3-5 short sentences total)." ) class NovaSonicIntegration: def __init__(self, call: Call, client: BedrockRuntimeClient): self._call = call self._client = client self._stream = None self.prompt_name = str(uuid.uuid4()) self.content_name = str(uuid.uuid4()) self.audio_content_name = str(uuid.uuid4()) self._send_to_nova_task: Optional[asyncio.Task] = None self._recv_from_nova_task: Optional[asyncio.Task] = None self._is_active = False async def _on_hangup(self, evt): if not self._is_active: return logger.info("Call terminated - cleaning up Nova session") self._is_active = False if self._send_to_nova_task: self._send_to_nova_task.cancel() if self._recv_from_nova_task: self._recv_from_nova_task.cancel() tasks = [t for t in (self._send_to_nova_task, self._recv_from_nova_task) if t] if tasks: await asyncio.gather(*tasks, return_exceptions=True) try: await self.end_session() except Exception: pass async def send_event(self, event_json: str): event = InvokeModelWithBidirectionalStreamInputChunk( value=BidirectionalInputPayloadPart(bytes_=event_json.encode("utf-8")) ) await self._stream.input_stream.send(event) async def start_session(self): self._stream = await self._client.invoke_model_with_bidirectional_stream( InvokeModelWithBidirectionalStreamOperationInput(model_id=MODEL_ID) ) self._is_active = True await self.send_event( """ { "event": { "sessionStart": { "inferenceConfiguration": { "maxTokens": 1024, "topP": 0.9, "temperature": 0.7 } } } } """ ) await self.send_event( f""" {{ "event": {{ "promptStart": {{ "promptName": "{self.prompt_name}", "textOutputConfiguration": {{ "mediaType": "text/plain" }}, "audioOutputConfiguration": {{ "mediaType": "audio/lpcm", "sampleRateHertz": 24000, "sampleSizeBits": 16, "channelCount": 1, "voiceId": "matthew", "encoding": "base64", "audioType": "SPEECH" }} }} }} }} """ ) await self.send_event( f""" {{ "event": {{ "contentStart": {{ "promptName": "{self.prompt_name}", "contentName": "{self.content_name}", "type": "TEXT", "interactive": false, "role": "SYSTEM", "textInputConfiguration": {{ "mediaType": "text/plain" }} }} }} }} """ ) await self.send_event( f""" {{ "event": {{ "textInput": {{ "promptName": "{self.prompt_name}", "contentName": "{self.content_name}", "content": "{SYSTEM_PROMPT}" }} }} }} """ ) await self.send_event( f""" {{ "event": {{ "contentEnd": {{ "promptName": "{self.prompt_name}", "contentName": "{self.content_name}" }} }} }} """ ) await self.send_event( f""" {{ "event": {{ "contentStart": {{ "promptName": "{self.prompt_name}", "contentName": "{self.audio_content_name}", "type": "AUDIO", "interactive": true, "role": "USER", "audioInputConfiguration": {{ "mediaType": "audio/lpcm", "sampleRateHertz": 24000, "sampleSizeBits": 16, "channelCount": 1, "audioType": "SPEECH", "encoding": "base64" }} }} }} }} """ ) async def end_session(self): if not self._stream: return try: await self.send_event( f""" {{ "event": {{ "contentEnd": {{ "promptName": "{self.prompt_name}", "contentName": "{self.audio_content_name}" }} }} }} """ ) await self.send_event( f""" {{ "event": {{ "promptEnd": {{ "promptName": "{self.prompt_name}" }} }} }} """ ) await self.send_event("""{ "event": { "sessionEnd": {} } }""") except Exception: pass finally: if self._stream: await self._stream.input_stream.close() self._stream = None async def run(self): self._call.on_hangup(self._on_hangup) await self.start_session() self._send_to_nova_task = asyncio.create_task(self.stream_to_nova()) self._recv_from_nova_task = asyncio.create_task(self.receive_audio_from_nova()) await asyncio.gather( self._send_to_nova_task, self._recv_from_nova_task, return_exceptions=True, ) async def stream_to_nova(self): try: async for audio_chunk in self._call.caller.audio_stream(): if not self._is_active: break blob = base64.b64encode(audio_chunk).decode("utf-8") await self.send_event( f""" {{ "event": {{ "audioInput": {{ "promptName": "{self.prompt_name}", "contentName": "{self.audio_content_name}", "content": "{blob}" }} }} }} """ ) except (ConnectionClosed, CallClosedError): pass except asyncio.CancelledError: raise except Exception: logger.exception("Error in stream to Nova") raise async def receive_audio_from_nova(self): try: while self._is_active: if not self._stream: await asyncio.sleep(0.1) continue output = await self._stream.await_output() result = await output[1].receive() if result.value and result.value.bytes_: json_data = json.loads(result.value.bytes_.decode("utf-8")) if "event" not in json_data: continue evt = json_data["event"] if "textOutput" in evt: text_content = evt["textOutput"]["content"] if '{ "interrupted" : true }' in text_content: await self._call.clear_send_audio_buffer() logger.debug("Nova interrupted - cleared send buffer") elif "audioOutput" in evt: audio_bytes = base64.b64decode(evt["audioOutput"]["content"]) try: await self._call.send_audio(audio_bytes) except BufferFullError: logger.warning( "Send buffer full - drop chunk or raise buffer_size" ) except (ConnectionClosed, asyncio.CancelledError): pass except CallClosedError: logger.debug("Call closed; stopping stream from Nova") except Exception: if self._is_active: logger.exception("Error in stream from Nova") raise async def main(): aws_config = Config( endpoint_uri=f"https://bedrock-runtime.{REGION}.amazonaws.com", region=REGION, aws_credentials_identity_resolver=EnvironmentCredentialsResolver(), ) bedrock_client = BedrockRuntimeClient(config=aws_config) 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: result = await call.answer() if not result: logger.error( "Answer failed %s: %s (%s)", call.id, result.error_message, result.error_code, ) return await NovaSonicIntegration(call, bedrock_client).run() except Exception: logger.exception("Error in Nova Sonic bridge") await call.close() raise await sm.run_forever() if __name__ == "__main__": asyncio.run(main()) ``` ## Step 4: Run the bridge ```bash python nova_sonic_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:** Both AgentDuet and Nova audio configs above use 24 kHz mono LPCM. Keep them aligned. - **Interrupt signal:** Nova signals barge-in inside `textOutput` content containing `{ "interrupted" : true }` - respond with `await call.clear_send_audio_buffer()`. - **Session teardown:** On hangup, cancel tasks *before* `end_session()` so you do not race writes on a closing stream. - **IAM / region:** Model availability varies by region; start with `us-east-1` and confirm Nova Sonic access. - **`BufferFullError`:** Increase `buffer_size` or drop chunks if Bedrock bursts faster than the phone path drains. ## Related - [Gemini Live](/integrations/gemini-live) - [OpenAI Realtime](/integrations/open-ai-realtime) - [Audio Streaming](/concepts/audio-streaming)