Webhooks

Cognethics sends real-time event notifications to your infrastructure via HTTPS webhooks. Every webhook is cryptographically signed with HMAC-SHA256, retried with exponential backoff on failure, and logged for debugging.

Overview

A webhook is an HTTP callback that Cognethics triggers when certain events occur in your organization. Instead of polling for changes, you can register a webhook endpoint and receive push notifications.

Key features:

HMAC-SHA256 signing — verify payload authenticity before processing
Exponential-backoff retries — automatic retry on failure (up to 5 attempts)
Delivery tracking — inspect every delivery attempt (status, response code, error message)
Event filtering — subscribe only to events you care about
Custom headers — inject authentication tokens or routing headers
Per-endpoint configuration — timeout, retry count, and headers per webhook

Getting Started

1. Create a webhook endpoint

Your endpoint must:

Accept HTTP POST requests
Return a 2xx status code within 30 seconds (default timeout)
Validate the X-Governance-Signature header before processing

Example (Python/Flask):

from flask import Flask, request
import hmac
import hashlib
import json
import time

app = Flask(__name__)

# Your webhook secret (store securely in environment)
WEBHOOK_SECRET = "whk_YOUR_SECRET_HERE"

@app.route('/webhooks/governance', methods=['POST'])
def handle_webhook():
    # Validate signature
    signature_header = request.headers.get('X-Governance-Signature', '')
    if not verify_signature(request.data, signature_header, WEBHOOK_SECRET):
        return {'error': 'Invalid signature'}, 401

    # Parse payload
    payload = request.get_json()
    event_type = payload['event_type']
    event_data = payload['data']

    # Process event
    print(f"Received {event_type} for org {payload['organization_id']}")

    return {'status': 'ok'}, 200

def verify_signature(body, signature_header, secret, tolerance_seconds=300):
    """Verify HMAC-SHA256 signature."""
    try:
        # Parse signature header: "t=timestamp,v1=signature"
        parts = dict(p.split('=') for p in signature_header.split(','))
        timestamp = int(parts.get('t', 0))
        received_sig = parts.get('v1', '')

        # Check timestamp tolerance (prevent replay attacks)
        current_time = int(time.time())
        if abs(current_time - timestamp) > tolerance_seconds:
            return False

        # Reconstruct message: "timestamp.json_payload"
        payload_json = json.dumps(json.loads(body), sort_keys=True, separators=(',', ':'))
        message = f"{timestamp}.{payload_json}"

        # Compute expected signature
        expected_sig = hmac.new(
            secret.encode('utf-8'),
            message.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

        # Constant-time comparison
        return hmac.compare_digest(expected_sig, received_sig)
    except Exception as e:
        print(f"Signature verification error: {e}")
        return False

2. Register the webhook via the API

POST to /api/v1/risk-assessment/governance/webhooks/ with:

curl -X POST https://gtm.cognethics.com/api/v1/risk-assessment/governance/webhooks/ \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "My Compliance Monitor",
    "description": "Notify our security team of governance decisions",
    "url": "https://your-domain.com/webhooks/governance",
    "events": ["governance.decision.block", "governance.decision.allow"],
    "filters": {
      "risk_levels": ["high", "critical"]
    },
    "custom_headers": {
      "Authorization": "Bearer YOUR_SECRET_TOKEN"
    },
    "max_retries": 3,
    "retry_delay_seconds": 60,
    "timeout_seconds": 30
  }'

Response:

{
  "id": "whk_550e8400e29b41d4a716446655440000",
  "name": "My Compliance Monitor",
  "url": "https://your-domain.com/webhooks/governance",
  "events": ["governance.decision.block", "governance.decision.allow"],
  "filters": {
    "risk_levels": ["high", "critical"]
  },
  "secret": "whk_secret_550e8400e29b41d4a716446655440000",
  "is_active": true,
  "created_at": "2026-05-21T15:30:00Z"
}

Store the secret securely. You'll need it to verify signatures. You can regenerate it at any time via PATCH.

3. Test the webhook

Use the Send test event button on the webhook config page (/admin/integrations/webhooks/) to dispatch a sample payload through the real delivery pipeline. The resulting WebhookDelivery row is flagged is_test=true and surfaces in the delivery history alongside production deliveries. You can also trigger a real governance decision in your system and inspect the delivery via the delivery-history API.

Webhook Events

