Kommo + Airtable: клиентская база и проекты из выигранных сделок

Airtable - гибридный инструмент между spreadsheet и базой данных. Используется как CRM-расширение, операционная база, трекер проектов. Нативной интеграции с Kommo нет. Строим через Airtable REST API v0 с Personal Access Token.

Что строим

1. Сделка выиграна -> создать запись в Airtable базе «Client Onboarding»
2. Статус записи обновился в Airtable -> Note в Kommo о прогрессе онбординга
3. Синхронизация поля «Дата следующего контакта» между Airtable и Kommo

Аутентификация: Personal Access Token

Airtable перешёл с API Key на Personal Access Token (PAT). API Key устарел и будет отключён.

import requests

AIRTABLE_PAT = "pat_XXXXXXX"  # Personal Access Token из Airtable Account Settings
BASE_ID   = "appXXXXXXXXXXXX"  # ID базы из URL или через /v0/meta/bases
TABLE_NAME = "Client Onboarding"  # или Table ID: tblXXXXXXXXX

at_session = requests.Session()
at_session.headers.update({
    "Authorization": f"Bearer {AIRTABLE_PAT}",
    "Content-Type": "application/json",
})

AT_BASE = f"https://api.airtable.com/v0/{BASE_ID}"

PAT имеет granular scopes: data.records:read, data.records:write, schema.bases:read. Для интеграции минимум: data.records:read + data.records:write.

Создание записи при выигрыше сделки

def create_airtable_record(deal: dict, contact: dict) -> str:
    """Create Airtable record from Kommo deal. Returns record ID."""
    payload = {
        "fields": {
            "Client Name":   contact.get("name", ""),
            "Company":       contact.get("company", ""),
            "Email":         contact.get("email", ""),
            "Deal Value":    deal.get("price", 0),
            "Kommo Deal ID": str(deal["id"]),       # строка! Airtable не имеет int64
            "Deal Name":     deal.get("name", ""),
            "Won Date":      deal.get("closed_at", "")[:10] if deal.get("closed_at") else "",
            "Status":        "Not Started",          # Airtable Single Select - должен совпасть точно
        }
    }
    r = at_session.post(f"{AT_BASE}/{TABLE_NAME}", json=payload)
    r.raise_for_status()
    return r.json()["id"]  # recXXXXXXXX

Имена полей в Airtable case-sensitive и должны совпадать точно с именами в базе. Типы полей также важны: Single Select принимает только значения из предустановленного списка - передача неизвестного значения вернёт ошибку 422.

Webhook Kommo -> создание записи:

from flask import Flask, request

app = Flask(__name__)

@app.route("/kommo/webhook", methods=["POST"])
def kommo_webhook():
    data = request.json
    for lead in data.get("leads", {}).get("update", []):
        if lead.get("status_id") == WON_STATUS_ID:
            deal    = get_kommo_deal(lead["id"])
            contact = get_kommo_deal_contact(lead["id"])
            rec_id  = create_airtable_record(deal, contact)
            # Сохранить маппинг deal_id -> airtable_record_id для обратной синхронизации
            save_mapping(lead["id"], rec_id)
    return "ok", 200

Airtable Webhook: cursor-based, не push

Это ключевое отличие Airtable от большинства платформ. Airtable webhook - это не push: Airtable уведомляет что изменения есть, но не отправляет сами данные. Вы должны самостоятельно запросить изменения через cursor.

Создание webhook:

def create_airtable_webhook(notification_url: str) -> dict:
    """Register Airtable webhook. Returns webhook config with cursor."""
    payload = {
        "notificationUrl": notification_url,
        "specification": {
            "options": {
                "filters": {
                    "fromSources": ["client", "publicApi"],
                    "dataTypes": ["tableData"],
                    "recordChangeScope": TABLE_ID,
                },
                "includes": {
                    "includeCellValuesInFieldIds": ["Status", "Next Contact Date"],
                }
            }
        }
    }
    r = at_session.post(
        f"https://api.airtable.com/v0/bases/{BASE_ID}/webhooks",
        json=payload,
    )
    r.raise_for_status()
    return r.json()

Обработка уведомления (cursor polling):

import redis  # хранение cursor между запросами

WEBHOOK_ID = "ach_XXXXXXXXX"  # из create_airtable_webhook ответа
redis_client = redis.Redis()

