Webhooks Setup

import os
import requests

api_key = os.environ["ENGRAMMA_API_KEY"]
base_url = "https://api.engramma-memory.com"

# Register a webhook for multiple events
response = requests.post(
    f"{base_url}/v1/webhooks",
    headers={
        "X-API-Key": api_key,
        "Content-Type": "application/json"
    },
    json={
        "url": "https://your-app.com/hooks/engramma",
        "events": ["regime.changed", "consolidation.completed", "pattern.limit.warning"]
    }
)

webhook = response.json()
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
import os
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 detected: {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

import os
import requests

api_key = os.environ["ENGRAMMA_API_KEY"]
base_url = "https://api.engramma-memory.com"
headers = {"X-API-Key": api_key}

# List all webhooks
response = requests.get(f"{base_url}/v1/webhooks", headers=headers)
webhooks = response.json()
for wh in webhooks:
    print(f"{wh['id']}: {wh['url']} -> {wh['events']}")

# List available event types
response = requests.get(f"{base_url}/v1/webhooks/events", headers=headers)
events = response.json()
print(f"Available events: {events}")

# Get a specific webhook
response = requests.get(f"{base_url}/v1/webhooks/wh_abc123", headers=headers)
print(response.json())

# Update a webhook (change events)
response = requests.patch(
    f"{base_url}/v1/webhooks/wh_abc123",
    headers={**headers, "Content-Type": "application/json"},
    json={"events": ["regime.changed", "anomaly.detected"]}
)
print(response.json())

# Check delivery history
response = requests.get(f"{base_url}/v1/webhooks/wh_abc123/deliveries", headers=headers)
deliveries = response.json()
for d in deliveries:
    print(f"  {d['timestamp']}: {d['event']} -> {d['status_code']} ({d['duration_ms']}ms)")

# Delete a webhook
response = requests.delete(f"{base_url}/v1/webhooks/wh_abc123", headers=headers)
print(f"Deleted: {response.status_code == 204}")