Перейти к содержимому

Звонки: автоответчик и голосовой бот

Обновлено 17 октября 2026

Что получится: автоответчик на Python. Он отвечает на входящий звонок, проигрывает приветствие, записывает голосовое сообщение звонящего — до 30 секунд или пока тот не положит трубку — и сохраняет запись. На том же каркасе строится голосовой бот: звук идёт к вам кадрами в реальном времени, ответ уходит обратно в звонок.

incomingCall (offer) ──► answerCall + приветствие ──► callStateChanged: active
                                                          │
                     WebSocket callAudio ◄── кадры PCM ───┤  голос звонящего
                     WebSocket callAudio ─── кадры PCM ──►│  ваш ответ (бот)
                                                          │
                     через 30 с hangUpCall ──► callStateChanged: ended + запись

Что понадобится

  • канал MAX или WhatsApp, у которого в getCapabilities call_audio: true. В Telegram звонки приходят только событиями;
  • вебхук канала с включённым incomingCallWebhook — Шаг 4. Уведомления на свой сервер;
  • Python 3.10+, pip install fastapi uvicorn requests websockets;
  • приветствие в MP3 или WAV по публичной ссылке — или загрузите его методом uploadFile.

Как идёт звонок

  1. Звонок приходит уведомлением incomingCall со статусом offer. idMessage — id звонка, callId.
  2. Пока звонит (около 90 секунд), приложение отвечает answerCall или отклоняет rejectCall. Не ответили — звонок звонит на телефонах номера как обычно.
  3. Ход разговора приходит уведомлениями callStateChanged: connecting → active → ended. В ended — причина и ссылка на запись.

Подробно — в статье Звонки: как это работает.

Шаг 1. Загрузите приветствие

curl

curl -X POST "$GRATIO_URL/waInstance$GRATIO_ID/uploadFile/$GRATIO_TOKEN" \
  -H 'Content-Type: audio/mpeg' -H 'GA-Filename: greeting.mp3' \
  --data-binary @greeting.mp3

Python

import os
import requests

BASE = f"{os.environ['GRATIO_URL']}/waInstance{os.environ['GRATIO_ID']}"
TOKEN = os.environ["GRATIO_TOKEN"]

url = f"{BASE}/uploadFile/{TOKEN}"
headers = {"Content-Type": "audio/mpeg", "GA-Filename": "greeting.mp3"}
with open("greeting.mp3", "rb") as f:
    response = requests.post(url, data=f, headers=headers, timeout=120)
print(response.status_code, response.json())

Node.js

import { readFile } from "node:fs/promises";
const { GRATIO_URL, GRATIO_ID, GRATIO_TOKEN } = process.env;
const base = `${GRATIO_URL}/waInstance${GRATIO_ID}`;

const url = `${base}/uploadFile/${GRATIO_TOKEN}`;
const response = await fetch(url, {
  method: "POST",
  headers: { "Content-Type": "audio/mpeg", "GA-Filename": "greeting.mp3" },
  body: await readFile("greeting.mp3"),
});
console.log(response.status, await response.json());

PHP

<?php
$base = getenv('GRATIO_URL') . '/waInstance' . getenv('GRATIO_ID');
$url = $base . '/uploadFile/' . getenv('GRATIO_TOKEN');

$ch = curl_init($url);
curl_setopt_array($ch, [
    CURLOPT_POST => true,
    CURLOPT_HTTPHEADER => ['Content-Type: audio/mpeg', 'GA-Filename: greeting.mp3'],
    CURLOPT_POSTFIELDS => file_get_contents('greeting.mp3'),
    CURLOPT_RETURNTRANSFER => true,
    CURLOPT_TIMEOUT => 120,
]);
$response = json_decode(curl_exec($ch), true);
echo curl_getinfo($ch, CURLINFO_HTTP_CODE), PHP_EOL;
print_r($response);

Вы увидите ссылку urlFile — сохраните её в GREETING_URL. Gratio возьмёт файл из своего хранилища без повторного скачивания.

Шаг 2. Ответ на звонок

