What are Webhooks?

Webhooks are automated messages sent from Capmo to a URL you provide when specific events occur in your projects. Instead of you repeatedly asking our API if something new happened, webhooks proactively notify your system in near real-time.
To set up a webhook, you need to go to the Company Administration Settings -> Webhooks, and provide:

A name for the webhook.
A secret or token for authentication (see below).
The URL endpoint on your server where Capmo should send notifications.
The specific events you want to be notified about

You can use the Webhook creation and editing UI to test the connection to your server.

Authentication

When creating a webhook, you provide a secret/token. Capmo will use this secret to sign the event message payload that is sent to your server. Your application should use this secret to verify that the payload in the incoming request genuinely originated from Capmo. This ensures the security and integrity of the notifications you receive.

The signature is passed in the X-Capmo-Signature header, and the contents of the message part of the request body should be validated against it.

Responses

Your server should respond with 2xx status code on success, any other status code will be treated as a failure.

If an outbound webhook request from Capmo to your server fails, the system includes a retry mechanism designed to ensure successful delivery.

Examples of server implementation

const fastify = require("fastify")({ logger: true });
const crypto = require("crypto");

const WEBHOOK_SECRET = "your-webhook-secret";

fastify.post("/webhook", async (request, reply) => {
  const signature = request.headers["x-capmo-signature"];
  const payload = request.body.message;
  const computedSignature = crypto
    .createHmac("sha256", WEBHOOK_SECRET)
    .update(payload)
    .digest("base64");

  if (signature !== computedSignature) {
    reply.code(403);
    return { status: "error", message: "Invalid signature" };
  }

  return { status: "success", message: "Webhook received successfully" };
});

const start = async () => {
  try {
    await fastify.listen({ port: 3008, host: "0.0.0.0" });
    console.log(`Server is running on ${fastify.server.address().port}`);
  } catch (err) {
    fastify.log.error(err);
    process.exit(1);
  }
};

start();

<?php

define('WEBHOOK_SECRET', 'your-webhook-secret');
define('SIGNATURE_HEADER', 'X-Capmo-Signature');
define('HMAC_ALGORITHM', 'sha256');
define('EXPECTED_PATH', '/webhook');

$requestPath = strtok($_SERVER['REQUEST_URI'], '?');
if ($requestPath !== EXPECTED_PATH) {
    http_response_code(404);
    header('Content-Type: application/json; charset=utf-8');
    echo json_encode(['status' => 'error', 'message' => 'Not Found']);
    exit;
}

if ($_SERVER['REQUEST_METHOD'] !== 'POST') {
    http_response_code(405);
    header('Content-Type: application/json; charset=utf-8');
    echo json_encode(['status' => 'error', 'message' => 'Method Not Allowed']);
    exit;
}

$receivedSignature = null;
$headerKeyToCheck = 'HTTP_' . strtoupper(str_replace('-', '_', SIGNATURE_HEADER));
if (isset($_SERVER[$headerKeyToCheck])) {
    $receivedSignature = $_SERVER[$headerKeyToCheck];
} elseif (function_exists('getallheaders')) {
    $headers = getallheaders();
    if ($headers !== false) {
        foreach ($headers as $name => $value) {
            if (strcasecmp($name, SIGNATURE_HEADER) === 0) {
                $receivedSignature = $value;
                break;
            }
        }
    }
}

$rawBody = file_get_contents('php://input');
$data = json_decode($rawBody, true);
$messageToSign = (string) $data['message'];
$computedSignatureBytes = hash_hmac(HMAC_ALGORITHM, $messageToSign, WEBHOOK_SECRET, true);
$computedSignature = base64_encode($computedSignatureBytes);

if (!hash_equals($receivedSignature, $computedSignature)) {
    http_response_code(403);
    header('Content-Type: application/json; charset=utf-8');
    echo json_encode(['status' => 'error', 'message' => 'Invalid signature']);
    error_log("Webhook Error: Invalid signature on " . EXPECTED_PATH . ". Received: $receivedSignature, Computed (from message field): $computedSignature");
    exit;
}

error_log("Webhook received successfully on " . EXPECTED_PATH . " (Validated against 'message' field)");

http_response_code(200);
header('Content-Type: application/json; charset=utf-8');
echo json_encode(['status' => 'success', 'message' => 'Webhook received successfully']);
exit;

?>

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"crypto/subtle"
	"encoding/base64"
	"encoding/json"
	"io"
	"log"
	"net/http"
)

const (
	webhookSecret   = "your-webhook-secret"
	signatureHeader = "X-Capmo-Signature"
	listenAddress   = ":3008"
	webhookEndpoint = "/webhook"
)

type webhookPayload struct {
	Message string `json:"message"`
}

