Flowfile Node Development

Flowfile node development

When NOT to use this skill

• Editing flow_graph.py execution/caching/scheduling internals without adding a node type → flowfile-architecture-contract (DAG engine, executor decision table, hash/cache invariants).
• Debugging a failure whose cause is unclear (not "which file do I edit") → flowfile-debugging-playbook first, then come back here once you know it's a node-registration problem.
• General Vue/store/Tauri conventions unrelated to node settings components → flowfile-frontend-conventions.
• Deep dive on flowfile_frame's expression system (Expr, _repr_str, _ff_repr, selectors) beyond "should this be a native node or polars-code" → flowfile-frame-and-codegen.
• Systematic flow-vs-generated-code parity bugs (Export to Python) → flowfile-codegen-parity-campaign.
• Running tests / CI markers in general → flowfile-testing-and-validation.
• Env vars / feature flags (FLOWFILE_OFFLOAD_TO_WORKER, FEATURE_FLAG_AI, etc.) → flowfile-config-and-flags.

All facts below verified in-repo as of 2026-07-03 (v0.12.7), branch feature/claude-skills. Re-verification commands are in the final section — run them before trusting a line number in a stale checkout.

0. The mental model — four layers, one node

A "node" is not one artifact. Adding my_node (fictional example used throughout) touches up to four independent registries that only agree by naming convention, not by a shared enum or generated code:

flowfile_core        settings schema (input_schema.py) + add_<type> method on FlowGraph
                      + NodeTemplate (configs/node_store/nodes.py) + NODE_TYPE_TO_SETTINGS_CLASS
                            │  served over HTTP as GET /node_list + POST /update_settings/?node_type=...
                            ▼
flowfile_frontend     settings Vue component at a convention path, resolved by import.meta.glob
                      (main desktop/web app only — NOT flowfile_wasm)
flowfile_frame        optional: a FlowFrame/Expr method that emits either a native add_<type> node
                      or a generated polars-code string
flowfile_wasm         optional, only if the node must run standalone-in-browser: Python engine fn +
                      TS settings component + palette entry (own registries, no HTTP, no flowfile_core)

Nothing here is enforced by a compiler or a single source of truth. Grep is the index. The canonical command to find every place a node type is wired:

grep -n "def add_" flowfile_core/flowfile_core/flowfile/flow_graph.py           # core graph methods
grep -n '"<node_type>"' flowfile_core/flowfile_core/configs/node_store/nodes.py  # template
grep -n "<node_type>" flowfile_core/flowfile_core/schemas/schemas.py             # settings-class registry
find flowfile_frontend/src/renderer/app/components/nodes/node-types/elements -iname '*<TitleCase>*'

1. Layer 1 — flowfile_core (the only layer that is ever mandatory)

1.1 The string-convention dispatch trap (read this before writing any code)

Adding a node from the UI/API goes through POST /update_settings/?node_type=<node_type> → add_generic_settings (flowfile_core/flowfile_core/routes/routes.py::add_generic_settings). It resolves everything by string manipulation, at request time, with no import-time check:

# routes.py::add_generic_settings (verified)
input_data["user_id"] = current_user.id
node_type = camel_case_to_snake_case(node_type)                 # e.g. "fuzzyMatch" -> "fuzzy_match"
add_func = getattr(flow, "add_" + node_type)                    # AttributeError if no add_fuzzy_match method
setting_name_ref = "node" + node_type.replace("_", "")          # "fuzzy_match" -> "nodefuzzymatch"
...
ref = get_node_model(setting_name_ref)                          # scans input_schema.__dict__ for a
                                                                  # class whose *lowercased* name matches
...
try:
    add_func(parsed_input)
except Exception as e:
    raise HTTPException(419, str(f"error: {e}")) from e         # non-standard status code, verified

get_node_model (routes.py) literally does for ref_name, ref in inspect.getmodule(input_schema).__dict__.items(): if ref_name.lower() == setting_name_ref: return ref.

Consequences you must design around:

• Your settings class name MUST be exactly Node + the node_type with underscores stripped, case-insensitively: text_to_rows → class NodeTextToRows (lowercases to nodetexttorows, matches "node" + "texttorows"). Get one letter wrong and get_node_model silently returns None → the route responds 404 "could not find the interface".
• Your graph method MUST be named add_<node_type> exactly (snake_case, matching what the frontend sends as node_type). Get it wrong and you get AttributeError inside add_generic_settings, which is not caught — it propagates as an unhandled 500, or if it happens after add_func is fetched but the call itself raises inside your method, you get the 419 above.
• None of this is checked until the endpoint is actually hit. A typo ships clean through code review, ruff, and mypy... because none of them look inside a runtime getattr/__dict__ scan.

