Wazzapo для Python

Официальная библиотека для API v2 Wazzapo: сообщения WhatsApp, шаблоны, контакты, сделки, переписки и вебхуки. Лицензия MIT.

• подсказки в редакторе по-русски: у каждого метода и поля описание из документации API;
• повторы на 429, 5xx и сбоях сети с учётом Retry-After;
• Idempotency-Key на каждый POST: повтор не отправит сообщение дважды;
• ошибки отдельными классами с code, message, details, request_id;
• списки: одна страница или все подряд итератором;
• проверка подписи вебхуков;
• синхронный и async клиент, одна зависимость (httpx), Python 3.9+.

Установка

pip install wazzapo

Первое сообщение за 5 минут

1. В кабинете Wazzapo откройте раздел «API» и создайте ключ. Для пробы берите тестовый ключ wz_test_…: он ничего не отправляет и не меняет, а отвечает как настоящий.
2. Положите ключ в переменную окружения:

export WAZZAPO_API_KEY=wz_test_...

3. Узнайте phone_number_id своего номера и отправьте сообщение:

from wazzapo import Wazzapo

client = Wazzapo()  # ключ из WAZZAPO_API_KEY, можно передать Wazzapo(api_key="...")

# В списке все каналы аккаунта (WhatsApp, Telegram, OLX, почта...): берём подключённый WhatsApp
numbers = client.phone_numbers.list()["data"]
whatsapp = next(
    n for n in numbers
    if n["provider_type"] in ("meta_cloud", "evolution_api") and n["status"] == "connected"
)
phone_number_id = whatsapp["phone_number_id"]

# Текст доходит, только если клиент писал вам за последние 24 часа
sent = client.messages.send({
    "phone_number_id": phone_number_id,
    "to": "77071234567",
    "type": "text",
    "text": {"body": "Айгуль, здравствуйте! Запись на 25 октября в 15:00 подтверждена."},
})
print(sent["message_id"])

4. Первое сообщение клиенту (или через 24 часа тишины) отправляйте одобренным шаблоном:

client.messages.send({
    "phone_number_id": phone_number_id,
    "to": "77071234567",
    "type": "template",
    "template": {
        "name": "zapis_napominanie",
        "language": {"code": "ru"},
        "components": [
            {
                "type": "body",
                "parameters": [
                    {"type": "text", "text": "Айгуль"},
                    {"type": "text", "text": "25 октября, 15:00"},
                ],
            }
        ],
    },
})

Список шаблонов и их статус: client.templates.list(phone_number_id=phone_number_id, status="APPROVED").

Как устроены методы

• client.<ресурс>.<действие>(...): client.contacts.create, client.deals.move, client.phone_numbers.get_business_profile.
• id из адреса передаётся первым аргументом: client.contacts.get("6a7d...").
• Тело запроса: словарь первым аргументом после id (client.contacts.update("6a7d...", {"name": "Айгуль"})) или body=; поля подсказывает редактор. Параметры списка: именованные аргументы.
• Ответ: обычный словарь, как в документации. Типы для своего кода: from wazzapo.types import SendMessageIn, DealOut.
• У каждого метода есть настройки одного вызова: timeout= (секунды), headers= (свои заголовки, например {"X-Request-Id": "order-4521"}; ключ и Idempotency-Key они не заменяют), max_retries=; у POST ещё idempotency_key=.

deal = client.deals.create(
    {
        "pipeline_id": "6712a0c4e1b2c3d4e5f60718",
        "title": "Заявка с сайта: кухня на заказ",
        "contact": {"name": "Айгуль Сейткали", "phone": "77071234567"},
        "amount": 15000000,  # в тиынах: 150 000 тенге
        "currency": "KZT",
        "external_id": "site-order-4521",
    },
    idempotency_key="site-order-4521",  # повтор с тем же ключом не создаст вторую сделку
)

Загрузка и скачивание файлов:

# путь, bytes или открытый файл; путь к несуществующему файлу сразу даёт FileNotFoundError
media = client.media.upload("schet_1024.pdf", phone_number_id=phone_number_id)
client.messages.send({"phone_number_id": phone_number_id, "to": "77071234567", "type": "document",
                      "document": {"id": media["media_id"], "filename": "schet_1024.pdf"}})
data: bytes = client.media.download(media["media_id"], phone_number_id=phone_number_id)

Ошибки

from wazzapo import OutsideWindowError, RateLimitError, ValidationError, WazzapoError

try:
    client.messages.send({...})
except OutsideWindowError:
    ...  # клиент не писал 24 часа: отправьте шаблон
