Webhooks
Webhooks let external systems receive push notifications when platform events occur. Register an HTTPS endpoint and select which event types you want to subscribe to. Deliveries are signed with HMAC-SHA256 when you provide a secret.
Base URL: https://api.ainative.studio
Supported Event Types
| Event | Triggered when |
|---|---|
memory.stored | A new memory record is persisted |
vectors.indexed | Vectors are indexed in the database |
agent.task.complete | An agent swarm task finishes |
mcp.server.ready | An MCP server instance becomes available |
memory.consolidated | A memory consolidation cycle completes |
Endpoints
Register a Webhook
POST /api/v1/webhooks/register
Subscribe a URL to one or more event types. An optional signing secret enables payload verification on your end. The secret is stored as a one-way hash and cannot be retrieved after registration.
Auth required.
Request body
| Field | Type | Required | Description |
|---|---|---|---|
url | string | Yes | HTTPS URL to receive POST requests |
events | string[] | Yes | One or more event types from the supported list |
secret | string | No | Signing secret (8–256 chars). When provided, each delivery includes X-ZeroDB-Signature. |
curl -X POST https://api.ainative.studio/api/v1/webhooks/register \
-H "Authorization: Bearer <your_api_key>" \
-H "Content-Type: application/json" \
-d '{
"url": "https://my-system.example.com/webhooks/ainative",
"events": ["memory.stored", "agent.task.complete"],
"secret": "my-secret-at-least-8-chars"
}'
import httpx
resp = httpx.post(
"https://api.ainative.studio/api/v1/webhooks/register",
headers={"Authorization": "Bearer <your_api_key>"},
json={
"url": "https://my-system.example.com/webhooks/ainative",
"events": ["memory.stored", "agent.task.complete"],
"secret": "my-secret-at-least-8-chars",
},
)
webhook = resp.json()
print(webhook["webhook_id"])
Response 201 Created
{
"webhook_id": "550e8400-e29b-41d4-a716-446655440099",
"url": "https://my-system.example.com/webhooks/ainative",
"events": ["agent.task.complete", "memory.stored"],
"created_at": "2026-04-25T10:00:00Z"
}
List Registered Webhooks
GET /api/v1/webhooks/
Return all active webhook subscriptions owned by the authenticated user.
Auth required.
curl https://api.ainative.studio/api/v1/webhooks/ \
-H "Authorization: Bearer <your_api_key>"
Response
{
"webhooks": [
{
"webhook_id": "550e8400-e29b-41d4-a716-446655440099",
"url": "https://my-system.example.com/webhooks/ainative",
"events": ["agent.task.complete", "memory.stored"],
"is_active": true,
"created_at": "2026-04-25T10:00:00Z"
}
],
"count": 1
}
Delete a Webhook
DELETE /api/v1/webhooks/{webhook_id}
Deactivate a webhook subscription. Returns 404 if the webhook does not exist or belongs to another user.
Auth required.
curl -X DELETE \
https://api.ainative.studio/api/v1/webhooks/550e8400-e29b-41d4-a716-446655440099 \
-H "Authorization: Bearer <your_api_key>"
Response
{
"webhook_id": "550e8400-e29b-41d4-a716-446655440099",
"status": "deleted"
}
Payload Format
All webhook deliveries are HTTP POST requests to your registered URL with Content-Type: application/json.
{
"event_type": "agent.task.complete",
"occurred_at": "2026-04-25T10:05:23Z",
"data": {
"task_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
"status": "completed",
"agent_types": ["backend", "qa"],
"owner_id": "user-uuid"
}
}
Signature Verification
When you register a webhook with a secret, every delivery includes an X-ZeroDB-Signature header:
X-ZeroDB-Signature: sha256=<hex_digest>
The signature is computed as HMAC-SHA256(secret, raw_request_body).
Verify in Python
import hashlib
import hmac
def verify_signature(payload_bytes: bytes, signature_header: str, secret: str) -> bool:
"""Return True if the signature matches the payload."""
expected = "sha256=" + hmac.new(
secret.encode(),
payload_bytes,
hashlib.sha256,
).hexdigest()
return hmac.compare_digest(expected, signature_header)
# In your webhook handler:
from fastapi import Request, Header, HTTPException
async def handle_webhook(
request: Request,
x_zerodb_signature: str = Header(...),
):
body = await request.body()
if not verify_signature(body, x_zerodb_signature, secret="my-secret-at-least-8-chars"):
raise HTTPException(status_code=401, detail="Invalid signature")
# Process the event...
Verify in Node.js
const crypto = require("crypto");
function verifySignature(payload, signatureHeader, secret) {
const expected =
"sha256=" +
crypto.createHmac("sha256", secret).update(payload).digest("hex");
return crypto.timingSafeEqual(
Buffer.from(expected),
Buffer.from(signatureHeader)
);
}
// In your Express handler:
app.post("/webhooks/ainative", (req, res) => {
const raw = req.rawBody; // ensure body-parser is configured to expose rawBody
const sig = req.headers["x-zerodb-signature"];
if (!verifySignature(raw, sig, "my-secret-at-least-8-chars")) {
return res.status(401).send("Invalid signature");
}
const event = req.body;
// handle event...
res.sendStatus(200);
});
Always use a timing-safe comparison (hmac.compare_digest / crypto.timingSafeEqual) to prevent timing attacks.
Retry Policy
The platform does not currently implement automatic retries for failed webhook deliveries. Your endpoint should return a 2xx status code promptly. If your endpoint is unavailable, the delivery is lost.
For critical workflows, consider:
- Implementing idempotent handlers using the
event_type+ a unique field indata - Polling the relevant API endpoint as a fallback (e.g.
GET /api/v1/agent-swarm/tasks/{task_id})
Luma Integration Webhooks
The platform also exposes Luma event webhooks for internal use. These receive guest registration events from Luma and sync contacts to Resend.
| Endpoint | Purpose |
|---|---|
POST /api/v1/webhooks/luma/guest-registered | Handle Luma guest registration events |
POST /api/v1/webhooks/luma/webhook-test | Validate Luma webhook configuration |
GET /api/v1/webhooks/luma/health | Health check for the Luma webhook handler |
Luma webhooks require a valid X-Luma-Signature header for HMAC-SHA256 verification and are authenticated using the LUMA_WEBHOOK_SECRET environment variable.