Automatisch 核心概念详解：App、Connection、Flow 与 Step

Automatisch 核心概念详解App、Connection、Flow 与 Step【免费下载链接】automatischThe open source Zapier alternative. Build workflow automation without spending time and money.项目地址: https://gitcode.com/GitHub_Trending/au/automatischAutomatisch 是一款开源的工作流自动化平台自托管 Zapier 的替代方案用于把 Twitter、Slack、GitHub 等第三方服务串联成自动化业务流程。在创建第一个 Flow流程之前理解其四大核心概念至关重要App应用、Connection连接、Flow流程与Step步骤。本文以官方文档 key-concepts.md 为主线结合仓库源码深入剖析这四个概念的实现原理帮助你快速建立 Automatisch 的领域模型为构建自己的首个自动化流程打下基础。概念总览Automatisch 的领域模型非常精简四个概念层层递进概念一句话概括类比App可接入的第三方服务如 Twitter、Slack你日常使用的各种 SaaS 产品Connection某个 App 的登录凭据集合你在该服务中的某个账号Flow编排多个步骤的自动化业务流程一份待执行的任务清单StepFlow 中的单个环节触发器或动作清单上的每一项任务下面逐一展开。App可接入第三方服务的基石App 是你可以通过 Automatisch 使用的第三方服务例如 Twitter、GitHub、Slack 等。Automatisch 的目标就是把不同的 App 连接起来帮助你构建工作流——因此在使用其他任何概念时你都离不开 App。你可以查看完整的可用应用列表。官方文档还提示如果列表中没有你需要的服务可以通过请求新集成页面提交需求官方会收集所有请求并对呼声最高的集成优先排期开发。从源码看 App 的形态每个 App 在packages/backend/src/apps/目录下都有一个独立的子目录以 Slack 为例slack/index.js 通过defineApp定义了一个完整的 App 描述对象export default defineApp({ name: Slack, key: slack, iconUrl: {BASE_URL}/apps/slack/assets/favicon.svg, authDocUrl: {DOCS_URL}/apps/slack/connection, supportsConnections: true, baseUrl: https://slack.com, apiBaseUrl: https://slack.com/api, primaryColor: #4a154b, beforeRequest: [addAuthHeader], auth, actions, dynamicData, dynamicFields, });这里的defineApp定义在 packages/backend/src/helpers/define-app.js其实现非常朴素——直接返回传入的 App 定义对象export default function defineApp(appDefinition) { return appDefinition; }它本质上是一个类型约定层用来统一规范 App 描述的结构名称、key、认证方式、动作、动态字段等便于引擎和前端界面统一消费。actions、dynamicFields、dynamicData分别对应这个 App 提供的动作指令、动态表单字段与动态下拉数据源。Connection凭据的存放与加密要使用某个 App你需要先添加一个 Connection。Connection 本质上是一个存储指定服务凭据的地方比如 consumer key、consumer secret 等让 Automatisch 能代表你连接第三方应用。当你点击 Add connection 并选择一个 App 时界面会提示你填写该连接所必需的字段认证方式因 App 而异可能是 OAuth2 授权、API Key也可能是 Basic Auth。如果你在同一个 App 下有多个账号可以添加多个 Connection每个 Connection 独立对应一套凭据。从源码看 Connection 的实现packages/backend/src/models/connection.js 中Connection 模型的 JSON Schema 定义了关键字段key对应 App 的唯一标识、data加密后的凭据串、formattedData解密后的明文凭据对象、verified凭据是否通过验证、draft是否为草稿等。安全性是 Connection 设计的核心源码通过 AES 对称加密在入库前对凭据进行加密、在查询后即时解密encryptData() { if (!this.eligibleForEncryption()) return; this.data AES.encrypt( JSON.stringify(this.formattedData), appConfig.encryptionKey ).toString(); delete this.formattedData; } decryptData() { if (!this.eligibleForDecryption()) return; this.formattedData JSON.parse( AES.decrypt(this.data, appConfig.encryptionKey).toString(enc) ); }加密密钥来自appConfig.encryptionKey即服务端配置的APP_SECRET_KEY之类的加密密钥对应$beforeInsert/$beforeUpdate钩子执行加密、$afterFind钩子执行解密。此外Connection 还负责凭据有效性校验verifyAndUpdateConnection()调用对应 App 的app.auth.verifyCredentials($)验证凭据成功后把verified置为true、draft置为falsetestAndUpdateConnection()周期性复验连接是否仍然有效通过app.auth.isStillVerified($)判断并更新verified状态checkEligibilityForCreation()创建连接前检查该 App 是否被管理员禁用、是否只允许使用预定义认证客户端等app-config.js 中对应的appConfig配置项可对这些能力进行管控。这也解释了为什么创建连接时系统往往会先做一次凭据测试——Connection 的verified状态就是那次测试的结果。Flow编排业务流程的核心Flow 是 Automatisch 中最重要的部分它是通过连接多个 Step 来编排业务工作流的地方。官方文档给出的经典例子是这样一个 Flow搜索推文Search tweets按 Automatisch 关键词搜索推文发送消息到频道Send a message to channel把搜到的推文发布到指定的 Slack 频道。一个 Flow 中通常不止两个 Step你可以根据自己的业务需要串联任意多个环节。从源码看 Flow 的实现packages/backend/src/models/flow.js 中Flow 模型与 Step 是一对多关系steps关联按position升序排列并通过triggerStep仅type trigger的那个步骤与executions执行记录等关联暴露核心业务结构。新建 Flow 时会自动创建两个初始步骤createInitialSteps()的实现恰好印证了文档中第一步必须是触发器、后续是动作的规则async createInitialSteps() { await Step.query().insert({ flowId: this.id, type: trigger, position: 1 }); await Step.query().insert({ flowId: this.id, type: action, position: 2 }); }也就是说新建 Flow 默认就有一个位于 position 1 的 Trigger 步骤和一个位于 position 2 的 Action 步骤占位等待用户配置。Flow 的状态管理同样值得注意populateStatusProperty根据active字段与用户额度把 Flow 划分为draft草稿、published已发布、paused已暂停三种状态updateStatus()在发布时会为轮询型 Trigger 注册定时任务通过Engine.runInBackground把 Flow 交给队列调度在停用时移除对应的 repeatable job。执行间隔也是 Flow 级配置executionInterval的可选值为1、2、5、10、15、30、60分钟默认 15 分钟并通过getExecutionIntervalAsCron()映射为对应的 cron 表达式从源码看非企业版许可证下执行间隔会被限制为默认的 15 分钟。流程如何被执行packages/backend/src/engine/index.js 是流程执行的入口。run()首先通过buildFlowContext构建包含flow、triggerStep、actionSteps的完整上下文然后调用getInitialData获取触发器产生的初始数据对轮询型 Trigger 而言即本次拉取到的新记录列表最后遍历每条数据调用iterateSteps依次执行各 Action 步骤。对于需要定时运行的 FlowrunInBackground()会把任务交给 Bull 队列flowQueue异步调度。这意味着 Flow 的执行引擎天然支持轮询触发、Webhook 触发、手动测试触发等多种启动方式。Step流程的组成单元Step 是 Flow 中的单个环节。在上面的例子中搜索推文和发送消息到频道分别是这个 Flow 中的两个 Step。Step 分为两种类型Trigger触发器步骤启动整个 Flow 的环节例如搜索推文。可以把 Trigger 理解为流程的起点——没有 Trigger 的 Flow 无法被调度执行。Action动作步骤定义如何处理来自上游步骤的数据例如例子中的发送消息到频道。Action 位于 Trigger 之后可以串联多个。每个 Flow 的第一步必须是 Trigger 步骤后续步骤必须是 Action 步骤。从源码看 Step 的实现packages/backend/src/models/step.js 中Step 模型的 JSON Schema 明确限定了type只能取action或trigger并提供statusincomplete/completed、position步骤在 Flow 中的位置、parameters步骤参数、connectionId步骤绑定的连接等字段。模型还提供了便捷判断get isTrigger() { return this.type trigger; } get isAction() { return this.type action; }getTriggerCommand()/getActionCommand()会根据步骤的appKey与key到对应 App 的triggers/actions列表中找到该步骤对应的指令定义getNextStep()则按position 1定位下一个步骤——这正是 Flow 顺序执行链路的实现基础。Trigger 的类型约束在 packages/backend/src/helpers/define-trigger.js 中可以看到一个合法的 Trigger 必须满足以下条件之一否则定义时会直接抛错带有pollInterval轮询型触发器如每 15 分钟拉取一次type webhookWebhook 型触发器等待外部回调属于内置调度器类型everyNMinutes、everyHour、everyDay、everyWeek、everyMonth是mcpTool类型的 MCP 触发器。对应的校验代码为const isWebhookOrPoll triggerDefinition.pollInterval || triggerDefinition.type webhook; const schedulerTriggers [everyNMinutes, everyHour, everyDay, everyWeek, everyMonth]; const isSchedulerTrigger schedulerTriggers.includes(triggerDefinition.key); const isMcpTrigger triggerDefinition.key mcpTool; const haveValidTriggerType isWebhookOrPoll || isSchedulerTrigger || isMcpTrigger; if (!haveValidTriggerType) { throw new Error( Trigger must have a poll interval or be a webhook for ${triggerDefinition.key} ); }用真实代码验证文档示例文档示例中的两个步骤在仓库中都有真实实现Trigger搜索推文。twitter 应用的 search-tweets 触发器 声明了pollInterval: 15每 15 分钟轮询一次、必填参数searchTerm并在run()中调用searchTweets($)拉取包含关键词的最新推文export default defineTrigger({ name: Search tweets, key: searchTweets, pollInterval: 15, description: Triggers when there is a new tweet containing a specific keyword, phrase, username or hashtag., arguments: [ { label: Search Term, key: searchTerm, type: string, required: true }, ], async run($) { await searchTweets($); }, });Action发送消息到频道。slack 应用的 post-message 动作 读取步骤参数中的channel目标频道 ID与message消息文本调用 Slack 的chat.postMessageAPI 发送消息并支持可选的sendAsBot以机器人身份发送、botName机器人名称与botIcon机器人头像配置const postMessage async ($) { const { parameters } $.step; const channelId parameters.channel; const text parameters.message; ... const response await $.http.post(/chat.postMessage, data, { additionalProperties: customConfig, }); ... $.setActionItem(message); };两者组合起来就构成了文档中所描述的搜索推文 → 发送到 Slack 频道的完整自动化场景。四个概念如何协同工作理解这四个概念之后一条完整的工作流数据链路也就清晰了选定一个App如 Twitter为它创建Connection填入或授权凭据通过验证后保存并加密存储新建Flow系统自动生成 position 1 的 Trigger 步骤与 position 2 的 Action 步骤为 Trigger 步骤选择 App 与其触发器如searchTweets并绑定 Connection、配置参数为 Action 步骤选择 App 与其动作如 Slack 的sendAMessageToChannel、绑定 Connection、配置参数并可通过{{step.stepId.key}}引用上游步骤的输出数据测试并发布 Flow引擎engine/index.js按 Trigger 类型轮询 / Webhook / 调度器触发执行数据按position顺序在步骤间传递。延伸阅读创建你的第一个 Flow将上述概念付诸实践的完整操作指南可用的应用列表查看目前支持的全部 App 与集成关键概念英文原文官方文档对应章节想深入源码可从 packages/backend/src/models/ 下的connection.js、flow.js、step.js三个模型文件开始再结合 packages/backend/src/engine/index.js 理解执行链路。【免费下载链接】automatischThe open source Zapier alternative. Build workflow automation without spending time and money.项目地址: https://gitcode.com/GitHub_Trending/au/automatisch创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考