localflow-sample-node

一个功能齐全的 LocalFlow 最小示例节点仓库，演示了节点开发的关键交互和最佳实践。

本仓库适合以下场景：

🎓 学习 LocalFlow 节点开发 — 从零理解节点的完整生命周期
🧪 验证节点加载流程 — 确认 LocalFlow 能正确发现、加载和执行节点
📋 作为开发模板 — 基于此仓库快速创建自己的节点仓库

快速开始

将本仓库克隆到 LocalFlow 的节点目录：

# 克隆到 LocalFlow 的用户节点目录
git clone https://github.com/localflow-app/localflow-sample-node.git

在 LocalFlow 中通过节点浏览器添加 demo_node 节点到工作流。
配置「问候语」参数（或使用默认值 Hello LocalFlow!），运行工作流。
观察节点执行过程：控制台输出日志 + UI 进度条更新 + 输出数据传递。

仓库结构

localflow-sample-node/
├── demo_node/            # 示例节点目录
│   ├── node.json         # 节点元数据配置
│   └── node.py           # 节点执行脚本
├── manifest.json         # 仓库清单（LocalFlow 发现节点的入口）
├── LICENSE               # Apache 2.0 许可证
└── README.md             # 本文档

核心规则：每个节点是一个独立子目录，目录名即 node_type（蛇形命名 snake_case），目录内至少包含 node.json 和 node.py 两个文件。

逐文件解读

manifest.json — 仓库清单

manifest.json 是 LocalFlow 发现节点的入口，位于仓库根目录：

{
  "repo_name": "localflow-sample-node",
  "repo_url": "https://github.com/localflow-app/localflow-sample-node",
  "snapshot_version": "1.0.0",
  "snapshot_commit": "",
  "nodes": [
    "demo_node"
  ]
}

字段	说明
`repo_name`	仓库名称，用于在 LocalFlow 中标识
`repo_url`	仓库的远程地址
`snapshot_version`	快照版本号
`snapshot_commit`	快照对应的 Git commit SHA（由 LocalFlow 自动填充）
`nodes`	所有节点 `node_type` 的列表，必须与节点目录名一致
`min_app_version`	❌ 可选，LocalFlow 最低版本要求，低于此版本不加载该仓库

manifest.json 的作用：

发现节点：LocalFlow 启动时读取 manifest.json 的 nodes 列表，然后逐个加载对应目录
版本管理：snapshot_version + snapshot_commit 用于检测远程仓库是否有新版本
仓库元信息：repo_name 和 repo_url 用于 UI 显示和 GitHub 交互

⚠️ 新增节点后务必更新此文件，否则 LocalFlow 无法发现新节点。

node.json — 节点元数据

demo_node/node.json 定义了节点的身份、配置项和依赖：

{
  "node_type": "demo_node",
  "name": "Demo 节点",
  "description": "一个用于验证 localflow 节点加载流程的示例节点",
  "category": "示例",
  "version": "1.0.0",
  "entry_file": "node.py",
  "dependencies": [],
  "config_schema": {
    "greeting": {
      "type": "string",
      "label": "问候语",
      "default": "Hello LocalFlow!",
      "placeholder": "输入问候语文本",
      "description": "节点执行时输出的问候内容"
    }
  },
  "input_schema": {},
  "output_schema": {
    "demo_output": {
      "type": "string",
      "description": "问候语输出"
    }
  },
  "metadata": {
    "node_kind": "demo"
  },
  "examples": [
    {
      "title": "默认问候",
      "config": {
        "greeting": "Hello LocalFlow!"
      }
    }
  ],
  "input_example": {},
  "output_example": {
    "demo_output": "Hello LocalFlow!"
  }
}

字段说明：

字段	类型	必需	说明
`node_type`	string	✅	节点唯一标识，必须与目录名一致，使用蛇形命名
`name`	string	✅	节点显示名称
`description`	string	✅	节点功能描述
`category`	string	✅	节点分类，用于节点浏览器分组
`version`	string	✅	语义化版本号（SemVer）
`entry_file`	string	❌	入口文件名，默认 `"node.py"`
`dependencies`	string[]	❌	pip 依赖包列表，如 `["requests", "pandas"]`
`config_schema`	object	❌	配置项定义，LocalFlow 会据此自动生成 UI 表单
`metadata`	object	❌	附加元数据。系统使用 `metadata.node_kind` 标识特殊节点类型
`input_schema`	object	❌	输入变量 schema，用于工作流引擎校验上游输出
`output_schema`	object	❌	输出变量 schema
`examples`	array	❌	使用示例列表，每项含 `title` 和 `config`，显示在属性面板
`input_example`	object	❌	输入样例数据
`output_example`	object	❌	输出样例数据

