Reference

Code Examples

Production-ready examples for common Voicebip integrations. Every example includes HMAC-SHA256 signature verification using the X-Voicebip-Signature header to ensure webhook authenticity.

Express.js BYOM Webhook Handler

A Node.js Express server that receives call.transcription events, verifies the HMAC signature, and returns a BYOM response.

const express = require("express");
const crypto = require("crypto");
const app = express();
const SIGNING_SECRET = process.env.VOICEBIP_SIGNING_SECRET;
// Parse raw body for HMAC verification
app.use(express.json({
  verify: (req, _res, buf) => { req.rawBody = buf; }
}));
function verifySignature(req) {
  const signature = req.headers["x-voicebip-signature"];
  const timestamp = req.headers["x-voicebip-timestamp"];
  if (!signature || !timestamp) return false;
  // Reject replays older than 5 minutes
  if (Math.abs(Date.now() / 1000 - parseInt(timestamp, 10)) > 300) return false;
  // Signed payload is "{timestamp}.{body}"
  const signedPayload = Buffer.concat([
    Buffer.from(timestamp + "."),
    req.rawBody,
  ]);
  const expected = "sha256=" + crypto
    .createHmac("sha256", SIGNING_SECRET)
    .update(signedPayload)
    .digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(expected),
    Buffer.from(signature)
  );
}
app.post("/webhook", (req, res) => {
  if (!verifySignature(req)) {
    return res.status(401).json({ error: "Invalid signature" });
  }
  const { event_id, event_type, payload } = req.body;
  // Deduplicate using event_id (idempotency)
  // In production, check event_id against a cache or database
  if (event_type === "call.transcription") {
    const transcript = payload.text;
    console.log(`[${event_id}] Caller said: ${transcript}`);
    return res.json({
      text: `You said: ${transcript}. How can I help you today?`,
      end_call: false,
    });
  }
  res.json({ status: "ok" });
});
app.listen(3000, () => console.log("Webhook handler listening on port 3000"));

Flask BYOM Webhook Handler

A Python Flask server handling BYOM webhook events with HMAC-SHA256 signature verification.

import hmac
import hashlib
import os
import time
from flask import Flask, request, jsonify, abort
app = Flask(__name__)
SIGNING_SECRET = os.environ["VOICEBIP_SIGNING_SECRET"]
def verify_signature(req):
    signature = req.headers.get("X-Voicebip-Signature", "")
    timestamp = req.headers.get("X-Voicebip-Timestamp", "")
    if not signature or not timestamp:
        return False
    # Reject replays older than 5 minutes
    if abs(time.time() - int(timestamp)) > 300:
        return False
    # Signed payload is "{timestamp}.{body}"
    signed_payload = f"{timestamp}.".encode() + req.get_data()
    expected = "sha256=" + hmac.new(
        SIGNING_SECRET.encode(), signed_payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)
@app.route("/webhook", methods=["POST"])
def webhook():
    if not verify_signature(request):
        abort(401, description="Invalid signature")
    payload = request.json
    event_type = payload.get("event_type")
    event_id = payload.get("event_id")
    # Deduplicate using event_id (idempotency)
    # In production, check event_id against a cache or database
    if event_type == "call.transcription":
        transcript = payload["payload"]["text"]
        return jsonify({
            "text": f"You said: {transcript}. How can I help?",
            "end_call": False,
        })
    return jsonify({"status": "ok"})
if __name__ == "__main__":
    app.run(port=3000)

Go Webhook Handler

A Go net/http handler with constant-time HMAC-SHA256 signature comparison.

