Webhooks

Configure tenant webhooks, understand event types, and verify MemoryOS webhook signatures.

MemoryOS sends signed operational events to a tenant webhook URL for quota alerts, mode changes, and processing notifications.

Event types

Event	Meaning
quota.warning	Tenant crossed the alert threshold
quota.critical	Remaining quota is critically low
quota.exhausted	Quota exhausted, mode action applied
quota.reset	Monthly counters reset
mode.changed	Tenant mode changed
processing.delayed	Queue delay crossed the warning threshold
processing.recovered	Processing recovered after a delay

Delivery headers

Header	Meaning
X-MemoryOS-Event	Event name
X-MemoryOS-Timestamp	Delivery timestamp
X-MemoryOS-Signature	HMAC-SHA256 hex digest of the raw request body

Example payload

{
  "event": "quota.warning",
  "tenant_id": "23a1f8b4-4b40-4067-8be0-e3501301b8d8",
  "timestamp": "2026-04-17T10:30:00Z",
  "data": {
    "remaining_pct": 0.17,
    "threshold_pct": 0.2,
    "reset_at": "2026-05-01T00:00:00Z",
    "upgrade_url": "https://app.memoryos.io/pricing"
  },
  "memoryos_version": "1.0"
}

Verify the signature

Verify against the raw request body bytes — not a re-serialized JSON object.

Python

import hashlib
import hmac

def verify_memoryos_webhook(body_bytes: bytes, signature_header: str, webhook_secret: str) -> bool:
    expected = hmac.new(webhook_secret.encode("utf-8"), body_bytes, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature_header)

FastAPI handler example

import json
import os
import hashlib
import hmac

from fastapi import FastAPI, HTTPException, Request

app = FastAPI()

def verify_memoryos_webhook(body_bytes: bytes, signature_header: str, webhook_secret: str) -> bool:
    expected = hmac.new(webhook_secret.encode("utf-8"), body_bytes, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature_header)

@app.post("/webhooks/memoryos")
async def memoryos_webhook(request: Request) -> dict:
    raw_body = await request.body()
    signature = request.headers.get("X-MemoryOS-Signature")

    if not signature:
        raise HTTPException(status_code=401, detail="Missing signature")

    if not verify_memoryos_webhook(raw_body, signature, os.environ["MEMORYOS_WEBHOOK_SECRET"]):
        raise HTTPException(status_code=401, detail="Invalid signature")

    payload = json.loads(raw_body.decode("utf-8"))

    if payload["event"] == "quota.warning":
        print("Quota warning:", payload["data"])
    elif payload["event"] == "processing.delayed":
        print("Processing delay:", payload["data"])

    return {"received": True}