x-abv-signature: t=1720701136,s=0123abcdef...

# Example: Use event ID to deduplicate
processed_events = set()  # In production, use Redis or database

def handle_webhook(event):
    event_id = event["id"]

    # Skip if already processed
    if event_id in processed_events:
        return {"status": "already_processed"}

    # Process event
    send_slack_notification(event)

    # Mark as processed
    processed_events.add(event_id)

    return {"status": "success"}

🔔 Prompt Updated

Name: movie-critic
Version: 3
Action: created
Labels: production, latest
Tags: entertainment

Commit message: "Improved critic persona"

View in ABV →

https://your-domain.com/api/webhooks/abv-prompts

X-API-Key: your-api-key
Authorization: Bearer your-token

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
from typing import Dict, Any

app = FastAPI()

class WebhookPayload(BaseModel):
    id: str
    timestamp: str
    type: str
    action: str
    prompt: Dict[str, Any]

@app.post("/api/webhooks/abv-prompts")
async def handle_abv_webhook(payload: WebhookPayload):
    """Process ABV prompt webhook."""
    print(f"Received {payload.action} event for prompt {payload.prompt['name']} v{payload.prompt['version']}")

    # Your custom logic here:
    # - Send notification to Microsoft Teams
    # - Trigger CI/CD pipeline
    # - Update documentation
    # - Log to audit system
    # etc.

    # Return 2xx to acknowledge receipt
    return {"status": "received", "event_id": payload.id}

import express from 'express';

const app = express();
app.use(express.json());

app.post('/api/webhooks/abv-prompts', (req, res) => {
  const { id, action, prompt } = req.body;

  console.log(`Received ${action} event for prompt ${prompt.name} v${prompt.version}`);

  // Your custom logic here

  res.status(200).json({ status: 'received', event_id: id });
});

app.listen(3000, () => console.log('Webhook handler listening on port 3000'));

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": "2024-07-10T10:30:00Z",
  "type": "prompt-version",
  "apiVersion": "v1",
  "action": "created",
  "prompt": {
    "id": "prompt_abc123",
    "name": "movie-critic",
    "version": 3,
    "projectId": "xyz789",
    "labels": ["production", "latest"],
    "prompt": "As a {{criticLevel}} movie critic, rate {{movie}} out of 10.",
    "type": "text",
    "config": { "model": "gpt-4o", "temperature": 0.7 },
    "commitMessage": "Improved critic persona",
    "tags": ["entertainment"],
    "createdAt": "2024-07-10T10:30:00Z",
    "updatedAt": "2024-07-10T10:30:00Z"
  }
}

x-abv-signature: t=1720701136,s=a1b2c3d4e5f6...

import hmac
import hashlib
from typing import Optional


def verify_abv_signature(
    raw_body: str,
    signature_header: str,
    secret: str,
) -> bool:
    """
    Validate a ABV webhook/event signature.

    Parameters
    ----------
    raw_body : str
        The request body exactly as received (no decoding or reformatting).
    signature_header : str
        The value of the `x-abv-signature` header, e.g. "t=1720701136,s=0123abcd...".
    secret : str
        Your ABV signing secret.

    Returns
    -------
    bool
        True if the signature is valid, otherwise False.
    """
    # Split "t=timestamp,s=signature" into the two expected key/value chunks
    try:
        ts_pair, sig_pair = signature_header.split(",", 1)
    except ValueError:  # wrong format / missing comma
        return False

    # Extract values (everything after the first "=")
    if "=" not in ts_pair or "=" not in sig_pair:
        return False
    timestamp = ts_pair.split("=", 1)[1]
    received_sig_hex = sig_pair.split("=", 1)[1]

    # Recreate the message and compute the expected HMAC-SHA256 hex digest
    message = f"{timestamp}.{raw_body}".encode("utf-8")
    expected_sig_hex = hmac.new(
        secret.encode("utf-8"), message, hashlib.sha256
    ).hexdigest()

    # Use constant-time comparison on the *decoded* byte strings
    try:
        return hmac.compare_digest(
            bytes.fromhex(received_sig_hex), bytes.fromhex(expected_sig_hex)
        )
    except ValueError:  # received_sig_hex isn't valid hex
        return False

