> 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.

# Telnyx

Fork a Telnyx call's audio to a relay that forwards the callee's audio to Noesis and receives the answers.

Telnyx media streaming sends a call's audio to a WebSocket as JSON messages. Run a relay that takes the callee's audio from those messages and forwards it to Noesis.

## 1. Stream the call to your relay

With TeXML, add a `<Stream>` inside `<Start>` so the call carries on:

```xml
<Response>
  <Start>
    <Stream url="wss://relay.example.com/telnyx" track="inbound_track"/>
  </Start>
  <!-- your existing call flow continues here -->
</Response>
```

With Call Control, start streaming on the answered call with `stream_url` set to `wss://relay.example.com/telnyx` and `stream_track` set to `inbound_track`. Keep the default audio format, PCMU (μ-law) at 8 kHz.

## 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):

```python
# telnyx_relay.py: Telnyx media stream -> your server -> Noesis
import asyncio, base64, json

import websockets

from noesis_relay import NoesisCall

ENCODINGS = {"PCMU": "mulaw", "PCMA": "alaw"}

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

async def handle(telnyx):
    call = None
    try:
        async for raw in telnyx:
            msg = json.loads(raw)
            if msg.get("event") == "start":
                fmt = msg["start"].get("media_format", {})
                call = NoesisCall(on_answer,
                                  encoding=ENCODINGS.get(fmt.get("encoding"), "mulaw"),
                                  sample_rate=fmt.get("sample_rate", 8000),
                                  call_id=msg["start"].get("call_control_id"))
                await call.open()
            elif msg.get("event") == "media" and call is not None:
                if msg["media"].get("track", "inbound") == "inbound":
                    await call.send(base64.b64decode(msg["media"]["payload"]))
            elif msg.get("event") == "stop":
                break
    except websockets.ConnectionClosed:
        pass
    if call is not None:
        await call.close()

async def main():
    async with websockets.serve(handle, "0.0.0.0", 8080):
        await asyncio.Future()                      # run until stopped

asyncio.run(main())
```

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