获取工作流

获取输入参数

获取待确认操作

检查文件是否存在

定义字段属性 Args: default: 默认值 required: 是否必填（默认 False） description: 字段描述 Example: class My

解决一个确认请求 Args: confirm_id: 确认 ID approved: 是否批准 sudo_password: sudo 密码（仅在 require_sudo=True 时需要）

调用工具 Args: name: 工具名称 params: 参数（字符串会被解析为 JSON） Returns: 工具执行结果

添加普通边（支持一对多并行） 当一个节点连接多个目标时，这些目标会并行执行。 Args: from_node: 源节点名（可以是 "__start__" 表示工作流入口）

获取 logger（各模块使用） Args: name: 模块名（通常使用 __name__） Example: logger = get_logger(__name__) logger.i

执行节点（包含条件检查和重试逻辑） Args: state: 工作流状态 context: 工作流上下文 Returns:

获取指定技能 Args: name: 技能名称 Returns: Skill 对象，不存在返回 None

添加输入参数

添加节点

Validate and update the todo list. Validation Rules: - Each item must have: content, status - Status must be

清空所有注册

获取全局配置

注册自定义状态字段

启动服务器

获取当前输出通道

解析技能目录 Args: skill_dir: 技能目录路径 Returns: Skill 对象 Raises: FileNotFoundError: SKILL.md 不存

执行工作流 Args: inputs: 输入数据 context: 上下文（用于子工作流继承） stream: 是否流式输出 thread_id: 会话 ID

按描述匹配技能（用于自动选择） Args: query: 查询字符串 Returns: 匹配的 Skill 列表（按相关度排序）

删除文件

关闭所有连接

获取全局数据库管理器

创建上下文副本（用于子工作流）

执行 PostgreSQL SQL

发布工作流为 API

添加字段

执行 Skill 脚本 Args: skill_name: 技能名称 script_path: 脚本路径 args: 命令行参数 timeout: 超时时间 Retur

获取所有工作流

获取会话管理器

中止任务 Returns: True 如果成功中止，False 如果任务不存在或已完成

获取输出键

导出工作流结构（用于可视化） Returns: 工作流结构字典，包含： - id: 工作流 ID - name: 工作流名称 - vers

注销工具

内部：推送事件

获取 OpenAI function calling 格式的工具列表 用于 LLM 调用

主动推送消息到指定 Channel 的用户。 供工作流/定时任务调用，实现模型主动发消息。 Args: channel_name: Channel 名称（如 "feishu_sales"） user

返回当前时间戳（秒）

获取指定模型配置

获取指定名称的节点 Args: name: 节点名称 Returns: 节点对象，如果不存在返回 None

Get built-in skills directory path.

获取单例实例

检查 Redis 是否可用

从环境变量加载配置（兼容旧方式）

获取全局文件存储实例

获取调试会话