Skip the poll loop. Subscribe to webhooks and Amps delivers action lifecycle events to your endpoint as they happen. Two events fire: push.completed and push.failed. Both arrive only when an action reaches a terminal state, typically within 10 seconds in live (3 minutes in sandbox).Every delivery is signed with an HMAC-SHA256 signature, retried with exponential backoff on non-2xx, and stamped with an event ID delivered in the svix-id header for idempotency.
Add the webhook URL in the Amps AI Dashboard under Webhooks settings. Pick the events to subscribe to (push.completed, push.failed) and copy the webhook secret. Store it in an environment variable; never commit it.
When a battery action completes, you receive a POST with this body. The event type and event ID are delivered in the svix-id, svix-timestamp, and svix-signature headers, not the body:
Every delivery includes three headers (svix-id, svix-timestamp, svix-signature). Verify the signature before trusting the body; the SDK reads all three. The signature is HMAC-SHA256 of {id}.{timestamp}.{rawBody} keyed by your webhook secret.
Node
Python
curl (replay test)
import express from "express";import { Webhook } from "svix";const app = express();const wh = new Webhook(process.env.WEBHOOK_SECRET);const seen = new Set();app.post("/webhooks", express.raw({ type: "application/json" }), (req, res) => { let payload; try { payload = wh.verify(req.body, req.headers); } catch (err) { return res.status(400).send("Invalid signature"); } const eventId = req.headers["svix-id"]; if (seen.has(eventId)) { return res.status(200).send("duplicate"); } seen.add(eventId); // Route by inspecting the payload shape: completed payloads carry `result`, failed payloads carry `error`. if (payload.result) { handleCompleted(payload); } else if (payload.error) { handleFailed(payload); } res.status(200).send("OK");});
import osfrom flask import Flask, request, jsonifyfrom svix.webhooks import Webhook, WebhookVerificationErrorapp = Flask(__name__)wh = Webhook(os.environ["WEBHOOK_SECRET"])seen = set()@app.route("/webhooks", methods=["POST"])def webhooks(): raw = request.get_data() try: payload = wh.verify(raw, request.headers) except WebhookVerificationError: return jsonify({"error": "Invalid signature"}), 400 event_id = request.headers.get("svix-id") if event_id in seen: return jsonify({"status": "duplicate"}), 200 seen.add(event_id) if "result" in payload: handle_completed(payload) elif "error" in payload: handle_failed(payload) return jsonify({"status": "ok"}), 200
Webhooks are at-least-once. Retries fire on any non-2xx response, a timeout (>30s), or a transient network failure. Use the svix-id header as the idempotency key and persist seen IDs alongside the work the webhook triggers.
async function handleCompleted(payload, eventId) { const seen = await db.events.findOne({ eventId }); if (seen) return; await db.events.insertOne({ eventId, actionId: payload.actionId, deviceId: payload.deviceId, completedAt: payload.completedAt, }); await notifyUserActionCompleted(payload);}
In production, store seen IDs in an external store. In-process memory will not survive across instances.
Backed off and surfaced in the dashboard delivery log.
Always respond 200 OK as soon as the signature verifies and the event is recorded. Push business logic (email, dashboard updates) to async workers so the response stays inside the 30-second window.
