Skip to main content
Datagrid webhooks let you subscribe to asynchronous events and receive signed HTTP callbacks at your endpoint.

​
Register a webhook

Create a webhook with an HTTPS endpoint and one or more event types. 
import os
from datagrid_ai import Datagrid

client = Datagrid(api_key=os.environ["DATAGRID_API_KEY"])

webhook = client.webhooks.create(
    url="https://example.com/datagrid/webhooks",
    events=["knowledge.processing.completed"],
)

print(webhook.id)
print(webhook.secret)  # only returned at creation time
Store secret immediately. It is shown only on creation and is used for signature verification.

​
List active webhooks for an event

Use this endpoint to retrieve enabled webhook subscriptions for a specific event type.
curl
curl -X GET "https://api.datagrid.com/v1/webhooks/active?event_type=knowledge.processing.completed" \
  -H "Authorization: Bearer $DATAGRID_API_KEY"

​
Event payload format

Webhook deliveries use JSON with this shape: 
{
  "id": "evt_123",
  "event_type": "knowledge.processing.completed",
  "timestamp": 1744230000,
  "data": {
    "knowledge_id": "kn_123",
    "teamspace_id": "org_123",
    "scope": "private",
    "status": "ready",
    "row_counts": {
      "tables": 2,
      "rows": 1000
    }
  }
}
  • id: unique event id, use for idempotency/deduplication.
  • event_type: currently knowledge.processing.completed.
  • timestamp: event timestamp (Unix seconds).
  • data: event-specific payload.

​
Signature verification

Each delivery includes a Datagrid-Signature header: 
t=<timestamp>,v1=<hex_hmac_sha256>
To verify:
  1. Read the raw request body exactly as received.
  2. Parse t and v1 from Datagrid-Signature.
  3. Compute HMAC SHA-256 using your webhook secret over: <t>.<raw_body>.
  4. Compare computed digest with v1 using constant-time comparison.
  5. Reject stale timestamps (recommended tolerance: 5 minutes).
import hmac
import hashlib
import time

def verify(signature_header: str, raw_body: bytes, secret: str) -> bool:
    parts = dict(item.split("=", 1) for item in signature_header.split(","))
    timestamp = parts["t"]
    received = parts["v1"]

    # Optional replay protection
    if abs(int(time.time()) - int(timestamp)) > 300:
      return False

    signed = f"{timestamp}.".encode("utf-8") + raw_body
    expected = hmac.new(
        secret.encode("utf-8"),
        signed,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(expected, received)

​
Retry behavior and best practices

When delivery fails (network timeout/error or non-2xx response), Datagrid retries asynchronously. Best practices:
  • Return a 2xx quickly, then process asynchronously.
  • Treat webhook handling as at least once delivery and dedupe by event id.
  • Keep signature verification on every request.
  • Log failures with webhook id, event id, and event type.
  • Use enabled=false (update webhook) to temporarily pause delivery without deleting configuration.

​
Endpoints