Azure Communication Services

Panoramica

Questo approccio assegna al tuo agente un numero di telefono. Un chiamante lo compone, Azure Communication Services (ACS) risponde con streaming multimediale bidirezionale e un piccolo bridge inoltra l’audio PCM tra ACS e l’agente ElevenLabs usando il protocollo WebSocket dell’agente standard. È il modello di contact center/IVR, analogo a un’implementazione con SIP trunking, con ACS come operatore.

Si connette inoltre a Teams in due modi: un utente Teams con un Calling Plan può chiamare direttamente il numero ACS, oppure puoi esporre il numero tramite Teams Phone Extensibility affinché le chiamate a un account risorsa Teams vengano instradate ad ACS.

Come funziona

L’audio è PCM 16 kHz mono su entrambi i lati (il formato di input/output dell’agente è pcm_16000), quindi viene trasferito come base64 senza ricampionamento.

Il bridge espone queste route:

Requisiti

1. Una sottoscrizione Azure a pagamento (MCA / EA / Pay-As-You-Go): le sottoscrizioni gratuite, di prova o sponsorizzate non possono acquistare numeri.
2. Una risorsa Azure Communication Services.
3. Un host HTTPS per il bridge con un WebSocket pubblico (Azure Container Apps, App Service o una VM).
4. Un agente ElevenLabs impostato su PCM 16000 Hz su entrambi i lati: formato di output TTS nella scheda Voce, formato audio di input dell’utente nella scheda Avanzate.

Autorizzazioni e ruoli

Passaggio 1 — Esegui il provisioning della risorsa ACS e del numero

RG=my-rg
# Register providers (once)
az provider register -n Microsoft.Communication --wait
az provider register -n Microsoft.EventGrid --wait
# Create the ACS resource
az communication create --name my-acs --resource-group $RG \
  --location global --data-location unitedstates

Acquista un numero nella risorsa (Portale → risorsa ACS → Numeri di telefono → Ottieni, oppure usa l’SDK phone-numbers). Per un agente che risponde alle chiamate, è sufficiente un numero con chiamate in entrata; aggiungi la funzionalità in uscita se vuoi usare anche /api/outboundCall.

Per verificare dalla CLI (richiede az extension add --name communication) e recuperare la stringa di connessione che il bridge usa come ACS_CONNECTION_STRING:

CONN=$(az communication list-key -n my-acs -g $RG --query primaryConnectionString -o tsv)
az communication phonenumber list --connection-string "$CONN" --query "[].phoneNumber"

Passaggio 2 — Distribuisci il bridge

Il bridge è una piccola app Flask + flask-sock che usa azure-communication-callautomation. Ecco il nucleo del flusso in entrata:

from azure.communication.callautomation import (
    CallAutomationClient, MediaStreamingOptions, StreamingTransportType,
    MediaStreamingContentType, MediaStreamingAudioChannelType, AudioFormat,
)
@app.route("/api/incomingCall", methods=["POST"])
def incoming_call():
    for event in request.get_json():
        # Event Grid subscription validation handshake
        if event.get("eventType") == "Microsoft.EventGrid.SubscriptionValidationEvent":
            return jsonify({"validationResponse": event["data"]["validationCode"]})
        if event.get("eventType") == "Microsoft.Communication.IncomingCall":
            client = CallAutomationClient.from_connection_string(ACS_CONNECTION_STRING)
            client.answer_call(
                incoming_call_context=event["data"]["incomingCallContext"],
                callback_url=f"https://{HOST}/api/callbacks",
                media_streaming=MediaStreamingOptions(
                    transport_url=f"wss://{HOST}/ws",
                    transport_type=StreamingTransportType.WEBSOCKET,
                    content_type=MediaStreamingContentType.AUDIO,
                    audio_channel_type=MediaStreamingAudioChannelType.MIXED,
                    start_media_streaming=True,
                    enable_bidirectional=True,
                    audio_format=AudioFormat.PCM16_K_MONO,
                ),
            )
    return jsonify({"status": "ok"})

Nel socket /ws, inoltra PCM16 in entrambe le direzioni: invia i frame ACS AudioData a ElevenLabs come {"user_audio_chunk": "<base64>"} e rimanda l’audio dell’agente come {"Kind":"AudioData","AudioData":{"Data":"<base64>"},"StopAudio":null}. Il primo frame inviato da ACS è AudioMetadata (il formato negoziato): registralo e ignoralo. Il lato ElevenLabs usa il protocollo WebSocket dell’agente standard.

