AMD Detection API

1Getting Started

The service delivers real-time, streaming answering-machine detection for telephony. You stream 8 kHz audio over WebSocket and receive a detection result as soon as the model reaches a confident decision.

How it works

1. Register your server's public IP via the Registration API (once, during setup).
2. Connect to ws://api.amdy.io:2700 with your API key in the header.
3. Send one JSON config frame describing the call.
4. Stream raw PCM audio in binary frames.
5. Receive an empty reply (keep streaming) or a non-empty result (stop).

Requirements

2API Key & Authentication

Your API key is the same credential for both IP registration and WebSocket detection.

API keys are 64-character hex strings. Retrieve yours from the Settings page on the portal.

Passing the key

Use either the Authorization header or the X-API-Key header:

HTTP / Registration API

Authorization: Bearer YOUR_API_KEY
# or
X-API-Key: YOUR_API_KEY

WebSocket

headers: { "X-API-Key": "YOUR_API_KEY" }

3IP Registration API

Register your server's public IP so it can reach the WebSocket detection endpoint. Every server that will stream audio must have its public IP registered first. Registration is a one-time step per server — the install scripts handle it automatically; for manual integrations, call this endpoint once during setup.

curl -X POST https://app.amdy.io/api/v1/ips/register \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"ip": "203.0.113.42", "description": "prod-dialer-01"}'

import requests

resp = requests.post(
    "https://app.amdy.io/api/v1/ips/register",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    json={"ip": "203.0.113.42", "description": "prod-dialer-01"},
)
resp.raise_for_status()
print(resp.json())   # {"ok": true, "activated": true, "ip": "203.0.113.42", ...}

# Omit "ip" — the server reads X-Forwarded-For automatically
curl -X POST https://app.amdy.io/api/v1/ips/register \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"description": "auto-registered"}'

{
  "ok": true,
  "activated": true,
  "account_activated_now": false,
  "client_id": 1042,
  "ip": "203.0.113.42",
  "registered": { "ip": "203.0.113.42", "status": "registered" }
}

Managing IPs

To view, rename, or remove registered IPs, use the IP management page in the portal. There is no public DELETE endpoint — removals are portal-only.

4WebSocket Detection API

The real-time detection endpoint your registered servers connect to.

Python

import websockets
headers = {"X-API-Key": "YOUR_API_KEY"}
ws = await websockets.connect("ws://api.amdy.io:2700", extra_headers=headers)

Node.js

const WebSocket = require('ws');
const ws = new WebSocket('ws://api.amdy.io:2700', {
  headers: { 'X-API-Key': 'YOUR_API_KEY' }
});

5Configuration Message

The first frame you send after connecting — one JSON text frame, then switch to binary audio.

{
  "config": {
    "sample_rate": 8000,
    "VID": "unique_call_id_12345",
    "immediate_detection": false,
    "stage_results": false
  }
}

6Streaming Protocol

Send audio binary frames, read a reply after each one.

1. Send the JSON config frame (text).
2. Stream audio as binary frames. ~1 second per chunk (8000 samples) is recommended.
3. After each chunk, read the server reply:
   "" (empty string) — keep streaming (standard mode).
   STAGE-… — intermediate progress frame; keep streaming (stage_results mode).
   Any other non-empty string — final result, stop and close.
4. If audio runs out before a result, signal end-of-audio: send {"eof": 1} (text) or an empty binary frame. The server returns a forced result immediately.

Buffer thresholds

7Response Format

A dash-delimited string: CLASSIFICATION-DURATION_SECONDS-CONFIDENCE

Parsing

cls, dur, conf = result.split("-")   # never more than 3 parts — classification never contains dashes
is_human = cls == "HUMAN"
duration  = float(dur)
confidence = float(conf)

8Classification Values

The statuses the production server returns today (42-class greeting model). Everything ending in AMD is a machine.

Coming with the next greeting model

These changes are not live yet. Production returns the list above until the new 46-class greeting model is deployed. Until then, the new statuses below come back as plain AMD or another machine status.

If your dialplan or reports match on FWDVMTONEAMD or PLVMSFORAMD, treat them the same as VERIZONAMD and PLVMSAMD now, so nothing changes for you when the new model ships. Anything ending in AMD is still a machine.

9Detection Modes

Three independent flags control how the server behaves during the cascade.

Standard mode (default)

Streams through the full cascade (1k → 2k → 3k → 4k → 8k → 12k → 16k → 24k → 32k + greeting at 4.5 s of speech). Returns "" after each chunk until a high-confidence verdict. Accuracy ~99%, typical latency 3–4 seconds.

{"config": {"sample_rate": 8000, "VID": "...", "immediate_detection": false}}

Immediate mode

Returns at first confident detection (4k–8k). Accuracy ~94–96%, latency 0.5–1 second. Best for high-volume, latency-sensitive routing.

{"config": {"sample_rate": 8000, "VID": "...", "immediate_detection": true}}

Stage results mode

Instead of silent "" replies between checkpoints, the server streams a STAGE-… progress frame after each cascade stage completes. The final result frame is identical to standard mode — no parsing changes required for existing consumers. Combine with standard mode for diagnostic visibility; combine with immediate mode for fast routing with progress frames.

{"config": {"sample_rate": 8000, "VID": "...", "stage_results": true}}