config_schema 配置项类型：

类型	说明	额外字段
`string`	字符串输入	`placeholder` 输入框占位文本
`int`	整数输入	—
`float`	浮点数输入	—
`bool`	布尔开关	—
`enum`	枚举选择	必须提供 `options` 数组

每个配置项支持以下子字段：

子字段	类型	说明
`label`	string	显示标签（必填）
`type`	string	字段类型（必填）
`default`	any	默认值
`description`	string	字段说明文字，UI 中显示为帮助提示
`placeholder`	string	输入框占位文本（仅 `string` 类型）

node.json 高级字段

以上 input_schema、output_schema、metadata、examples、input_example、output_example 为可选字段，适用于需要与工作流引擎深度集成的节点：

input_schema / output_schema：定义节点的输入输出接口。AI 服务（ai_chat_service.py）读取这些 schema 以理解节点的数据协议，自动生成正确的连线代码
metadata：附加元数据。系统通过 metadata.node_kind 识别特殊节点类型。例如 Playwright 脚本节点的 node_kind: "playwright_script" 会被节点属性面板和注册表用于分支判断（node_base.py:169、node_registry.py:171、node_properties.py:272）
examples：使用示例列表，每项包含 title 和 config，显示在节点属性面板的"使用示例"区域（node_properties.py:320-321）
input_example / output_example：样例数据，供调试和 AI 节点生成参考

动态 config_schema

对于绝大多数节点，config_schema 是 node.json 中的静态 JSON。但少数特殊节点类型（如 Playwright 脚本节点）采用 动态 config_schema 模式：

用户在 UI 中输入一段 Python 脚本，其中包含 {{url}}、{{limit}} 等占位符
系统通过 extract_playwright_params() 提取占位符列表
build_playwright_config_schema() 基于占位符动态构建 schema，自动生成对应的配置表单字段

# 伪代码：动态 schema 生成逻辑
param_names = ["url", "limit"]  # 从脚本中提取
for name in param_names:
    schema[name] = {
        "type": "string",
        "label": name.replace("_", " ").title(),
        "default": "",
        "placeholder": f"填写 {name}",
    }

此模式在注册时通过 node_registry.py:171-172 替换静态 schema 实现。对于普通自定义节点，使用静态 JSON 即可，无需关注此模式。

node.py — 节点执行逻辑

demo_node/node.py 是节点的核心执行脚本：

def execute(self, input_data):
    """Demo 节点 - 输出问候语"""
    greeting = self.config.get("greeting", "Hello LocalFlow!")
    print(f"[Demo] 收到输入数据: {list(input_data.keys())}")
    report_progress(50, "正在生成问候语...")
    print(f"[Demo] 生成问候语: {greeting}")
    report_progress(100, "完成")
    return {**input_data, "demo_output": greeting}

这段简短的代码涵盖了 LocalFlow 节点的 全部关键交互：

交互	代码	说明
读取配置	`self.config.get("greeting", ...)`	从 `config_schema` 获取用户在 UI 中设置的值
接收输入	`input_data` 参数	获取上游节点传递的数据字典
进度报告	`report_progress(50, "...")`	向 UI 报告执行进度
传递输出	`return {**input_data, ...}`	将结果传递给下游节点

关键交互详解

配置读取：self.config

self.config 是一个字典，包含用户在 UI 中为 config_schema 定义的参数所设置的运行时值。

# 读取配置，提供默认值作为兜底
greeting = self.config.get("greeting", "Hello LocalFlow!")

要点：

键名与 node.json 中 config_schema 的键名一一对应
始终使用 .get(key, default) 而非 [key]，避免 KeyError
默认值应与 config_schema 中的 default 字段保持一致

数据接收：input_data

input_data 是一个字典，包含上游所有节点输出的合并数据。

# 查看上游传递了哪些数据
print(f"收到输入数据: {list(input_data.keys())}")

# 读取特定上游输出
some_value = input_data.get("upstream_key", None)

要点：

如果节点是工作流的第一个节点，input_data 为空字典 {}
上游节点的输出键名由上游节点的返回值决定
使用 .get(key, default) 安全访问，避免因上游数据缺失而崩溃

数据传递：返回值

