Flowfile Custom Node Authoring

Flowfile custom-node authoring

Generate one .py file that defines a user custom node via the node_designer SDK, from a description of what the node should do. The file is the whole node: the visual Node Designer round-trips it, the worker (or a Docker kernel) executes its process(), and the community registry publishes it as-is.

When NOT to use this skill

• Adding a built-in node type (a first-party transform wired across flowfile_core settings schema + FlowGraph + frontend registry + flowfile_frame + wasm) → flowfile-node-development. That is four registries agreeing by naming convention; this skill is one self-contained file.
• Publishing / browsing / installing shared nodes, the community registry, the index bot, device-flow PR publishing → see the community-nodes contract in the root CLAUDE.md and flowfile_core/flowfile/community_nodes/.
• Kernel container lifecycle, image builds, FLOWFILE_KERNEL_IMAGE_*, kernel↔core auth → flowfile-config-and-flags + flowfile_core/flowfile_core/kernel/.
• Debugging why an existing node opens in an error state, KernelRequiredError, exec-vs-AST scan failures → flowfile-debugging-playbook.
• Deep FlowFrame/Expr / generated-code parity → flowfile-frame-and-codegen.

All facts below verified in-repo on 2026-07-12, branch feature/add-custom-node-store. Re-verification commands are in §7 — run them before trusting a line number in a stale checkout.

0. Mental model — one file, one class

my_node.py
├─ import polars as pl
├─ from flowfile import node_designer as nd          # the ONLY flowfile import allowed
├─ class MyNodeSettings(nd.NodeSettings):  ...        # optional: the settings UI
└─ class MyNode(nd.CustomNodeBase):                   # exactly ONE such subclass per file
      node_name / metadata fields ...
      settings_schema: MyNodeSettings = MyNodeSettings()
      def process(self, *inputs: pl.LazyFrame) -> pl.LazyFrame: ...

• Canonical import: from flowfile import node_designer as nd. The SDK lives in shared/node_designer/ and is import-pure (no flowfile_core, DB, or migrations). A node file may import only the SDK + stdlib + third-party libraries — any other flowfile/flowfile_core.* import raises a contract error in worker/dry-run contexts (shared/node_designer/loading.py:22-26,48-54).
• Exactly one CustomNodeBase subclass per file — more raise NodeLoadError (loading.py:146-156).
• On disk: user nodes live in ~/.flowfile/user_defined_nodes/. Community installs write a flat <id>.py there.
• Where it runs: in-process/worker by default (environment="local"), or in an isolated Docker kernel (environment="kernel"). See §3.

1. The generation recipe (description → node)

Follow these steps in order. Each maps to a section below.

1. Classify the transform. From the description extract: how many inputs (default 1) and outputs (default 1); which knobs the user should be able to configure (columns, numbers, choices, toggles, secrets); whether it needs heavy/third-party libraries (sklearn, xgboost, lightgbm, statsmodels).
2. Pick the environment (§3). Pure-polars / stdlib → environment="local" (omit the field). Needs ML libs or isolation → environment="kernel" + dependencies=[...] (plain PyPI specs only). Never set execution_location — that is a core-level concept, not an SDK field.
3. Design settings_schema (§2.3–2.5). One control per knob, grouped into Sections. Keep every kwarg a designer literal (§4.4).
4. Write process(self, *inputs) (§2.2, §2.6). Read settings via self.settings_schema.<section>.<component>.value; .collect() when a library needs eager data; int(...)-cast NumericInput/SliderInput values; for kernel nodes import heavy libs inside process and never reference SDK types; return a LazyFrame/DataFrame, or a dict keyed by output_names for multi-output.
5. Add metadata + examples (§2.1). node_name (required), plus node_category/title/intro; author/version/tags for publishing. Add example_inputs + example_settings (both required to install/publish; both must be literals).
6. Self-check against guardrails (§4). No DENY constructs. Import only the SDK + third-party. One bare CustomNodeBase subclass with node_name + process. PNG icon only for published bundles.
7. Verify (§6). Ruff-clean the file; dry-run it via the designer Test tab or dry_run.py; place it in ~/.flowfile/user_defined_nodes/.

