callAudio

Во время разговора, который ведёт сервер Gratio, приложение слушает собеседника и говорит ему в реальном времени: для голосового бота, распознавания речи или своей записи разговора.

Параметры

Подключение

Подключайтесь после answerCall или makeCall, когда звонок в connecting или active. Адрес в стиле Green API:

wss://api.gratio.net/waInstance{idInstance}/callAudio/{apiTokenInstance}?callId={callId}

Или ключ — заголовком:

wss://api.gratio.net/v1/instances/{idInstance}/callAudio?callId={callId}
Authorization: Bearer {apiTokenInstance}

Вместо заголовка можно добавить к адресу &token={apiTokenInstance}.

Кадры от сервера

Первый кадр:

{"event": "start", "callId": "31A9ADA2-3889-4B2B-9D88-6A7142875D7F", "sampleRate": 16000, "channels": 1, "format": "s16le"}

Последний кадр — endReason как в объекте звонка, null, если причина неизвестна:

{"event": "end", "callId": "31A9ADA2-3889-4B2B-9D88-6A7142875D7F", "endReason": "hangup"}

Очередь звука переполнена:

{"event": "error", "code": "audio_queue_full", "message": "в очереди уже 300 с звука"}

Медленный слушатель теряет старые кадры: сервер держит для него около 5 секунд звука. Разговор от этого не страдает.

Кадры от приложения

Закрыли сокет — разговор продолжается: подключитесь снова или положите трубку через hangUpCall.

Пример: ответить, записать и положить трубку

Скрипт отвечает на входящий звонок, говорит приветствие, пишет голос собеседника в WAV и через 20 секунд кладёт трубку. Нужны Python 3.10+ и pip install requests websockets. Кроме GRATIO_URL, GRATIO_ID и GRATIO_TOKEN задайте GREETING_URL — ссылку на файл приветствия (mp3, wav или ogg), например из uploadFile.

Запуск — с callId из уведомления incomingCall со status: "offer", пока звонок звонит:

python3 answer_call.py 31A9ADA2-3889-4B2B-9D88-6A7142875D7F

Файл answer_call.py:

"""Ответить на звонок, сказать приветствие, записать собеседника в WAV и положить трубку."""

import asyncio
import contextlib
import json
import os
import sys
import wave

import requests
import websockets
from websockets.exceptions import ConnectionClosed, InvalidHandshake

API = os.environ["GRATIO_URL"]  # https://api.gratio.net
ID = os.environ["GRATIO_ID"]
TOKEN = os.environ["GRATIO_TOKEN"]
GREETING_URL = os.environ["GREETING_URL"]  # приветствие: mp3, wav или ogg по ссылке
TALK_SECONDS = 20  # столько слушаем после подключения, потом кладём трубку


def call_method(name: str, payload: dict) -> dict:
    url = f"{API}/waInstance{ID}/{name}/{TOKEN}"
    response = requests.post(url, json=payload, timeout=60)
    data = response.json()
    if response.status_code != 200:
        sys.exit(f"{name}: {response.status_code} {data.get('code')} {data.get('message')}")
    return data


async def listen(call_id: str, path: str) -> None:
    # https://… → wss://…, http://… → ws://…
    url = "ws" + API.removeprefix("http") + f"/waInstance{ID}/callAudio/{TOKEN}?callId={call_id}"
    async with websockets.connect(url) as ws:
        start = json.loads(await ws.recv())  # первый кадр — формат звука
        print("Звук пошёл:", start["format"], start["sampleRate"], "Гц")

        async def hang_up_later() -> None:
            await asyncio.sleep(TALK_SECONDS)
            with contextlib.suppress(ConnectionClosed):
                await ws.send(json.dumps({"command": "hangup"}))

        timer = asyncio.create_task(hang_up_later())
        with wave.open(path, "wb") as wav:
            wav.setnchannels(start["channels"])
            wav.setsampwidth(2)  # s16le — 2 байта на отсчёт
            wav.setframerate(start["sampleRate"])
            try:
                async for frame in ws:
                    if isinstance(frame, bytes):  # голос собеседника, около 20 мс
                        wav.writeframes(frame)
                        continue
                    event = json.loads(frame)
                    if event["event"] == "end":
                        print("Звонок закончился:", event["endReason"])
                        break
                    print("Событие:", event)
            finally:
                timer.cancel()


def main() -> None:
    call_id = sys.argv[1]
    call = call_method("answerCall", {
        "callId": call_id,
        "playUrl": GREETING_URL,
        "record": False,  # пишем сами, запись на сервере не нужна
        "maxDurationSeconds": TALK_SECONDS + 60,
    })
    print("Ответили, состояние:", call["state"])
    path = f"call-{call_id}.wav"
    try:
        asyncio.run(listen(call_id, path))
    except InvalidHandshake as e:
        sys.exit(f"Сокет не открылся: {e}")
    print("Голос собеседника —", path)


if __name__ == "__main__":
    main()

Вы увидите:

Ответили, состояние: connecting
Звук пошёл: s16le 16000 Гц
Звонок закончился: hangup
Голос собеседника — call-31A9ADA2-3889-4B2B-9D88-6A7142875D7F.wav

Голосовой бот устроен так же: кадры собеседника — в распознавание речи, ответ синтезом — двоичными кадрами в тот же сокет, clear — когда собеседник перебивает. Пошагово — в руководстве Звонки: автоответчик и голосовой бот.

Отказ в подключении

Сервер проверяет ключ и звонок до установки соединения. При отказе рукопожатие WebSocket не проходит: сервер отвечает HTTP 403, кода закрытия у такого отказа нет. Браузер покажет закрытие с кодом 1006.

Если звонок закончился в момент подключения, сервер может открыть сокет и сразу прислать end.

Особенности каналов

• MAX и WhatsApp — один формат звука: PCM s16le, моно, 16 кГц.
• Telegram. Звука звонков нет — подключение отклоняется.
• Звонок через API — только голосовой: видео не передаётся.

Что дальше

• answerCall — ответить, чтобы подключиться к звуку.
• playAudio — сказать файлом вместо потока.
• Звонки: автоответчик и голосовой бот — руководство с распознаванием и синтезом.