AudioSocket no Asterisk: do zero ao stream de áudio em Python

Autor: Rafael Camargo | Pilar: AI Telephony | Stack: Asterisk 20 LTS (20.8.x) + Python 3.11+

---

TL;DR

• AudioSocket é a camada TCP binária do Asterisk para streaming de áudio bidirecional: setup em ~20 linhas de Python, latência mínima, zero overhead HTTP.
• Este tutorial cobre dialplan no Asterisk 20 LTS, servidor asyncio do zero, integração com Google STT pt-BR, e os 4 comportamentos que a documentação oficial omite.
• Use AudioSocket para protótipos e IVRs 1:1 com Voice AI; para fluxos com transferência ou mais de 200 chamadas simultâneas, avalie ARI ExternalMedia primeiro.

---

O que é AudioSocket e por que importa agora

O AudioSocket (app_audiosocket) é um módulo do Asterisk que estabelece uma conexão TCP entre o servidor de telefonia e uma aplicação externa, passando áudio bruto nos dois sentidos. Sem AGI, sem AMI, sem ARI — apenas TCP binário com frames de 20ms de PCM.

A razão de isso ser relevante em 2026: a maioria dos frameworks de Voice AI que conectam LLMs a chamadas telefônicas reais — Pipecat, LiveKit Agents, integrações custom com o OpenAI Realtime — precisam de um ponto de entrada de áudio bruto com latência mínima. O AudioSocket é exatamente esse ponto. Uma conexão TCP, frames constantes a 8kHz, zero overhead de protocolo HTTP.

A documentação oficial do Asterisk (wiki.asterisk.org/wiki/display/AST/AudioSocket, acesso 2026-05-12) cobre o caso básico. O que ela não cobre: o comportamento quando sua aplicação Python cai no meio de uma chamada, por que você vai perder frames silenciosamente se não dimensionar o servidor corretamente, e como integrar o stream com um STT em tempo real sem quebrar o event loop.

Este tutorial cobre os três.

---

A anatomia do protocolo AudioSocket

O protocolo AudioSocket foi especificado originalmente por Leland Lucius para a CyCoreSystems (github.com/CyCoreSystems/audiosocket, acesso 2026-05-12). O formato de cada mensagem tem três campos:

+----------+------------------+---------------------+
| Tipo     | Comprimento      | Payload             |
| (1 byte) | (2 bytes, BE)    | (variável)          |
+----------+------------------+---------------------+

Os tipos definidos pela especificação:

Tipo	Hex	Direção	Payload
HANGUP	0x00	Asterisk → App	vazio
UUID	0x01	Asterisk → App	36 bytes, UUID da chamada
SLIN	0x10	Bidirecional	320 bytes de PCM 16-bit signed, 8kHz
ERROR	0xFF	Asterisk → App	código de erro

O primeiro pacote que você recebe sempre é o UUID (tipo 0x01). Se o primeiro pacote não for um UUID, a conexão não veio do Asterisk ou o módulo está mal configurado.

Cada frame SLIN tem exatamente 320 bytes: 160 amostras de 16-bit a 8kHz = 20ms de áudio. O Asterisk envia frames independentemente de haver fala ativa — silêncio são zeros. A cadência é constante enquanto a chamada está ativa.

O que a documentação não menciona: não há controle de fluxo. Se sua aplicação demorar mais de aproximadamente 200ms para consumir o socket, o Asterisk começa a descartar frames sem aviso (estimativa prática do autor; não há threshold documentado na spec do módulo app_audiosocket). O único buffer é o SO_RCVBUF do socket TCP.

---

Configurando o Asterisk

Verificando se o módulo está carregado

asterisk -rx "module show like audiosocket"

Saída esperada:

app_audiosocket.so         AudioSocket Application      Running

Se o módulo não aparecer, adicione ao /etc/asterisk/modules.conf:

[modules]
load = app_audiosocket.so

Reinicie o Asterisk ou carregue via CLI:

asterisk -rx "module load app_audiosocket.so"

Dialplan básico

No extensions.conf:

[from-pstn]
exten => s,1,NoOp(Chamada recebida - ${CALLERID(num)})
 same => n,Answer()
 same => n,TryExec(AudioSocket(${UNIQUEID},127.0.0.1:9999))
 same => n,GotoIf($["${TRYSTATUS}" = "FAILED"]?fallback,s,1)
 same => n,Hangup()

[fallback]
exten => s,1,Playback(sorry-service-unavailable)
 same => n,Hangup()

Três pontos sobre este dialplan:

Ponto 1: o Asterisk se conecta ativamente ao endereço que você fornece. Não é o inverso. Sua aplicação Python precisa estar ouvindo antes de qualquer chamada chegar.

Ponto 2: Answer() antes de AudioSocket() é obrigatório. O módulo app_audiosocket não atende a chamada automaticamente — sem Answer(), o canal não é estabelecido e o stream não começa.

Ponto 3: app_audiosocket não define uma variável de status no Asterisk 20 LTS. Para detectar falha de conexão (servidor Python offline), use TryExec(): ele captura o retorno -1 do AudioSocket() e define TRYSTATUS=FAILED em vez de desligar o canal imediatamente. Sem TryExec, uma queda da sua aplicação Python resulta no desligamento abrupto da chamada — sem mensagem ao usuário.

Configuração de codec para evitar transcoding

O AudioSocket sempre trabalha com SLIN (PCM linear 16-bit, 8kHz). Se o endpoint SIP negocia G.711 ou G.722, o Asterisk faz transcoding internamente. Em volume alto, esse transcoding custa CPU.

Para endpoints que você controla, configure para aceitar SLIN diretamente:

[endpoint-ai]
type=endpoint
; outras configurações do endpoint
allow=!all,slin

Para SIP trunks de provedores brasileiros (Brava, Voxbeam, Vono), o trunk provavelmente negocia G.711 µ-law — você não tem controle sobre o codec do lado do provedor. O transcoding vai acontecer. Dimensione o Asterisk levando isso em conta.

---

Servidor Python do zero

Por que asyncio e não threads

Com threads Python e o GIL, o throughput cai quando você tem muitas chamadas simultâneas e processamento de áudio concorrente. Com asyncio, o event loop gerencia toda a concorrência no mesmo thread Python — o gargalo real em Voice AI é I/O (envio para STT via rede), não CPU de processamento de bytes.

Uma exceção: se você rodar inferência local (VAD, STT on-premise), o processamento é CPU-bound e vai bloquear o event loop. Nesse caso, use asyncio.get_event_loop().run_in_executor() com um ThreadPoolExecutor ou ProcessPoolExecutor.

Servidor base

import asyncio
import struct
import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("audiosocket")

HOST = "0.0.0.0"
PORT = 9999

TYPE_HANGUP = 0x00
TYPE_UUID   = 0x01
TYPE_SLIN   = 0x10
TYPE_ERROR  = 0xFF

FRAME_BYTES = 320  # 160 amostras × 2 bytes = 20ms @ 8kHz


async def read_packet(reader: asyncio.StreamReader) -> tuple[int, bytes]:
    """Lê um pacote AudioSocket completo do stream."""
    header = await reader.readexactly(3)
    ptype = header[0]
    plen = struct.unpack(">H", header[1:3])[0]
    payload = await reader.readexactly(plen) if plen > 0 else b""
    return ptype, payload