execute 方法必须返回一个字典，该字典将合并到数据流中传递给下游节点。

# ✅ 推荐：展开 input_data，确保上游数据继续向下传递
return {**input_data, "demo_output": greeting}

# ❌ 不推荐：丢弃上游数据
return {"demo_output": greeting}

要点：

使用 {**input_data, "new_key": value} 模式，确保上游数据不丢失
输出键名建议使用具有描述性的名称（如 demo_output、file_path），避免与上游键冲突
如果新键与上游键重名，新值会覆盖上游值

进度报告：report_progress()

report_progress() 由 LocalFlow 运行时自动注入，无需 import，可直接在 execute 中调用。

# 报告进度：百分比 + 描述信息
report_progress(50, "正在生成问候语...")
report_progress(100, "完成")

要点：

percent 范围 0–100，超出会自动截断
message 为可选参数，在 UI 上显示当前步骤描述
适用于耗时操作（循环处理、网络请求、文件读写等）
如果不调用，UI 显示默认的旋转动画指示器
不影响执行性能，可放心在循环中使用

高级模式：子进程执行

对于需要隔离子环境、大型第三方依赖（如 Playwright、Selenium）或长时间运行的节点，execute() 可以不直接执行逻辑，而是生成一个独立的 Python 脚本并通过 subprocess.run() 启动子进程执行。

何时使用子进程模式

节点依赖需要独立安装（如 playwright、selenium）
执行可能阻塞 UI 或需要超时控制
需要隔离的进程环境（环境变量、工作目录）

LocalFlow 子进程协议

LocalFlow 内置了标准的子进程执行模板（node_base.py:214-248），生成的脚本框架会自动处理通信。核心机制：

┌─────────────────────────────────────────┐
│  NodeShim（父进程端）                     │
│  1. 嵌入 NODE_CONFIG 到脚本              │
│  2. 将 input_data 写入 stdin             │
│  3. 启动 subprocess.run(capture_output)  │
│  4. 从 stdout 解析 ###JSON_OUTPUT###     │
│  5. 从 ###PROGRESS## 解析进度并转发 UI    │
└──────────────┬──────────────────────────┘
               │ subprocess.run()
               ▼
┌─────────────────────────────────────────┐
│  子进程脚本                              │
│  - 读取 stdin → input_data              │
│  - 实例化 NodeShim → self.config        │
│  - 调用 execute(shim, input_data)       │
│  - print("###JSON_OUTPUT###")           │
│  - print(json.dumps(output_data))       │
│  - print("###JSON_OUTPUT_END###")       │
└─────────────────────────────────────────┘

在子进程模式下，节点作者可使用的通信机制：

功能	方式	说明
输出数据	`print("###JSON_OUTPUT###")` + JSON + `print("###JSON_OUTPUT_END###")`	由 NodeShim 模板自动生成，无需手动编写
进度报告	`print("###PROGRESS##{...}")` 或直接调用 `report_progress()`	模板中已定义 `report_progress()` 函数，内部使用此标记
错误退出	`sys.exit(1)`	父进程捕获非零返回码，将 stderr 作为错误信息

完整示例：子进程包装器模式

def execute(self, input_data):
    import json, subprocess, sys, os
    from pathlib import Path

    # 1. 生成运行时脚本
    runtime_script = '''#!/usr/bin/env python
import json, sys
NODE_CONFIG = {config_json}
def report_progress(pct, msg=""):
    print(f"###PROGRESS##{{json.dumps({{...}})}}")
{user_logic}
def main():
    input_data = json.loads(sys.stdin.read())
    class NodeShim:
        def __init__(self, cfg): self.config = cfg
    output = execute(NodeShim(NODE_CONFIG), input_data)
    print("###JSON_OUTPUT###")
    print(json.dumps(output))
    print("###JSON_OUTPUT_END###")
if __name__ == "__main__":
    main()
'''

    script_path = Path.cwd() / "runtime_script.py"
    script_path.write_text(runtime_script, encoding="utf-8")

    # 2. 子进程执行
    proc = subprocess.run(
        [sys.executable, str(script_path)],
        input=json.dumps(input_data),
        capture_output=True, text=True, timeout=120,
        cwd=os.getcwd()
    )

    # 3. 解析结果
    if proc.returncode != 0:
        raise RuntimeError(proc.stderr.strip())
    start = proc.stdout.find("###JSON_OUTPUT###")
    end = proc.stdout.find("###JSON_OUTPUT_END###")
    payload = json.loads(proc.stdout[start+17:end].strip())
    return {**input_data, **payload}