governance.decision.block

Triggered when a governance decision blocks an action. The system evaluated policies against the request and determined it violates one or more rules.

Event structure:

{
  "event_id": "dec_550e8400e29b41d4a716446655440000",
  "event_type": "governance.decision.block",
  "timestamp": "2026-05-21T15:30:00Z",
  "organization_id": "org_550e8400e29b41d4a716446655440000",
  "api_version": "2024-12-31",
  "data": {
    "decision_id": "dec_550e8400e29b41d4a716446655440000",
    "decision": "BLOCK",
    "decision_type": "agent_action",
    "primary_reason": "Violation of data-access policy: classified resource",
    "ai_system": {
      "id": "sys_550e8400e29b41d4a716446655440000",
      "name": "Research Agent v2"
    },
    "policies_evaluated": 12,
    "policies_violated": 1,
    "latency_ms": 45,
    "trace_id": "trace_550e8400e29b41d4a716446655440000"
  }
}

governance.decision.allow

Triggered when a governance decision allows an action. The system evaluated policies and found no violations.

Same structure as governance.decision.block, with "decision": "ALLOW".

governance.violation.detected

Triggered when a policy violation is detected outside of a decision context (e.g., post-hoc audit, model drift alert).

Event structure:

{
  "event_id": "vio_550e8400e29b41d4a716446655440000",
  "event_type": "governance.violation.detected",
  "timestamp": "2026-05-21T15:30:00Z",
  "organization_id": "org_550e8400e29b41d4a716446655440000",
  "api_version": "2024-12-31",
  "data": {
    "violation_id": "vio_550e8400e29b41d4a716446655440000",
    "policy_name": "No High-Risk Data Access",
    "severity": "high",
    "ai_system": {
      "id": "sys_550e8400e29b41d4a716446655440000",
      "name": "Research Agent v2"
    },
    "detected_at": "2026-05-21T15:29:30Z"
  }
}

governance.oversight.requested

Triggered when a governance decision is escalated to human oversight (e.g., ambiguous case, manual review needed).

Event structure:

{
  "event_id": "ovr_550e8400e29b41d4a716446655440000",
  "event_type": "governance.oversight.requested",
  "timestamp": "2026-05-21T15:30:00Z",
  "organization_id": "org_550e8400e29b41d4a716446655440000",
  "api_version": "2024-12-31",
  "data": {
    "request_id": "ovr_550e8400e29b41d4a716446655440000",
    "status": "pending",
    "priority": "high",
    "oversight_reason": "Decision confidence below threshold",
    "sla_deadline": "2026-05-22T15:30:00Z"
  }
}

Delivery & Retries

When you register a webhook, Cognethics will attempt delivery whenever a matching event occurs.

Retry Logic

If your endpoint returns a non-2xx status code or times out, Cognethics retries with exponential backoff:

Attempt	Delay	Total elapsed
1	immediate	0s
2	1 second	1s
3	2 seconds	3s
4	4 seconds	7s
5	8 seconds	15s
After 5 attempts	—	marked FAILED

The retry_delay_seconds field in the webhook config sets the base delay; the formula is base * (2 ^ (attempt - 1)).

Delivery Status

Every delivery attempt is logged with:

Status — PENDING, RETRYING, SUCCESS, or FAILED
Response code — HTTP status code from your endpoint (null if connection failed)
Response body — first 1000 characters of your endpoint's response
Error message — details if delivery failed
Attempt count — which retry this was
Next retry at — timestamp of the next retry (null if no retry scheduled)

You can inspect delivery history via the API:

curl https://gtm.cognethics.com/api/v1/risk-assessment/governance/webhooks/whk_ID/deliveries/ \
  -H "Authorization: Bearer YOUR_API_TOKEN"

Response:

{
  "count": 42,
  "results": [
    {
      "id": "del_550e8400e29b41d4a716446655440000",
      "event_type": "governance.decision.block",
      "event_id": "dec_550e8400e29b41d4a716446655440000",
      "status": "success",
      "response_status_code": 200,
      "response_body": "{\"received\": true}",
      "error_message": null,
      "attempt_count": 1,
      "created_at": "2026-05-21T15:30:00Z",
      "delivered_at": "2026-05-21T15:30:01Z",
      "next_retry_at": null
    },
    {
      "id": "del_660e8400e29b41d4a716446655440001",
      "event_type": "governance.decision.allow",
      "event_id": "dec_660e8400e29b41d4a716446655440001",
      "status": "retrying",
      "response_status_code": 500,
      "response_body": "Internal Server Error",
      "error_message": "HTTP 500: Internal Server Error",
      "attempt_count": 2,
      "created_at": "2026-05-21T15:29:00Z",
      "delivered_at": null,
      "next_retry_at": "2026-05-21T15:30:05Z"
    }
  ]
}