from fastapi import FastAPI, Request, HTTPException, Header

WEBHOOK_SECRET = "your-signing-secret-from-abv"

@app.post("/api/webhooks/abv-prompts")
async def handle_webhook(
    request: Request,
    x_abv_signature: str = Header(...)
):
    # Read raw body (important: don't decode or parse yet)
    raw_body = await request.body()

    # Verify signature
    if not verify_abv_signature(raw_body.decode(), x_abv_signature, WEBHOOK_SECRET):
        raise HTTPException(status_code=401, detail="Invalid signature")

    # Parse JSON after verification
    payload = await request.json()

    # Process webhook...
    return {"status": "received"}

import crypto from "crypto";

export function verifyABVSignature(
  rawBody: string,
  signatureHeader: string,
  secret: string
): boolean {
  // Parse header: "t=timestamp,s=signature"
  const [tsPair, sigPair] = signatureHeader.split(",");
  if (!tsPair || !sigPair) return false;

  const timestamp = tsPair.split("=")[1];
  const receivedSig = sigPair.split("=")[1];

  // Recreate signature
  const expectedSig = crypto
    .createHmac("sha256", secret)
    .update(`${timestamp}.${rawBody}`, "utf8")
    .digest("hex");

  // Constant-time comparison
  try {
    return crypto.timingSafeEqual(
      Buffer.from(receivedSig, "hex"),
      Buffer.from(expectedSig, "hex")
    );
  } catch {
    return false;  // receivedSig not valid hex
  }
}

import express from 'express';
import { verifyABVSignature } from './verify-signature';

const WEBHOOK_SECRET = process.env.ABV_WEBHOOK_SECRET!;

const app = express();

// Use express.text() to get raw body as string
app.post('/api/webhooks/abv-prompts',
  express.text({ type: 'application/json' }),
  (req, res) => {
    const signature = req.headers['x-abv-signature'] as string;

    // Verify signature
    if (!verifyABVSignature(req.body, signature, WEBHOOK_SECRET)) {
      return res.status(401).json({ error: 'Invalid signature' });
    }

    // Parse JSON after verification
    const payload = JSON.parse(req.body);

    // Process webhook...
    res.json({ status: 'received', event_id: payload.id });
  }
);

import time

def verify_abv_signature_with_timestamp(raw_body, signature_header, secret, tolerance=300):
    # ... existing verification code ...

    # After signature verification passes, check timestamp
    current_time = int(time.time())
    timestamp_int = int(timestamp)

    if abs(current_time - timestamp_int) > tolerance:
        # Reject if timestamp is > 5 minutes old (replay attack protection)
        return False

    return True

# Correct: convert hex to bytes before comparison
return hmac.compare_digest(
    bytes.fromhex(received_sig_hex),
    bytes.fromhex(expected_sig_hex)
)

🚨 Production Prompt Updated

Name: customer-support/billing
Version: 12
Action: created
Label: production

Commit: "Updated refund policy for EU customers"

@channel - New billing prompt deployed

from elasticsearch import Elasticsearch

es = Elasticsearch(["https://your-elasticsearch-cluster"])

@app.post("/api/webhooks/abv-prompts")
async def sync_to_elasticsearch(payload: WebhookPayload):
    # Index prompt in Elasticsearch
    es.index(
        index="prompts",
        id=payload.prompt["id"],
        document={
            "name": payload.prompt["name"],
            "version": payload.prompt["version"],
            "content": payload.prompt["prompt"],
            "labels": payload.prompt["labels"],
            "tags": payload.prompt["tags"],
            "updated_at": payload.timestamp
        }
    )

    return {"status": "synced"}

{
  "event_id": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": "2024-07-10T10:30:00Z",
  "action": "created",
  "resource_type": "prompt",
  "resource_id": "prompt_abc123",
  "resource_name": "customer-support/gdpr-request-handler",
  "version": 5,
  "labels": ["production"],
  "commit_message": "Updated GDPR data deletion flow per legal review",
  "actor": "[email protected]",
  "changes": {
    "before": {...},
    "after": {...}
  }
}