package main
import (
	"crypto/hmac"
	"crypto/sha256"
	"crypto/subtle"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"math"
	"net/http"
	"os"
	"strconv"
	"time"
)
var signingSecret = os.Getenv("VOICEBIP_SIGNING_SECRET")
type WebhookEvent struct {
	EventID   string          `json:"event_id"`
	EventType string          `json:"event_type"`
	Channel   string          `json:"channel"`
	AgentID   string          `json:"agent_id"`
	Timestamp string          `json:"timestamp"`
	Payload   json.RawMessage `json:"payload"`
}
type TranscriptionPayload struct {
	CallID     string  `json:"call_id"`
	Text       string  `json:"text"`
	IsFinal    bool    `json:"is_final"`
	Confidence float64 `json:"confidence"`
}
type BYOMResponse struct {
	Text    string `json:"text"`
	EndCall bool   `json:"end_call"`
}
// verifySignature checks HMAC-SHA256 over "{timestamp}.{body}".
func verifySignature(timestamp string, body []byte, signature string) bool {
	// Reject replays older than 5 minutes.
	ts, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil {
		return false
	}
	if math.Abs(float64(time.Now().Unix()-ts)) > 300 {
		return false
	}
	// Signed payload is "{timestamp}.{body}"
	mac := hmac.New(sha256.New, []byte(signingSecret))
	mac.Write([]byte(timestamp + "."))
	mac.Write(body)
	expected := "sha256=" + hex.EncodeToString(mac.Sum(nil))
	return subtle.ConstantTimeCompare([]byte(expected), []byte(signature)) == 1
}
func webhookHandler(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "failed to read body", http.StatusBadRequest)
		return
	}
	defer r.Body.Close()
	signature := r.Header.Get("X-Voicebip-Signature")
	timestamp := r.Header.Get("X-Voicebip-Timestamp")
	if !verifySignature(timestamp, body, signature) {
		http.Error(w, "invalid signature", http.StatusUnauthorized)
		return
	}
	var event WebhookEvent
	if err := json.Unmarshal(body, &event); err != nil {
		http.Error(w, "invalid JSON", http.StatusBadRequest)
		return
	}
	// Deduplicate using event.EventID (idempotency)
	// In production, check EventID against a cache or database
	switch event.EventType {
	case "call.transcription":
		var tp TranscriptionPayload
		if err := json.Unmarshal(event.Payload, &tp); err != nil {
			http.Error(w, "invalid payload", http.StatusBadRequest)
			return
		}
		log.Printf("[%s] Caller said: %s", event.EventID, tp.Text)
		resp := BYOMResponse{
			Text:    fmt.Sprintf("You said: %s. How can I help?", tp.Text),
			EndCall: false,
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(resp)
	default:
		w.Header().Set("Content-Type", "application/json")
		w.Write([]byte(`{"status":"ok"}`))
	}
}
func main() {
	http.HandleFunc("/webhook", webhookHandler)
	log.Println("Webhook handler listening on :3000")
	log.Fatal(http.ListenAndServe(":3000", nil))
}

Next.js API Route

A Next.js 14 App Router POST handler with HMAC-SHA256 signature verification.

// app/api/webhook/route.ts
import { NextRequest, NextResponse } from "next/server";
import crypto from "crypto";
const SIGNING_SECRET = process.env.VOICEBIP_SIGNING_SECRET!;
function verifySignature(body: string, timestamp: string, signature: string): boolean {
  // Reject replays older than 5 minutes
  if (Math.abs(Date.now() / 1000 - parseInt(timestamp, 10)) > 300) return false;
  // Signed payload is "{timestamp}.{body}"
  const signedPayload = `${timestamp}.${body}`;
  const expected =
    "sha256=" +
    crypto.createHmac("sha256", SIGNING_SECRET).update(signedPayload).digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(expected),
    Buffer.from(signature)
  );
}
export async function POST(req: NextRequest) {
  const body = await req.text();
  const signature = req.headers.get("x-voicebip-signature") ?? "";
  const timestamp = req.headers.get("x-voicebip-timestamp") ?? "";
  if (!verifySignature(body, timestamp, signature)) {
    return NextResponse.json({ error: "Invalid signature" }, { status: 401 });
  }
  const event = JSON.parse(body);
  // Deduplicate using event.event_id (idempotency)
  // In production, check event_id against a cache or database
  if (event.event_type === "call.transcription") {
    const transcript = event.payload.text;
    console.log(`[${event.event_id}] Caller said: ${transcript}`);
    return NextResponse.json({
      text: `You said: ${transcript}. How can I help you today?`,
      end_call: false,
    });
  }
  return NextResponse.json({ status: "ok" });
}

Voice + OpenAI Integration

An Express handler that receives a call.transcription event, forwards the transcript to the OpenAI Chat Completions API, and returns the response. Includes HMAC signature verification.