💡 Playwright 节点 是子进程模式的典型实现。它的 execute() 由 build_playwright_inline_wrapper_source()（playwright_node_utils.py:131）自动生成，除了标准通信外还额外注入了 LF_INPUT_DATA、LF_DOWNLOAD_DIR、LF_ARTIFACTS_DIR 等运行时变量，并对 Playwright 浏览器进行了 monkey-patch（自动拦截下载事件）。

最佳实践

1. 始终传递上游数据

# ✅ 好的做法
return {**input_data, "my_result": result}

# ❌ 不好的做法 — 下游节点将无法访问上游数据
return {"my_result": result}

2. 安全访问配置和输入

# ✅ 带默认值，避免 KeyError
value = self.config.get("key", "default")
data = input_data.get("key", None)

# ❌ 直接索引，可能抛出异常
value = self.config["key"]
data = input_data["key"]

3. 为耗时操作报告进度

def execute(self, input_data):
    items = input_data.get("items", [])
    results = []
    for i, item in enumerate(items):
        results.append(process(item))
        report_progress(int((i + 1) / len(items) * 100), f"处理中 {i+1}/{len(items)}")
    return {**input_data, "results": results}

4. 声明第三方依赖

如果节点使用了第三方库，必须在 node.json 的 dependencies 中声明：

{
  "dependencies": ["requests", "beautifulsoup4"]
}

LocalFlow 会自动安装声明的依赖，无需用户手动操作。

5. 使用 print() 输出调试日志

print() 的输出会显示在 LocalFlow 的控制台中，适合输出调试信息：

print(f"[MyNode] 开始处理，共 {len(items)} 条数据")

注意：如果节点使用 子进程执行模式，print() 的输出会被 subprocess.run(capture_output=True) 捕获，不会实时显示在控制台。子进程结束后，stdout 会通过 result['script_stdout'] 字段透传给父进程进行统一输出。如果需要实时日志，使用 report_progress() 或 sys.stderr.write()。

6. 错误处理

建议在节点中捕获异常并返回错误信息，而非让异常直接崩溃：

def execute(self, input_data):
    try:
        result = do_something()
        return {**input_data, "success": True, "result": result}
    except Exception as e:
        return {**input_data, "success": False, "error": str(e)}

7. 定义输入输出 Schema

为 input_schema 和 output_schema 填写准确的类型和描述有助于：

AI 服务理解节点的数据接口，生成正确的工作流连线
工作流引擎在运行前校验上下游数据兼容性
UI 显示输出数据的结构提示

{
  "output_schema": {
    "result": { "type": "string", "description": "处理结果" },
    "count":  { "type": "integer", "description": "数量" }
  }
}

创建自己的节点仓库

基于本仓库创建你自己的节点仓库只需三步：

1. 复制仓库结构

your-node-repo/
├── your_node/            # 你的节点目录
│   ├── node.json         # 修改为你的节点元数据
│   └── node.py           # 修改为你的节点逻辑
├── manifest.json         # 修改 repo_name、repo_url 和 nodes
└── LICENSE

2. 修改关键文件

manifest.json：

{
  "repo_name": "your-node-repo",
  "repo_url": "https://github.com/your-username/your-node-repo",
  "snapshot_version": "1.0.0",
  "snapshot_commit": "",
  "nodes": [
    "your_node"
  ]
}

your_node/node.json：

{
  "node_type": "your_node",
  "name": "你的节点",
  "description": "节点的功能描述",
  "category": "分类名称",
  "version": "1.0.0",
  "entry_file": "node.py",
  "dependencies": [],
  "config_schema": {}
}

your_node/node.py：

def execute(self, input_data):
    result = "your logic here"
    return {**input_data, "output_key": result}

3. 在 LocalFlow 中使用

将节点目录放入 LocalFlow 的节点目录，重启应用即可在节点浏览器中看到新节点。

许可证

本项目采用 Apache License 2.0 开源许可证。

localflow-sample-node

A minimal, fully functional LocalFlow sample node repository that demonstrates key interactions and best practices for node development.

This repository is ideal for:

🎓 Learning LocalFlow node development — Understand the complete node lifecycle from scratch
🧪 Verifying node loading — Confirm LocalFlow can correctly discover, load, and execute nodes
📋 Using as a development template — Quickly create your own node repository based on this one

Quick Start

Clone this repository into LocalFlow's node directory:

git clone https://github.com/localflow-app/localflow-sample-node.git

