"""Generate Langflow-export-format flow JSON fixtures. Uses lfx's own serializers (Component.to_frontend_node / build_custom_component_template), so the output is byte-faithful to what the Langflow UI would export. Run with the backend venv (has the `lfx` dev dep):: uv run python scripts/make_flow_fixtures.py Writes to two places: - backend/tests/fixtures/flows/ (test fixtures, never deployed as-is) - backend/flows/ (canonical source of truth, synced into DB + LFX) Then verify:: uv run lfx validate --level 4 backend/tests/fixtures/flows/ """ from __future__ import annotations import importlib import json import sys import uuid from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent.parent FIXTURE_DIR = REPO_ROOT / "backend" / "tests" / "fixtures" / "flows" FLOW_DIR = REPO_ROOT / "backend" / "flows" HANDLE_QUOTE = "\u0153" # ReactFlow handle-id quoting used by Langflow exports # -------------------------------------------------------------------------- # node builders # -------------------------------------------------------------------------- def _normalize(node_dict: dict, node_id: str, comp_name: str, x: int, y: int) -> dict: # to_frontend_node() returns an envelope {"data": {...}, "id": ...}; # build_custom_component_template() returns the bare node dict. if "data" not in node_dict: node_dict = { "data": {"node": node_dict, "type": comp_name, "id": node_id}, "id": node_id, } else: node_dict["id"] = node_id node_dict["data"]["id"] = node_id node_dict["data"]["type"] = comp_name node_dict["data"]["showNode"] = True node_dict["measured"] = {"width": 320, "height": 200} node_dict["position"] = {"x": x, "y": y} node_dict["type"] = "genericNode" return node_dict def builtin_node( module_path: str, class_name: str, node_id: str, values: dict | None = None, position: tuple[int, int] = (0, 0), ) -> dict: mod = importlib.import_module(module_path) cls = getattr(mod, class_name) inst = cls() node_dict = _normalize(inst.to_frontend_node(), node_id, cls.name, *position) for field, value in (values or {}).items(): node_dict["data"]["node"]["template"][field]["value"] = value return node_dict def custom_node( code: str, node_id: str, values: dict | None = None, position: tuple[int, int] = (0, 0), ) -> dict: from lfx.custom import validate as validate_mod from lfx.custom.utils import build_custom_component_template cls = validate_mod.create_class(code, validate_mod.extract_class_name(code)) inst = cls() inst._code = code bare, _ = build_custom_component_template(inst) node_dict = _normalize(bare, node_id, "CustomComponent", *position) for field, value in (values or {}).items(): node_dict["data"]["node"]["template"][field]["value"] = value return node_dict def kirche_node( node_id: str, values: dict | None = None, position: tuple[int, int] = (0, 0), ) -> dict: """A node from our GUI-editable category lib (backend/components/kirche). Built with lfx's own frontend-node serialization, exactly like ``Component.to_frontend_node()`` — minus its final ``Template.from_dict`` round-trip: lfx 1.12.x cannot re-instantiate ``MultiselectInput`` (it emits ``list``, the constructor forbids extras; their own builtins fail the same way). The template dict before that round-trip is complete and faithful, so we use it directly and append the ``code`` field by hand (mirroring the shape ``build_custom_component_template`` produces). """ sys.path.insert(0, str(REPO_ROOT / "backend" / "components")) from kirche.form_inputs import CustomInputForm from lfx.template.frontend_node.custom_components import ( ComponentFrontendNode, ) comp_name = getattr(CustomInputForm, "name", None) or "CustomInputForm" inst = CustomInputForm() frontend_node = ComponentFrontendNode.from_inputs(**inst.get_template_config(inst)) inst._map_parameters_on_frontend_node(frontend_node) node = frontend_node.to_dict(keep_name=False) node = inst._update_template(node) inst._map_parameters_on_template(node["template"]) if not getattr(inst, "_code", None): inst.set_class_code() node["template"]["code"] = { "advanced": True, "api_editable": False, "dynamic": True, "fileTypes": [], "info": "", "list": False, "load_from_db": False, "multiline": True, "name": "code", "password": False, "placeholder": "", "required": True, "show": True, "title_case": False, "type": "code", "value": inst._code, } node_dict = { "data": {"node": node, "type": comp_name, "id": node_id}, "id": node_id, } node_dict = _normalize(node_dict, node_id, comp_name, *position) for field, value in (values or {}).items(): node_dict["data"]["node"]["template"][field]["value"] = value # lfx validate --level 4 errors on required inputs with no value and no # incoming edge. Our form fields are filled at runtime via the # form_values envelope (collect() merges), never via edges — so required # template fields carry a single-space sentinel value. The form parser # (app/flows/input_schema.py) treats whitespace-only values as "no # default", keeping the web form clean. for field_def in node_dict["data"]["node"]["template"].values(): if ( isinstance(field_def, dict) and field_def.get("required") and field_def.get("value") in (None, "", [], {}) ): field_def["value"] = " " return node_dict # -------------------------------------------------------------------------- # edges # -------------------------------------------------------------------------- def _output_def(node: dict, output_name: str) -> dict: for out in node["data"]["node"]["outputs"]: if out["name"] == output_name: return out raise KeyError(f"output {output_name!r} not in {node['id']}") def _input_def(node: dict, input_name: str) -> dict: field = node["data"]["node"]["template"][input_name] return field def edge(src: dict, src_output: str, dst: dict, dst_input: str) -> dict: out = _output_def(src, src_output) inp = _input_def(dst, dst_input) source_handle = { "dataType": src["data"]["type"], "id": src["id"], "name": src_output, "output_types": out.get("types", []), } target_handle = { "fieldName": dst_input, "id": dst["id"], "inputTypes": inp.get("input_types", []), "type": inp.get("type", "other"), } sh = json.dumps(source_handle).replace('"', HANDLE_QUOTE) th = json.dumps(target_handle).replace('"', HANDLE_QUOTE) return { "animated": False, "className": "", "data": {"sourceHandle": source_handle, "targetHandle": target_handle}, "id": f"xy-edge__{sh}-{th}", "selected": False, "source": src["id"], "sourceHandle": sh, "target": dst["id"], "targetHandle": th, } # -------------------------------------------------------------------------- # flow document # -------------------------------------------------------------------------- def flow_doc(name: str, description: str, nodes: list[dict], edges: list[dict]) -> dict: return { "data": { "nodes": nodes, "edges": edges, "viewport": {"x": 0, "y": 0, "zoom": 1}, }, "description": description, "endpoint_name": None, "icon": None, "icon_bg_color": None, "id": str(uuid.uuid5(uuid.NAMESPACE_URL, f"kirche-einfach/{name}")), "is_component": False, "locked": False, "mcp_enabled": False, "name": name, "tags": [], "webhook": False, } # -------------------------------------------------------------------------- # custom component code # -------------------------------------------------------------------------- RULE_CHECK_CODE = """from lfx.custom import Component from lfx.io import MessageTextInput, Output from lfx.schema.message import Message import json class RuleCheckComponent(Component): display_name = "Rule Check" description = "Bewertet Betrag gegen Schwelle aus JSON-Eingabe." inputs = [ MessageTextInput(name="payload", display_name="Payload"), ] outputs = [ Output(display_name="Verdict", name="verdict", method="check"), ] def _text(self, value) -> str: if value is None: return "" if hasattr(value, "text"): return value.text or "" return str(value) def check(self) -> Message: try: data = json.loads(self._text(self.payload) or "{}") except Exception: data = {} if not isinstance(data, dict): data = {} # Die API ruft Flows immer mit Huelle auf # {"payload": {...}, "run_id": ..., ...}; rohe Payloads gehen auch. if isinstance(data.get("payload"), dict): data = data["payload"] try: betrag = float(data.get("betrag", 0) or 0) except (TypeError, ValueError): betrag = 0.0 try: schwelle = float(data.get("schwelle", 200000) or 200000) except (TypeError, ValueError): schwelle = 200000.0 return Message(text="PRUEFUNG" if betrag > schwelle else "OK") """ ENVELOPE_CODE = """from lfx.custom import Component from lfx.io import MessageTextInput, Output from lfx.schema.message import Message import json class EnvelopeComponent(Component): display_name = "Envelope" description = "Zerlegt die Eingabe-Huelle in Payload und Laufzeit-Kontext." inputs = [ MessageTextInput(name="raw", display_name="Raw Input"), ] outputs = [ Output(display_name="Payload", name="payload", method="get_payload"), Output(display_name="Run ID", name="run_id", method="get_run_id"), Output( display_name="Callback URL", name="callback_url", method="get_callback_url", ), Output( display_name="Continuation", name="continuation", method="get_continuation", ), Output( display_name="Callback Token", name="callback_token", method="get_callback_token", ), ] def _text(self, value) -> str: if value is None: return "" if hasattr(value, "text"): return value.text or "" return str(value) def _env(self) -> dict: if not hasattr(self, "_env_cache"): try: self._env_cache = json.loads(self._text(self.raw) or "{}") except Exception: self._env_cache = {} if not isinstance(self._env_cache, dict): self._env_cache = {} return self._env_cache def get_payload(self) -> Message: return Message(text=json.dumps(self._env().get("payload", {}))) def get_run_id(self) -> Message: return Message(text=str(self._env().get("run_id", ""))) def get_callback_url(self) -> Message: return Message(text=str(self._env().get("callback_url", ""))) def get_continuation(self) -> Message: value = self._env().get("continuation_flow_id") return Message(text=str(value) if value else "") def get_callback_token(self) -> Message: return Message(text=str(self._env().get("callback_token", ""))) """ APPROVAL_CALLBACK_CODE = """from lfx.custom import Component from lfx.io import MessageTextInput, Output from lfx.schema.message import Message class ApprovalCallbackComponent(Component): display_name = "Approval Callback" description = "Meldet einen Sonderfall per HTTP an unsere API und terminiert." inputs = [ MessageTextInput(name="trigger", display_name="Trigger"), MessageTextInput(name="run_id", display_name="Run ID"), MessageTextInput(name="callback_url", display_name="Callback URL"), MessageTextInput( name="callback_token", display_name="Callback Token", ), MessageTextInput( name="continuation_flow_id", display_name="Continuation Flow ID", ), MessageTextInput(name="facts", display_name="Facts"), ] outputs = [ Output(display_name="Result", name="result", method="notify"), ] def _text(self, value) -> str: if value is None: return "" if hasattr(value, "text"): return value.text or "" return str(value) def notify(self) -> Message: import httpx import json trigger = self._text(self.trigger).strip() if not trigger: # Zweig wurde nicht genommen: kein Seiteneffekt. return Message(text="") try: facts = json.loads(self._text(self.facts) or "{}") except Exception: facts = {} url = self._text(self.callback_url).rstrip("/") + "/api/approvals" body = { "run_id": self._text(self.run_id), "title": f"Freigabe erforderlich ({trigger})", "payload": {"verdict": trigger, "facts": facts}, "continuation_flow_id": self._text(self.continuation_flow_id) or None, } headers = {"X-Callback-Token": self._text(self.callback_token)} try: response = httpx.post(url, json=body, headers=headers, timeout=10.0) response.raise_for_status() except Exception as exc: return Message(text=f"CALLBACK_FAILED: {exc}") return Message(text="PENDING_HUMAN_APPROVAL") """ # -------------------------------------------------------------------------- # flows # -------------------------------------------------------------------------- ROUTER_VALUES = { "operator": "equals", "match_text": "PRUEFUNG", "true_case_message": "Sonderfall: menschliche Pruefung erforderlich", "false_case_message": "Standardfall: automatisch freigegeben", } def build_simple_rule() -> dict: entry = builtin_node( "lfx.components.input_output.chat", "ChatInput", "ChatInput-entry", position=(0, 200), ) rule = custom_node(RULE_CHECK_CODE, "CustomComponent-rule", position=(400, 200)) router = builtin_node( "lfx.components.flow_controls.conditional_router", "ConditionalRouterComponent", "ConditionalRouter-router", values=ROUTER_VALUES, position=(800, 200), ) out_ok = builtin_node( "lfx.components.input_output.chat_output", "ChatOutput", "ChatOutput-ok", position=(1200, 60), ) out_sonder = builtin_node( "lfx.components.input_output.chat_output", "ChatOutput", "ChatOutput-sonder", position=(1200, 340), ) nodes = [entry, rule, router, out_ok, out_sonder] edges = [ edge(entry, "message", rule, "payload"), edge(rule, "verdict", router, "input_text"), edge(router, "false_result", out_ok, "input_value"), edge(router, "true_result", out_sonder, "input_value"), ] return flow_doc( "simple_rule", "Regelpruefung ohne Seiteneffekt (Spike-Fixture, kein LLM).", nodes, edges, ) def build_approval_handoff() -> dict: entry = builtin_node( "lfx.components.input_output.chat", "ChatInput", "ChatInput-entry", position=(0, 300), ) envelope = custom_node(ENVELOPE_CODE, "CustomComponent-envelope", position=(400, 300)) rule = custom_node(RULE_CHECK_CODE, "CustomComponent-rule", position=(800, 300)) router = builtin_node( "lfx.components.flow_controls.conditional_router", "ConditionalRouterComponent", "ConditionalRouter-router", values=ROUTER_VALUES, position=(1200, 300), ) callback = custom_node( APPROVAL_CALLBACK_CODE, "CustomComponent-callback", position=(1600, 160), ) out_pending = builtin_node( "lfx.components.input_output.chat_output", "ChatOutput", "ChatOutput-pending", position=(2000, 160), ) out_ok = builtin_node( "lfx.components.input_output.chat_output", "ChatOutput", "ChatOutput-ok", position=(2000, 440), ) nodes = [entry, envelope, rule, router, callback, out_pending, out_ok] edges = [ edge(entry, "message", envelope, "raw"), edge(envelope, "payload", rule, "payload"), edge(rule, "verdict", router, "input_text"), edge(router, "true_result", callback, "trigger"), edge(envelope, "run_id", callback, "run_id"), edge(envelope, "callback_url", callback, "callback_url"), edge(envelope, "callback_token", callback, "callback_token"), edge(envelope, "continuation", callback, "continuation_flow_id"), edge(envelope, "payload", callback, "facts"), edge(callback, "result", out_pending, "input_value"), edge(router, "false_result", out_ok, "input_value"), ] return flow_doc( "approval_handoff", "Sonderfall meldet sich per HTTP an unsere API und terminiert (Spike).", nodes, edges, ) def build_intake_form() -> dict: entry = builtin_node( "lfx.components.input_output.chat", "ChatInput", "ChatInput-entry", position=(0, 200), ) form = kirche_node( "CustomInputForm-intake", position=(400, 200), ) out = builtin_node( "lfx.components.input_output.chat_output", "ChatOutput", "ChatOutput-collected", position=(800, 200), ) nodes = [entry, form, out] edges = [ edge(entry, "message", form, "form_values"), edge(form, "form_data", out, "input_value"), ] return flow_doc( "intake_form", "Structured intake via CustomInputForm (parsed into a web form).", nodes, edges, ) def main() -> None: FIXTURE_DIR.mkdir(parents=True, exist_ok=True) FLOW_DIR.mkdir(parents=True, exist_ok=True) flows = { "simple_rule.json": build_simple_rule(), "approval_handoff.json": build_approval_handoff(), "intake_form.json": build_intake_form(), } for filename, doc in flows.items(): payload = json.dumps(doc, indent=2, ensure_ascii=False) + "\n" for target in (FIXTURE_DIR / filename, FLOW_DIR / filename): target.write_text(payload) print(f"wrote {target}") if __name__ == "__main__": sys.exit(main())