async def handle_call(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    peer = writer.get_extra_info("peername")
    call_uuid = None

    try:
        # Primeiro pacote obrigatoriamente UUID
        ptype, payload = await asyncio.wait_for(read_packet(reader), timeout=5.0)
        if ptype != TYPE_UUID:
            logger.error(f"{peer}: esperado UUID (0x01), recebido 0x{ptype:02x}")
            return

        call_uuid = payload.decode("ascii", errors="replace")
        logger.info(f"Chamada iniciada UUID={call_uuid} peer={peer}")

        while True:
            ptype, payload = await read_packet(reader)

            if ptype == TYPE_HANGUP:
                logger.info(f"Hangup recebido UUID={call_uuid}")
                break
            elif ptype == TYPE_ERROR:
                logger.error(f"Erro AudioSocket UUID={call_uuid} payload={payload.hex()}")
                break
            elif ptype == TYPE_SLIN:
                await process_audio(call_uuid, payload)

    except asyncio.IncompleteReadError:
        # Asterisk fechou o socket sem enviar HANGUP — tratar como encerramento normal
        logger.info(f"Conexão encerrada (sem HANGUP) UUID={call_uuid or peer}")
    except asyncio.TimeoutError:
        logger.error(f"Timeout aguardando UUID de {peer}")
    except Exception as exc:
        logger.exception(f"Erro inesperado UUID={call_uuid or peer}: {exc}")
    finally:
        writer.close()
        await writer.wait_closed()
        logger.info(f"Socket fechado UUID={call_uuid or peer}")


async def process_audio(call_uuid: str, pcm: bytes) -> None:
    """
    Processa um frame de 20ms de PCM 16-bit signed @ 8kHz.
    Aqui você envia para STT, acumula buffer, etc.
    """
    samples = struct.unpack(f"<{len(pcm) // 2}h", pcm)
    rms = (sum(s * s for s in samples) / len(samples)) ** 0.5
    if rms > 300:
        logger.debug(f"UUID={call_uuid} voz detectada RMS={rms:.0f}")
    # Nota: RMS é proxy de exemplo para atividade de voz.
    # Para VAD de produção, use webrtcvad (ver seção Próximos passos) —
    # a diferença de acurácia em ambientes com ruído de fundo é significativa.


async def main():
    server = await asyncio.start_server(handle_call, HOST, PORT)
    addrs = [s.getsockname() for s in server.sockets]
    logger.info(f"AudioSocket server ouvindo em {addrs}")
    async with server:
        await server.serve_forever()


if __name__ == "__main__":
    asyncio.run(main())

Este servidor aceita múltiplas conexões simultâneas (uma por chamada). Cada chamada é uma coroutine independente. O asyncio.wait_for(..., timeout=5.0) no UUID evita conexões TCP abertas que nunca enviam dados.

---

Os erros que a documentação não menciona

Frames descartados silenciosamente

O Asterisk não avisa quando descarta frames por socket cheio. Se o seu servidor Python travar por mais de ~200ms em qualquer operação síncrona dentro do loop de leitura, frames serão perdidos — e o único sintoma é áudio truncado no STT.

Como detectar: monitore RMS médio por chamada. Quedas abruptas para zero entre períodos de fala ativa indicam descarte de frames.

Como evitar: toda operação de I/O dentro do handle_call precisa ser await. Nunca use time.sleep(), requests.post() (síncrono), ou qualquer operação bloqueante dentro de uma coroutine sem run_in_executor.

Se precisar de mais buffer, aumente o SO_RCVBUF do socket do servidor:

import socket

async def main():
    server = await asyncio.start_server(
        handle_call, HOST, PORT,
        reuse_address=True,
    )
    for sock in server.sockets:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 262144)  # 256KB
    async with server:
        await server.serve_forever()

Asterisk fecha o socket sem enviar HANGUP

Em algumas builds do Asterisk 18.x, o módulo pode fechar o socket TCP sem enviar o pacote 0x00. O asyncio.StreamReader.readexactly() vai lançar asyncio.IncompleteReadError.

Trate IncompleteReadError como encerramento normal — o código acima já faz isso. Não trate como falha de comunicação ou você vai gerar alertas falsos.

CPU alta no Asterisk com múltiplas chamadas

Consumo elevado de CPU no processo asterisk com app_audiosocket sob carga concorrente é um problema documentado: a issue #234 do repositório oficial reportou o bug (“app AudioSocket High CPU Usage”), com correção via PR #270. Antes de investigar em produção, verifique se a sua build do Asterisk 20 LTS inclui o cherry-pick desse PR — confirme com asterisk -rx "core show version" e compare com o CHANGES do branch 20.

IPv6 no dialplan falha silenciosamente

Se o seu servidor Python escuta em ::1 (loopback IPv6) e você escreve AudioSocket(::1:9999,...) no dialplan, o Asterisk vai falhar ao conectar sem log de erro claro. Use sempre IPv4 no parâmetro do AudioSocket(), ou configure um alias IPv4 explícito no host.

---

Reconexão e tratamento de falhas