const express = require("express");
const crypto = require("crypto");
const OpenAI = require("openai");
const app = express();
const SIGNING_SECRET = process.env.VOICEBIP_SIGNING_SECRET;
const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
app.use(express.json({
  verify: (req, _res, buf) => { req.rawBody = buf; }
}));
function verifySignature(req) {
  const signature = req.headers["x-voicebip-signature"];
  const timestamp = req.headers["x-voicebip-timestamp"];
  if (!signature || !timestamp) return false;
  // Reject replays older than 5 minutes
  if (Math.abs(Date.now() / 1000 - parseInt(timestamp, 10)) > 300) return false;
  // Signed payload is "{timestamp}.{body}"
  const signedPayload = Buffer.concat([
    Buffer.from(timestamp + "."),
    req.rawBody,
  ]);
  const expected = "sha256=" + crypto
    .createHmac("sha256", SIGNING_SECRET)
    .update(signedPayload)
    .digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(expected),
    Buffer.from(signature)
  );
}
app.post("/webhook", async (req, res) => {
  if (!verifySignature(req)) {
    return res.status(401).json({ error: "Invalid signature" });
  }
  const { event_type, payload } = req.body;
  if (event_type === "call.transcription") {
    // Build messages from conversation_history (last 20 turns included by Voicebip)
    const messages = [
      {
        role: "system",
        content:
          "You are a helpful Nigerian customer support agent for Voicebip. " +
          "Be concise — the caller is on a voice call. Keep responses under 2 sentences.",
      },
    ];
    if (payload.conversation_history) {
      for (const turn of payload.conversation_history) {
        messages.push({
          role: turn.role === "caller" ? "user" : "assistant",
          content: turn.text,
        });
      }
    }
    // Add the current transcription
    messages.push({ role: "user", content: payload.text });
    const completion = await openai.chat.completions.create({
      model: "gpt-4o",
      messages,
      max_tokens: 150,
      temperature: 0.7,
    });
    const reply = completion.choices[0].message.content;
    return res.json({
      text: reply,
      end_call: false,
    });
  }
  res.json({ status: "ok" });
});
app.listen(3000, () => console.log("OpenAI webhook handler on port 3000"));

Voice + Claude Integration

A Python handler that receives a call.transcription event, forwards the transcript to the Anthropic Messages API, and returns the response. Includes HMAC signature verification.

import hmac
import hashlib
import os
import time
import anthropic
from flask import Flask, request, jsonify, abort
app = Flask(__name__)
SIGNING_SECRET = os.environ["VOICEBIP_SIGNING_SECRET"]
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
def verify_signature(req):
    signature = req.headers.get("X-Voicebip-Signature", "")
    timestamp = req.headers.get("X-Voicebip-Timestamp", "")
    if not signature or not timestamp:
        return False
    # Reject replays older than 5 minutes
    if abs(time.time() - int(timestamp)) > 300:
        return False
    # Signed payload is "{timestamp}.{body}"
    signed_payload = f"{timestamp}.".encode() + req.get_data()
    expected = "sha256=" + hmac.new(
        SIGNING_SECRET.encode(), signed_payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)
@app.route("/webhook", methods=["POST"])
def webhook():
    if not verify_signature(request):
        abort(401, description="Invalid signature")
    payload = request.json
    event_type = payload.get("event_type")
    if event_type == "call.transcription":
        inner = payload["payload"]
        # Build messages from conversation_history (last 20 turns included by Voicebip)
        messages = []
        if inner.get("conversation_history"):
            for turn in inner["conversation_history"]:
                messages.append({
                    "role": "user" if turn["role"] == "caller" else "assistant",
                    "content": turn["text"],
                })
        # Add the current transcription
        messages.append({"role": "user", "content": inner["text"]})
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=150,
            system=(
                "You are a helpful Nigerian customer support agent for Voicebip. "
                "Be concise — the caller is on a voice call. Keep responses under 2 sentences."
            ),
            messages=messages,
        )
        reply = response.content[0].text
        return jsonify({
            "text": reply,
            "end_call": False,
        })
    return jsonify({"status": "ok"})
if __name__ == "__main__":
    app.run(port=3000)

For the lowest latency, use Voicebip’s hosted AI mode instead of BYOM. Hosted mode runs the AI provider inside Voicebip’s infrastructure with sub-500ms turn times. BYOM adds your webhook round-trip to the total latency. See Best Practices for latency optimization tips.