can't setup rtclient for callautomation-azure-openai-voice

[crieux]

Hi,
Thanks for adding the repo callautomation-azure-openai-voice, I'll be glad to test it.
However, there is a bug when I try to setup my python environement using the instruction in the repo's README file :

pip install -r requirements.txt

Everything is okay with the first command, but the second one returns a error :

pip install -r ./aoai-whl/rtclient-0.5.1-py3-none-any.whl

File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.11_3.11.2544.0_x64__qbz5n2kfra8p0\Lib\encodings\cp1252.py", line 15, in decode
    return codecs.charmap_decode(input,errors,decoding_table)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
UnicodeDecodeError: 'charmap' codec can't decode byte 0x90 in position 84: character maps to <undefined>

I'm working on Windows 11 and using a python 3.11 version.
Am I doing something wrong ?
Thank you

---

This issue is for a: (mark with an x)

- [ x ] bug report -> please search issues before submitting
- [ ] feature request
- [ ] documentation issue or request
- [ ] regression (a behavior that used to work and stopped in a new release)

[crieux]

Got this fixed using the following command :
pip install -U wheel ./aoai-whl/rtclient-0.5.1-py3-none-any.whl

[venkatasaive]

@crieux thankyou very much. it worked fine.

[venkatasaive]

Got this fixed using the following command : pip install -U wheel ./aoai-whl/rtclient-0.5.1-py3-none-any.whl

hi @crieux i am trying to stop media streaming when the key words are detected like 'agent' so that i can transfer call to external agent. but the issue is that i am unable to stop the media streaming and also when i close the websocket it is causing issues for the another uses calling the application. this is my code --- main.py from quart import Quart, Response, request, json, redirect, websocket
from azure.eventgrid import EventGridEvent, SystemEventNames
from urllib.parse import urlencode, urljoin, urlparse, urlunparse
from logging import INFO
from azure.communication.callautomation import (
MediaStreamingOptions,
AudioFormat,
MediaStreamingTransportType,
MediaStreamingContentType,
MediaStreamingAudioChannelType,
PhoneNumberIdentifier,
)
from azure.communication.callautomation.aio import (
CallAutomationClient
)
import uuid
from azure.core.messaging import CloudEvent

from azureOpenAIService import init_websocket, start_conversation
from mediaStreamingHandler import process_websocket_message_async
from threading import Thread

Your ACS resource connection string

ACS_CONNECTION_STRING = "ACS_string"

Callback events URI to handle callback events.

CALLBACK_URI_HOST = "https://gddh5wtf.inc1.devtunnels.ms:8080"
CALLBACK_EVENTS_URI = CALLBACK_URI_HOST + "/api/callbacks"

acs_client = CallAutomationClient.from_connection_string(ACS_CONNECTION_STRING)
call_connection_id = None
callback_uri = None
transfer_call_agent=False
app = Quart(name)

@app.route("/api/incomingCall", methods=['POST'])
async def incoming_call_handler():
app.logger.info("incoming event data")
for event_dict in await request.json:
global callback_uri
event = EventGridEvent.from_dict(event_dict)
app.logger.info("incoming event data --> %s", event.data)
if event.event_type == SystemEventNames.EventGridSubscriptionValidationEventName:
app.logger.info("Validating subscription")
validation_code = event.data['validationCode']
validation_response = {'validationResponse': validation_code}
return Response(response=json.dumps(validation_response), status=200)
elif event.event_type =="Microsoft.Communication.IncomingCall":
app.logger.info("Incoming call received: data=%s",
event.data)
if event.data['from']['kind'] =="phoneNumber":
caller_id = event.data['from']["phoneNumber"]["value"]
else :
caller_id = event.data['from']['rawId']
app.logger.info("incoming call handler caller id: %s",
caller_id)
incoming_call_context=event.data['incomingCallContext']
guid =uuid.uuid4()
query_parameters = urlencode({"callerId": caller_id})
callback_uri = f"{CALLBACK_EVENTS_URI}/{guid}?{query_parameters}"

            parsed_url = urlparse(CALLBACK_EVENTS_URI)
            websocket_url = urlunparse(('wss',parsed_url.netloc,'/ws','', '', ''))

            app.logger.info("callback url: %s",  callback_uri)
            app.logger.info("websocket url: %s",  websocket_url)

            media_streaming_options = MediaStreamingOptions(
                    transport_url=websocket_url,
                    transport_type=MediaStreamingTransportType.WEBSOCKET,
                    content_type=MediaStreamingContentType.AUDIO,
                    audio_channel_type=MediaStreamingAudioChannelType.MIXED,
                    start_media_streaming=True,
                    enable_bidirectional=True,
                    audio_format=AudioFormat.PCM24_K_MONO)
            
            answer_call_result = await acs_client.answer_call(incoming_call_context=incoming_call_context,
                                                        operation_context="incomingCall",
                                                        callback_url=callback_uri, 
                                                        media_streaming=media_streaming_options)
            app.logger.info("Answered call for connection id: %s",
                            answer_call_result.call_connection_id)
        return Response(status=200)