Add the demo_node to your workflow via the LocalFlow node browser.
Configure the "Greeting" parameter (or use the default Hello LocalFlow!) and run the workflow.
Observe the node execution: console log output + UI progress bar updates + output data passing.

Repository Structure

localflow-sample-node/
├── demo_node/            # Sample node directory
│   ├── node.json         # Node metadata configuration
│   └── node.py           # Node execution script
├── manifest.json         # Repository manifest (entry point for LocalFlow node discovery)
├── LICENSE               # Apache 2.0 License
└── README.md             # This document

Core rule: Each node is an independent subdirectory. The directory name is the node_type (snake_case), and must contain at least node.json and node.py.

File-by-File Walkthrough

manifest.json — Repository Manifest

manifest.json is the entry point for LocalFlow to discover nodes, located at the repository root:

{
  "repo_name": "localflow-sample-node",
  "repo_url": "https://github.com/localflow-app/localflow-sample-node",
  "snapshot_version": "1.0.0",
  "snapshot_commit": "",
  "nodes": [
    "demo_node"
  ]
}

Field	Description
`repo_name`	Repository name, used for identification in LocalFlow
`repo_url`	Remote URL of the repository
`snapshot_version`	Snapshot version number
`snapshot_commit`	Git commit SHA for the snapshot (auto-filled by LocalFlow)
`nodes`	List of all node `node_type` values, must match node directory names
`min_app_version`	❌ Optional. Minimum LocalFlow version required; the repo won't load if the app is older

Purpose of manifest.json:

Node discovery: LocalFlow reads the nodes list from manifest.json on startup, then loads each node directory
Version management: snapshot_version + snapshot_commit are used to detect new versions in the remote repository
Repository metadata: repo_name and repo_url are used for UI display and GitHub interactions

⚠️ Always update this file when adding new nodes, otherwise LocalFlow won't discover them.

node.json — Node Metadata

demo_node/node.json defines the node's identity, configuration, and dependencies:

{
  "node_type": "demo_node",
  "name": "Demo 节点",
  "description": "一个用于验证 localflow 节点加载流程的示例节点",
  "category": "示例",
  "version": "1.0.0",
  "entry_file": "node.py",
  "dependencies": [],
  "config_schema": {
    "greeting": {
      "type": "string",
      "label": "问候语",
      "default": "Hello LocalFlow!",
      "placeholder": "输入问候语文本",
      "description": "节点执行时输出的问候内容"
    }
  },
  "input_schema": {},
  "output_schema": {
    "demo_output": {
      "type": "string",
      "description": "问候语输出"
    }
  },
  "metadata": {
    "node_kind": "demo"
  },
  "examples": [
    {
      "title": "默认问候",
      "config": {
        "greeting": "Hello LocalFlow!"
      }
    }
  ],
  "input_example": {},
  "output_example": {
    "demo_output": "Hello LocalFlow!"
  }
}

Field Reference:

Field	Type	Required	Description
`node_type`	string	✅	Unique node identifier, must match directory name, snake_case
`name`	string	✅	Display name for the node
`description`	string	✅	Node functionality description
`category`	string	✅	Node category for grouping in the node browser
`version`	string	✅	Semantic version (SemVer)
`entry_file`	string	❌	Entry file name, defaults to `"node.py"`
`dependencies`	string[]	❌	pip dependency list, e.g. `["requests", "pandas"]`
`config_schema`	object	❌	Configuration definition; LocalFlow auto-generates UI forms from this
`metadata`	object	❌	Additional metadata. The system uses `metadata.node_kind` to identify special node types
`input_schema`	object	❌	Input variable schema for upstream validation and AI auto-wiring
`output_schema`	object	❌	Output variable schema
`examples`	array	❌	Usage examples list, each with `title` and `config`, shown in the properties panel
`input_example`	object	❌	Sample input data
`output_example`	object	❌	Sample output data

config_schema Parameter Types:

Type	Description	Extra Fields
`string`	String input	`placeholder` input hint text
`int`	Integer input	—
`float`	Float input	—
`bool`	Boolean toggle	—
`enum`	Enum selection	Must provide `options` array

Each parameter supports these sub-fields:

Sub-field	Type	Description
`label`	string	Display label (required)
`type`	string	Field type (required)
`default`	any	Default value
`description`	string	Field description, shown as a help tooltip in the UI
`placeholder`	string	Input placeholder text (`string` type only)

Advanced node.json Fields