@app.route("/airtable/notification", methods=["POST"])
def airtable_notification():
    """Airtable sends notification without payload. Must poll for changes."""
    cursor = redis_client.get(f"airtable_cursor_{WEBHOOK_ID}")
    cursor_param = {"cursor": int(cursor)} if cursor else {}

    r = at_session.get(
        f"https://api.airtable.com/v0/bases/{BASE_ID}/webhooks/{WEBHOOK_ID}/payloads",
        params=cursor_param,
    )
    r.raise_for_status()
    response = r.json()

    for payload in response.get("payloads", []):
        changed_records = payload.get("changedFieldsByRecord", {})
        for record_id, changes in changed_records.items():
            if "Status" in changes:
                new_status = changes["Status"]["current"]["value"]
                deal_id = get_deal_id_by_record(record_id)
                if deal_id:
                    add_kommo_note(deal_id, f"Airtable: статус изменён на '{new_status}'")

    # Сохранить новый cursor
    new_cursor = response.get("cursor")
    if new_cursor:
        redis_client.set(f"airtable_cursor_{WEBHOOK_ID}", new_cursor)

    return "ok", 200

Webhook Airtable истекает через 7 дней. Нужно обновлять через POST /bases/{id}/webhooks/{wh_id}/refresh.

Чтение и обновление записей

def get_record(record_id: str) -> dict:
    r = at_session.get(f"{AT_BASE}/{TABLE_NAME}/{record_id}")
    r.raise_for_status()
    return r.json()["fields"]

def update_record(record_id: str, fields: dict) -> dict:
    """PATCH update - only specified fields changed."""
    r = at_session.patch(
        f"{AT_BASE}/{TABLE_NAME}/{record_id}",
        json={"fields": fields},
    )
    r.raise_for_status()
    return r.json()

# Пример: синхронизировать дату следующего контакта из Kommo
def sync_next_contact_date(deal_id: int, next_contact: str):
    record_id = get_record_id_by_deal(deal_id)
    if record_id:
        update_record(record_id, {"Next Contact Date": next_contact})

Airtable API rate limit: 5 запросов в секунду на base. При bulk-синхронизации добавьте time.sleep(0.2) между запросами или батчируйте через endpoint PATCH /v0/{baseId}/{tableId} (до 10 записей за раз).

Реальный кейс

Digital-агентство с 25-30 новых клиентов в месяц вело две системы: Kommo как CRM для продаж и Airtable как операционную базу для delivery-команды. Данные копировались вручную при выигрыше сделки - задержка 1-2 часа, ошибки в 15% случаев (неверный email, неполное имя).

После интеграции:

• Запись в Airtable создаётся автоматически при переходе сделки в Won (<10 секунд)
• Delivery-команда видит актуальные данные без ожидания
• Обновление статуса в Airtable отражается в Kommo Notes - продажи знают о прогрессе

Экономия: ~45 минут ручной работы в день на переносе данных.

Для кого подходит

Компании, у которых продажи работают в CRM (Kommo), а операционная команда использует Airtable как рабочий инструмент. Типичный сценарий: продажи, маркетинг или HR-команды, которые привыкли к Airtable как к гибкому трекеру.

Для более специализированных инструментов управления проектами - Kommo + Asana, Kommo + Notion, Kommo + Linear.

Часто задаваемые вопросы

Почему Airtable webhook cursor-based, а не push?

Airtable разработан как база данных с версионированием изменений. Cursor-based подход позволяет не пропускать события при недоступности вашего сервиса: вы всегда можете запросить изменения начиная с последнего известного cursor. Push-webhook без cursor рискует потерять события при downtime.

Как работать с Linked Records (связанные записи)?

В payload изменений Airtable linked record выглядит как массив record ID (["recXXX", "recYYY"]). Чтобы получить данные linked записи - отдельный запрос к таблице. Для интеграции с Kommo удобнее денормализовать данные в плоские поля (например, хранить имя клиента строкой, а не ссылкой).

Как обновить Airtable webhook через 7 дней?

Webhook истекает, если не обновлять. Добавьте в cron каждые 6 дней: POST /v0/bases/{id}/webhooks/{wh_id}/refresh. Если webhook истёк - создать новый через тот же create_airtable_webhook() и обновить cursor в Redis.

Можно ли синхронизировать файлы (вложения) из Kommo в Airtable?

Да, через Airtable Attachments field - передаёте массив {"url": "...", "filename": "..."}. URL должен быть публично доступен. Файлы из Kommo (Notes attachments) можно проксировать через ваш сервис с временной ссылкой.

Итог

Ключевые особенности интеграции Kommo + Airtable:

• PAT аутентификация (не API Key - устарел), granular scopes
• Имена полей и Single Select значения case-sensitive и должны совпадать точно
• Webhook: уведомление без данных, нужен cursor-polling для получения изменений
• Webhook истекает через 7 дней - нужен refresh job в cron

Если у вас Airtable как операционная база и Kommo как CRM - опишите задачу команде Exceltic.dev. Разберём стек и настроим двустороннюю синхронизацию.