Stage frame format

Format: STAGE-<stage>-<CLS>-<duration>-<confidence>. Stage frames occupy the same reply slot as empty strings. A frame always starts with STAGE- and is never a valid final classification token. If multiple cascade stages complete between two audio chunks, they are semicolon-joined in a single frame.

Handling stage frames in Python

msg = await ws.recv()
if not msg:
    pass                          # empty — keep streaming (standard mode)
elif msg.startswith("STAGE-"):
    for frame in msg.split(";"):
        _, stage, cls, dur, conf = frame.split("-", 4)
        print(f"  {stage}: {cls} @ {dur}s ({float(conf):.0%}")   # optional
    # keep streaming regardless
else:
    cls, dur, conf = msg.split("-")   # final result
    break

10Code Examples

Python

import asyncio, json, websockets, numpy as np, soundfile as sf

async def detect(audio_file, api_key, immediate=False):
    uri = "ws://api.amdy.io:2700"
    async with websockets.connect(uri, extra_headers={"X-API-Key": api_key}) as ws:
        # 1) Send config
        await ws.send(json.dumps({"config": {
            "sample_rate": 8000,
            "VID": f"CALL_{int(asyncio.get_event_loop().time()*1000)}",
            "immediate_detection": immediate,
        }}))

        # 2) Load and resample to 8 kHz int16
        audio, sr = sf.read(audio_file)
        if audio.ndim > 1: audio = audio.mean(axis=1)
        if sr != 8000:
            from scipy import signal
            audio = signal.resample(audio, int(len(audio)*8000/sr))
        pcm = (audio * 32767).astype(np.int16)

        # 3) Stream 1-second chunks, stop on non-empty reply
        CHUNK = 8000
        for i in range(0, len(pcm), CHUNK):
            await ws.send(pcm[i:i+CHUNK].tobytes())
            try:
                res = await asyncio.wait_for(ws.recv(), timeout=0.5)
            except asyncio.TimeoutError:
                continue
            if res:
                return parse(res)

        # 4) Audio exhausted — force a result
        await ws.send(json.dumps({"eof": 1}))
        return parse(await ws.recv())

def parse(res):
    res = res if isinstance(res, str) else res.decode()
    cls, dur, conf = res.split("-")
    return {"classification": cls, "duration": float(dur), "confidence": float(conf)}

# Usage
r = asyncio.run(detect("call.wav", "YOUR_API_KEY"))
print(r["classification"], f'{r["confidence"]:.2%}')
is_human = r["classification"] == "HUMAN"

Node.js

const WebSocket = require('ws');
const fs = require('fs');
const wav = require('node-wav');

function detect(audioFile, apiKey, immediate = false) {
  return new Promise((resolve, reject) => {
    const ws = new WebSocket('ws://api.amdy.io:2700', {
      headers: { 'X-API-Key': apiKey }
    });
    const CHUNK = 8000;
    let pcm, offset = 0;

    const next = () => {
      if (offset < pcm.length) {
        ws.send(pcm.slice(offset, offset += CHUNK).buffer);
      } else {
        ws.send(JSON.stringify({ eof: 1 }));   // force result
      }
    };

    ws.on('open', () => {
      ws.send(JSON.stringify({
        config: { sample_rate: 8000, VID: `CALL_${Date.now()}`, immediate_detection: immediate }
      }));
      const d = wav.decode(fs.readFileSync(audioFile));
      pcm = Int16Array.from(d.channelData[0], x => Math.max(-32768, Math.min(32767, x * 32767)));
      next();
    });

    ws.on('message', (data) => {
      const res = data.toString();
      if (res) {
        const [cls, dur, conf] = res.split('-');
        ws.close();
        resolve({ classification: cls, duration: +dur, confidence: +conf });
      } else {
        next();
      }
    });

    ws.on('error', reject);
  });
}

detect('call.wav', 'YOUR_API_KEY')
  .then(r => console.log(r.classification, (r.confidence * 100).toFixed(2) + '%'));

Bash (one-shot registration)

#!/bin/bash
# Register this server's IP — run once during setup
API_KEY="YOUR_API_KEY"
DESCRIPTION="$(hostname)"

curl -sfX POST https://app.amdy.io/api/v1/ips/register \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d "{\"description\": \"$DESCRIPTION\"}" | python3 -m json.tool

11Audio Format

All audio must be 8 kHz · 16-bit · mono · raw PCM (signed little-endian).

FFmpeg conversion

ffmpeg -i input.mp3 -ar 8000 -ac 1 -f s16le -acodec pcm_s16le output.raw

Python conversion

import soundfile as sf, numpy as np
from scipy import signal

def to_pcm(path):
    a, sr = sf.read(path)
    if a.ndim > 1: a = a.mean(axis=1)
    if sr != 8000: a = signal.resample(a, int(len(a) * 8000 / sr))
    return (a * 32767).astype(np.int16)

12Error Handling

WebSocket close codes and recovery strategies.

Common issues

Retry pattern

import time

def detect_with_retry(audio_file, api_key, max_retries=3):
    for attempt in range(max_retries):
        try:
            return asyncio.run(detect(audio_file, api_key))
        except Exception as e:
            if attempt == max_retries - 1: raise
            time.sleep(2 ** attempt)   # 1s, 2s, 4s

13Support