The optional fields input_schema, output_schema, metadata, examples, input_example, and output_example are for nodes that need deep integration with the workflow engine:

input_schema / output_schema: Define the node's data interface. The AI service (ai_chat_service.py) reads these schemas to understand the node's data protocol and auto-generate correct wiring code
metadata: Additional metadata. The system uses metadata.node_kind to identify special node types. For example, Playwright script nodes use node_kind: "playwright_script" which triggers branching logic in the node properties panel and registry (node_base.py:169, node_registry.py:171, node_properties.py:272)
examples: Usage examples, each with title and config, displayed in the "Examples" section of the node properties panel (node_properties.py:320-321)
input_example / output_example: Sample data for debugging and AI-generated node reference

Dynamic config_schema

For the vast majority of nodes, config_schema is a static JSON in node.json. However, some special node types (like Playwright script nodes) use a dynamic config_schema pattern:

The user enters a Python script in the UI containing {{url}}, {{limit}} placeholders
The system extracts placeholder names via extract_playwright_params()
build_playwright_config_schema() dynamically builds a schema from the placeholders, auto-generating corresponding form fields

# Pseudocode: dynamic schema generation
param_names = ["url", "limit"]  # extracted from script
for name in param_names:
    schema[name] = {
        "type": "string",
        "label": name.replace("_", " ").title(),
        "default": "",
        "placeholder": f"Enter {name}",
    }

This pattern is implemented via node_registry.py:171-172, which replaces the static schema at registration time. For regular custom nodes, static JSON is sufficient — no need to worry about this pattern.

node.py — Node Execution Logic

demo_node/node.py is the core execution script:

def execute(self, input_data):
    """Demo 节点 - 输出问候语"""
    greeting = self.config.get("greeting", "Hello LocalFlow!")
    print(f"[Demo] 收到输入数据: {list(input_data.keys())}")
    report_progress(50, "正在生成问候语...")
    print(f"[Demo] 生成问候语: {greeting}")
    report_progress(100, "完成")
    return {**input_data, "demo_output": greeting}

This short code covers all key interactions of a LocalFlow node:

Interaction	Code	Description
Read config	`self.config.get("greeting", ...)`	Get runtime values set by the user in the UI via `config_schema`
Receive input	`input_data` parameter	Get the data dictionary passed from upstream nodes
Report progress	`report_progress(50, "...")`	Report execution progress to the UI
Pass output	`return {**input_data, ...}`	Pass results to downstream nodes

Key Interactions Explained

Reading Configuration: self.config

self.config is a dictionary containing the runtime values set by the user for the parameters defined in config_schema.

# Read config with a fallback default value
greeting = self.config.get("greeting", "Hello LocalFlow!")

Key points:

Keys correspond one-to-one with config_schema keys in node.json
Always use .get(key, default) instead of [key] to avoid KeyError
Default values should be consistent with the default field in config_schema

Receiving Data: input_data

input_data is a dictionary containing the merged output data from all upstream nodes.

# Check what data upstream has passed
print(f"Received input data: {list(input_data.keys())}")

# Read a specific upstream output
some_value = input_data.get("upstream_key", None)

Key points:

If the node is the first in the workflow, input_data is an empty dictionary {}
Upstream output key names are determined by the upstream node's return value
Use .get(key, default) for safe access to avoid crashes from missing upstream data

Passing Data: Return Value

The execute method must return a dictionary, which is merged into the data flow and passed to downstream nodes.

# ✅ Recommended: spread input_data to ensure upstream data continues downstream
return {**input_data, "demo_output": greeting}

# ❌ Not recommended: discards upstream data
return {"demo_output": greeting}

Key points:

Use the {**input_data, "new_key": value} pattern to ensure no upstream data is lost
Use descriptive output key names (e.g. demo_output, file_path) to avoid conflicts with upstream keys
If a new key has the same name as an upstream key, the new value overwrites the upstream value

Progress Reporting: report_progress()

report_progress() is automatically injected by the LocalFlow runtime. No import is needed — just call it directly in execute.

# Report progress: percentage + description
report_progress(50, "Generating greeting...")
report_progress(100, "Done")

Key points:

percent range is 0–100; values outside this range are automatically clamped
message is optional; displayed in the UI as the current step description
Use for long-running operations (loops, network requests, file I/O, etc.)
If not called, the UI shows a default spinner indicator
Does not affect execution performance; safe to use in loops

Advanced: Subprocess Execution