@app.route('/api/callbacks/', methods=['POST'])
async def callbacks(contextId):
for event in await request.json:
# Parsing callback events
global call_connection_id
event_data = event['data']
call_connection_id = event_data["callConnectionId"]
app.logger.info(f"Received Event:-> {event['type']}, Correlation Id:-> {event_data['correlationId']}, CallConnectionId:-> {call_connection_id}")
if event['type'] == "Microsoft.Communication.CallConnected":
call_connection_properties = await acs_client.get_call_connection(call_connection_id).get_call_properties()
media_streaming_subscription = call_connection_properties.media_streaming_subscription
app.logger.info(f"MediaStreamingSubscription:--> {media_streaming_subscription}")
app.logger.info(f"Received CallConnected event for connection id: {call_connection_id}")
app.logger.info("CORRELATION ID:--> %s", event_data["correlationId"])
app.logger.info("CALL CONNECTION ID:--> %s", event_data["callConnectionId"])
elif event['type'] == "Microsoft.Communication.MediaStreamingStarted":
app.logger.info(f"Media streaming content type:--> {event_data['mediaStreamingUpdate']['contentType']}")
app.logger.info(f"Media streaming status:--> {event_data['mediaStreamingUpdate']['mediaStreamingStatus']}")
app.logger.info(f"Media streaming status details:--> {event_data['mediaStreamingUpdate']['mediaStreamingStatusDetails']}")
elif event['type'] == "Microsoft.Communication.MediaStreamingStopped":
app.logger.info(f"Media streaming content type:--> {event_data['mediaStreamingUpdate']['contentType']}")
app.logger.info(f"Media streaming status:--> {event_data['mediaStreamingUpdate']['mediaStreamingStatus']}")
app.logger.info(f"Media streaming status details:--> {event_data['mediaStreamingUpdate']['mediaStreamingStatusDetails']}")
elif event['type'] == "Microsoft.Communication.MediaStreamingFailed":
app.logger.info(f"Code:->{event_data['resultInformation']['code']}, Subcode:-> {event_data['resultInformation']['subCode']}")
app.logger.info(f"Message:->{event_data['resultInformation']['message']}")
elif event['type'] == "Microsoft.Communication.CallDisconnected":
pass
return Response(status=200)

New endpoint to trigger call transfer

@app.route('/api/triggerTransfer', methods=['POST'])
async def trigger_transfer():
global call_connection_id,callback_uri,transfer_call_agent
if not call_connection_id:
app.logger.error("No active call connection to transfer.")
return Response("No active call to transfer.", status=400)

try:
    # Extract target from request or use a default agent phone number
    request_data = await request.get_json()
    agent_phone_number = "+61417125566"  # Replace as needed
    transfer_destination =PhoneNumberIdentifier(agent_phone_number)
    call_connection_client = acs_client.get_call_connection(call_connection_id)
    # await acs_client.get_call_connection(call_connection_id).stop_media_streaming(operation_context="stopMediaStreamingContext")
    caller_id_number = PhoneNumberIdentifier("+61721422160")
    transfer_call_agent =True
    # await call_connection_client.add_participant(
    #     target_participant=transfer_destination,source_caller_id_number=caller_id_number
    # )
    # Transfer the call to the agent
    # Check the ACS Call Automation SDK for exact parameters/method calls for call transfer
    # transfer_result = await acs_client.transfer_call(
    #     call_connection_id=call_connection_id,
    #     target=agent_phone_number
    # )

    app.logger.info(f"Call successfully transferred to agent:")
    return Response("Call transfer initiated.", status=200)
except Exception as e:
    app.logger.error(f"Failed to transfer call: {e}")
    return Response("Failed to transfer call.", status=500)

WebSocket.

@app.websocket('/ws')
async def ws():
global transfer_call_agent
print("Client connected to WebSocket")
await init_websocket(websocket)
if not transfer_call_agent:
await start_conversation()
# else:
# pass
while True:
try:
# Receive data from the client
data = await websocket.receive()
await process_websocket_message_async(data)
except Exception as e:
print(f"WebSocket connection closed: {e}")
break
else:
pass

@app.route('/')
def home():
return 'Hello ACS CallAutomation!'

