> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.voicebip.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.voicebip.com/_mcp/server.

# Code Examples

> Production-ready webhook handlers in Express.js, Flask, Go, and Next.js, plus LLM integration examples with OpenAI and Claude.

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.

```javascript
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.

```python
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.

```go
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.

```typescript
// 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.

```javascript
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.

```python
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)
```

> **Tip**
>
> 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](/reference/best-practices) for latency optimization tips.