Signing & Verification

How Signing Works

When Cognethics delivers a webhook, it computes an HMAC-SHA256 signature of the payload and includes it in the X-Governance-Signature header.

Header format:

X-Governance-Signature: t=1526326882,v1=abc123def456abc123def456abc123def456abc123def456abc123def456abc1

How it's computed:

Get the current Unix timestamp: t
Serialize the JSON payload with sorted keys and no whitespace: payload_json
Construct the message: message = f"{t}.{payload_json}"
Compute HMAC-SHA256: signature = hmac_sha256(secret, message)
Return header: t={t},v1={signature_hex}

In addition to the signature header, every delivery includes:

X-Governance-Event: {event_type}
X-Governance-Delivery: {delivery_id}
User-Agent: A7-Governance-Webhook/1.0
Content-Type: application/json

How to Verify

On receipt, verify the signature before processing.

Python:

import hmac
import hashlib
import json
import time

def verify_signature(body_bytes, signature_header, secret, tolerance_seconds=300):
    """
    Verify HMAC-SHA256 signature.

    Args:
        body_bytes: Raw request body (bytes)
        signature_header: X-Governance-Signature header value (string)
        secret: Webhook secret (string)
        tolerance_seconds: Max age of signature (default 5 minutes, prevent replay)

    Returns:
        True if signature is valid, False otherwise
    """
    try:
        # Parse header
        parts = dict(p.split('=') for p in signature_header.split(','))
        timestamp = int(parts.get('t', 0))
        received_sig = parts.get('v1', '')

        # Check timestamp (prevent replay attacks)
        current_time = int(time.time())
        if abs(current_time - timestamp) > tolerance_seconds:
            return False

        # Reconstruct message
        payload_json = json.dumps(json.loads(body_bytes), sort_keys=True, separators=(',', ':'))
        message = f"{timestamp}.{payload_json}"

        # Compute expected signature
        expected_sig = hmac.new(
            secret.encode('utf-8'),
            message.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

        # Constant-time comparison
        return hmac.compare_digest(expected_sig, received_sig)
    except Exception as e:
        print(f"Signature verification error: {e}")
        return False

Node.js:

const crypto = require('crypto');
const express = require('express');

const app = express();
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;

// Capture raw body before express.json() parses it
app.use((req, res, next) => {
  let rawBody = '';
  req.setEncoding('utf8');
  req.on('data', (chunk) => { rawBody += chunk; });
  req.on('end', () => {
    req.rawBody = rawBody;
    next();
  });
});

app.post('/webhooks/governance', (req, res) => {
  const signatureHeader = req.headers['x-governance-signature'];
  const body = req.rawBody;

  if (!verifySignature(body, signatureHeader, WEBHOOK_SECRET)) {
    return res.status(401).json({ error: 'Signature verification failed' });
  }

  const payload = JSON.parse(body);
  // Process payload...

  res.json({ status: 'ok' });
});

function verifySignature(body, signatureHeader, secret, toleranceSeconds = 300) {
  try {
    const parts = {};
    signatureHeader.split(',').forEach((part) => {
      const [key, value] = part.split('=');
      parts[key] = value;
    });

    const timestamp = parseInt(parts.t || '0', 10);
    const receivedSig = parts.v1 || '';

    const currentTime = Math.floor(Date.now() / 1000);
    if (Math.abs(currentTime - timestamp) > toleranceSeconds) return false;

    const parsed = JSON.parse(body);
    const payloadJson = JSON.stringify(parsed, Object.keys(parsed).sort());
    const message = `${timestamp}.${payloadJson}`;

    const expectedSig = crypto
      .createHmac('sha256', secret)
      .update(message)
      .digest('hex');

    return crypto.timingSafeEqual(
      Buffer.from(expectedSig, 'utf8'),
      Buffer.from(receivedSig, 'utf8')
    );
  } catch (error) {
    console.error('Signature verification error:', error);
    return false;
  }
}

Why Verify?

Signature verification ensures:

Authenticity — the webhook came from Cognethics, not an attacker spoofing the endpoint
Integrity — the payload was not modified in transit
Freshness — the webhook is not a replay of an old event (via timestamp validation)

Always verify before processing.

Management

List webhooks

GET /api/v1/risk-assessment/governance/webhooks/

Get a single webhook

GET /api/v1/risk-assessment/governance/webhooks/{id}/

Update a webhook

PATCH /api/v1/risk-assessment/governance/webhooks/{id}/

You can update the URL, events, filters, headers, retry config, etc. To rotate the secret:

PATCH /api/v1/risk-assessment/governance/webhooks/{id}/ \
  -d '{ "secret": "whk_new_secret_..." }'

Disable a webhook

PATCH /api/v1/risk-assessment/governance/webhooks/{id}/ \
  -d '{ "is_active": false }'

Inactive webhooks do not deliver events.

Delete a webhook

DELETE /api/v1/risk-assessment/governance/webhooks/{id}/

Filtering

When you register a webhook, you can filter which events it receives:

{
  "events": ["governance.decision.block"],
  "filters": {
    "ai_system_ids": ["sys_550e8400e29b41d4a716446655440000"],
    "risk_levels": ["high", "critical"],
    "decision_types": ["agent_action"]
  }
}

events — list of event types. If omitted or empty, the webhook receives all governance event types.
ai_system_ids — limit to events from specific AI systems. If omitted, all systems.
risk_levels — limit to high-risk or critical events. If omitted, all risk levels.
decision_types — limit to specific decision types (e.g., agent_action). If omitted, all types.

Filters are AND-ed: the event must match all specified filters to trigger the webhook.

Platform Event Types

In addition to the governance events documented above, the following platform-wide event types are available. Subscribe by adding them to the events array on your webhook.

Event type	When it fires
`document.uploaded`	A new document is uploaded or imported.
`document.processed`	Document processing completes (text extraction + chunking).
`agent.completed`	An agent instance retires successfully.
`agent.failed`	An agent instance terminates with an error.
`mission.started`	A mission transitions out of planning into active.
`mission.completed`	A mission is marked completed.
`mission.budget_exceeded`	A mission's accumulated cost crosses its budget ceiling.
`audit.entry_created`	A new entry is appended to the hash-chained governance audit ledger.
`claim.submitted`	An insurance claim is submitted to a payer.
`invoice.received`	A vendor invoice is recorded.
`ticket.created`	A new ITSM ticket is created.

Each event uses the same envelope shape (event_id / event_type / timestamp / organization_id / data / api_version) and the same HMAC-SHA256 signing as the governance events. The data block carries event-specific fields; the entity_crud/core/webhook_event/list MCP handler returns the full registry programmatically.

Troubleshooting

Webhook not triggering

Check is_active — inactive webhooks never deliver.
Check events — the event type must be in the webhook's events list (or events must be empty).
Check filters — the event must match all filters.
Inspect delivery history — fetch /api/v1/risk-assessment/governance/webhooks/{id}/deliveries/ to see if Cognethics tried to deliver.

Delivery failing

Check response code — is your endpoint returning 2xx?
Check timeout — is your endpoint responding within timeout_seconds?
Check signature — are you validating the signature correctly? Test with the example code above.
Check retry schedule — Cognethics will retry up to 5 times. Check next_retry_at in the delivery record.

"Invalid signature" error

Verify the secret — make sure you're using the correct secret from the webhook record.
Save the raw body — don't parse and re-serialize the body; use the raw bytes you received.
Check the timestamp — if the signature is more than 5 minutes old, it's rejected by default.

API Reference

Create a webhook

POST /api/v1/risk-assessment/governance/webhooks/

Request body:

{
  "name": "string (required, max 255 chars)",
  "description": "string (optional)",
  "url": "https://... (required, must be HTTPS)",
  "events": ["governance.decision.block", "..."],
  "filters": {
    "ai_system_ids": ["..."],
    "risk_levels": ["high", "critical"],
    "decision_types": ["..."]
  },
  "custom_headers": {
    "Authorization": "Bearer ...",
    "X-Custom-Header": "value"
  },
  "max_retries": 3,
  "retry_delay_seconds": 60,
  "timeout_seconds": 30
}

Response: 201 Created with webhook object.