# /// 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/it-agent.json # find mistakes before n8n sees the file uv run tools/flow.py push my-work/it-agent.json # import + publish into n8n (production URL) uv run tools/flow.py call my-it-agent requests/faq-vpn.json uv run tools/flow.py test cases/lab01.json [--repeat 3] Run it from the day folder (Last3Mon/67). 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, } 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 cmd_push(args: argparse.Namespace) -> int: path = resolve(args.file) workflow = load_json(path) errors, warnings = check_workflow(workflow) for warning in warnings: print(f"{WARN} {warning}") if errors: for error in errors: print(f"{BAD} {error}") print(f"{BAD} not pushed. Fix the errors first (uv run tools/flow.py check {args.file}).") return 1 if not container_running(): die(f"n8n container '{CONTAINER}' is not running. {START_HINT}") 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']}"))) mine = set(webhooks(workflow)) # Another published workflow with the same webhook path would block this one. 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: existing = json.loads(exported) if exported.strip() else [] except json.JSONDecodeError: existing = [] for other in existing: if other.get("id") != workflow["id"] and other.get("active") and mine & 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") 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; a restart makes the running n8n load the new version (about 5 s). docker("restart", CONTAINER) ready = wait_until_ready(sorted(mine)) print(f"{OK} pushed '{workflow['name']}' (id {workflow['id']})") for method, hook in sorted(mine): 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 # ---------------------------------------------------------------- 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 {} status, payload, seconds = http("POST", f"{N8N_URL}/webhook/{args.path.strip('/')}", body) print(f"HTTP {status} · {seconds:.1f}s") 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 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", "{}")) for run in range(1, args.repeat + 1): status, payload, seconds = http("POST", f"{N8N_URL}/webhook/{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))) results.append({"case": case.get("name"), "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) 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)") 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() 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)") 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/67/00-DAY67-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 a workflow file") item.add_argument("file") item.set_defaults(run=cmd_push) item = commands.add_parser("call", help="POST one request to a webhook path") item.add_argument("path", help="webhook path, e.g. my-it-agent") item.add_argument("request", nargs="?", help="request file or inline JSON") 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.set_defaults(run=cmd_test) args = parser.parse_args() return args.run(args) if __name__ == "__main__": sys.exit(main())