Примеры: Python

Python 3.10+ с requests и flask.

pip install requests flask

Клиент X-Hub

# xhub_client.py
import os, uuid, requests

BASE_URL = 'https://api.x-hub.online/api/v1'
HEADERS = {
    'X-Api-Key': os.environ['XHUB_API_KEY'],
    'X-Api-Secret': os.environ['XHUB_API_SECRET'],
}


class XHubError(Exception):
    pass


def create_payment(amount_rub, external_id, description, metadata=None):
    r = requests.post(
        f'{BASE_URL}/payments',
        headers={
            **HEADERS,
            'Idempotency-Key': str(uuid.uuid4()),
            'Content-Type': 'application/json',
        },
        json={
            'amount_rub': amount_rub,
            'external_id': external_id,
            'payment_method': 'sbp',
            'description': description,
            'metadata': metadata or {},
        },
        timeout=15,
    )
    if not r.ok:
        raise XHubError(f'X-Hub {r.status_code}: {r.json().get("error")}')
    return r.json()


def get_balance():
    r = requests.get(f'{BASE_URL}/balance', headers=HEADERS, timeout=10)
    r.raise_for_status()
    return r.json()

Webhook receiver (Flask)

# webhook_server.py
import os, hmac, hashlib, json
from flask import Flask, request, abort

app = Flask(__name__)
processed = set()  # в проде — Redis / БД

WEBHOOK_SECRET = os.environ['XHUB_WEBHOOK_SECRET']


@app.route('/webhook/xhub', methods=['POST'])
def webhook():
    raw = request.get_data()

    sig = request.headers.get('X-Webhook-Signature', '')
    timestamp = request.headers.get('X-Webhook-Timestamp', '')

    # Подпись: HMAC-SHA256(timestamp + "." + body, secret)
    signature_data = timestamp.encode() + b'.' + raw
    expected = 'sha256=' + hmac.new(
        WEBHOOK_SECRET.encode(), signature_data, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(sig, expected):
        abort(401, 'Invalid signature')

    event = json.loads(raw)
    dedupe_key = f"{event.get('payment_id') or event.get('withdrawal_id')}:{event.get('status')}"

    if dedupe_key in processed:
        return 'OK', 200
    processed.add(dedupe_key)

    if event['event'] == 'payment.status_changed':
        if event['status'] == 'paid':
            grant_access(event['external_id'], event.get('metadata', {}))

    return 'OK', 200


def grant_access(external_id, metadata):
    print(f'Granting access for {external_id}')


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)