Webhooks Setup

import os
from engramma_cloud import EngrammaClient

client = EngrammaClient(api_key=os.environ["ENGRAMMA_API_KEY"])

# Register a webhook for multiple events
webhook = client.webhooks.create(
    url="https://your-app.com/hooks/engramma",
    events=["regime.changed", "consolidation.completed", "pattern.limit.warning"]
)

print(f"Webhook ID: {webhook.id}")
print(f"Secret: {webhook.secret}")
print(f"Events: {webhook.events}")

# Save the secret — you'll need it to verify signatures

import hmac
import hashlib
from flask import Flask, request, jsonify

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["ENGRAMMA_WEBHOOK_SECRET"]

def verify_signature(payload: bytes, signature: str) -> bool:
    """Verify the webhook signature from Engramma."""
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

@app.route("/hooks/engramma", methods=["POST"])
def handle_webhook():
    # Verify the signature
    signature = request.headers.get("X-Engramma-Signature")
    if not verify_signature(request.data, signature):
        return jsonify({"error": "Invalid signature"}), 401
    
    event = request.json
    event_type = event["event"]
    data = event["data"]
    
    if event_type == "regime.changed":
        print(f"Regime changed: {data['from']} → {data['to']}")
        print(f"Reason: {data['reason']}")
        # Alert your team if anomaly detected
        if data["to"] == "anomaly":
            send_alert(f"⚠️ Engramma anomaly: {data['reason']}")
    
    elif event_type == "consolidation.completed":
        print(f"Consolidation: {data['patterns_before']} → {data['patterns_after']}")
        print(f"Merged: {data['merged']}, Pruned: {data['pruned']}")
    
    elif event_type == "pattern.limit.warning":
        print(f"Pattern limit warning: {data['current']}/{data['limit']} ({data['percentage']}%)")
        # Notify that an upgrade may be needed
    
    return jsonify({"received": True}), 200

# List all webhooks
webhooks = client.webhooks.list()
for wh in webhooks:
    print(f"{wh.id}: {wh.url} → {wh.events}")

# Check delivery history
deliveries = client.webhooks.deliveries(webhook_id="wh_abc123")
for d in deliveries:
    print(f"  {d.timestamp}: {d.event} → {d.status_code} ({d.duration_ms}ms)")

# Delete a webhook
client.webhooks.delete(webhook_id="wh_abc123")