#!/usr/bin/env python3 """n8n workflow health watchdog. Uses only the public n8n REST API (/api/v1) with the X-N8N-API-KEY header. For each workflow id it asks three questions: 1. GET {base}/api/v1/workflows/{id} -> is it active? 2. GET {base}/api/v1/executions?workflowId={id}&status=success&limit=1 3. GET {base}/api/v1/executions?workflowId={id}&status=error&limit=5 Verdicts, checked in this order: INACTIVE the workflow's active flag is false STALE no successful execution, or the newest one is older than --max-silence-min ERRORING the newest error execution is newer than the newest success OK none of the above Read-only by default. Alerts and reactivation only happen with --apply. The API key is read from the N8N_API_KEY environment variable. Exit code: 0 all OK, 1 at least one workflow not OK, 2 could not query the API. """ import argparse import json import math import os import sys import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone API_PREFIX = "/api/v1" # overridable with --api-prefix TIMEOUT_S = 20 class ApiError(Exception): pass # ---------------------------------------------------------------- HTTP layer # These two functions are the only place that touches the network. # Mock mode replaces them; everything below them is shared by live and mock. def http_get(url, headers): req = urllib.request.Request(url, headers=headers, method="GET") try: with urllib.request.urlopen(req, timeout=TIMEOUT_S) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: raise ApiError("HTTP %s from GET %s" % (e.code, urllib.parse.urlsplit(url).path)) except (urllib.error.URLError, OSError, ValueError) as e: raise ApiError("GET %s failed: %s" % (urllib.parse.urlsplit(url).path, e)) def http_post(url, headers, body): data = json.dumps(body).encode("utf-8") if body is not None else b"" h = dict(headers) h["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=h, method="POST") try: with urllib.request.urlopen(req, timeout=TIMEOUT_S) as resp: raw = resp.read().decode("utf-8") return {"status": resp.status, "body": raw[:200]} except urllib.error.HTTPError as e: raise ApiError("HTTP %s from POST %s" % (e.code, urllib.parse.urlsplit(url).hostname)) except (urllib.error.URLError, OSError) as e: raise ApiError("POST to %s failed: %s" % (urllib.parse.urlsplit(url).hostname, e)) def make_mock_transport(sample_dir): """Return (get, post) that read fixtures instead of calling a server.""" def load(name): path = os.path.join(sample_dir, name) if not os.path.isfile(path): raise ApiError("no fixture %s" % name) with open(path, encoding="utf-8") as f: return json.load(f) def mock_get(url, headers): parts = urllib.parse.urlsplit(url) q = urllib.parse.parse_qs(parts.query) path = parts.path if path.endswith("/executions"): return load("executions_%s_%s.json" % (q["workflowId"][0], q["status"][0])) if "/workflows/" in path: return load("workflow_%s.json" % path.rsplit("/", 1)[1]) raise ApiError("mock: unknown path %s" % path) def mock_post(url, headers, body): return {"status": 200, "body": "mock"} return mock_get, mock_post # ---------------------------------------------------------------- logic def parse_ts(s): return datetime.fromisoformat(s.replace("Z", "+00:00")).astimezone(timezone.utc) def newest(executions): """Newest execution by startedAt. Does not rely on the API's ordering.""" items = [e for e in (executions or {}).get("data", []) if e.get("startedAt")] return max(items, key=lambda e: parse_ts(e["startedAt"])) if items else None def finish_time(execution): return parse_ts(execution.get("stoppedAt") or execution["startedAt"]) def evaluate(workflow, success_resp, error_resp, now, max_silence_min): wid = workflow.get("id") name = workflow.get("name") or wid active = bool(workflow.get("active")) last_ok = newest(success_resp) last_err = newest(error_resp) recent_errors = len((error_resp or {}).get("data", [])) minutes = None errors_since = 0 if last_ok: ok_time = finish_time(last_ok) minutes = int(math.floor((now - ok_time).total_seconds() / 60.0 + 0.5)) errors_since = sum( 1 for e in error_resp.get("data", []) if e.get("startedAt") and parse_ts(e["startedAt"]) > ok_time ) else: errors_since = recent_errors if not active: verdict, detail = "INACTIVE", "workflow is not active" elif minutes is None: verdict, detail = "STALE", "no successful execution found" elif minutes > max_silence_min: verdict, detail = "STALE", "last success %d min ago (limit %d)" % (minutes, max_silence_min) elif last_err and parse_ts(last_err["startedAt"]) > finish_time(last_ok): verdict, detail = "ERRORING", "last success %d min ago" % minutes else: verdict, detail = "OK", "last success %d min ago" % minutes if verdict in ("STALE", "ERRORING") and errors_since: detail += ", %d error(s) since" % errors_since return { "id": wid, "name": name, "active": active, "verdict": verdict, "minutes_since_success": minutes, "recent_errors": recent_errors, "errors_since_success": errors_since, "message": "[n8n watchdog] %s (%s): %s - %s" % (name, wid, verdict, detail), } def check_workflow(base, wid, headers, now, max_silence_min): q = urllib.parse.quote(wid, safe="") wf = http_get("%s%s/workflows/%s" % (base, API_PREFIX, q), headers) ok = http_get("%s%s/executions?workflowId=%s&status=success&limit=1" % (base, API_PREFIX, q), headers) err = http_get("%s%s/executions?workflowId=%s&status=error&limit=5" % (base, API_PREFIX, q), headers) return evaluate(wf, ok, err, now, max_silence_min) def format_table(rows): cols = [("ID", "id"), ("NAME", "name"), ("ACTIVE", "active"), ("VERDICT", "verdict"), ("MIN SINCE OK", "minutes_since_success"), ("RECENT ERRORS", "recent_errors")] cells = [[h for h, _ in cols]] for r in rows: cells.append([("-" if r[k] is None else str(r[k]).lower() if isinstance(r[k], bool) else str(r[k])) for _, k in cols]) widths = [max(len(row[i]) for row in cells) for i in range(len(cols))] return "\n".join(" ".join(c.ljust(w) for c, w in zip(row, widths)).rstrip() for row in cells) def send_alert(webhook, field, message, apply): host = urllib.parse.urlsplit(webhook).hostname if not apply: print("dry-run: would POST alert to %s: %s" % (host, message)) return http_post(webhook, {}, {field: message}) print("alert sent to %s for: %s" % (host, message)) def reactivate(base, wid, headers, apply): url = "%s%s/workflows/%s/activate" % (base, API_PREFIX, urllib.parse.quote(wid, safe="")) if not apply: print("dry-run: would POST %s" % urllib.parse.urlsplit(url).path) return http_post(url, headers, None) print("activate requested for %s (check the n8n editor to confirm)" % wid) def run(args): global http_get, http_post, API_PREFIX API_PREFIX = "/" + args.api_prefix.strip("/") headers = {} now = datetime.now(timezone.utc) if args.mock: http_get, http_post = make_mock_transport(args.mock) if args.now: now = parse_ts(args.now) else: with open(os.path.join(args.mock, "now.txt"), encoding="utf-8") as f: now = parse_ts(f.read().strip()) else: key = os.environ.get("N8N_API_KEY") if not key: print("error: set the N8N_API_KEY environment variable", file=sys.stderr) return 2 headers = {"X-N8N-API-KEY": key, "Accept": "application/json"} webhook = args.alert_webhook or os.environ.get("ALERT_WEBHOOK_URL") base = args.base_url.rstrip("/") rows, failures = [], [] for wid in args.workflow_id: try: rows.append(check_workflow(base, wid, headers, now, args.max_silence_min)) except ApiError as e: failures.append("%s: %s" % (wid, e)) if rows: print(format_table(rows)) for f in failures: print("could not check %s" % f, file=sys.stderr) bad = [r for r in rows if r["verdict"] != "OK"] for r in bad: print(r["message"]) try: if webhook: send_alert(webhook, args.alert_field, r["message"], args.apply) if r["verdict"] == "INACTIVE" and args.auto_reactivate: reactivate(base, r["id"], headers, args.apply) except ApiError as e: failures.append("%s: %s" % (r["id"], e)) print("action failed for %s: %s" % (r["id"], e), file=sys.stderr) if failures: return 2 return 1 if bad else 0 def parse_args(argv): p = argparse.ArgumentParser(description="Check that n8n workflows are active and still succeeding.") p.add_argument("--base-url", default="http://localhost:5678", help="n8n base URL, no /api/v1 suffix") p.add_argument("--api-prefix", default="/api/v1", help="API path prefix; change it if the docs for your n8n version differ") p.add_argument("--workflow-id", action="append", required=True, help="workflow id, repeatable") p.add_argument("--max-silence-min", type=int, default=30, help="STALE if the last success is older than this") p.add_argument("--alert-webhook", help="incoming webhook URL (or set ALERT_WEBHOOK_URL)") p.add_argument("--alert-field", default="text", help="JSON field for the message: text (Slack, RocketChat) or content (Discord)") p.add_argument("--auto-reactivate", action="store_true", help="POST /activate for INACTIVE workflows (needs --apply)") p.add_argument("--apply", action="store_true", help="really send alerts and activate calls; default is dry-run") p.add_argument("--mock", metavar="SAMPLE_DIR", help="read fixtures from this directory instead of calling n8n") p.add_argument("--now", help="mock mode only: ISO timestamp to use as the current time") return p.parse_args(argv) def main(argv=None): return run(parse_args(argv)) if __name__ == "__main__": sys.exit(main())