For nodes that need isolated environments, large third-party dependencies (e.g., Playwright, Selenium), or long-running operations, execute() can launch a child process via subprocess.run() instead of running logic directly in-process.

When to Use Subprocess Mode

Node dependencies need separate installation (e.g., playwright, selenium)
Execution may block the UI or needs timeout control
An isolated process environment (env vars, working directory) is required

LocalFlow Subprocess Protocol

LocalFlow has a built-in standard subprocess execution template (node_base.py:214-248). The generated script framework handles communication automatically:

┌─────────────────────────────────────────┐
│  NodeShim (parent process side)          │
│  1. Embed NODE_CONFIG into the script    │
│  2. Write input_data to stdin            │
│  3. Launch subprocess.run(capture_output)│
│  4. Parse ###JSON_OUTPUT### from stdout  │
│  5. Parse ###PROGRESS## markers → UI     │
└──────────────┬──────────────────────────┘
               │ subprocess.run()
               ▼
┌─────────────────────────────────────────┐
│  Child process script                    │
│  - Read stdin → input_data              │
│  - Instantiate NodeShim → self.config   │
│  - Call execute(shim, input_data)       │
│  - print("###JSON_OUTPUT###")           │
│  - print(json.dumps(output_data))       │
│  - print("###JSON_OUTPUT_END###")       │
└─────────────────────────────────────────┘

Communication mechanisms available in subprocess mode:

Feature	Method	Description
Output data	`print("###JSON_OUTPUT###")` + JSON + `print("###JSON_OUTPUT_END###")`	Auto-generated by NodeShim template; no manual coding needed
Progress reporting	`print("###PROGRESS##{...}")` or call `report_progress()` directly	The `report_progress()` function is already defined in the template
Error exit	`sys.exit(1)`	Parent captures non-zero return code, uses stderr as error message

Complete Example: Subprocess Wrapper Pattern

def execute(self, input_data):
    import json, subprocess, sys, os
    from pathlib import Path

    # 1. Generate runtime script
    runtime_script = '''#!/usr/bin/env python
import json, sys
NODE_CONFIG = {config_json}
def report_progress(pct, msg=""):
    print(f"###PROGRESS##{{json.dumps({{...}})}}")
{user_logic}
def main():
    input_data = json.loads(sys.stdin.read())
    class NodeShim:
        def __init__(self, cfg): self.config = cfg
    output = execute(NodeShim(NODE_CONFIG), input_data)
    print("###JSON_OUTPUT###")
    print(json.dumps(output))
    print("###JSON_OUTPUT_END###")
if __name__ == "__main__":
    main()
'''

    script_path = Path.cwd() / "runtime_script.py"
    script_path.write_text(runtime_script, encoding="utf-8")

    # 2. Execute child process
    proc = subprocess.run(
        [sys.executable, str(script_path)],
        input=json.dumps(input_data),
        capture_output=True, text=True, timeout=120,
        cwd=os.getcwd()
    )

    # 3. Parse results
    if proc.returncode != 0:
        raise RuntimeError(proc.stderr.strip())
    start = proc.stdout.find("###JSON_OUTPUT###")
    end = proc.stdout.find("###JSON_OUTPUT_END###")
    payload = json.loads(proc.stdout[start+17:end].strip())
    return {**input_data, **payload}

💡 Playwright node is a typical subprocess mode implementation. Its execute() is auto-generated by build_playwright_inline_wrapper_source() (playwright_node_utils.py:131). Beyond standard communication, it injects runtime variables (LF_INPUT_DATA, LF_DOWNLOAD_DIR, LF_ARTIFACTS_DIR) and monkey-patches Playwright browsers to auto-handle downloads.

Best Practices

1. Always Pass Upstream Data

# ✅ Good practice
return {**input_data, "my_result": result}

# ❌ Bad practice — downstream nodes cannot access upstream data
return {"my_result": result}

2. Safely Access Configuration and Input

# ✅ With default values, avoiding KeyError
value = self.config.get("key", "default")
data = input_data.get("key", None)

# ❌ Direct indexing, may raise exceptions
value = self.config["key"]
data = input_data["key"]

3. Report Progress for Long-Running Operations

def execute(self, input_data):
    items = input_data.get("items", [])
    results = []
    for i, item in enumerate(items):
        results.append(process(item))
        report_progress(int((i + 1) / len(items) * 100), f"Processing {i+1}/{len(items)}")
    return {**input_data, "results": results}

4. Declare Third-Party Dependencies

If your node uses third-party libraries, you must declare them in node.json:

{
  "dependencies": ["requests", "beautifulsoup4"]
}

LocalFlow will automatically install declared dependencies — no manual action required from users.

5. Use print() for Debug Logging

print() output is displayed in the LocalFlow console, suitable for debug information:

print(f"[MyNode] Starting processing, {len(items)} items total")

Note: If your node uses subprocess execution mode, print() output is captured by subprocess.run(capture_output=True) and won't appear in real time. After the child process finishes, stdout is forwarded through the result['script_stdout'] field for unified output. For real-time logging, use report_progress() or sys.stderr.write().

6. Error Handling

It's recommended to catch exceptions and return error information rather than letting exceptions crash the node:

def execute(self, input_data):
    try:
        result = do_something()
        return {**input_data, "success": True, "result": result}
    except Exception as e:
        return {**input_data, "success": False, "error": str(e)}

7. Define Input/Output Schemas

Providing accurate types and descriptions for input_schema and output_schema helps:

The AI service understand your node's data interface and generate correct workflow wiring
The workflow engine validate upstream/downstream data compatibility before execution
The UI display structure hints for output data

{
  "output_schema": {
    "result": { "type": "string", "description": "Processing result" },
    "count":  { "type": "integer", "description": "Item count" }
  }
}

Creating Your Own Node Repository

Create your own node repository based on this one in three steps:

1. Copy the Repository Structure

your-node-repo/
├── your_node/            # Your node directory
│   ├── node.json         # Modify with your node metadata
│   └── node.py           # Modify with your node logic
├── manifest.json         # Update repo_name, repo_url, and nodes
└── LICENSE

2. Modify Key Files

manifest.json:

{
  "repo_name": "your-node-repo",
  "repo_url": "https://github.com/your-username/your-node-repo",
  "snapshot_version": "1.0.0",
  "snapshot_commit": "",
  "nodes": [
    "your_node"
  ]
}

your_node/node.json:

{
  "node_type": "your_node",
  "name": "Your Node",
  "description": "Description of your node's functionality",
  "category": "Category Name",
  "version": "1.0.0",
  "entry_file": "node.py",
  "dependencies": [],
  "config_schema": {}
}

your_node/node.py:

def execute(self, input_data):
    result = "your logic here"
    return {**input_data, "output_key": result}

3. Use in LocalFlow

Place the node directory into LocalFlow's node directory and restart the application. Your node will appear in the node browser.

License

This project is licensed under the Apache License 2.0.

Name		Name	Last commit message	Last commit date
demo_node		demo_node
README.md		README.md
manifest.json		manifest.json

Latest commit

History

Folders and files

Repository files navigation

localflow-sample-node

目录

快速开始

仓库结构

逐文件解读

manifest.json — 仓库清单

node.json — 节点元数据

node.json 高级字段

动态 config_schema

node.py — 节点执行逻辑

关键交互详解

配置读取：self.config

数据接收：input_data

数据传递：返回值

进度报告：report_progress()

高级模式：子进程执行

何时使用子进程模式

LocalFlow 子进程协议

完整示例：子进程包装器模式

最佳实践

1. 始终传递上游数据

2. 安全访问配置和输入

3. 为耗时操作报告进度

4. 声明第三方依赖

5. 使用 print() 输出调试日志

6. 错误处理

7. 定义输入输出 Schema

创建自己的节点仓库

1. 复制仓库结构

2. 修改关键文件

3. 在 LocalFlow 中使用

许可证

localflow-sample-node

Table of Contents

Quick Start

Repository Structure

File-by-File Walkthrough

manifest.json — Repository Manifest

node.json — Node Metadata

Advanced node.json Fields

Dynamic config_schema

node.py — Node Execution Logic

Key Interactions Explained

Reading Configuration: self.config

Receiving Data: input_data

Passing Data: Return Value

Progress Reporting: report_progress()

Advanced: Subprocess Execution

When to Use Subprocess Mode

LocalFlow Subprocess Protocol

Complete Example: Subprocess Wrapper Pattern

Best Practices

1. Always Pass Upstream Data

2. Safely Access Configuration and Input

3. Report Progress for Long-Running Operations

4. Declare Third-Party Dependencies

5. Use print() for Debug Logging

6. Error Handling

7. Define Input/Output Schemas

Creating Your Own Node Repository

1. Copy the Repository Structure

2. Modify Key Files

3. Use in LocalFlow

License

About

Resources

Stars

Watchers

Forks

Contributors

Languages