O AudioSocket não tem mecanismo de reconexão. Se sua aplicação Python cair durante uma chamada ativa, o Asterisk recebe ECONNREFUSED ou EPIPE e o AudioSocket() retorna com código de erro. Com o padrão TryExec do dialplan acima, TRYSTATUS será FAILED e a execução segue para o [fallback]. Do lado Python, a estratégia mais simples é um healthcheck HTTP separado que o seu supervisor de processo (systemd, supervisord) monitora:

from aiohttp import web

async def healthcheck(request):
    return web.Response(text="ok")

async def main():
    # servidor AudioSocket
    audio_server = await asyncio.start_server(handle_call, HOST, PORT)

    # healthcheck HTTP na porta 9998
    app = web.Application()
    app.router.add_get("/health", healthcheck)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "0.0.0.0", 9998)
    await site.start()

    logger.info("AudioSocket server pronto")
    async with audio_server:
        await audio_server.serve_forever()

O Asterisk não usa o healthcheck — mas seu load balancer ou sistema de monitoramento usa. Isso permite detectar uma queda do servidor Python antes que chamadas sejam roteadas para ele.

---

Integrando com STT em tempo real

O problema do buffer

Cada frame AudioSocket tem 20ms de áudio. Enviar cada frame individualmente para um STT em nuvem gera latência de rede alta (uma chamada HTTP por frame) e custo alto (muitas requisições pequenas).

O padrão é acumular frames até ter ~100–300ms de áudio antes de enviar. Com VAD (Voice Activity Detection), você envia apenas segmentos com voz ativa — em testes práticos do autor com chamadas de distribuição típica de silêncio (~50–60% do tempo sem fala ativa), a redução foi de ~60–70% no volume de áudio enviado para o STT.

Exemplo com Google Cloud Speech-to-Text streaming (pt-BR)

from google.cloud import speech_v1 as speech

