> For AI agents: the documentation index is at https://platform.detectif.ai/llms.txt. Every page is also available as Markdown by adding `.md` to its path.

# Asterisk

Send an Asterisk channel's audio to a relay with AudioSocket, and forward it to Noesis.

Asterisk's AudioSocket (Asterisk 18 and later) sends a channel's audio to a TCP server as 16-bit PCM at 8 kHz. Run a relay that accepts the AudioSocket connection and forwards the audio to Noesis.

## 1. Send the channel's audio to your relay

In the dialplan, give AudioSocket a UUID for the call and your relay's address:

```conf
exten => 100,1,Answer()
 same => n,AudioSocket(40325ec2-5efd-4bd3-805f-53576e581d13,relay.example.com:9092)
```

AudioSocket connects the channel to the relay until the connection closes. To listen to a call that continues with a person or bot at the same time, run AudioSocket on a spy channel of the person you called, for example with ChanSpy or an ARI snoop channel.

## 2. Forward the audio to Noesis

The relay uses `noesis_relay.py` from the [relay helper](https://platform.detectif.ai/docs/noesis/websocket-client/#relay-helper). Each AudioSocket message is a 1-byte type, a 2-byte length (big-endian) and the payload:

```python
# asterisk_relay.py: Asterisk AudioSocket -> your server -> Noesis
import asyncio, uuid

from noesis_relay import NoesisCall

HANGUP, CALL_UUID, AUDIO = 0x00, 0x01, 0x10

async def on_answer(event):
    if event["type"] == "classification":
        print(event["call_id"], event["label"])     # your own logic

async def handle(reader, writer):
    call = None
    try:
        while True:
            header = await reader.readexactly(3)
            kind, size = header[0], int.from_bytes(header[1:3], "big")
            payload = await reader.readexactly(size) if size else b""
            if kind == CALL_UUID:
                call = NoesisCall(on_answer, encoding="pcm_s16le", sample_rate=8000,
                                  call_id=str(uuid.UUID(bytes=payload)))
                await call.open()
            elif kind == AUDIO and call is not None:
                await call.send(payload)            # 16-bit PCM, 8 kHz, mono
            elif kind == HANGUP:
                break
    except asyncio.IncompleteReadError:
        pass                                        # Asterisk closed the connection
    if call is not None:
        await call.close()
    writer.close()

async def main():
    server = await asyncio.start_server(handle, "0.0.0.0", 9092)
    async with server:
        await server.serve_forever()

asyncio.run(main())
```

To not record calls, create each call with `record="false"` as well.