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

Event types

EventMeaning
quota.warningTenant crossed the alert threshold
quota.criticalRemaining quota is critically low
quota.exhaustedQuota exhausted, mode action applied
quota.resetMonthly counters reset
mode.changedTenant mode changed
processing.delayedQueue delay crossed the warning threshold
processing.recoveredProcessing recovered after a delay

Delivery headers

HeaderMeaning
X-MemoryOS-EventEvent name
X-MemoryOS-TimestampDelivery timestamp
X-MemoryOS-SignatureHMAC-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.
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}