import asyncio, json, os, queue, threading, websockets
from flask_sock import Sock
sock = Sock(app)
AGENT_ID = os.environ["ELEVENLABS_AGENT_ID"]
# US default; data residency: wss://api.eu.residency.elevenlabs.io, .in., or .sg.
EL_ORIGIN = os.environ.get("ELEVENLABS_ORIGIN", "wss://api.elevenlabs.io")
EL_WS = f"{EL_ORIGIN}/v1/convai/conversation?agent_id={AGENT_ID}"
@sock.route("/ws")
def media_stream(ws):
    loop = asyncio.new_event_loop()
    el = {"ws": None}
    to_acs = queue.Queue()  # outbound frames; only this handler thread touches `ws`
    async def el_session():
        async with websockets.connect(EL_WS) as elws:
            el["ws"] = elws
            await elws.send(json.dumps({"type": "conversation_initiation_client_data"}))
            async for msg in elws:
                data = json.loads(msg)
                kind = data.get("type")
                if kind == "audio":  # agent audio -> caller
                    b64 = data["audio_event"]["audio_base_64"]
                    to_acs.put({"Kind": "AudioData", "AudioData": {"Data": b64}, "StopAudio": None})
                elif kind == "ping":
                    await elws.send(json.dumps({"type": "pong", "event_id": data["ping_event"]["event_id"]}))
                elif kind == "interruption":  # barge-in
                    to_acs.put({"Kind": "StopAudio", "AudioData": None, "StopAudio": {}})
    threading.Thread(target=lambda: loop.run_until_complete(el_session()), daemon=True).start()
    # Keep all ACS-socket I/O on this one thread: receive with a short timeout,
    # then drain any audio the ElevenLabs thread queued. Sending from the other
    # thread would race flask-sock and corrupt the stream.
    try:
        while True:
            raw = ws.receive(timeout=0.02)  # None when no frame arrived this tick
            if raw:
                evt = json.loads(raw)
                if evt.get("kind") == "AudioData" and el["ws"]:  # caller audio -> agent
                    asyncio.run_coroutine_threadsafe(
                        el["ws"].send(json.dumps({"user_audio_chunk": evt["audioData"]["data"]})), loop)
            while not to_acs.empty():
                ws.send(json.dumps(to_acs.get_nowait()))
    except Exception:
        pass  # ACS socket closed

Distribuisci su Azure Container Apps e acquisisci l’FQDN pubblico:

az containerapp up --name acs-el-bridge --resource-group $RG \
  --source . --ingress external --target-port 8080 \
  --env-vars ELEVENLABS_AGENT_ID=$AGENT_ID \
    ELEVENLABS_ORIGIN=wss://api.elevenlabs.io
FQDN=$(az containerapp show -n acs-el-bridge -g $RG \
  --query properties.configuration.ingress.fqdn -o tsv)

Quindi imposta BRIDGE_PUBLIC_HOST=$FQDN e la stringa di connessione ACS (come secret) nell’app.

Passaggio 3 — Instrada IncomingCall al bridge

Crea una sottoscrizione Event Grid sulla risorsa ACS che invii IncomingCall al bridge. L’handshake di convalida del bridge (sopra) completa automaticamente la sottoscrizione.

ACS_ID=$(az communication show -n my-acs -g $RG --query id -o tsv)
az eventgrid event-subscription create \
  --name acs-incomingcall \
  --source-resource-id "$ACS_ID" \
  --endpoint "https://$FQDN/api/incomingCall" \
  --endpoint-type webhook \
  --included-event-types Microsoft.Communication.IncomingCall
# Verify — should print "Succeeded"
az eventgrid event-subscription show --name acs-incomingcall \
  --source-resource-id "$ACS_ID" --query provisioningState -o tsv

La sottoscrizione appare nel pannello Events della risorsa ACS:

Chiama il numero: risponderà l’agente.

Collegamento a Teams

• Chiamata diretta: un utente Teams con Teams Phone + un Calling Plan può chiamare il numero ACS come qualsiasi numero esterno.
• Account risorsa Teams (TPE): collega un account risorsa Teams alla risorsa ACS con Teams Phone Extensibility, affinché le chiamate all’account risorsa attivino lo stesso flusso IncomingCall → bridge.

Fine della chiamata

Quando l’agente termina la conversazione (ad esempio con lo strumento End Call), ElevenLabs chiude il WebSocket. Riaggancia il lato ACS affinché il chiamante non resti su una linea inattiva:

CallAutomationClient.from_connection_string(ACS_CONNECTION_STRING) \
    .get_call_connection(call_connection_id).hang_up(is_for_everyone=True)

Trasferimento assistito a un operatore

Gli strumenti di trasferimento nativi di ElevenLabs si applicano solo quando ElevenLabs gestisce la telefonia. In questo caso, quindi, l’agente attiva uno strumento client personalizzato (ad esempio transfer_to_human) che il bridge gestisce aggiungendo l’operatore alla chiamata in corso con add_participant (trasferimento assistito), anziché effettuare un trasferimento cieco:

conn = client.get_call_connection(call_connection_id)
conn.add_participant(
    PhoneNumberIdentifier(human_number),
    source_caller_id_number=PhoneNumberIdentifier(your_outbound_number),
    invitation_timeout=30,
)
# then mute the bot and skip the end-of-call hangup so the human's leg survives

ACS invia i callback AddParticipantSucceeded / AddParticipantFailed a /api/callbacks. Restituisci un client_tool_result all’agente, così potrà pronunciare il messaggio di passaggio. Consulta gli strumenti di sistema per la configurazione lato agente.

Risoluzione dei problemi

Link utili

• Panoramica di ACS Call Automation
• Streaming audio ACS
• Azioni di controllo chiamata (trasferimento / aggiunta partecipante)
• Teams Phone Extensibility
• Protocollo WebSocket dell’agente
• SIP trunking di ElevenLabs