1.2 Add-a-node checklist (core)

1. Settings model — flowfile_core/flowfile_core/schemas/input_schema.py. Subclass one of:
   • NodeSingleInput (input_schema.py:473) — adds depending_on_id: int | None = -1. Use for one-input transforms (filter, sort, select, formula...).
   • NodeMultiInput (input_schema.py:479) — adds depending_on_ids: list[int]. Use for join/union/concat-style nodes.
   • NodeBase (input_schema.py:428) directly — sources with no input (read, manual_input, database_reader...). NodeBase already carries flow_id, node_id, cache_results, pos_x/pos_y, description, node_reference, user_id, is_setup, output_field_config. Implement get_default_description(); add to_yaml_dict() only if you need curated YAML on save.
2. Register in NODE_TYPE_TO_SETTINGS_CLASS — flowfile_core/flowfile_core/schemas/schemas.py:27. It's a flat dict, "<node_type>": input_schema.Node<CamelCase>. Skip this and saved flows fail to reload with "Unknown node type" (used by NodeInformation.validate_setting_input when deserializing).
3. Template in flowfile_core/flowfile_core/configs/node_store/nodes.py::get_all_standard_nodes() — a NodeTemplate(...) entry in the big list. Required fields: item (== node_type, snake_case), input/output counts, node_type ("input"|"process"|"output"), transform_type ("narrow" → eligible for local-with-sampling execution; "wide" → forced remote/worker under optimize_for_downstream), laziness ("lazy"|"eager"|"conditional"), node_group (palette section), image (an <name>.svg filename), drawer_title/drawer_intro, tags (palette search). Add an entry to the node_defaults dict (nodes.py:755) only if the node should get pre-filled default settings on drop (checked via check_if_has_default_setting at add time, routes.py:594).
4. add_<node_type> method on FlowGraph (flowfile_core/flowfile_core/flowfile/flow_graph.py), decorated @with_history_capture(HistoryActionType.UPDATE_SETTINGS), which runs it inside FlowGraph.transaction (a standalone call records one undo step; inside an editor route, apply_operations or an AI batch it records nothing, because the outer transaction records the gesture), calling the one true factory self.add_node_step(node_id=..., function=_func, node_type="<node_type>", setting_input=..., input_node_ids=[...], schema_callback=...) (add_node_step itself lives at flow_graph.py:3378). _func receives one FlowDataEngine per input slot and returns a FlowDataEngine (or a NamedOutputs for multi-output nodes like filter_split).
   • _func must never write to setting_input. Runs and reads never write settings (TestReadsAndRunsDoNotWrite in tests/flowfile/test_history_invariants.py); reconcile on a per-call deepcopy as add_join/add_select do. Anything the drawer should see derived from the inputs belongs in a @setting_updator_method (flowfile/setting_generator/settings.py), which works on a deep copy.
   • _func must tolerate empty schema-only frames. Schema prediction runs your _func against FlowDataEngine.create_from_schema(...) placeholder inputs so the UI can show output columns without executing anything (flow_node.py:1060, _predicted_data_getter — surfaced via get_predicted_resulting_data). Don't assume real row data inside _func unless you gate on it.
   • Source-style nodes that need custom lifecycle (read/database_reader/manual_input) skip add_node_step and build/patch the FlowNode directly, registering via self.add_node_to_starting_list(node) + self._node_db[...] — see add_read (flow_graph.py:4646) as the template if you're building a source with special schema-callback logic.
5. If it's a network-source node, read §1.3 below before writing _func — do not fetch from core.
6. Frontend: node config component + path convention — §2.
7. Parity surfaces that silently lag (each has its own registry, none cross-check at build time): the code generator (flowfile_core/flowfile_core/flowfile/code_generator/, may need a new handler for Export-to-Python), the flowfile_frame API method (§3, optional), the WASM node set (§4, optional — the runnable palette + available:false teaser entries in flowfile_wasm/src/config/nodeCatalog.ts), the AI tools' node catalog (context for the AI agents to know the node exists — search flowfile_core/flowfile_core/ai/ for existing node references if the node should be AI-discoverable).
8. Tests — flowfile_core/tests/flowfile/ (e.g. test_basic_filter.py, test_filter_expressions.py are the pattern for a simple transform node). Assert: settings validate, the graph method adds a node with the right template counts, .collect()/execution produces correct data, and (if the node has a native vs polars-code split anywhere downstream) round-trip through save/reload.

1.3 Worked example — simple, single-input: filter

Read this to see the whole chain end to end for the simplest realistic case.

