# /// script # requires-python = ">=3.10" # dependencies = [] # /// """Course helper: check, push and test the n8n workflows that a CLI coding agent wrote. uv run tools/flow.py doctor # is everything ready? uv run tools/flow.py check my-work/notice.json # find mistakes before n8n sees the file uv run tools/flow.py push my-work/writer.json my-work/reviewer.json # import + publish (one n8n restart) uv run tools/flow.py off my-reviewer # unpublish by file or webhook path (a role that is down) uv run tools/flow.py call my-notice requests/notice-mail.json [-v] uv run tools/flow.py test cases/lab02.json -v [--repeat 3] # -v: show each request and response JSON Run it from the day folder (Last3Mon/69). n8n is the day 65 install (~/day65-n8n, Docker project day65-n8n), started from Last3Mon/65/n8n. Environment overrides: N8N_CONTAINER (default: the running n8n service of project day65-n8n, else day65-n8n-n8n-1) N8N_HOME (default ~/day65-n8n), N8N_URL (default http://localhost:5678) """ from __future__ import annotations import argparse import datetime as dt import hashlib import json import os import re import shutil import subprocess import sys import time import urllib.error import urllib.request import uuid from pathlib import Path ROOT = Path(__file__).resolve().parent.parent CACHE = Path(__file__).resolve().parent / ".cache" def n8n_home() -> Path: """Where the official n8n install lives: N8N_HOME, else Last3Mon/65/n8n/.env, else ~/day65-n8n.""" if os.getenv("N8N_HOME"): return Path(os.environ["N8N_HOME"]).expanduser() env_file = ROOT.parent / "65" / "n8n" / ".env" if env_file.exists(): for line in env_file.read_text(encoding="utf-8").splitlines(): if line.startswith("N8N_HOME="): value = line.split("=", 1)[1].strip().strip("'\"") if value: return (env_file.parent / Path(value).expanduser()).resolve() return Path.home() / "day65-n8n" N8N_HOME = n8n_home() START_HINT = "cd ~/aio-01d4-stu/Last3Mon/65/n8n && docker compose start (n8n install: ~/day65-n8n)" def find_container() -> str: """The n8n container of the day 65 project, whichever folder it was started from.""" if os.getenv("N8N_CONTAINER"): return os.environ["N8N_CONTAINER"] if shutil.which("docker"): result = subprocess.run(["docker", "ps", "-a", "--filter", "label=com.docker.compose.project=day65-n8n", "--filter", "label=com.docker.compose.service=n8n", "--format", "{{.Names}}"], capture_output=True, text=True) names = result.stdout.split() if names: return names[0] return "day65-n8n-n8n-1" CONTAINER = find_container() N8N_URL = os.getenv("N8N_URL", "http://localhost:5678").rstrip("/") # Versions written in AGENTS.md. Other versions still work if n8n knows them (warning only). RECOMMENDED = { "n8n-nodes-base.webhook": 2, "n8n-nodes-base.respondToWebhook": 1.1, "n8n-nodes-base.if": 2.2, "n8n-nodes-base.switch": 3.2, "n8n-nodes-base.set": 3.4, "n8n-nodes-base.code": 2, "n8n-nodes-base.httpRequest": 4.2, "n8n-nodes-base.httpRequestTool": 4.2, "@n8n/n8n-nodes-langchain.agent": 3.1, "@n8n/n8n-nodes-langchain.lmChatOpenAi": 1.2, } USES_DATA_API = False # day 69 labs do not call the data API AGENT_HOOKS: tuple[str, ...] = () # no ready-made agents on day 69: every role is built in the labs COURSE_CREDENTIALS = {"openAiCourse0001": "OpenAI (.env)"} # OpenAI only from day 66 (no Ollama) CONNECTION_TYPES = {"main", "ai_tool", "ai_languageModel", "ai_memory", "ai_outputParser"} SECRET = re.compile(r"\bsk-[A-Za-z0-9_-]{16,}") OK, BAD, WARN = "✔", "✘", "!" # ---------------------------------------------------------------- helpers def die(message: str) -> None: print(f"{BAD} {message}", file=sys.stderr) sys.exit(1) def docker(*args: str, stdin: str | None = None, check: bool = True) -> subprocess.CompletedProcess: if not shutil.which("docker"): die("docker command not found. Start Docker first.") result = subprocess.run(["docker", *args], input=stdin, capture_output=True, text=True) if check and result.returncode != 0: die(f"docker {' '.join(args[:2])} failed:\n{(result.stderr or result.stdout).strip()}") return result def n8n(*args: str, stdin: str | None = None, check: bool = True) -> subprocess.CompletedProcess: return docker("exec", "-i", CONTAINER, "n8n", *args, stdin=stdin, check=check) def container_running() -> bool: if not shutil.which("docker"): return False result = subprocess.run(["docker", "inspect", "-f", "{{.State.Running}}", CONTAINER], capture_output=True, text=True) return result.stdout.strip() == "true" def http(method: str, url: str, body: object | None = None, timeout: float = 300) -> tuple[int, object, float]: data = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8") request = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) start = time.monotonic() try: with urllib.request.urlopen(request, timeout=timeout) as response: status, raw = response.status, response.read() except urllib.error.HTTPError as error: status, raw = error.code, error.read() except (urllib.error.URLError, TimeoutError, ConnectionError) as error: return 0, {"error": str(getattr(error, "reason", error))}, time.monotonic() - start text = raw.decode("utf-8", errors="replace") try: payload: object = json.loads(text) if text.strip() else "" except json.JSONDecodeError: payload = text return status, payload, time.monotonic() - start def load_json(path: Path) -> object: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: die(f"file not found: {path}") except json.JSONDecodeError as error: die(f"{path} is not valid JSON (line {error.lineno}, column {error.colno}): {error.msg}") def resolve(path: str) -> Path: candidate = Path(path) if candidate.exists() or candidate.is_absolute(): return candidate return ROOT / path def node_versions() -> dict[str, list[float]]: """Node types and versions that this n8n really has (exported once, then cached).""" cache = CACHE / "node-versions.json" if cache.exists(): return json.loads(cache.read_text()) if not container_running(): return {} n8n("export:nodes", "--output=/tmp/course-nodes.json") raw = docker("exec", CONTAINER, "sh", "-c", "cat /tmp/course-nodes.json && rm -f /tmp/course-nodes.json").stdout versions: dict[str, set[float]] = {} for node in json.loads(raw): value = node.get("version") versions.setdefault(node["name"], set()).update(value if isinstance(value, list) else [value]) result = {name: sorted(values) for name, values in versions.items()} CACHE.mkdir(exist_ok=True) cache.write_text(json.dumps(result)) return result def walk_strings(value: object, where: str = ""): if isinstance(value, str): yield where, value elif isinstance(value, dict): for key, item in value.items(): yield from walk_strings(item, f"{where}.{key}" if where else key) elif isinstance(value, list): for index, item in enumerate(value): yield from walk_strings(item, f"{where}[{index}]") def webhooks(workflow: dict) -> list[tuple[str, str]]: found = [] for node in workflow.get("nodes", []): if node.get("type") == "n8n-nodes-base.webhook": parameters = node.get("parameters", {}) found.append((str(parameters.get("httpMethod", "GET")).upper(), str(parameters.get("path", "")).strip("/"))) return found # ---------------------------------------------------------------- check def check_workflow(workflow: object) -> tuple[list[str], list[str]]: errors: list[str] = [] warnings: list[str] = [] if not isinstance(workflow, dict): return ["the file must be one workflow object { name, nodes, connections }"], warnings if not str(workflow.get("name", "")).strip(): errors.append("missing workflow 'name'") nodes = workflow.get("nodes") connections = workflow.get("connections") if not isinstance(nodes, list) or not nodes: return errors + ["'nodes' must be a non-empty list"], warnings if not isinstance(connections, dict): errors.append("'connections' must be an object") connections = {} known = node_versions() names: dict[str, dict] = {} for index, node in enumerate(nodes): label = f"node #{index + 1}" if not isinstance(node, dict): errors.append(f"{label} is not an object") continue name = node.get("name") if not name: errors.append(f"{label} has no 'name'") continue label = f"node '{name}'" if name in names: errors.append(f"{label} appears twice (node names must be unique)") names[name] = node node_type, version = node.get("type"), node.get("typeVersion") if not node_type or version is None: errors.append(f"{label} needs 'type' and 'typeVersion'") elif known and node_type not in known: errors.append(f"{label}: type '{node_type}' does not exist in this n8n") elif known and version not in known[node_type]: errors.append(f"{label}: {node_type} has no typeVersion {version} (available: {known[node_type]})") elif node_type in RECOMMENDED and version != RECOMMENDED[node_type]: warnings.append(f"{label}: typeVersion {version} (AGENTS.md uses {RECOMMENDED[node_type]})") if not isinstance(node.get("parameters"), dict): errors.append(f"{label} needs 'parameters' (use {{}} when empty)") position = node.get("position") if not (isinstance(position, list) and len(position) == 2): errors.append(f"{label} needs 'position': [x, y]") for kind, credential in (node.get("credentials") or {}).items(): credential_id = (credential or {}).get("id") if credential_id not in COURSE_CREDENTIALS: errors.append(f"{label}: credential '{kind}' must use a course id {list(COURSE_CREDENTIALS)}, got {credential_id!r}") for where, text in walk_strings(node.get("parameters", {})): if SECRET.search(text): errors.append(f"{label}: parameters.{where} looks like an API key. Use a credential instead.") if "{{" in text and "}}" in text and not text.startswith("="): warnings.append(f"{label}: parameters.{where} has {{{{ }}}} but does not start with '=' (it stays plain text)") targets_of: dict[str, set[str]] = {} for source, outputs in connections.items(): if source not in names: errors.append(f"connections: source '{source}' is not a node name") continue if not isinstance(outputs, dict): errors.append(f"connections['{source}'] must be an object like {{\"main\": [[...]]}}") continue for kind, branches in outputs.items(): if kind not in CONNECTION_TYPES: errors.append(f"connections['{source}']: unknown connection type '{kind}'") for branch in branches or []: for link in branch or []: target = (link or {}).get("node") if target not in names: errors.append(f"connections['{source}'] points to '{target}', which is not a node name") elif link.get("type") != kind: errors.append(f"connections['{source}'] -> '{target}': link type must be '{kind}'") else: targets_of.setdefault(target, set()).add(kind) if kind != "main": targets_of.setdefault(f"{source}::from", set()).add(kind) has_respond = any(node.get("type") == "n8n-nodes-base.respondToWebhook" for node in names.values()) for name, node in names.items(): node_type = node.get("type", "") parameters = node.get("parameters", {}) if isinstance(node.get("parameters"), dict) else {} if node_type == "n8n-nodes-base.webhook": if not str(parameters.get("path", "")).strip("/"): errors.append(f"node '{name}': webhook needs a 'path'") if parameters.get("responseMode") == "responseNode" and not has_respond: errors.append(f"node '{name}': responseMode is 'responseNode' but there is no Respond to Webhook node") if node_type == "@n8n/n8n-nodes-langchain.agent" and "ai_languageModel" not in targets_of.get(name, set()): errors.append(f"node '{name}': AI Agent needs a chat model connected with 'ai_languageModel'") if node_type.endswith("Tool") or node_type.startswith("@n8n/n8n-nodes-langchain.tool"): if "ai_tool" not in targets_of.get(f"{name}::from", set()): errors.append(f"node '{name}': a tool must be connected to an AI Agent with 'ai_tool'") if node_type.startswith("@n8n/n8n-nodes-langchain.lmChat"): if "ai_languageModel" not in targets_of.get(f"{name}::from", set()): errors.append(f"node '{name}': a chat model must be connected to an AI Agent with 'ai_languageModel'") if node_type != "n8n-nodes-base.webhook" and not node_type.startswith("@n8n/n8n-nodes-langchain.lm") \ and not (node_type.endswith("Tool") or node_type.startswith("@n8n/n8n-nodes-langchain.tool")) \ and "main" not in targets_of.get(name, set()) and "trigger" not in node_type.lower(): warnings.append(f"node '{name}': nothing is connected into it (it never runs)") if not webhooks(workflow): warnings.append("no Webhook node: 'push' works, but 'call' and 'test' need a webhook path") return errors, warnings def cmd_check(args: argparse.Namespace) -> int: path = resolve(args.file) errors, warnings = check_workflow(load_json(path)) for warning in warnings: print(f"{WARN} {warning}") for error in errors: print(f"{BAD} {error}") if errors: print(f"{BAD} {path.name}: {len(errors)} error(s). Fix them and run check again.") return 1 print(f"{OK} {path.name}: no errors" + (f" ({len(warnings)} warning(s))" if warnings else "")) return 0 # ---------------------------------------------------------------- push def stable_id(path: Path) -> str: return "w" + hashlib.sha1(path.stem.encode("utf-8")).hexdigest()[:15] def wait_until_ready(paths: list[tuple[str, str]], seconds: int = 90) -> bool: deadline = time.monotonic() + seconds while time.monotonic() < deadline: status, _, _ = http("GET", f"{N8N_URL}/healthz", timeout=3) if status == 200: break time.sleep(1) else: return False for method, path in paths: while time.monotonic() < deadline: # A webhook that is registered for POST answers a GET with a hint, not "is not registered". status, payload, _ = http("GET" if method != "GET" else "HEAD", f"{N8N_URL}/webhook/{path}", timeout=5) text = json.dumps(payload, ensure_ascii=False) if "not registered for" in text or status not in (0, 404, 500, 502, 503): break time.sleep(1) time.sleep(2) # webhooks of the other published workflows finish registering right after return time.monotonic() < deadline def exported_workflows() -> list[dict]: exported = docker("exec", CONTAINER, "sh", "-c", "n8n export:workflow --all --output=/tmp/course-all.json >/dev/null 2>&1; " "cat /tmp/course-all.json 2>/dev/null; rm -f /tmp/course-all.json", check=False).stdout try: return json.loads(exported) if exported.strip() else [] except json.JSONDecodeError: return [] def prepare(path: Path) -> dict | None: workflow = load_json(path) errors, warnings = check_workflow(workflow) for warning in warnings: print(f"{WARN} {path.name}: {warning}") if errors: for error in errors: print(f"{BAD} {path.name}: {error}") return None workflow = dict(workflow) if not re.fullmatch(r"[A-Za-z0-9]{10,}", str(workflow.get("id", ""))): workflow["id"] = stable_id(path) workflow["active"] = False workflow.setdefault("settings", {}).setdefault("executionOrder", "v1") for node in workflow["nodes"]: node.setdefault("id", str(uuid.uuid5(uuid.NAMESPACE_URL, f"{workflow['id']}/{node['name']}"))) if node["type"] == "n8n-nodes-base.webhook": node.setdefault("webhookId", str(uuid.uuid5(uuid.NAMESPACE_URL, f"{workflow['id']}/webhook/{node['name']}"))) return workflow def cmd_push(args: argparse.Namespace) -> int: paths = [resolve(file) for file in args.files] workflows = [prepare(path) for path in paths] if None in workflows: print(f"{BAD} not pushed. Fix the errors first (uv run tools/flow.py check ).") return 1 if not container_running(): die(f"n8n container '{CONTAINER}' is not running. {START_HINT}") mine_ids = {workflow["id"] for workflow in workflows} hooks = {hook for workflow in workflows for hook in webhooks(workflow)} # Another published workflow with the same webhook path would block this one. for other in exported_workflows(): if other.get("id") not in mine_ids and other.get("active") and hooks & set(webhooks(other)): n8n("unpublish:workflow", f"--id={other['id']}") print(f"{WARN} unpublished '{other.get('name')}' ({other['id']}): it used the same webhook path") for workflow in workflows: docker("exec", "-i", CONTAINER, "sh", "-c", "cat > /tmp/course-workflow.json && n8n import:workflow --input=/tmp/course-workflow.json; " "status=$?; rm -f /tmp/course-workflow.json; exit $status", stdin=json.dumps(workflow, ensure_ascii=False)) n8n("publish:workflow", f"--id={workflow['id']}") # The CLI changes the database only; one restart makes the running n8n load the new versions (about 5 s). docker("restart", CONTAINER) ready = wait_until_ready(sorted(hooks)) for workflow in workflows: print(f"{OK} pushed '{workflow['name']}' (id {workflow['id']})") for method, hook in sorted(webhooks(workflow)): print(f" {method} {N8N_URL}/webhook/{hook}") if not ready: print(f"{WARN} n8n is still starting. Wait a few seconds before 'call' or 'test'.") return 0 def cmd_off(args: argparse.Namespace) -> int: """Unpublish workflows, given as files or webhook paths, to see how the others behave when one role is down.""" if not container_running(): die(f"n8n container '{CONTAINER}' is not running. {START_HINT}") exported = exported_workflows() wanted: dict[str, str] = {} for item in args.files: path = resolve(item) if (ROOT / item).exists() or Path(item).exists() else None if path: workflow = load_json(path) workflow_id = workflow.get("id") if re.fullmatch(r"[A-Za-z0-9]{10,}", str(workflow.get("id", ""))) else stable_id(path) wanted[workflow_id] = item hooks = {hook for _, hook in webhooks(workflow)} else: hooks = {item.strip("/")} for other in exported: if other.get("active") and hooks & {hook for _, hook in webhooks(other)}: wanted[other["id"]] = item active = {item.get("id") for item in exported if item.get("active")} for workflow_id, item in wanted.items(): if workflow_id in active: n8n("unpublish:workflow", f"--id={workflow_id}") print(f"{OK} unpublished {item} (id {workflow_id})") docker("restart", CONTAINER) wait_until_ready([]) print(f"{OK} n8n restarted. Turn it back on with: uv run tools/flow.py push ") return 0 # ---------------------------------------------------------------- call / test def request_body(spec: str | dict, base: Path = ROOT) -> object: if isinstance(spec, dict): return spec text = spec.strip() if text.startswith("{"): try: return json.loads(text) except json.JSONDecodeError as error: die(f"request is not valid JSON: {error.msg}") path = Path(text) if not path.exists(): path = base / text return load_json(path) def cmd_call(args: argparse.Namespace) -> int: body = request_body(args.request) if args.request else {} hook = args.path.strip('/') status, payload, seconds = http("POST", f"{N8N_URL}/webhook/{hook}", body) print(f"HTTP {status} · {seconds:.1f}s") if args.verbose: show_json(f"→ request POST /webhook/{hook}", body) show_json(f"← response HTTP {status}", payload) return 0 if 200 <= status < 300 else 1 print(json.dumps(payload, ensure_ascii=False, indent=2) if not isinstance(payload, str) else payload) return 0 if 200 <= status < 300 else 1 def field(payload: object, dotted: str) -> object: value = payload for part in dotted.split("."): if isinstance(value, dict): value = value.get(part) elif isinstance(value, list) and part.isdigit() and int(part) < len(value): value = value[int(part)] else: return None return value def as_text(value: object) -> str: return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) def judge(expect: dict, status: int, payload: object) -> list[str]: problems = [] if status != expect.get("status", 200): problems.append(f"status {status} (expected {expect.get('status', 200)})") for name, wanted in (expect.get("equals") or {}).items(): actual = field(payload, name) if actual != wanted: problems.append(f"{name} = {as_text(actual)[:60]} (expected {as_text(wanted)})") for name, words in (expect.get("contains") or {}).items(): words = [words] if isinstance(words, str) else words text = as_text(field(payload, name) or "") if not any(word in text for word in words): problems.append(f"{name} has none of {words}") for name, words in (expect.get("not_contains") or {}).items(): words = [words] if isinstance(words, str) else words text = as_text(field(payload, name) or "") leaked = [word for word in words if word in text] if leaked: problems.append(f"{name} must not have {leaked}") return problems def show_json(label: str, value: object) -> None: """Print one JSON value under a label, indented below the result line (call -v, test -v).""" text = json.dumps(value, ensure_ascii=False, indent=2) if not isinstance(value, str) else value print(f" {label}") for row in text.splitlines(): print(f" {row}") def cmd_test(args: argparse.Namespace) -> int: path = resolve(args.cases) suite = load_json(path) hook = (args.webhook or suite.get("webhook", "")).strip("/") cases = suite.get("cases", []) if not hook or not cases: die(f"{path.name} needs \"webhook\" and a non-empty \"cases\" list") print(f"POST {N8N_URL}/webhook/{hook} · {len(cases)} case(s) × {args.repeat}") results, passed = [], 0 for case in cases: body = request_body(case.get("body") or case.get("request", "{}")) case_hook = (case.get("webhook") or hook).strip("/") if not args.webhook else hook for run in range(1, args.repeat + 1): status, payload, seconds = http("POST", f"{N8N_URL}/webhook/{case_hook}", body) problems = judge(case.get("expect", {}), status, payload) passed += not problems mark = OK if not problems else BAD suffix = f" #{run}" if args.repeat > 1 else "" print(f"{mark} {case.get('name', '?')}{suffix} · {seconds:.1f}s" + ("" if not problems else " · " + "; ".join(problems))) if args.verbose: show_json(f"→ request POST /webhook/{case_hook}", body) show_json(f"← response HTTP {status}", payload) results.append({"case": case.get("name"), "webhook": case_hook, "run": run, "passed": not problems, "problems": problems, "seconds": round(seconds, 1), "status": status, "request": body, "response": payload}) total = len(results) print(f"Passed {passed}/{total}") runs = ROOT / "runs" runs.mkdir(exist_ok=True) stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S") report = runs / f"{path.stem}-{stamp}.json" report.write_text(json.dumps({"cases": str(path.name), "webhook": hook, "passed": passed, "total": total, "results": results}, ensure_ascii=False, indent=2), encoding="utf-8") print(f"report: {report.relative_to(ROOT)}") return 0 if passed == total else 1 # ---------------------------------------------------------------- doctor def cmd_doctor(args: argparse.Namespace) -> int: failures = 0 def line(good: bool, text: str, hint: str = "") -> None: nonlocal failures failures += not good print(f"{OK if good else BAD} {text}" + ("" if good or not hint else f"\n → {hint}")) installed = (N8N_HOME / "compose.yml").exists() print(f"{OK if installed else WARN} n8n install folder {N8N_HOME}" + ("" if installed else "\n → day 65 SETUP-N8N 5단계 (get.n8n.io 설치). 다른 곳에 설치했다면 Last3Mon/65/n8n/.env 에 N8N_HOME=<폴더>")) running = container_running() line(running, f"n8n container {CONTAINER}", START_HINT) status, _, _ = http("GET", f"{N8N_URL}/healthz", timeout=5) line(status == 200, f"n8n {N8N_URL}", "wait 10 seconds after starting n8n, then run doctor again") status, _, _ = http("GET", "http://localhost:8800/health", timeout=5) if USES_DATA_API: line(status == 200, "data API http://localhost:8800", "cd ~/aio-01d4-stu/Last3Mon/67/data-api && docker compose up -d --force-recreate (start n8n first)") else: print(f"{OK if status == 200 else WARN} data API http://localhost:8800 (not used today)") if running: probe = docker("exec", CONTAINER, "node", "-e", "fetch('http://data-api:8800/health').then(r=>console.log(r.status)).catch(()=>console.log(0))", check=False).stdout.strip() if USES_DATA_API: line(probe == "200", "n8n → http://data-api:8800", "data API must join the day65-n8n_default network (compose.yml)") exported = docker("exec", CONTAINER, "sh", "-c", "n8n export:credentials --all --output=/tmp/course-cred.json >/dev/null 2>&1; " "cat /tmp/course-cred.json 2>/dev/null; rm -f /tmp/course-cred.json", check=False).stdout try: ids = {item.get("id") for item in json.loads(exported)} if exported.strip() else set() except json.JSONDecodeError: ids = set() for credential_id, name in COURSE_CREDENTIALS.items(): line(credential_id in ids, f"credential '{name}' ({credential_id})", "cd ~/aio-01d4-stu/Last3Mon/65/n8n && bash register-credentials.sh") line(bool(node_versions()), "node catalog (for check)") for hook in AGENT_HOOKS: _, payload, _ = http("GET", f"{N8N_URL}/webhook/{hook}", timeout=5) line("not registered for" in json.dumps(payload, ensure_ascii=False), f"agent webhook /{hook}", "uv run tools/flow.py push ") agents = [name for name in ("codex", "claude", "agy") if shutil.which(name)] line(bool(agents), "CLI coding agent: " + (", ".join(agents) if agents else "none"), "install one of codex / claude / agy (see Lecture/69/00-DAY69-PREP.md)") print("ready" if not failures else f"{failures} item(s) to fix") return 0 if not failures else 1 def main() -> int: parser = argparse.ArgumentParser(description="Check, push and test course n8n workflows.") commands = parser.add_subparsers(dest="command", required=True) commands.add_parser("doctor", help="check n8n, data API, credentials and CLI agents").set_defaults(run=cmd_doctor) item = commands.add_parser("check", help="find mistakes in a workflow file") item.add_argument("file") item.set_defaults(run=cmd_check) item = commands.add_parser("push", help="import and publish workflow files (one n8n restart)") item.add_argument("files", nargs="+") item.set_defaults(run=cmd_push) item = commands.add_parser("off", help="unpublish workflow files or webhook paths (a role that is down)") item.add_argument("files", nargs="+") item.set_defaults(run=cmd_off) item = commands.add_parser("call", help="POST one request to a webhook path") item.add_argument("path", help="webhook path, e.g. my-notice") item.add_argument("request", nargs="?", help="request file or inline JSON") item.add_argument("-v", "--verbose", action="store_true", help="print the request JSON as well as the response") item.set_defaults(run=cmd_call) item = commands.add_parser("test", help="run a test case file") item.add_argument("cases") item.add_argument("--repeat", type=int, default=1, help="run every case N times") item.add_argument("--webhook", help="override the webhook path in the case file") item.add_argument("-v", "--verbose", action="store_true", help="print the request and response JSON of every run") item.set_defaults(run=cmd_test) args = parser.parse_args() return args.run(args) if __name__ == "__main__": sys.exit(main())