Azure Communication Services

1
from azure.communication.callautomation import (
2
    CallAutomationClient, MediaStreamingOptions, StreamingTransportType,
3
    MediaStreamingContentType, MediaStreamingAudioChannelType, AudioFormat,
4
)
5
6
@app.route("/api/incomingCall", methods=["POST"])
7
def incoming_call():
8
    for event in request.get_json():
9
        # Event Grid subscription validation handshake
10
        if event.get("eventType") == "Microsoft.EventGrid.SubscriptionValidationEvent":
11
            return jsonify({"validationResponse": event["data"]["validationCode"]})
12
13
        if event.get("eventType") == "Microsoft.Communication.IncomingCall":
14
            client = CallAutomationClient.from_connection_string(ACS_CONNECTION_STRING)
15
            client.answer_call(
16
                incoming_call_context=event["data"]["incomingCallContext"],
17
                callback_url=f"https://{HOST}/api/callbacks",
18
                media_streaming=MediaStreamingOptions(
19
                    transport_url=f"wss://{HOST}/ws",
20
                    transport_type=StreamingTransportType.WEBSOCKET,
21
                    content_type=MediaStreamingContentType.AUDIO,
22
                    audio_channel_type=MediaStreamingAudioChannelType.MIXED,
23
                    start_media_streaming=True,
24
                    enable_bidirectional=True,
25
                    audio_format=AudioFormat.PCM16_K_MONO,
26
                ),
27
            )
28
    return jsonify({"status": "ok"})

1
import asyncio, json, os, queue, threading, websockets
2
from flask_sock import Sock
3
4
sock = Sock(app)
5
AGENT_ID = os.environ["ELEVENLABS_AGENT_ID"]
6
# US default; data residency: wss://api.eu.residency.elevenlabs.io, .in., or .sg.
7
EL_ORIGIN = os.environ.get("ELEVENLABS_ORIGIN", "wss://api.elevenlabs.io")
8
EL_WS = f"{EL_ORIGIN}/v1/convai/conversation?agent_id={AGENT_ID}"
9
10
@sock.route("/ws")
11
def media_stream(ws):
12
    loop = asyncio.new_event_loop()
13
    el = {"ws": None}
14
    to_acs = queue.Queue()  # outbound frames; only this handler thread touches `ws`
15
16
    async def el_session():
17
        async with websockets.connect(EL_WS) as elws:
18
            el["ws"] = elws
19
            await elws.send(json.dumps({"type": "conversation_initiation_client_data"}))
20
            async for msg in elws:
21
                data = json.loads(msg)
22
                kind = data.get("type")
23
                if kind == "audio":  # agent audio -> caller
24
                    b64 = data["audio_event"]["audio_base_64"]
25
                    to_acs.put({"Kind": "AudioData", "AudioData": {"Data": b64}, "StopAudio": None})
26
                elif kind == "ping":
27
                    await elws.send(json.dumps({"type": "pong", "event_id": data["ping_event"]["event_id"]}))
28
                elif kind == "interruption":  # barge-in
29
                    to_acs.put({"Kind": "StopAudio", "AudioData": None, "StopAudio": {}})
30
31
    threading.Thread(target=lambda: loop.run_until_complete(el_session()), daemon=True).start()
32
33
    # Keep all ACS-socket I/O on this one thread: receive with a short timeout,
34
    # then drain any audio the ElevenLabs thread queued. Sending from the other
35
    # thread would race flask-sock and corrupt the stream.
36
    try:
37
        while True:
38
            raw = ws.receive(timeout=0.02)  # None when no frame arrived this tick
39
            if raw:
40
                evt = json.loads(raw)
41
                if evt.get("kind") == "AudioData" and el["ws"]:  # caller audio -> agent
42
                    asyncio.run_coroutine_threadsafe(
43
                        el["ws"].send(json.dumps({"user_audio_chunk": evt["audioData"]["data"]})), loop)
44
            while not to_acs.empty():
45
                ws.send(json.dumps(to_acs.get_nowait()))
46
    except Exception:
47
        pass  # ACS socket closed