Answering-machine detection (WebSocket)

curl --request GET \
  --url https://api.pyai.com/v1/amd/stream \
  --header 'Authorization: Bearer <token>'

const options = {method: 'GET', headers: {Authorization: 'Bearer <token>'}};

fetch('https://api.pyai.com/v1/amd/stream', options)
  .then(res => res.json())
  .then(res => console.log(res))
  .catch(err => console.error(err));

<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.pyai.com/v1/amd/stream",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "GET",
  CURLOPT_HTTPHEADER => [
    "Authorization: Bearer <token>"
  ],
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}

package main

import (
	"fmt"
	"net/http"
	"io"
)

func main() {

	url := "https://api.pyai.com/v1/amd/stream"

	req, _ := http.NewRequest("GET", url, nil)

	req.Header.Add("Authorization", "Bearer <token>")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}

require 'uri'
require 'net/http'

url = URI("https://api.pyai.com/v1/amd/stream")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Get.new(url)
request["Authorization"] = 'Bearer <token>'

response = http.request(request)
puts response.read_body

{
  "error": {
    "message": "<string>",
    "type": "<string>",
    "code": "invalid_request_error",
    "param": "<string>"
  }
}

{
  "error": {
    "message": "<string>",
    "type": "<string>",
    "code": "invalid_request_error",
    "param": "<string>"
  }
}

AMD

Answering-machine detection (WebSocket)

Realtime answering-machine detection over a WebSocket. This surface speaks Twilio’s Media Streams protocol natively (start / media / stop frames, G.711 μ-law 8 kHz base64, ~20 ms), so migrating from Twilio AMD is a one-line-TwiML change, point the call’s media at PyAI, keep your carrier and your code.

<Response><Start>
  <Stream url="wss://api.pyai.com/v1/amd/stream">
    <Parameter name="api_key" value="YOUR_PYAI_KEY"/>
    <Parameter name="aggressiveness" value="0.25"/>
    <Parameter name="webhook" value="https://you/amd-events"/>
  </Stream>
</Start>
  <!-- your existing call flow continues here -->
</Response>

Use <Start><Stream>, NOT <Connect><Stream>. <Start> forks the audio and TwiML continues to your next verb, so the call still goes where it was going; <Connect> hands the media path to the socket and blocks TwiML until the stream ends, and because AMD is listen-only and never sends audio back the caller would hear dead air and a dialer would never reach the agent. (<Connect> is correct for Omni, which is a two-way voice agent.) Drop machineDetection from the call and keep your carrier.

On a Twilio-originated stream TWILIO owns the socket and relays only media/mark frames, so the pushed amd event does not reach you: a Twilio integration must read the decision from the webhook <Parameter> (or GET /v1/amd/calls/{id}). The socket push is for clients that drive the socket themselves.

From Twilio, authenticate with the api_key <Parameter> shown above (Twilio strips query strings from the <Stream> URL and cannot send headers; PyAI verifies the key from the stream’s start frame before processing any audio, and closes connections that never present a valid key). Server-side clients may instead authenticate at the handshake with the Sec-WebSocket-Protocol: pyai.v1, pyai-key.<API_KEY> subprotocol pair or ?api_key=. Requires the amd:detect scope. Mid-call, PyAI pushes an amd decision event on the socket (and to the per-call TwiML webhook parameter): answered_by (the routing class: human, machine, sit_invalid, unknown), answered_by_twilio (Twilio’s exact AnsweredBy enum for drop-in routing parity), subtype (when available), party_detected, voicemail_ready, confidence, decision_ms, and a human-readable reason. party_detected is true for human or machine classification and false for unknown or invalid-number outcomes. voicemail_ready is false on classification events: detecting voicemail does not establish that recording has started or authorize dropping a message. decision_ms measures processed inbound audio through the decision, not elapsed time from carrier answer. A machine decision can carry a subtype (voicemail, ivr, screening, music); the stored call record folds that subtype into answered_by. Read the stored record with GET /v1/amd/calls/{id} or receive it on the account-wide amd.call.completed webhook (webhook_url in POST /v1/amd/config). The per-call aggressiveness <Parameter> overrides the account default from POST /v1/amd/config.

When the decision arrives

Measured over ~1,850 real answered calls (US telephony, 8 kHz μ-law), streamed at real time:

verdict	typical	9 in 10 by
`human`	~1.4 s	~3.0 s
`machine`	~2.2 s	~3.2 s

About 1 human decision in 4 lands under a second. There is a hard deadline at 6 s: if nothing is decisive by then you get a verdict anyway, unknown at a human-safe operating point. Size your fallback timer past 6 s, not past the typical case, and prefer reacting to the event over polling.

