#!/usr/bin/env python3 """Create Langflow global variables, API key, and both PM flows via API. Idempotent: variables are upserted; existing flows with the same name are deleted and recreated. Run AFTER deploy_files.sh. Needs secrets.local.env sourced into the environment (PM_IMAP_PASSWORD, PM_ESPO_API_KEY, PM_ESPO_BASE). Node/edge shapes and the two builtin_template category keys below were verified live against Langflow 1.10.0's GET /api/v1/all and the existing "vLLM Smoke Test" flow's edge dump (see task-10-report.md for the raw evidence). Notable deviations from the naive assumption: - Webhook lives in the "input_output" category, not "data". - Every custom Component's Data-typed `out` output reports its Langflow wire type as "JSON" (not "Data") in outputs[].types. - DataInput/JSONInput fields (`ctx_in`, `trigger`) report input_types as the full ["Data", "JSON"], not just ["Data"]. Edges are built from the live template's outputs[].types / template[field].input_types instead of hardcoded guesses, so any future Langflow version drift here is easy to re-derive by rerunning the Step-3 probe in the task brief. IMPORTANT (found while debugging Step 6, see task-10-report.md): Langflow's `Component` base class defines `ctx` as a reserved @property (returns `self.graph.context`, the flow-level shared context store). The Task-9 components originally named their inter-stage DataInput field "ctx", which silently shadows the framework property at attribute-read time -- the edge correctly delivers data into `self._attributes["ctx"]`, but `self.ctx` inside build_out() resolves to the *class property* instead (an empty dotdict), not the input. There is no error; the payload just silently comes back empty one hop downstream. Task 10 renamed the field to `ctx_in` in all five affected components (pm_extract.py, pm_match.py, pm_rules.py, pm_crm.py, pm_notify.py); the edges below target "ctx_in". """ import json import os import sys import time from pathlib import Path import requests BASE = "http://127.0.0.1:8090/api/v1" COMPONENTS = Path(__file__).resolve().parent.parent / "components" ENV_FILE = Path.home() / ".config" / "projekt-matching" / "env" GENERIC_VARS = { "PM_ESPO_BASE": os.environ.get("PM_ESPO_BASE", ""), "PM_NOTIFY_TO": "chancen@destengs.com", "PM_ALERT_TO": "chancen@destengs.com", "PM_THRESHOLD": "85", "PM_VLLM_BASE": "http://host.containers.internal:8081/v1", "PM_VLLM_MODEL": "AxionML/Qwen3.5-9B-NVFP4", "PM_FLOW2_ID": "placeholder", } SECRET_VARS = ("PM_IMAP_PASSWORD", "PM_ESPO_API_KEY") def login(): r = requests.get(f"{BASE}/auto_login", timeout=15) r.raise_for_status() token = r.json()["access_token"] return {"Authorization": f"Bearer {token}"} def upsert_variable(hdr, name, value, vtype): existing = {v["name"]: v["id"] for v in requests.get(f"{BASE}/variables/", headers=hdr, timeout=15).json()} body = {"name": name, "value": value, "type": vtype, "default_fields": []} if name in existing: requests.patch(f"{BASE}/variables/{existing[name]}", headers=hdr, json={"id": existing[name], "name": name, "value": value}, timeout=15).raise_for_status() else: requests.post(f"{BASE}/variables/", headers=hdr, json=body, timeout=15).raise_for_status() def get_api_key(hdr): r = requests.post(f"{BASE}/api_key/", headers=hdr, json={"name": "projekt-matching"}, timeout=15) r.raise_for_status() return r.json()["api_key"] def component_template(hdr, code): r = requests.post(f"{BASE}/custom_component", headers=hdr, json={"code": code}, timeout=60) r.raise_for_status() out = r.json() return out["data"], out["type"] def builtin_template(hdr, category, name): r = requests.get(f"{BASE}/all", headers=hdr, timeout=60) r.raise_for_status() return r.json()[category][name] def handle_str(d): return json.dumps(d, separators=(",", ":")).replace('"', "œ") def make_node(node_id, comp_type, template, x): return {"id": node_id, "type": "genericNode", "position": {"x": x, "y": 0}, "data": {"type": comp_type, "id": node_id, "node": template, "showNode": True}} def out_spec(template, out_name): """Look up an output's live wire types from a fetched node template.""" for o in template["outputs"]: if o["name"] == out_name: return o["types"] raise KeyError(f"output {out_name!r} not found") def in_spec(template, field_name): """Look up a field's live (input_types, type) from a node template.""" spec = template["template"][field_name] return spec["input_types"], spec["type"] def make_edge(src_id, src_type, out_name, out_types, tgt_id, field, input_types, field_type): sh = {"dataType": src_type, "id": src_id, "name": out_name, "output_types": out_types} th = {"fieldName": field, "id": tgt_id, "inputTypes": input_types, "type": field_type} return {"id": f"reactflow__edge-{src_id}-{tgt_id}", "source": src_id, "target": tgt_id, "sourceHandle": handle_str(sh), "targetHandle": handle_str(th), "data": {"sourceHandle": sh, "targetHandle": th}, "className": ""} def edge_between(src_id, src_type, src_tpl, out_name, tgt_id, tgt_tpl, field): out_types = out_spec(src_tpl, out_name) input_types, field_type = in_spec(tgt_tpl, field) return make_edge(src_id, src_type, out_name, out_types, tgt_id, field, input_types, field_type) def _matches_flow(f, name, endpoint): fname, fep = f["name"], f.get("endpoint_name") or "" return (fname == name or fname.startswith(f"{name} (") or fep == endpoint or (fep.startswith(f"{endpoint}-") and fep[len(endpoint) + 1:].isdigit())) def create_flow(hdr, name, endpoint, nodes, edges): # Delete every previous incarnation. Langflow auto-renames on collision # ("Name (1)", endpoint "endpoint-1") instead of rejecting, so a plain # exact-name match leaves suffixed zombies behind that then keep the # endpoint slot occupied on the next run -- match the auto-suffixed # variants (by name AND endpoint) too, and fail loudly if a delete # doesn't stick. Then wait until the listing no longer shows any of # them: the create-time uniqueness check otherwise still sees the # just-deleted rows and suffixes the new flow anyway (observed live # on 1.10.0). flows = requests.get(f"{BASE}/flows/", headers=hdr, timeout=30).json() for f in flows: if _matches_flow(f, name, endpoint): requests.delete(f"{BASE}/flows/{f['id']}", headers=hdr, timeout=15).raise_for_status() for _ in range(20): flows = requests.get(f"{BASE}/flows/", headers=hdr, timeout=30).json() if not any(_matches_flow(f, name, endpoint) for f in flows): break time.sleep(0.5) body = {"name": name, "endpoint_name": endpoint, "is_component": False, "data": {"nodes": nodes, "edges": edges, "viewport": {"x": 0, "y": 0, "zoom": 0.5}}} r = requests.post(f"{BASE}/flows/", headers=hdr, json=body, timeout=30) r.raise_for_status() created = r.json() # Belt-and-braces: if the name/endpoint still got suffixed, PATCH them # back. The PATCH can 404 transiently right after the POST (read-your- # writes lag observed live), so retry briefly. fixes = {} if created["name"] != name: fixes["name"] = name if (created.get("endpoint_name") or "") != endpoint: fixes["endpoint_name"] = endpoint if fixes: for attempt in range(10): pr = requests.patch(f"{BASE}/flows/{created['id']}", headers=hdr, json=fixes, timeout=15) if pr.status_code == 404: time.sleep(1) continue pr.raise_for_status() break else: raise RuntimeError( f"could not normalize flow {created['id']} to " f"name={name!r} endpoint={endpoint!r}") return created["id"] def main(): hdr = login() for name in SECRET_VARS: value = os.environ.get(name, "") if not value: # Langflow rejects empty Credential values outright ("Variable # value cannot be empty"). PM_ESPO_API_KEY legitimately comes # back empty here when secrets.local.env's grep against # /home/tlg/mkt/bewerb/.secrets/espocrm-api.md can't read that # chmod-600 file (see task-10-report.md). Use an obviously-fake # placeholder so the deploy can proceed -- stage_crm() only # calls espocrm with this value when ctx["decision"] == # "consider", so the reject-path smoke test (Step 6) is # unaffected; the accept-path / real CRM writes remain # unverified until a real key is supplied. value = f"UNSET-{name}-blocked-by-file-permissions" print(f"WARN: {name} is empty in the environment, using " f"placeholder credential value", file=sys.stderr) upsert_variable(hdr, name, value, "Credential") for name, value in GENERIC_VARS.items(): if name == "PM_FLOW2_ID": continue if not value: # Same empty-value rejection as above, for PM_ESPO_BASE. value = f"UNSET-{name}-blocked-by-file-permissions" print(f"WARN: {name} is empty in the environment, using " f"placeholder value", file=sys.stderr) upsert_variable(hdr, name, value, "Generic") api_key = get_api_key(hdr) # PMIngest reads the run-API key via load_from_db -> must exist as a # Credential global variable, not only in the host env file. upsert_variable(hdr, "PM_LANGFLOW_API_KEY", api_key, "Credential") def custom(fname): code = (COMPONENTS / fname).read_text() return component_template(hdr, code) # ---- Flow 2: TextInput -> PMFetch -> ... -> PMNotify -> TextOutput text_in = builtin_template(hdr, "input_output", "TextInput") text_out = builtin_template(hdr, "input_output", "TextOutput") chain = [("TextInput-pm2i", "TextInput", text_in)] for fname, nid in [("pm_fetch.py", "PMFetch-pm2a"), ("pm_extract.py", "PMExtract-pm2b"), ("pm_match.py", "PMMatch-pm2c"), ("pm_rules.py", "PMRules-pm2d"), ("pm_crm.py", "PMCrm-pm2e"), ("pm_notify.py", "PMNotify-pm2f")]: tpl, ctype = custom(fname) chain.append((nid, ctype, tpl)) chain.append(("TextOutput-pm2o", "TextOutput", text_out)) nodes2 = [make_node(nid, ctype, tpl, x=i * 380) for i, (nid, ctype, tpl) in enumerate(chain)] by_id = {nid: tpl for nid, _, tpl in chain} by_type = {nid: ctype for nid, ctype, _ in chain} edges2 = [ edge_between("TextInput-pm2i", "TextInput", by_id["TextInput-pm2i"], "text", "PMFetch-pm2a", by_id["PMFetch-pm2a"], "payload"), edge_between("PMFetch-pm2a", by_type["PMFetch-pm2a"], by_id["PMFetch-pm2a"], "out", "PMExtract-pm2b", by_id["PMExtract-pm2b"], "ctx_in"), edge_between("PMExtract-pm2b", by_type["PMExtract-pm2b"], by_id["PMExtract-pm2b"], "out", "PMMatch-pm2c", by_id["PMMatch-pm2c"], "ctx_in"), edge_between("PMMatch-pm2c", by_type["PMMatch-pm2c"], by_id["PMMatch-pm2c"], "out", "PMRules-pm2d", by_id["PMRules-pm2d"], "ctx_in"), edge_between("PMRules-pm2d", by_type["PMRules-pm2d"], by_id["PMRules-pm2d"], "out", "PMCrm-pm2e", by_id["PMCrm-pm2e"], "ctx_in"), edge_between("PMCrm-pm2e", by_type["PMCrm-pm2e"], by_id["PMCrm-pm2e"], "out", "PMNotify-pm2f", by_id["PMNotify-pm2f"], "ctx_in"), edge_between("PMNotify-pm2f", by_type["PMNotify-pm2f"], by_id["PMNotify-pm2f"], "out", "TextOutput-pm2o", by_id["TextOutput-pm2o"], "input_value"), ] flow2_id = create_flow(hdr, "PM Projekt bewerten", "pm-flow2", nodes2, edges2) upsert_variable(hdr, "PM_FLOW2_ID", flow2_id, "Generic") # ---- Flow 1: Webhook -> PMIngest -> TextOutput webhook = builtin_template(hdr, "input_output", "Webhook") tpl_ing, ctype_ing = custom("pm_ingest.py") nodes1 = [make_node("Webhook-pm1w", "Webhook", webhook, 0), make_node("PMIngest-pm1a", ctype_ing, tpl_ing, 380), make_node("TextOutput-pm1o", "TextOutput", text_out, 760)] edges1 = [ edge_between("Webhook-pm1w", "Webhook", webhook, "output_data", "PMIngest-pm1a", tpl_ing, "trigger"), edge_between("PMIngest-pm1a", ctype_ing, tpl_ing, "out", "TextOutput-pm1o", text_out, "input_value"), ] flow1_id = create_flow(hdr, "PM Ingest", "pm-ingest", nodes1, edges1) ENV_FILE.parent.mkdir(parents=True, exist_ok=True) ENV_FILE.write_text( f"LANGFLOW_API_KEY={api_key}\n" f"FLOW1_ID={flow1_id}\nFLOW2_ID={flow2_id}\n" f"FLOW1_WEBHOOK=http://127.0.0.1:8090/api/v1/webhook/pm-ingest\n") ENV_FILE.chmod(0o600) print(f"flow1={flow1_id} flow2={flow2_id}\nenv -> {ENV_FILE}") if __name__ == "__main__": sys.exit(main())