1. Settings: input_schema.NodeFilter(NodeSingleInput) (input_schema.py:542) → holds filter_input: transform_schema.FilterInput plus split_mode: bool for the pass/fail two-output variant.
2. Registered: NODE_TYPE_TO_SETTINGS_CLASS["filter"] = input_schema.NodeFilter (schemas.py:27 block).
3. Template: NodeTemplate(name="Filter data", item="filter", input=1, output=1, transform_type="narrow", node_type="process", laziness="lazy", image="filter.svg", ...) (configs/node_store/nodes.py, in the list built by get_all_standard_nodes).
4. Graph method add_filter (flow_graph.py:2224): _func(fl: FlowDataEngine) builds a polars-expr-transformer expression from either the advanced filter string or a BasicFilter locally (a run never writes settings; the drawer's pre-filled advanced expression comes from the display-only filter updator in setting_generator/settings.py), then returns fl.filter_split(expression) in split mode or fl.do_filter(expression) otherwise. Ends with self.add_node_step(..., node_type="filter", renew_schema=False, input_node_ids=[filter_settings.depending_on_id]).
5. Runtime dispatch: POST /update_settings/?node_type=filter → generic add_generic_settings (§1.1) — no filter-specific route exists.

1.4 Worked example — complex, source + two-input: read and join

• read (source, add_read at flow_graph.py:4646): settings NodeRead(NodeBase) carries a discriminated union InputTableSettings (csv/json/parquet/excel/ipc/ndjson/avro). _func() picks the execution path at run time: local/parquet/ipc/ndjson/utf-8-CSV stay in-core as a lazy scan (FlowDataEngine.create_from_path); exotic encodings and Excel go to the worker to convert to parquet first (FlowDataEngine.create_from_path_worker, an ExternalCreateFetcher). The schema callback is chosen by file type (declared fields win; else read-the-file per format; Excel uses a bounded 100-row sample) and is set on both node.schema_callback and node.user_provided_schema_callback so it survives a reset(). read is also the only node type with source-file change detection — the executor compares file-stat snapshots and reruns automatically if the underlying file changed on disk, and reset() deliberately preserves that tracking state.
• join (two-input, add_join at flow_graph.py:2537): settings NodeJoin(NodeMultiInput) → join_input: transform_schema.JoinInput. _func(main, right) deep-copies the join input, recomputes per-field is_available before calling main.join(...). Uses an explicit schema callback (calculate_join_schema from flowfile/schema_callbacks.py) instead of running _func on placeholder frames — this predicts the output schema without executing the join. Template: input=2, transform_type="wide" (wide ⇒ eligible to be forced onto the worker when optimizing downstream execution).

1.5 Network-source nodes: the worker does ALL external fetching

Rule, verified against three current implementations: a node whose _func talks to the outside world (HTTP API, OAuth provider, Kafka broker, a database over the network) must not perform that I/O inside flowfile_core. Build a worker-settings payload and dispatch it through an External*Fetcher (flowfile_core/flowfile_core/flowfile/flow_data_engine/subprocess_operations/subprocess_operations.py) — the fetcher POSTs the settings to flowfile_worker, which does the actual network round-trip in a spawned subprocess and hands back a cached, lazy result.

Canonical zero-local-branch template — add_google_analytics_reader (flow_graph.py:4254): _func() unconditionally builds ExternalGoogleAnalyticsFetcher(_build_worker_settings(), wait_on_completion=False) and calls .get_result(). There is no execution_location == "local" branch at all — GA4 always round-trips through the worker, full stop. The schema callback is derived purely from the selected metrics/dimensions (derive_schema, pure Python, no network call), so schema prediction never has to touch the worker either.

Default to the GA4 pattern for new network-source nodes. It is the smallest, least error-prone shape: one fetch path, one place secrets get decrypted (worker-side, just-in-time), nothing to keep in sync.

The nuance (verified, corrects a stricter-sounding rule of thumb): two existing readers — add_database_reader (flow_graph.py:3950) and add_rest_api_reader (flow_graph.py:4399) — do carry an execution_location == "local" branch that fetches in-process instead of going through the worker. This exists specifically to support --run-flow/CLI execution and other package/local modes where no worker process exists at all (core forces execution_location="local" and OFFLOAD_TO_WORKER=False in that mode). The REST API reader's local branch was added later than its worker-only fetch path and is explicitly modeled on the database reader (comment in source: # No worker service in local runs — fetch in-process (cf. add_database_reader).) — it is a deliberate, kept pattern, not a one-off legacy wart to avoid copying.

So the actual rule is:

• Never fetch externally when execution_location != "local" — that's the worker's job, no exceptions.
• Only add a gated if self.execution_location == "local": in-process fetch branch if your node has a concrete need to work without a worker (CLI/--run-flow, scheduled local execution). If you add one, mirror add_rest_api_reader's shape exactly: decrypt/resolve credentials the same way for both branches, and call the same underlying fetch function the worker would use (e.g. shared/rest_api/fetch.py::fetch_rest_api) so the two paths can't silently diverge.
• If you have no such requirement, don't add the branch — follow GA4, not the two exceptions.

1.6 Traps specific to this layer

• HTTP 419 (nonstandard, not a typo) is the status code add_generic_settings returns when your add_<type> method raises for any reason (routes.py::add_generic_settings). If a frontend network tab shows a bare "419" with no obvious cause, the bug is inside your node's _func/settings validation, not the transport.
• add_node_step deletes-then-recreates a node whose node_type changed but keeps it (via update_node) if the type is unchanged — don't assume "the node with this id" is stable across settings edits that change the node's own type.
• Nodes with input > 2 or multi=True in their template use append-connection semantics; templates with input <= 2 replace the main input on a new connection instead of appending (flow_node.py:689-693) — relevant if your node ever needs more than two ordered inputs.

2. Layer 2 — flowfile_frontend (main desktop/web UI)

Not required if the node is core-only (e.g. an internal/AI-orchestrated node with no user-facing settings). Required for anything a human drags onto the canvas.

2.1 The convention (no registry file, no build-time check)

There is no static list of node components. Two independent import.meta.glob calls, which must be kept pointing at the same tree, resolve a settings component purely from the node's item string:

1. app/composables/useDragAndDrop.ts:88 — import.meta.glob("../components/nodes/node-types/elements/**/*.vue"). getComponent() (useDragAndDrop.ts:158) builds the path elements/${camelCase(item)}/${TitleCase(item)}.vue and loads it for the VueFlow node itself.
2. app/components/nodes/GenericNode.vue:25 — import.meta.glob("./node-types/elements/**/*.vue"). loadDrawerComponent() (GenericNode.vue:47) applies the same convention for the settings-drawer component, via defineAsyncComponent with a 3s timeout and up to 3 retries.

TitleCase = each _-separated word of the node_type capitalized and joined: text_to_rows → dir textToRows, file TextToRows.vue. So text_to_rows resolves to exactly:

src/renderer/app/components/nodes/node-types/elements/textToRows/TextToRows.vue

If you get the path wrong, there is no build error. The glob still succeeds (it just doesn't find your file); loadDrawerComponent logs a console.error and the settings drawer silently never opens for that node. This is a pure-runtime failure mode — verify by actually opening the node's settings drawer in a running app, not just by compiling.

2.2 Add-a-node-UI checklist

1. Backend template already done (§1.2 step 3) — item, image, counts must already exist in /node_list.
2. Icon: add <image>.svg to src/renderer/app/features/designer/assets/icons/.
3. Settings component at exactly src/renderer/app/components/nodes/node-types/elements/<camelCase(item)>/<TitleCase(item)>.vue. Copy the filter/Filter.vue pattern:
   • Wrap the form in <generic-node-settings v-model="nodeXxx" @update:model-value="handleGenericSettingsUpdate" @request-save="saveSettings">.
   • Pull const { saveSettings, pushNodeData, handleGenericSettingsUpdate } = useNodeSettings({ nodeRef, onBeforeSave }) from app/composables/useNodeSettings.ts — it auto-sets is_setup = true and POSTs through nodeStore.updateSettings.
   • Implement loadNodeData(nodeId) (fetch via nodeStore.getNodeData(nodeId, false), hydrate setting_input) and end with defineExpose({ loadNodeData, pushNodeData, saveSettings }) — NodeSettingsDrawer.vue requires both hooks to exist on every drawer-entry component ("Universal Apply" contract).
   • Never save from loadNodeData: open-time fix-ups stay in the draft and go out with the next save. Closing the drawer saves only real user edits, except for a node whose NodeData.is_setup was false when the drawer loaded — then it saves what the drawer shows. setting_input alone can't tell: for a promise it may be a generated proposal that looks configured.
4. No registration step. Both globs auto-discover the new file by path.
5. Set prod_ready: false on the backend NodeTemplate while iterating — keeps it out of the production palette but it still loads fine in saved flows (useful for landing a half-finished node without exposing it).

2.3 Verify (don't just compile)

Drag the node from the palette, open its settings drawer, and confirm it loads — vue-tsc and npm run build:web will not catch a wrong component path. If you don't have a running desktop/web session available, at minimum grep for the exact expected path and confirm the file exists there with that exact casing (case-sensitive on Linux CI even if your local macOS filesystem is case-insensitive and would mask the bug).

3. Layer 3 — flowfile_frame (Python programmatic API)

Not required for most nodes — many core nodes (kernel/python_script, kafka_source, most ML nodes) have no flowfile_frame method and that's fine. Add one only when the node should be reachable from import flowfile_frame as ff code.