When you have a ~/.flowfile/user_defined_nodes/ reachable and the user wants it installed, write the file there directly; otherwise write to the path the user names (or the repo flowfile-community-nodes/nodes/<id>/node.py layout when they intend to publish).

2. The authoring contract

2.1 CustomNodeBase fields (shared/node_designer/custom_node.py:401-444)

A Pydantic model — declare attributes as annotated fields with defaults: node_name: str = "My Node".

requires_kernel: bool and kernel_id: str|None are deprecated — they map onto environment="kernel". Use environment.

2.2 process signature (custom_node.py:732-750)

def process(self, *inputs: pl.LazyFrame) -> pl.LazyFrame | pl.DataFrame:

• One pl.LazyFrame per connected input, in port order (inputs[0], inputs[1], …). .collect() inside process when you need eager data.
• Single output: return a LazyFrame or DataFrame (the framework normalizes).
• Multi-output (len(output_names) > 1): return dict[str, pl.LazyFrame | pl.DataFrame] keyed by the output_names.

2.3 Settings container — NodeSettings + Section

class MyNodeSettings(nd.NodeSettings):
    my_section: nd.Section = nd.Section(
        title="…", description="…", layout="vertical",   # or "horizontal"
        some_field=nd.TextInput(label="…"),
        another=nd.NumericInput(label="…", default=1.0),
    )

Section(title=None, description=None, hidden=False, layout="vertical", **components) — every keyword after the first four is a component that becomes a field (ui_components.py:254-317).

2.4 Control catalog (all subclass FlowfileInComponent; import as nd.<Name>)

options= for the selects is a list of bare strings or (value, label) tuples, or a dynamic marker class passed by class (not instance):

• nd.IncomingColumns — populate from the input frame's columns.
• nd.AvailableArtifacts — populate from upstream artifact names.
• nd.AvailableSecrets — the default for SecretSelector.

2.5 Column type filters — data_types= (shared/node_designer/types.py)

Groups: nd.Types.Numeric, .String, .AnyDate, .Boolean, .Binary, .Complex, .All. Specific: .Int/.Int64/…, .Float/.Float64, .Str, .Date, .Datetime, .Bool, .List, .Struct, etc. Accepts one value, a list, bare strings ("Numeric", "Int64", "int"/"float"/"str"), or a real pl.DataType. "ALL" = no filter. Both data_types=nd.Types.String and data_types=["Numeric"] are valid spellings.

2.6 Reading settings inside process()

Access: self.settings_schema.<section_attr>.<component_attr>.value. Gotchas:

• NumericInput / SliderInput .value is a number but is not coerced — it arrives as whatever was supplied (an int or a float, e.g. example_settings with "n_clusters": 3 yields int). Cast with int(...)/float(...) when you need a specific type.
• ColumnSelector(multiple=True).value → list[str]; multiple=False → a single column-name str.
• SecretSelector → read .secret_value (a SecretStr; call .get_secret_value()), not .value (which is the secret name). Local-only — secrets are not available inside kernels yet (§3.3).
• ColumnActionInput.value → a ColumnActionValue (.rows of ColumnActionRow, .group_by_columns, .order_by_column).

2.7 Dry runs — is_dry_run() and example_artifacts()

