Azure Communication Services

Überblick

Mit diesem Ansatz erhält Ihr Agent eine Telefonnummer. Ein Anrufer wählt sie, Azure Communication Services (ACS) nimmt den Anruf mit bidirektionalem Media Streaming an, und eine kleine Bridge leitet PCM-Audio zwischen ACS und dem ElevenLabs-Agenten über das standardmäßige Agent-WebSocket-Protokoll weiter. Dies entspricht dem Contact-Center-/IVR-Muster – derselben Struktur wie bei einer SIP-Trunking-Bereitstellung, wobei ACS der Carrier ist.

Außerdem verbindet es sich auf zwei Arten mit Teams: Ein Teams-Nutzer mit Calling Plan kann die ACS-Nummer direkt wählen, oder Sie schalten Teams Phone Extensibility vor die Nummer, sodass Anrufe an ein Teams-Ressourcenkonto in ACS weitergeleitet werden.

So funktioniert es

Audio ist auf beiden Seiten PCM 16 kHz Mono (das Ein-/Ausgabeformat des Agenten ist pcm_16000) und wird daher ohne Resampling als base64 durchgereicht.

Die Bridge stellt diese Routen bereit:

Voraussetzungen

1. Ein kostenpflichtiges Azure-Abonnement (MCA / EA / Pay-As-You-Go) – kostenlose Test- oder Sponsoring-Abonnements können keine Nummern kaufen.
2. Eine Azure Communication Services-Ressource.
3. Einen HTTPS-Host für die Bridge mit einem öffentlichen WebSocket (Azure Container Apps, App Service oder eine VM).
4. Einen ElevenLabs-Agenten, der auf beiden Seiten auf PCM 16000 Hz eingestellt ist: TTS-Ausgabeformat im Tab Voice, Audioeingabeformat des Nutzers im Tab Advanced.

Berechtigungen und Rollen

Schritt 1 – ACS-Ressource und Nummer bereitstellen

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

Kaufen Sie in der Ressource eine Nummer (Portal → Ihre ACS-Ressource → Phone numbers → Get oder über das phone-numbers SDK). Für einen Agenten, der Anrufe annimmt, genügt eine Nummer mit eingehenden Anrufen. Fügen Sie die Fähigkeit ausgehend hinzu, wenn Sie auch /api/outboundCall verwenden möchten.

Zur Überprüfung über die CLI (erfordert az extension add --name communication) und zum Abrufen der Verbindungszeichenfolge, die die Bridge als ACS_CONNECTION_STRING verwendet:

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

Schritt 2 – Bridge bereitstellen

Die Bridge ist eine kleine Flask- + flask-sock-App mit azure-communication-callautomation. Der Kern des eingehenden Ablaufs:

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"})

Leiten Sie auf dem Socket /ws PCM16 in beide Richtungen weiter: Leiten Sie ACS-AudioData-Frames als {"user_audio_chunk": "<base64>"} an ElevenLabs weiter und senden Sie das Audio des Agenten als {"Kind":"AudioData","AudioData":{"Data":"<base64>"},"StopAudio":null} zurück. Der erste von ACS gesendete Frame ist AudioMetadata (das ausgehandelte Format) – protokollieren und ignorieren Sie ihn. Die ElevenLabs-Seite nutzt das standardmäßige Agent-WebSocket-Protokoll.

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

Stellen Sie die Anwendung in Azure Container Apps bereit und erfassen Sie den öffentlichen FQDN:

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)

Setzen Sie anschließend BRIDGE_PUBLIC_HOST=$FQDN und die ACS-Verbindungszeichenfolge als Secret in der App.

Schritt 3 – IncomingCall zur Bridge weiterleiten

Erstellen Sie für die ACS-Ressource ein Event-Grid-Abonnement, das IncomingCall an die Bridge sendet. Der oben gezeigte Validierungs-Handshake der Bridge schließt das Abonnement automatisch ab.

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

Das Abonnement wird in der Ansicht Events der ACS-Ressource angezeigt:

Wählen Sie die Nummer – der Agent nimmt den Anruf an.

Mit Teams verbinden

• Direktwahl: Ein Teams-Nutzer mit Teams Phone + Calling Plan kann die ACS-Nummer wie jede externe Nummer wählen.
• Teams-Ressourcenkonto (TPE): Binden Sie ein Teams-Ressourcenkonto mit Teams Phone Extensibility an die ACS-Ressource, damit Anrufe an das Ressourcenkonto denselben Ablauf IncomingCall → Bridge auslösen.

Anrufende

Wenn der Agent die Unterhaltung beendet (z. B. über sein Tool End Call), schließt ElevenLabs den WebSocket. Legen Sie die ACS-Verbindung auf, damit der Anrufer nicht in einer toten Leitung bleibt:

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

Warme Übergabe an einen Menschen

Die nativen Übergabe-Tools von ElevenLabs gelten nur, wenn ElevenLabs die Telefonie bereitstellt. Hier löst der Agent daher ein benutzerdefiniertes Client-Tool aus (z. B. transfer_to_human), das Ihre Bridge verarbeitet, indem sie den Menschen mit add_participant zum aktiven Anruf hinzufügt (warm), statt den Anruf blind weiterzuleiten:

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 sendet AddParticipantSucceeded- / AddParticipantFailed-Callbacks an /api/callbacks. Geben Sie dem Agenten ein client_tool_result zurück, damit er seine Übergabezeile sagen kann. Die Konfiguration auf Agentenseite finden Sie unter System-Tools.

Fehlerbehebung

Nützliche Links

• ACS Call Automation – Überblick
• ACS-Audio-Streaming
• Anrufsteuerungsaktionen (Übertragen / Teilnehmer hinzufügen)
• Teams Phone Extensibility
• Agent-WebSocket-Protokoll
• ElevenLabs SIP-Trunking