if name == 'main':
app.logger.setLevel(INFO)
app.run(port=8080)
-------------------- this is my azureopenaiservice.py-------------------
import asyncio
import json
from rtclient import (
RTLowLevelClient,
SessionUpdateMessage,
ServerVAD,
SessionUpdateParams,
InputAudioBufferAppendMessage,
InputAudioTranscription,
)
from azure.core.credentials import AzureKeyCredential
import aiohttp
active_websocket = None
answer_prompt_system_template = "You are an AI assistant that helps people find information."
AZURE_OPENAI_SERVICE_ENDPOINT = ""
AZURE_OPENAI_SERVICE_KEY = ""
AZURE_OPENAI_DEPLOYMENT_MODEL_NAME = "gpt-4o-realtime-preview"
KEYWORDS = ['agent', 'help', 'support', 'transfer', 'operator']
TRIGGER_TRANSFER_URL = "https://gddh5wtf.inc1.devtunnels.ms:8080/api/triggerTransfer"
transferred_call = False
client= None
async def start_conversation():
global client
client = RTLowLevelClient(url=AZURE_OPENAI_SERVICE_ENDPOINT, key_credential=AzureKeyCredential(AZURE_OPENAI_SERVICE_KEY), azure_deployment=AZURE_OPENAI_DEPLOYMENT_MODEL_NAME)
await client.connect()
await client.send(
SessionUpdateMessage(
session=SessionUpdateParams(
instructions=answer_prompt_system_template,
turn_detection=ServerVAD(type="server_vad",silence_duration_ms=1000),
voice= 'shimmer',
input_audio_format='pcm16',
output_audio_format='pcm16',
input_audio_transcription=InputAudioTranscription(model="whisper-1")
)
)
)

asyncio.create_task(receive_messages(client))

async def send_audio_to_external_ai(audioData: str):
await client.send(message=InputAudioBufferAppendMessage(type="input_audio_buffer.append", audio=audioData, _is_azure=True))

async def receive_messages(client: RTLowLevelClient):
while not client.closed:
message = await client.recv()
global transferred_call
if message is None :
continue
if transferred_call:
print("Transfer has been initiated. Stopping message processing.")
break # Exit the loop to stop processing further messages
match message.type:
case "session.created":
print("Session Created Message")
print(f" Session Id: {message.session.id}")
pass
case "error":
print(f" Error: {message.error}")
pass
case "input_audio_buffer.cleared":
print("Input Audio Buffer Cleared Message")
pass
case "input_audio_buffer.speech_started":
print(f"Voice activity detection started at {message.audio_start_ms} [ms]")
await stop_audio()
pass
case "input_audio_buffer.speech_stopped":
pass
case "conversation.item.input_audio_transcription.completed":
print(f" User:-- {message.transcript}")
transcript = message.transcript.lower()
detected_keywords = [kw for kw in KEYWORDS if kw in transcript]
if detected_keywords:
print(f"Detected keywords: {detected_keywords}. Triggering call transfer.")
await trigger_call_transfer()
transferred_call = True
break
case "conversation.item.input_audio_transcription.failed":
print(f" Error: {message.error}")
case "response.done":
print("Response Done Message")
print(f" Response Id: {message.response.id}")
if message.response.status_details:
print(f" Status Details: {message.response.status_details.model_dump_json()}")
case "response.audio_transcript.done":
print(f" AI:-- {message.transcript}")
case "response.audio.delta":
await receive_audio_for_outbound(message.delta)
pass
case _:
pass
print("Exiting receive_messages coroutine.")
# Optionally, perform cleanup if needed
await client.close()
# asyncio.sleep(2.0)

async def trigger_call_transfer():
global active_websocket,client
# Make an HTTP call back to main.py's /api/triggerTransfer endpoint
# Optionally, you can send JSON data, such as the agent's phone number
data = {"agentNumber": "+1234567890"} # Adjust as needed
async with aiohttp.ClientSession() as session:
async with session.post(TRIGGER_TRANSFER_URL, json=data) as resp:
if resp.status == 200:
print("Call transfer initiated successfully.")
else:
print(f"Failed to initiate call transfer. Status: {resp.status}, Response: {await resp.text()}")
async def init_websocket(socket):
global active_websocket
active_websocket = socket

async def receive_audio_for_outbound(data):
try:
data = {
"Kind": "AudioData",
"AudioData": {
"Data": data
},
"StopAudio": None
}

    # Serialize the server streaming data
    serialized_data = json.dumps(data)
    await send_message(serialized_data)
    
except Exception as e:
    print(e)

async def stop_audio():
stop_audio_data = {
"Kind": "StopAudio",
"AudioData": None,
"StopAudio": {}
}

    json_data = json.dumps(stop_audio_data)
    await send_message(json_data)

async def send_message(message: str):
global active_websocket
try:
await active_websocket.send(message)
except Exception as e:
print(f"Failed to send message: {e}") ----the mediastreaminghandler.py is same------