What to do with each verdict

`answered_by`	subtype	do
`human`	—	connect the agent
`machine`	`voicemail`	safe to drop a message; wait for the record tone rather than assuming one
`machine`	`ivr`	a phone tree, do not drop a message — navigate or abandon
`machine`	`screening`	an AI screener (iPhone/Google) is relaying to a person; treat as a live-ish path, not voicemail
`machine`	`music`	tonal audio and nothing transcribed — hold music or ringback, but also a greeting we failed to transcribe. Keep waiting; do not read it as a positive hold-music signal, and do not gate a drop on it
`sit_invalid`	—	dead/invalid number, stop retrying it
`unknown`	`silence`	answered but nothing came down the line, retry later rather than burning an agent slot
`unknown`	—	no decisive evidence; your default decides

Gate your voicemail drop on the subtype, not on answered_by. voicemail, ivr, screening and music are all machine, and a message dropped into a phone tree or an AI screener is wasted. Drop only on voicemail: music means we heard tonal audio and never transcribed anything, which covers hold music and an untranscribed greeting alike, so it is not evidence either way. The subtype is on the call record and the amd.call.completed webhook, not on the wire event.

Choosing `aggressiveness`

The dial only changes what happens when the 6 s deadline is reached with nothing decisive; it never overrides positive evidence. Measured on the same corpus, calls that reach that deadline are roughly 85% machines, so:

live-agent dialers (a person hearing voicemail is cheap, a machine reaching an agent wastes a seat): keep it low. You get unknown and your own default decides.
automated voicemail-drop bots (nobody is waiting): raise it. You trade about 1 wrong machine in 7 of those deadline cases for never stalling.

Undecided calls are ~2% of answered calls at the default operating point, so this dial affects a small tail either way.

Billed per answered call (amd.calls), the first 5,000 answered calls/month are free, then $0.004/answered call; AMD bundled with PyAI telephony/Omni is included.

GET

amd

stream

Answering-machine detection (WebSocket)

curl --request GET \
  --url https://api.pyai.com/v1/amd/stream \
  --header 'Authorization: Bearer <token>'

const options = {method: 'GET', headers: {Authorization: 'Bearer <token>'}};

fetch('https://api.pyai.com/v1/amd/stream', options)
  .then(res => res.json())
  .then(res => console.log(res))
  .catch(err => console.error(err));

<?php

$curl = curl_init();

curl_setopt_array($curl, [
  CURLOPT_URL => "https://api.pyai.com/v1/amd/stream",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 30,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "GET",
  CURLOPT_HTTPHEADER => [
    "Authorization: Bearer <token>"
  ],
]);

$response = curl_exec($curl);
$err = curl_error($curl);

curl_close($curl);

if ($err) {
  echo "cURL Error #:" . $err;
} else {
  echo $response;
}

package main

import (
	"fmt"
	"net/http"
	"io"
)

func main() {

	url := "https://api.pyai.com/v1/amd/stream"

	req, _ := http.NewRequest("GET", url, nil)

	req.Header.Add("Authorization", "Bearer <token>")

	res, _ := http.DefaultClient.Do(req)

	defer res.Body.Close()
	body, _ := io.ReadAll(res.Body)

	fmt.Println(string(body))

}

require 'uri'
require 'net/http'

url = URI("https://api.pyai.com/v1/amd/stream")

http = Net::HTTP.new(url.host, url.port)
http.use_ssl = true

request = Net::HTTP::Get.new(url)
request["Authorization"] = 'Bearer <token>'

response = http.request(request)
puts response.read_body

{
  "error": {
    "message": "<string>",
    "type": "<string>",
    "code": "invalid_request_error",
    "param": "<string>"
  }
}

{
  "error": {
    "message": "<string>",
    "type": "<string>",
    "code": "invalid_request_error",
    "param": "<string>"
  }
}

Authorizations

Authorization

string

header

required

Use Authorization: Bearer pyai_live_... (or pyai_test_...).

Response

WebSocket upgrade, Twilio Media Streams protocol; PyAI emits amd decision events.

MCP

Cast

Dub

Console creations

Identity

Call Integrations

Sandbox

Startup Program

Models

Speech To Text (Hear)

Text To Speech (Speak)

Agents

Clone

Transcription Jobs

Trace

Recap

Integrations

AMD

Omni

Knowledge Bases

Telephony

WhatsApp

Answering-machine detection (WebSocket)

When the decision arrives

What to do with each verdict

Choosing `aggressiveness`

Authorizations

Response