"""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 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" 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 # -------------------------------------------------------------------------- # 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", ), ] 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 "") ''' 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="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, } try: response = httpx.post(url, json=body, 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, "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 main() -> None: FIXTURE_DIR.mkdir(parents=True, exist_ok=True) flows = { "simple_rule.json": build_simple_rule(), "approval_handoff.json": build_approval_handoff(), } for filename, doc in flows.items(): path = FIXTURE_DIR / filename path.write_text(json.dumps(doc, indent=2, ensure_ascii=False) + "\n") print(f"wrote {path}") if __name__ == "__main__": sys.exit(main())