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