async def handle_call_stt(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    client = speech.SpeechAsyncClient()

    config = speech.RecognitionConfig(
        encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
        sample_rate_hertz=8000,
        language_code="pt-BR",
        enable_automatic_punctuation=True,
    )
    streaming_config = speech.StreamingRecognitionConfig(
        config=config, interim_results=True
    )

    audio_queue: asyncio.Queue[bytes | None] = asyncio.Queue()

    async def audio_generator():
        yield speech.StreamingRecognizeRequest(streaming_config=streaming_config)
        buffer = b""
        while True:
            chunk = await audio_queue.get()
            if chunk is None:
                break
            buffer += chunk
            if len(buffer) >= 1600:  # ~100ms (5 frames)
                yield speech.StreamingRecognizeRequest(audio_content=buffer)
                buffer = b""
        if buffer:
            yield speech.StreamingRecognizeRequest(audio_content=buffer)

    async def print_transcripts():
        async for response in await client.streaming_recognize(
            requests=audio_generator()
        ):
            for result in response.results:
                alt = result.alternatives[0]
                status = "FINAL" if result.is_final else "parcial"
                logger.info(f"[{status}] {alt.transcript} (conf={alt.confidence:.2f})")

    # recebe UUID
    ptype, payload = await asyncio.wait_for(read_packet(reader), timeout=5.0)
    if ptype != TYPE_UUID:
        return
    call_uuid = payload.decode("ascii", errors="replace")

    transcript_task = asyncio.create_task(print_transcripts())

    try:
        while True:
            ptype, payload = await read_packet(reader)
            if ptype == TYPE_HANGUP:
                await audio_queue.put(None)
                break
            elif ptype == TYPE_SLIN:
                await audio_queue.put(payload)
    except asyncio.IncompleteReadError:
        await audio_queue.put(None)
    finally:
        await transcript_task
        writer.close()
        await writer.wait_closed()

Nota de compliance: se você transcreve e armazena o conteúdo de chamadas, a LGPD exige consentimento explícito do titular, definição de prazo de retenção, e mecanismo de anonimização ou exclusão sob demanda. Adicione o consentimento no fluxo de URA antes de ativar a transcrição. Esta nota é de caráter informativo geral; consulte assessoria jurídica para análise do seu caso específico.

Latência E2E estimada neste stack

Etapa	Latência típica
SIP signaling + RTP decode no Asterisk	5–15ms
Granularidade do frame AudioSocket	20ms
Acumulação de buffer (~100ms)	100ms
Google STT TTFB pt-BR — streaming (estimativa prática do autor; latências variam por região e carga da API)	200–400ms
Subtotal até primeira hipótese parcial	~325–535ms

Para não soar mecânico em chamada real, o threshold de voice-to-voice (fim da fala do usuário → início de resposta audível do agente) é 1.200ms no P95 — referência prática adotada em benchmarks de frameworks como Pipecat (github.com/pipecat-ai/pipecat) e baseada em estudos de usabilidade de sistemas de diálogo interativo. Neste stack, sobram ~665–875ms para LLM + TTS. Com modelos de resposta rápida (GPT-4o-mini, Claude Haiku) e TTS streaming, isso é viável.

---

AudioSocket vs ARI ExternalMedia: quando usar cada um

Esta é a comparação que qualquer artigo sobre AudioSocket precisa fazer.

Critério	AudioSocket	ARI ExternalMedia
Protocolo	TCP binário direto	HTTP + WebSocket
Complexidade de setup	Baixa	Alta
Latência adicional	Mínima	~50–150ms
Reconexão automática	Não	Sim
Controle do dialplan em tempo real	Limitado	Completo
Suporte a múltiplas pernas/bridge	Não	Sim
Versão mínima Asterisk	16.x	12.x (ARI)
Documentação disponível	Escassa	Extensa
Adequado para call center com transferências	Não	Sim
Adequado para IVR simples com IA	Sim	Sim, mas excessivo

AudioSocket é a escolha certa quando:
– Você está prototipando integração de Voice AI com Asterisk.
– O fluxo é 1:1, sem transferências, sem hold, sem conferência.
– Latência mínima é crítica e você não precisa controlar o dialplan dinamicamente.

ARI ExternalMedia é a escolha certa quando:
– Você precisa transferir a chamada para um agente humano de dentro da aplicação.
– O fluxo tem múltiplas pernas (chamada em conferência, gravação separada de pernas).
– Você já usa ARI para controle de chamadas e quer adicionar processamento de áudio.
– Você precisa de reconexão automática gerenciada pelo Asterisk.

Para a maioria dos deployments de Voice AI em produção com volume real e fluxos de call center, o ARI ExternalMedia acaba sendo a escolha mais adequada — mesmo com a complexidade adicional de setup. O AudioSocket é o caminho mais rápido para validar se sua integração de áudio com o Asterisk funciona antes de investir em ARI.

---

Quando não usar AudioSocket

1. Call centers com transferências ou filas

O AudioSocket() não tem como redirecionar a chamada para outra fila ou agente humano enquanto está ativo. Para transferir, você precisa encerrar o socket (e a aplicação retorna para o dialplan), o que interrompe o fluxo de áudio. Em call centers de produção, isso é inaceitável.

2. Mais de 200 chamadas simultâneas por instância Asterisk

Sob carga alta, o app_audiosocket adiciona overhead de CPU no Asterisk além do que o transcoding e RTP já consomem. Teste o limite na sua infraestrutura antes de assinar SLA. Com Asterisk rodando em VPS de 4 vCPUs, 100–150 chamadas simultâneas com transcoding G.711→SLIN é um teto razoável para começar a estressar (estimativa prática do autor em VPS de 4 vCPUs com Asterisk 20 LTS).

3. Áudio bidirecional com barge-in (interrupção do usuário)

O protocolo suporta envio de áudio de volta para o Asterisk (frames SLIN tipo 0x10 enviados pela aplicação). A parte difícil é sincronização: se o usuário interrompe o TTS, sua aplicação precisa parar de enviar áudio imediatamente. Controlar jitter e timing de barge-in via AudioSocket é manual. O ARI ExternalMedia tem abstrações melhores para isso via controle de bridge.

4. Rede instável entre Asterisk e aplicação Python

Sem controle de fluxo e sem reconexão automática, uma oscilação de rede derruba a chamada sem aviso. AudioSocket só é adequado quando Asterisk e sua aplicação estão na mesma LAN ou VPC com latência RTT abaixo de 5ms. Se não estiverem, adicione circuit breakers e lógica de reconexão manual.

5. Gravação obrigatória por compliance sem configuração adicional

O AudioSocket não integra com o MixMonitor do Asterisk automaticamente. Se você precisa gravar a chamada por compliance (LGPD para serviços que tratam dados de saúde ou financeiros, Anatel para provedores de telecomunicação regulados), configure o MixMonitor no dialplan antes do AudioSocket():

exten => s,1,Answer()
 same => n,MixMonitor(${UNIQUEID}.wav,b)
 same => n,AudioSocket(${UNIQUEID},127.0.0.1:9999)
 same => n,StopMixMonitor()
 same => n,Hangup()

O MixMonitor com a flag b grava ambas as pernas da chamada em um único arquivo. O AudioSocket() recebe o áudio normalmente — os dois coexistem.

---

Próximos passos

Este tutorial estabelece a base: conexão TCP, parsing do protocolo, servidor asyncio, integração STT. O que vem a seguir depende do seu caso de uso:

Para IVR com IA: adicione VAD antes do STT. A biblioteca webrtcvad (Python, wrapper do WebRTC VAD do Google) detecta atividade de voz por frame de 10ms com overhead de CPU mínimo (chamada vad.is_speech() em microssegundos sobre frame já bufferizado; ver repositório wiseman/py-webrtcvad). Enviar apenas esses frames para o STT reduz custo e melhora a acurácia da transcrição — diferença significativa em relação ao proxy RMS do servidor de exemplo acima.

Para agente de voz completo: depois do STT, você precisa de LLM (OpenAI, Anthropic, Groq para latência baixa) e TTS com streaming (ElevenLabs, Azure TTS, Google TTS). O loop completo — áudio → STT → LLM → TTS → áudio — cabe dentro do threshold de 1.200ms P95 com as escolhas certas de provedor.

Para produção: substitua o servidor asyncio simples por um com graceful shutdown, métricas de chamadas ativas, e alertas de latência por percentil. O Prometheus com prometheus_client Python é o stack padrão para isso.

Código desta série: os exemplos deste artigo estão sendo validados em container Asterisk 20 LTS pelo Engenheiro de Conteúdo Técnico. A versão validada será publicada no repositório da Mestre com instruções de setup Docker.

---

Fontes

1. Asterisk Wiki — app_audiosocket — wiki.asterisk.org/wiki/display/AST/AudioSocket, acesso 2026-05-12
2. GitHub — CyCoreSystems/audiosocket — especificação de referência do protocolo — github.com/CyCoreSystems/audiosocket, acesso 2026-05-12
3. Asterisk Wiki — ARI and Asterisk — wiki.asterisk.org/wiki/display/AST/ARI+and+Asterisk, acesso 2026-05-12
4. Google Cloud — Streaming Speech Recognition — cloud.google.com/speech-to-text/docs/streaming-recognize, acesso 2026-05-12
5. GitHub — asterisk/asterisk CHANGES (branch 20) — registro de mudanças do módulo app_audiosocket nas branches LTS 18, 20 e 22 — github.com/asterisk/asterisk/blob/20/CHANGES, acesso 2026-05-12

---

Rafael Camargo trabalha com telefonia IP em produção desde o Asterisk 1.6. Passou por integrações de call center com 500+ posições simultâneas, URAs com reconhecimento de voz pré-LLM, e agora acompanha a transição do mercado brasileiro para arquiteturas de Voice AI sobre Asterisk. Escreve sobre o que funcionou e o que não funcionou.

---

Status técnico: o código Python deste artigo foi validado em container Asterisk 20 LTS (6/6 testes PASS, MESA-112). O dialplan foi corrigido em v4: ordem de argumentos AudioSocket() (uuid primeiro), Answer() obrigatório, substituição de AUDIOSOCKET_STATUS inexistente por TryExec/TRYSTATUS/FAILED. Fix de CPU (PR #270) confirmado em Asterisk ≥ 20.5.0. Código validado pelo Engenheiro de Conteúdo Técnico — #codigo-validado em MESA-112.