Файл voice.py. Вебхук ловит offer, отвечает на звонок с приветствием и ставит таймер: через 30 секунд — положить трубку.

import asyncio
import os
import requests
from fastapi import FastAPI, Header, HTTPException

BASE = f"{os.environ['GRATIO_URL']}/waInstance{os.environ['GRATIO_ID']}"
TOKEN = os.environ["GRATIO_TOKEN"]
HOOK_TOKEN = os.environ["GRATIO_HOOK_TOKEN"]
GREETING_URL = os.environ["GREETING_URL"]
VOICEMAIL_SECONDS = 30

app = FastAPI()


def call(method: str, payload: dict) -> dict:
    r = requests.post(f"{BASE}/{method}/{TOKEN}", json=payload, timeout=60)
    r.raise_for_status()
    return r.json()


async def hang_up_later(call_id: str) -> None:
    await asyncio.sleep(VOICEMAIL_SECONDS)
    try:
        call("hangUpCall", {"callId": call_id})
    except requests.HTTPError:
        pass                                           # звонящий уже положил трубку


@app.post("/gratio/hook")
async def hook(body: dict, authorization: str = Header("")):
    if authorization != f"Bearer {HOOK_TOKEN}":
        raise HTTPException(status_code=401)
    kind = body["typeWebhook"]
    if kind == "incomingCall" and body["status"] == "offer":
        call_id = body["idMessage"]
        answered = call("answerCall", {"callId": call_id, "playUrl": GREETING_URL,
                                       "record": True, "maxDurationSeconds": VOICEMAIL_SECONDS + 15})
        print("Ответили:", body["chatId"], answered["state"])
        asyncio.create_task(hang_up_later(call_id))
    elif kind == "callStateChanged" and body["state"] == "ended":
        rec = body.get("recording")
        print("Звонок окончен:", body["chatId"], body["endReason"], rec and rec["downloadUrl"])
        if rec:
            save(body["callId"], rec["downloadUrl"])
    return {"ok": True}


def save(call_id: str, url: str) -> None:
    audio = requests.get(url, timeout=60).content     # ссылка подписанная, токен не нужен
    with open(f"voicemail-{call_id}.wav", "wb") as f:
        f.write(audio)

Запустите:

uvicorn voice:app --port 8080

Вы увидите при звонке на номер канала: приветствие в трубке, через 30 секунд — конец разговора. В консоли:

Ответили: 8695836 connecting
Звонок окончен: 8695836 hangup https://api.gratio.net/calls/7100000001/31A9ADA2-…/….wav

…и файл voicemail-31A9ADA2-….wav — голос звонящего, WAV, моно, 16 кГц.

Шаг 3. Живой звук для голосового бота

Чтобы разговаривать, а не только записывать, откройте WebSocket callAudio, когда звонок в connecting или active. Сервер шлёт голос звонящего двоичными кадрами PCM s16le, моно, 16 кГц, примерно по 20 мс. Ваши кадры того же формата Gratio проигрывает звонящему.

import json
import websockets


async def talk(call_id: str) -> None:
    url = (f"{BASE.replace('https://', 'wss://')}/callAudio/{TOKEN}?callId={call_id}")
    async with websockets.connect(url) as ws:
        async for frame in ws:
            if isinstance(frame, str):
                event = json.loads(frame)
                if event.get("event") == "end":
                    break
                continue
            text = recognize(frame)                    # ваше распознавание речи по кадрам
            if text:
                await ws.send(json.dumps({"command": "clear"}))   # собеседник перебил — сбросить
                for chunk in synthesize(answer(text)):           # ваш синтез: PCM s16le 16 кГц моно
                    await ws.send(chunk)

recognize, answer и synthesize — ваши распознавание, логика и синтез речи. Запускайте talk задачей сразу после answerCall вместо таймера hang_up_later. Положить трубку из сокета — {"command": "hangup"}.

Форматы кадров, ошибки подключения и полный пример — callAudio.

Исходящий звонок

Тот же каркас работает и в обратную сторону — makeCall звонит клиенту и проигрывает файл, например напоминание о записи. Не чаще раза в 20 секунд на канал.

Что дальше

Была ли статья полезна?