except ValidationError as e:
    print(e.fields)  # [{"field": "body.to", "message": "Field required"}]
except RateLimitError as e:
    print(e.retry_after)  # SDK уже повторил запрос max_retries раз
except WazzapoError as e:
    print(e.status, e.code, e.message, e.request_id)  # request_id сообщите в поддержку

Все классы наследуют WazzapoError. Если ответ пришёл не от API (например, страница прокси с 502), code берётся по статусу: BAD_GATEWAY, SERVICE_UNAVAILABLE и т.д.

Все коды и их значения: раздел «Коды ошибок» в документации.

Списки

page = client.contacts.list(tag="vip", limit=100)  # одна страница: data, next_cursor, has_more
if page["has_more"]:
    next_page = client.contacts.list(tag="vip", limit=100, cursor=page["next_cursor"])

for contact in client.contacts.list_all(tag="vip"):  # все страницы, по одной за раз
    print(contact["phone"])

Async

import asyncio
from wazzapo import AsyncWazzapo

async def main() -> None:
    async with AsyncWazzapo() as client:
        await client.messages.send({...})
        async for deal in client.deals.list_all(pipeline_id="6712a0c4e1b2c3d4e5f60718"):
            print(deal["title"])

asyncio.run(main())

Вебхуки

Создайте вебхук и сохраните секрет: он показывается один раз.

hook = client.webhooks.create({"url": "https://example.kz/wazzapo", "event_types": ["message.incoming", "deal.created"]})
secret = hook["secret"]

Проверяйте подпись на сыром теле запроса, до разбора JSON. verify сверяет подпись и время (допуск 5 минут) и возвращает событие. Без секрета (пустая строка или None, например не задана переменная окружения) verify сразу бросает WebhookVerificationError: так поддельный запрос не пройдёт. Та же ошибка, если подписи нет, она чужая, запрос старый или тело не JSON.

FastAPI:

from fastapi import FastAPI, HTTPException, Request
from wazzapo import webhooks

app = FastAPI()

@app.post("/wazzapo")
async def wazzapo_hook(request: Request) -> dict[str, bool]:
    try:
        event = webhooks.verify(await request.body(), request.headers, secret)
    except webhooks.WebhookVerificationError:
        raise HTTPException(status_code=400)
    if event["event"] == "message.incoming":
        print(event["data"]["message"]["text"])
    return {"ok": True}

Flask:

from flask import Flask, abort, request
from wazzapo import webhooks

app = Flask(__name__)

@app.post("/wazzapo")
def wazzapo_hook() -> dict[str, bool]:
    try:
        # verify принимает любые заголовки с items(): request.headers Flask, dict, заголовки FastAPI
        event = webhooks.verify(request.get_data(), request.headers, secret)
    except webhooks.WebhookVerificationError:
        abort(400)
    return {"ok": True}

Ответьте 2xx быстрее 10 секунд, долгую работу делайте в фоне. Одно событие может прийти дважды (повтор доставки): у всех повторов один и тот же заголовок X-Wazzapo-Delivery-Id (webhooks.DELIVERY_ID_HEADER), по нему и отсеивайте дубли.

delivery_id = request.headers.get(webhooks.DELIVERY_ID_HEADER)
if already_processed(delivery_id):  # ваша проверка: таблица, Redis и т.п.
    return {"ok": True}

Для своих тестов подпись ставит webhooks.sign(body, secret), а webhooks.verify(..., now=...) проверяет на заданное время.

Тестовый ключ

Ключ wz_test_… проходит все проверки, но ничего не отправляет в WhatsApp и не меняет данные: сообщения получают id wamid.SANDBOX…. Чтение работает по-настоящему. client.is_test показывает, какой ключ в работе.

Настройки

Wazzapo(
    api_key="wz_live_...",            # или WAZZAPO_API_KEY
    base_url="https://wazzapo.kz",   # или WAZZAPO_BASE_URL
    timeout=30,                      # секунды на запрос
    max_retries=2,                   # повторы на 429, 5xx и сбоях сети
    http_client=None,                # свой httpx.Client: прокси, свои заголовки
)

Ключ не попадает в логи и repr (ни клиента, ни его настроек): видны только первые 8 знаков, они же в client.api_key_hint. Отладка запросов: logging.getLogger("wazzapo").setLevel(logging.DEBUG).

Разработка

Код в src/wazzapo/_generated собирается из описания API: cd sdk && python -m codegen python. Руками его не правят.

uv venv && uv pip install -e ".[dev]"
pytest
mypy src/wazzapo
WAZZAPO_TEST_API_KEY=wz_test_... pytest tests/test_live.py