A dry run runs your real process() against example_inputs, but sandboxes its side effects. Two paths do this: the designer Test tab (user_defined/dry_run.py) and the community-node CI harness (community_nodes/dry_run_local.py, run by the registry's PR checks).

example_inputs must build a real frame. Polars infers each column's dtype from its first value, so "x1": [5, 6.5, 4.7, 6.9] infers Int64 and then rejects 6.5. Write 5.0. Bundle validation constructs every entry and reports EXAMPLES_INVALID (validation.py::_check_example_inputs_build), so this fails at validate time rather than as a polars traceback mid-dry-run.

flowfile_ctx.is_dry_run() -> bool — True only in a dry run; False in a real flow run, a single-file export, and an exported project. Branch on it to skip work that only makes sense against real data:

if not flowfile_ctx.is_dry_run():
    notify_downstream(result)

example_artifacts() — a node that reads an artifact another node publishes (a predict node loading a train node's model) has nothing to read in a dry run, so it fails on a missing artifact instead of exercising its real path. Override the hook to supply one:

def example_artifacts(self) -> dict:
    """Objects the dry-run seeds into the artifact store before process() runs."""
    from sklearn.ensemble import RandomForestClassifier   # heavy imports inside, as in process()
    ...
    return {"automl_model": {"pipeline": pipeline, "task": "classification", ...}}

• It is a method, not a class attribute — the objects (a fitted pipeline, a tokenizer) are not AST-extractable literals, and _check_examples only requires example_inputs/example_settings to be literals.
• Return a flat {name: obj} dict. Each entry is seeded into both the flow-local store (read_artifact) and the global store (get_global), in the CI harness and the Test tab alike.
• Build the objects in-process. Loading a fixture from disk or the network makes the security scanner attach fs_read / network to the node's consent disclosure (§4.2), and an inlined base64 model blob over 1024 chars is a hard FF-SEC-011 deny (§4.1).
• Only dry runs call it; a real flow run never does.

What the harness fakes. The dry-run flowfile_ctx is a faithful in-memory stand-in, not a no-op: publish_global returns a real positive id and versions on republish, get_global/read_artifact raise KeyError on a miss, publish_artifact raises ValueError on a duplicate name, and get_shared_location returns a writable temp path. Catalog APIs (read_catalog_table, list_catalogs, …) need a running server and raise a clear error — guard them with if not flowfile_ctx.is_dry_run():. Anything unmodelled falls through to a no-op returning None. A drift guard (tests/flowfile/community_nodes/test_dry_run_ctx_parity.py) fails CI if a new kernel API is added without a matching fake.

flowfile_ctx is only bound for environment="kernel" nodes. In a local node it is an undefined name at runtime.

3. Local vs kernel execution

3.1 Choosing

• environment="local" (default): runs in the Flowfile process/worker. Full SDK + polars available; secrets resolve; no Docker. Use for pure-polars/stdlib transforms.
• environment="kernel": runs in an isolated Docker kernel. Use when you need heavy libs (sklearn, xgboost, lightgbm, statsmodels) or isolation. Declare dependencies (plain PyPI specs) — auto-installed only for kernel nodes. A kernel node must be bound to a kernel before it runs; the user picks the kernel (and image flavour) in the UI. Unbound → KernelRequiredError (raised in flow_graph.py).

3.2 What is different inside a kernel (the real path)

A kernel node's source never runs directly. Core AST-generates a self-contained script (user_defined/kernel_codegen.py::generate_kernel_script, invoked from flow_graph.py). Consequences you must design around:

• SDK imports are stripped; only the node's own third-party imports + polars/json/logging/sys survive. Do not reference SDK types inside process.
• Only process + nested/helper defs are kept; all class-level attribute assignments are dropped. Put every bit of logic inside process — nothing computed at class-body scope exists at kernel runtime.
• Settings are baked as JSON and rebuilt behind a proxy, so self.settings_schema.<section>.<component>.value still works. Every settings value must be JSON-serializable or generation fails (KernelCodegenError).
• Inputs arrive as parquet scans (pl.scan_parquet per input); .collect() for eager work. Outputs are marshalled back via an injected flowfile_ctx.publish_output epilogue.
• flowfile_ctx is an injected runtime global (undefined in the plain file — that's expected; linters will flag it). Surface includes log_info(...) (surfaces in the Test panel), read_inputs/read_first, publish_artifact(name, obj) / read_artifact(name) (cloudpickle-backed) for persisting a trained model across nodes, publish_global / get_global for persisting one across flows, and is_dry_run() (§2.7).