type webhookResponse struct {
	Status  string `json:"status"`
	Message string `json:"message"`
}

func main() {
	http.HandleFunc(webhookEndpoint, webhookHandler)

	log.Printf("Server starting on %s%s\n", listenAddress, webhookEndpoint)
	err := http.ListenAndServe(listenAddress, nil)
	if err != nil {
		log.Fatalf("Server failed to start: %v", err)
	}
}

func webhookHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		sendJSONError(w, "Method Not Allowed", http.StatusMethodNotAllowed)
		return
	}

	receivedSignatureB64 := r.Header.Get(signatureHeader)

	bodyBytes, _ := io.ReadAll(r.Body)
	defer r.Body.Close()

	var payload webhookPayload
	json.Unmarshal(bodyBytes, &payload)

	computedHMAC := calculateHMACSHA256([]byte(payload.Message), []byte(webhookSecret))
	computedSignatureB64 := base64.StdEncoding.EncodeToString(computedHMAC)

	receivedSignatureBytes, _ := base64.StdEncoding.DecodeString(receivedSignatureB64)

	isValid := subtle.ConstantTimeCompare(receivedSignatureBytes, computedHMAC) == 1

	if !isValid {
		log.Printf("Webhook Error: Invalid signature. Received: %s, Computed (from message): %s\n", receivedSignatureB64, computedSignatureB64)
		sendJSONError(w, "Invalid signature", http.StatusForbidden)
		return
	}

	log.Printf("Webhook received successfully! (Validated against 'message' field: %q)\n", payload.Message)

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	json.NewEncoder(w).Encode(webhookResponse{
		Status:  "success",
		Message: "Webhook received successfully",
	})
}

func calculateHMACSHA256(data []byte, secret []byte) []byte {
	h := hmac.New(sha256.New, secret)
	h.Write(data)
	return h.Sum(nil)
}

func sendJSONError(w http.ResponseWriter, message string, statusCode int) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(statusCode)
	json.NewEncoder(w).Encode(webhookResponse{
		Status:  "error",
		Message: message,
	})
}

#!/usr/bin/env python3

import hmac
import hashlib
import base64
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
import logging
import binascii

WEBHOOK_SECRET = b"your-webhook-secret"
SIGNATURE_HEADER = "X-Capmo-Signature"
HMAC_ALGORITHM = hashlib.sha256
EXPECTED_PATH = "/webhook"
HOST = "0.0.0.0"
PORT = 3008


class WebhookHandler(BaseHTTPRequestHandler):
    def _send_json_response(self, status_code, response_data):
        """Helper to send JSON responses."""
        self.send_response(status_code)
        self.send_header("Content-type", "application/json; charset=utf-8")
        self.end_headers()
        self.wfile.write(json.dumps(response_data).encode("utf-8"))

    def do_POST(self):
        """Handles POST requests."""
        if self.path != EXPECTED_PATH:
            response = {"status": "error", "message": "Not Found"}
            self._send_json_response(404, response)
            logging.warning(f"Received request for wrong path: {self.path}")
            return

        received_signature_b64 = self.headers.get(SIGNATURE_HEADER)
        content_length = int(self.headers.get("Content-Length", 0))
        body_bytes = self.rfile.read(content_length)

        body_string = body_bytes.decode("utf-8")
        data = json.loads(body_string)

        message_to_sign = str(data["message"])
        message_bytes = message_to_sign.encode("utf-8")

        computed_hmac = hmac.new(WEBHOOK_SECRET, message_bytes, HMAC_ALGORITHM)
        computed_digest = computed_hmac.digest()
        computed_signature_b64 = base64.b64encode(computed_digest).decode("utf-8")

        received_signature_bytes = base64.b64decode(
            received_signature_b64.encode("utf-8")
        )

        if not hmac.compare_digest(received_signature_bytes, computed_digest):
            logging.warning(
                f"Webhook Error: Invalid signature. Received: {received_signature_b64}, Computed (from message): {computed_signature_b64}"
            )
            response = {"status": "error", "message": "Invalid signature"}
            self._send_json_response(403, response)
            return

        logging.info(
            f"Webhook received successfully! (Validated against 'message' field: '{message_to_sign}')"
        )

        response = {"status": "success", "message": "Webhook received successfully"}
        self._send_json_response(200, response)


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
    )
    server_address = (HOST, PORT)
    httpd = HTTPServer(server_address, WebhookHandler)
    logging.info(f"Starting webhook server on {HOST}:{PORT}{EXPECTED_PATH}...")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        logging.info("Server shutting down.")
        httpd.server_close()

Payload examples

You can view example payloads for all events with Webhooks enabled in the Capmo Webhook API section of this documentation (see the sidebar).