Python пример

Production-ready клиент для inbound + FastAPI handler для callback'ов.

Зависимости

pip install httpx fastapi uvicorn

Клиент: залить лид

# cartelcrm_client.py
import hmac
import hashlib
import json
import time
import os
from typing import Any
import httpx


class CartelCrmClient:
    def __init__(self, base_url: str, api_key: str, hmac_secret: str):
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.hmac_secret = hmac_secret.encode()
        self._client = httpx.Client(timeout=15.0)

    def submit_lead(
        self,
        payload: dict[str, Any],
        idempotency_key: str,
    ) -> tuple[int, dict[str, Any]]:
        # Сериализуем один раз и шлём ту же строку.
        # separators=(',', ':') — компактный JSON без лишних пробелов.
        body = json.dumps(
            payload,
            separators=(',', ':'),
            ensure_ascii=False,
        ).encode('utf-8')

        ts = str(int(time.time()))
        sig = hmac.new(self.hmac_secret, body, hashlib.sha256).hexdigest()

        resp = self._client.post(
            f'{self.base_url}/v1/inbound/leads',
            content=body,
            headers={
                'Authorization': f'Bearer {self.api_key}',
                'X-Signature': f'sha256={sig}',
                'X-Timestamp': ts,
                'Idempotency-Key': idempotency_key,
                'Content-Type': 'application/json',
            },
        )
        return resp.status_code, resp.json()

    def get_status(self, external_id: str) -> tuple[int, dict[str, Any]]:
        ts = str(int(time.time()))
        # Подпись пустого body
        sig = hmac.new(self.hmac_secret, b'', hashlib.sha256).hexdigest()

        resp = self._client.get(
            f'{self.base_url}/v1/inbound/leads/{external_id}/status',
            headers={
                'Authorization': f'Bearer {self.api_key}',
                'X-Signature': f'sha256={sig}',
                'X-Timestamp': ts,
            },
        )
        return resp.status_code, resp.json()


# Использование
if __name__ == '__main__':
    client = CartelCrmClient(
        base_url=os.environ['CARTELCRM_BASE_URL'],
        api_key=os.environ['CARTELCRM_API_KEY'],
        hmac_secret=os.environ['CARTELCRM_HMAC_SECRET'],
    )
    status, body = client.submit_lead(
        {
            'external_id': 'aff42-lead-001',
            'phone': '+380501234567',
            'first_name': 'Иван',
            'country': 'UA',
            'source': 'facebook_ads',
        },
        idempotency_key='aff42-lead-001-r0',
    )
    print(status, body)

С retry / backoff (tenacity)

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class RetryableError(Exception): ...

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(min=1, max=30),
    retry=retry_if_exception_type(RetryableError),
)
def submit_with_retry(client, payload, idempotency_key):
    status, body = client.submit_lead(payload, idempotency_key)
    if status == 200:
        return ('accepted', body)
    if status == 409:
        return ('duplicate', body)
    if status == 429 or status >= 500:
        raise RetryableError(f'retryable: {status}')
    # 4xx (кроме 429) — финал
    return ('rejected', body)

Idempotency-Key — один на лид

При retry не меняйте ключ. Тот же ключ — гарантия отсутствия дублей. Меняйте только между разными лидами.

Сервер: callback handler (FastAPI)

# callback_server.py
import hmac
import hashlib
import os
import time
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

CALLBACK_SECRET = os.environ['CARTELCRM_CALLBACK_SECRET'].encode()


def verify_signature(raw_body: bytes, sig_header: str, secret: bytes) -> bool:
    clean = sig_header[len('sha256='):] if sig_header.startswith('sha256=') else sig_header
    expected = hmac.new(secret, raw_body, hashlib.sha256).hexdigest()
    # timing-safe сравнение
    return hmac.compare_digest(clean, expected)


@app.post('/cartelcrm/callback')
async def callback(request: Request):
    raw_body = await request.body()
    sig = request.headers.get('x-signature', '')
    ts_raw = request.headers.get('x-timestamp', '0')

    try:
        ts = int(ts_raw)
    except ValueError:
        raise HTTPException(401, 'missing_timestamp')

    # Anti-replay
    if abs(time.time() - ts) > 300:
        raise HTTPException(401, 'timestamp_out_of_range')

    # Signature
    if not verify_signature(raw_body, sig, CALLBACK_SECRET):
        raise HTTPException(403, 'bad_signature')

    # Now parse — после верификации
    import json
    event = json.loads(raw_body)

    # Идемпотентность: ключ — (internal_id, event, timestamp)
    id_key = f"{event.get('internal_id')}:{event.get('event')}:{event.get('timestamp')}"
    if already_seen(id_key):
        return {'ok': True, 'replay': True}

    enqueue(event)
    return {'ok': True}


def already_seen(key: str) -> bool:
    # Implement: Redis SETNX with 24h TTL, etc.
    return False


def enqueue(event: dict):
    # Implement: put on your queue / DB
    pass

Запуск:

uvicorn callback_server:app --host 0.0.0.0 --port 8080

HMAC: проверка соответствия

Один раз посчитайте контрольную подпись в shell и убедитесь, что Python даёт то же:

echo -n '{"phone":"+380501234567"}' | openssl dgst -sha256 -hmac 'test_secret'

import hmac, hashlib
body = b'{"phone":"+380501234567"}'
secret = b'test_secret'
print(hmac.new(secret, body, hashlib.sha256).hexdigest())

Должны быть одинаковые hex-строки. Если разные — почти всегда виновата сериализация JSON (порядок ключей, пробелы, ensure_ascii).