צד המערכת
כל מה שצריך להגדיר בתוך הדשבורד לפני שה־Webhook הראשון נשלח. URL, מפתח חתימה, מדיניות ניסיון חוזר ומסיכת שדות. הפניית מטען מלאה וקטע אימות ל־Node.js בהמשך.
איפה ה־Webhooks נמצאים בדשבורד
פתיחת מפתחים → Webhooks. לכל סוכן קולי ו־WhatsApp בחשבון יש שורה משלו עם שלושה דברים שניתן לשלוט עליהם:
- URL — הנקודה הציבורית שאליה אנחנו שולחים POST בסיום כל שיחה.
- מפתח חתימה — לחישוב חתימת ה־HMAC. לחיצה על הצגה מאפשרת העתקה.
- מתג הפעלה — משהה את השליחות מבלי לאבד את ה־URL והסוד.
כותרות שאנחנו שולחים
בכל POST מופיעות ארבע כותרות מותאמות:
X-Webhook-Event— סוג האירוע, לדוגמהconversation.completed.X-Webhook-Id— מזהה שליחות ייחודי. ל־idempotency.X-Webhook-Timestamp— שניות Unix. לדחות כל מה שמעל 5 דקות.X-Webhook-Signature—sha256=<hex>— HMAC על גוף הבקשה הגולמי.
אימות HMAC (Node.js)
חשוב לוודא את החתימה לפני כל דבר אחר. לדחות אי־התאמות ולדחות כל מה שמעל 5 דקות:
// Node.js — אימות HMAC ודחיית מטענים מעל 5 דקות
import crypto from "node:crypto";
export function verifyWebhook(req, rawBody, secret) {
const sig = req.headers["x-webhook-signature"] ?? "";
const ts = req.headers["x-webhook-timestamp"] ?? "0";
if (Math.abs(Date.now() / 1000 - Number(ts)) > 300) return false;
const expected = "sha256=" + crypto
.createHmac("sha256", secret)
.update(rawBody, "utf8")
.digest("hex");
return crypto.timingSafeEqual(Buffer.from(sig), Buffer.from(expected));
}חשוב להשתמש ב־req.rawBody ולא ב־JSON המפוענח — פענוח JSON מחליף את סדר המפתחות ושובר את ה־digest.
מסיכת שדות — בחירה אילו שדות יישלחו
פתיחת מפתחים → Webhooks → שדות. המסיכה תקפה לכל החשבון (חלה על כל סוכן). אפשר להוריד תמלילים מסיבות פרטיות, או לשלוח רק עלות ומזהה סוכן למערכת חיוב שלא צריכה את כל המטען.
ברירת המחדל שולחת הכול — מומלץ לכבות רק מה שה־receiver באמת לא צריך.
אימות חתימת HMAC
import hmac
import hashlib
import time
from typing import Mapping
def verify_webhook(headers: Mapping[str, str], raw_body: bytes, secret: str) -> bool:
"""Return True iff the request is a fresh, correctly-signed webhook."""
sig = headers.get("x-webhook-signature", "")
ts = headers.get("x-webhook-timestamp", "0")
# Reject anything older than 5 minutes (replay-attack guard).
try:
if abs(time.time() - int(ts)) > 300:
return False
except ValueError:
return False
expected = "sha256=" + hmac.new(
secret.encode("utf-8"),
raw_body,
hashlib.sha256,
).hexdigest()
# Constant-time comparison so signing-side timing attacks are off the table.
return hmac.compare_digest(sig, expected)קליטת אירוע webhook
import os
import json
from flask import Flask, request, abort
from verify_webhook import verify_webhook
app = Flask(__name__)
SECRET = os.environ["AICALL_WEBHOOK_SECRET"]
@app.post("/webhooks/aicall")
def receive():
raw = request.get_data() # Bytes — DO NOT use request.json here.
if not verify_webhook(request.headers, raw, SECRET):
abort(401)
event = json.loads(raw)
kind = event.get("event")
if kind == "agent.conversation.completed":
# event["conversation"] holds duration, transcript, cost.
record_call(event["conversation"])
elif kind == "agent.call.failed":
notify_oncall(event["conversation"])
else:
# Unknown event type — log and ACK so we don't retry forever.
app.logger.info("ignored event kind=%s", kind)
# 2xx ACK — anything else triggers exponential-backoff retries.
return ("", 204)
def record_call(conv): ...
def notify_oncall(conv): ...