Azure Communication Services

Visão geral

Essa abordagem dá ao seu agente um número de telefone. Quem liga disca para ele, o Azure Communication Services (ACS) atende com streaming de mídia bidirecional, e uma pequena ponte retransmite o áudio PCM entre o ACS e o agente da ElevenLabs usando o protocolo WebSocket do agente padrão. É o padrão de central de atendimento / URA — a mesma estrutura de uma implantação de tronco SIP, com o ACS como operadora.

Ele também se conecta ao Teams de duas formas: um usuário do Teams com um Plano de Chamadas pode discar diretamente para o número do ACS, ou você pode colocar Teams Phone Extensibility na frente do número para que chamadas a uma conta de recurso do Teams sejam roteadas para o ACS.

Como funciona

O áudio é PCM mono de 16 kHz em ambas as pontas (o formato de entrada/saída do agente é pcm_16000), então ele é transmitido como base64 sem reamostragem.

A ponte expõe estas rotas:

Requisitos

1. Uma assinatura paga do Azure (MCA / EA / Pay-As-You-Go) — assinaturas gratuitas/de avaliação/patrocinadas não podem comprar números.
2. Um recurso do Azure Communication Services.
3. Um host HTTPS para a ponte com um WebSocket público (Azure Container Apps, App Service ou uma VM).
4. Um agente da ElevenLabs configurado para PCM 16000 Hz em ambas as pontas: formato de saída TTS na aba Voice e formato de áudio de entrada do usuário na aba Advanced.

Permissões e funções

Etapa 1 — Provisionar o recurso ACS e o número

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

Compre um número no recurso (Portal → seu recurso ACS → Phone numbers → Get, ou o SDK de phone-numbers). Para um agente que atende chamadas, basta um número com chamadas recebidas; adicione a capacidade de saída se também quiser usar /api/outboundCall.

Para verificar pela CLI (requer az extension add --name communication) e buscar a string de conexão que a ponte usa como 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"

Etapa 2 — Implantar a ponte

A ponte é um pequeno app Flask + flask-sock que usa azure-communication-callautomation. O núcleo do fluxo de entrada:

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

No socket /ws, retransmita PCM16 nas duas direções: encaminhe frames AudioData do ACS para a ElevenLabs como {"user_audio_chunk": "<base64>"} e envie o áudio do agente de volta como {"Kind":"AudioData","AudioData":{"Data":"<base64>"},"StopAudio":null}. O primeiro frame enviado pelo ACS é AudioMetadata (o formato negociado) — registre-o em log e ignore-o. O lado da ElevenLabs usa o protocolo WebSocket do agente padrão.

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

Implante no Azure Container Apps e capture o FQDN público:

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)

Em seguida, defina BRIDGE_PUBLIC_HOST=$FQDN e a string de conexão do ACS (como segredo) no app.

Etapa 3 — Rotear IncomingCall para a ponte

Crie uma assinatura do Event Grid no recurso ACS que envie IncomingCall para a ponte. O handshake de validação da ponte (acima) conclui a assinatura automaticamente.

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

A assinatura aparece na tela Events do recurso ACS:

Disque para o número — o agente atende.

Como conectar ao Teams

• Discagem direta: um usuário do Teams com Teams Phone + Plano de Chamadas pode discar para o número ACS como qualquer número externo.
• Conta de recurso do Teams (TPE): vincule uma conta de recurso do Teams ao recurso ACS com Teams Phone Extensibility para que chamadas à conta de recurso disparem o mesmo fluxo IncomingCall → ponte.

Fim da chamada

Quando o agente encerra a conversa (por exemplo, com a ferramenta End Call), a ElevenLabs fecha o WebSocket. Desligue a ponta do ACS para que quem ligou não fique em uma linha sem resposta:

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

Transferência assistida para uma pessoa

As ferramentas nativas de transferência da ElevenLabs só se aplicam quando a ElevenLabs controla a telefonia. Portanto, aqui o agente dispara uma ferramenta de cliente personalizada (por exemplo, transfer_to_human) que sua ponte processa adicionando a pessoa à chamada ativa com add_participant (assistida), em vez de uma transferência cega:

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

O ACS emite callbacks AddParticipantSucceeded / AddParticipantFailed para /api/callbacks. Retorne um client_tool_result ao agente para que ele possa dizer sua frase de transferência. Consulte ferramentas do sistema para ver a configuração do lado do agente.

Solução de problemas

Links úteis

• Visão geral do ACS Call Automation
• Streaming de áudio do ACS
• Ações de controle de chamadas (transferir / adicionar participante)
• Teams Phone Extensibility
• Protocolo WebSocket do agente
• Tronco SIP da ElevenLabs