如果你想快速搭建一套 Agent 工作流又不想为每个语言环境分别维护 SDK那“只用 TOML 定义配置 通过 Webhook 通信”的设计会很值得参考。这个思路最早出现在 Show HN 的一条项目介绍上An agent engine with no SDK, just TOML and webhooks。它把 Agent 引擎做成了一个“配置驱动、事件驱动”的轻量层接入方不用安装任何 SDK只需要提交一个 TOML 描述文件然后在自己的系统里暴露一个 Webhook 接收通知即可。这篇文章会围绕这个设计理念展开先解释 Agent Engine、TOML、Webhook 三个核心概念再说明为什么要去掉 SDK接着用一套最小可运行的 Python 示例带你从零搭建一个“TOML 定义 Agent 工作流 Webhook 触发与回调”的引擎雏形。文章末尾还有常见问题排查和安全建议适合想自研轻量 Agent 编排平台或者对低代码化 AI 工作流感兴趣的同学。1. Agent Engine、TOML 与 Webhook 到底是做什么的1.1 Agent Engine把“智能体能力”变成可编排的工作流Agent Engine智能体引擎是一种运行环境负责接收一个任务拆解成步骤再调用不同类型的执行单元完成步骤最终返回结果。它和传统函数调用最大的区别在于执行单元往往不是写死在代码里的而是通过配置和协议来描述的。举个例子一个典型的 Agent 工作流可能包含接收一段用户提问。调用大模型生成回复。判断回复是否需要查数据库。如果需要执行 SQL 查询。把结果拼装成最终答案。如果把这些步骤全部通过硬编码写在一个类里系统会很难扩展。Agent Engine 的思路是把这些步骤抽成可配置的工作流Workflow每个步骤可以指向一个 Agent也可以指向一个外部动作比如发送 HTTP 请求、调用内部工具。1.2 TOML一种适合描述 Agent 配置的轻量格式TOMLToms Obvious Minimal Language是一种配置文件格式设计目标是“易于阅读、语义明确、能无歧义地映射为哈希表”。一个最简单的 TOML 文件长这样name demo-agent version 1.0.0 [agent] model demo-model max_tokens 1024TOML 在 Agent Engine 中的定位是“工作流和智能体的描述语言”。它不需要用户学习新的配置语法也不用写代码只需要声明“有哪些 Agent”“每个 Agent 用什么模型”“工作流有哪些步骤”即可。相比 JSONTOML 对人和 diff 更友好相比 YAMLTOML 的缩进和类型规则更严格不容易踩到“同一个 key 在不同引号下类型不一致”这类坑。因此在“配置复杂、希望减少解释成本”的场景里TOML 是一个合适的中立选择。1.3 Webhook让外部系统不装 SDK 也能与引擎协作Webhook 的含义是“反向 API”。通常我们调用 API 是主动向服务器发请求而 Webhook 是服务器在某些事件发生时通过 HTTP POST 请求通知第三方。它的本质是一个回调 URL。在 Agent Engine 中Webhook 承担了两个关键职责作为引擎的输入GitLab、GitHub、工单系统、支付系统等事件源把事件 POST 到引擎指定的 Webhook 地址触发对应工作流。作为引擎的输出工作流执行完成后引擎将结果 POST 回调用方提供的回调地址。这样一来接入方只需要维护两个 HTTP 地址不需要引入引擎的客户端 SDK也没有语言绑定、版本冲突、依赖升级问题。2. 为什么选择“无 SDK”架构2.1 SDK 集成方式带来的普遍问题SDKSoftware Development Kit本身是一个非常好的抽象它把复杂的网络通信、序列化、签名、失败重试等逻辑封装成函数。但在 Agent Engine 这类偏“平台化”的组件里SDK 集成模式会带来几个容易被低估的成本语言绑定成本引擎如果只提供 Java SDK那 Python、Go、Node.js 团队都要自己维护一个客户端。版本同步成本SDK 与引擎核心版本的兼容关系需要严格管理否则经常出现“SDK 更新了启动报 NoSuchMethodError”的问题可以参考许多 Android SDK、Vivado SDK 类工具链的兼容性痛点。升级推广成本业务方不升级 SDK就拿不到新能力升级 SDK又要重新回归测试。代码侵入成本业务系统需要引入依赖、初始化客户端、维护连接池使原本可以依靠配置完成的事情被迫进入了代码层。2.2 TOML Webhook 架构的优势让“无 SDK 架构”成立的核心是用 TOML 描述“做什么”用 Webhook 解决“怎么通信”两者组合起来就形成了一套事件驱动、配置驱动的轻量协议。这种架构有四个明显优势跨语言任何能发送 HTTP POST 请求、能解析 TOML 的语言都能接入前提是引擎没有隐藏依赖。最小化接入成本接入方只要写一个 TOML 文件、提供两个 URLSDK 和客户端库都不需要。方便可视化编排配置本身是纯文本可以被上层 UI 直接编辑和保存适合做低代码 Agent 工作流平台。故障边界清晰引擎只依赖 HTTP 协议调用方可以通过重试、超时、幂等等方式控制可靠性不再受制于某个 SDK 内部的连接管理。2.3 适用场景与边界这套设计不是万能的它更适合以下几类场景已有多个异构系统比如 Java 业务服务 Python AI 服务 Node.js 工单服务希望统一接入 Agent 能力。团队希望以“配置变更”而不是“代码发版”来调整 Agent 工作流。事件驱动型任务比如“代码变更 - 自动 review”“工单创建 - 生成摘要 - 回填业务系统”。如果是低延迟双向流式对话、强类型 RPC、需要复杂事务补偿的场景那纯 Webhook TOML 的模型就需要扩展比如叠加 WebSocket、消息队列或注册中心。它的价值在于简单和通用而不是替代所有中间件。3. 环境准备与项目结构3.1 运行环境为了演示一个最小可运行的 Agent Engine我会用 Python 来实现引擎主体因为 Python 3.11 之后内置了tomllib模块可以直接解析 TOML不需要额外安装解析库。# 建议环境 # Python 3.11 # 可选Flask 用于接收 Webhook pip install flask requests需要说明的是本文给出的代码是以“演示引擎设计思想”为目的并不是某个已发布项目的源码。实际项目中你可以用 Go、Java、Node.js 实现同样的解析和路由逻辑思路完全一致。3.2 推荐目录结构一个最小可运行的参考工程可以这样组织agent-engine-demo/ ├── engine.py # 引擎核心逻辑加载配置、路由、执行工作流 ├── webhook_server.py # Webhook 接收服务接收外部事件 ├── callbacks.py # 回调客户端向业务系统发送执行结果 ├── configs/ │ └── demo-agent.toml # Agent 与工作流配置 └── requirements.txt # Python 依赖这种结构的好处是配置、引擎逻辑、网络入口三者分离。你可以把configs/目录放到独立的配置中心或 Git 仓库后续做配置审计和版本回滚都很方便。3.3 依赖说明flask提供 Webhook 接收服务方便我们快速验证 HTTP POST 请求。requests用于引擎执行完成后向业务系统回传结果。内置模块tomllib解析 TOML、hmac签名校验、hashlib摘要算法。如果你使用的 Python 版本低于 3.11可以安装tomli作为兼容替代# Python 3.10 及以下 try: import tomllib except ModuleNotFoundError: import tomli as tomllib4. 核心配置与语法拆解4.1 用一个 TOML 文件描述整条 Agent 工作流下面是一份完整的demo-agent.toml你可以把它放到configs/目录下。# 文件路径configs/demo-agent.toml [engine] name demo-engine call_back_url https://business.example.com/callback [agent.code_reviewer] type llm model your-model-name system_prompt 你是一名资深代码审查专家请从可读性、安全性和性能三个角度 分析下方补丁并以 Markdown 格式输出审查意见。 max_tokens 2048 temperature 0.3 [workflow.code_review] description 收到代码仓库的 Merge Request 事件后执行代码审查 trigger { type webhook, path /webhooks/code-review, method POST } [[workflow.code_review.steps]] agent code_reviewer input {{ event.patch }} [[workflow.code_review.steps]] type webhook url {{ engine.call_back_url }} method POST payload { review_result: {{ steps[0].output }} }4.2 TOML 关键字段解释这段配置分为三个部分。[engine]定义引擎级信息name引擎名称只用于日志展示。call_back_url工作流执行完成后引擎需要把结果回调给哪个地址。这里用{{ engine.call_back_url }}引用避免在步骤里重复写 URL。[agent.code_reviewer]定义 Agenttype llm当前 Agent 类型是大模型调用。实际项目里还可能有tool、http、human_approval等类型。model模型名称示例中用了your-model-name占位需要替换成你实际能访问的模型。system_prompt系统提示词用三引号字符串保持多行格式。max_tokens/temperature调用大模型时控制输出长度和随机性。[workflow.code_review]定义工作流trigger表示该工作流被哪个 Webhook 事件触发。path是引擎接收事件的 URL 路径method指定 HTTP 方法。steps执行步骤列表使用 TOML 的数组格式按顺序执行。第一步agent code_reviewer表示调用名为code_reviewer的 Agent并把{{ event.patch }}作为输入。第二步type webhook表示向{{ engine.call_back_url }}发送结果。这里的{{ ... }}是模板语法不是 TOML 内置能力。引擎在运行时会把eventWebhook 事件体、steps步骤执行记录、engine引擎级配置注入模板上下文再进行字符串替换。这样配置里就可以相对自然地引用动态数据而不需要写代码。4.3 为什么选择“数组 内联表”描述步骤TOML 里表示步骤列表有两种常用方式[[workflow.code_review.steps]] agent code_reviewer input {{ event.patch }}和[workflow.code_review.steps] 1 { agent code_reviewer, input {{ event.patch }} } 2 { type webhook, url {{ engine.call_back_url }} }第一种方式适合步骤较多、字段较多的情况读起来像表格第二种适合快速定义一个简单顺序。推荐使用第一种因为后续每一步可能增加超时、重试、条件分支等字段二维表结构扩展性更好。5. 完整实战案例搭建一个 Webhook 触发的代码审查 Agent5.1 业务场景假设你的开发团队使用 GitLab希望在每次提交 Merge RequestMR时自动调用大模型进行代码审查审查结果再回传给业务系统。所有协调都通过 HTTP 完成GitLab 发送 MR 事件到引擎的/webhooks/code-review。引擎解析事件体渲染模板调用代码审查 Agent。引擎把审查结果 POST 到业务回调地址。接下来我会一步步写出可运行的最小实现。你可以直接复制到本地运行再根据实际场景调整。5.2 实现引擎核心逻辑文件路径engine.pyimport json import tomllib from pathlib import Path def load_config(path: str) - dict: 读取并解析 TOML 配置文件 with open(path, rb) as f: return tomllib.load(f) def find_workflow(config: dict, path: str, method: str): 根据 Webhook 的 path 和 method 找到对应的工作流 workflows config.get(workflow, {}) for name, wf in workflows.items(): trigger wf.get(trigger, {}) if trigger.get(path) path and trigger.get(method, POST).upper() method.upper(): return name, wf return None, None def render_template(template: str, context: dict) - str: 极简模板渲染把 {{ key.subkey }} 替换为上下文中的值 这里只做演示实际项目建议使用 Jinja2 result template while {{ in result and }} in result: start result.find({{) end result.find(}}, start) 2 expr result[start 2:end - 2].strip() value context for part in expr.split(.): if part.isdigit(): value value[int(part)] else: value value.get(part, ) result result[:start] str(value) result[end:] return resultfind_workflow是路由映射的关键函数。它遍历所有工作流的trigger如果发现path和method都匹配就把这条工作流返回给调用方。render_template是模板渲染的极简实现仅用于理解原理生产环境建议直接使用 Jinja2避免自己处理边界条件。继续补充步骤执行逻辑def run_workflow(config: dict, wf_name: str, wf: dict, event: dict) - dict: 按顺序执行工作流中的每一步 agents config.get(agent, {}) context { event: event, engine: config.get(engine, {}), steps: [], } for step in wf.get(steps, []): if agent in step: agent_conf agents.get(step[agent]) if not agent_conf: raise RuntimeError(fAgent not found: {step[agent]}) # 这里简化处理实际项目会在这里调用大模型服务 # 下面用一段字符串模拟 LLM 输出结果 prompt render_template(step.get(input, ), context) output f[模拟 LLM 输出] 已收到输入长度 {len(prompt)}正在生成审查意见... context[steps].append({agent: step[agent], output: output}) elif step.get(type) webhook: url render_template(step.get(url, ), context) payload_raw json.dumps(step.get(payload, {})) payload_str render_template(payload_raw, context) payload json.loads(payload_str) context[steps].append({webhook_url: url, payload: payload}) return context这段代码演示了一个极简的步骤执行器。真实引擎中agent步骤通常会封装对不同模型提供方的 HTTP 调用webhook步骤会把结果 POST 给目标系统。这里的重点是执行器本身不包含任何业务代码它只依据 TOML 一步步调度。5.3 实现 Webhook 接收服务文件路径webhook_server.pyimport hmac import hashlib from flask import Flask, request, jsonify import engine app Flask(__name__) # 配置一个 Webhook 签名密钥实际生产环境应该从环境变量或密钥管理系统读取 WEBHOOK_SECRET your-webhook-secret app.route(/) def index(): return {status: ok} app.route(/webhooks/code-review, methods[POST]) def code_review(): body request.get_data() # 1. 校验签名防止伪造事件 signature request.headers.get(X-Signature, ) expected hmac.new( WEBHOOK_SECRET.encode(), body, hashlib.sha256, ).hexdigest() if not hmac.compare_digest(signature, expected): return jsonify({error: invalid signature}), 401 # 2. 解析事件 event request.get_json(silentTrue) or {} # 3. 加载配置 config engine.load_config(configs/demo-agent.toml) # 4. 路由到工作流 wf_name, wf engine.find_workflow(config, /webhooks/code-review, POST) if not wf: return jsonify({error: workflow not found}), 404 # 5. 执行工作流 result engine.run_workflow(config, wf_name, wf, event) # 6. 在演示中直接返回执行结果生产环境可以先返回 202 再异步执行 return jsonify({workflow: wf_name, result: result}), 200 if __name__ __main__: app.run(host0.0.0.0, port8080)这个 Webhook 服务做了四件事校验签名、解析事件、找到对应工作流、执行工作流。需要注意演示里我直接在 HTTP 请求线程里执行了工作流这对长耗时任务不友好生产环境通常会把事件先放入消息队列再异步执行同时立即返回202 Accepted。5.4 运行与验证先把项目目录准备好pip install flask requests python webhook_server.py服务启动后另开一个终端发送测试请求curl -X POST http://127.0.0.1:8080/webhooks/code-review \ -H Content-Type: application/json \ -H X-Signature: 计算出的签名 \ -d {patch: diff --git a/src/main.py b/src/main.py\nprint(1)}签名可以用 Python 快速计算import hmac, hashlib body b{patch: diff --git a/src/main.py b/src/main.py\\nprint(1)} print(hmac.new(byour-webhook-secret, body, hashlib.sha256).hexdigest())把输出值替换进X-Signature请求头就能看到类似下面的返回{ workflow: code_review, result: { steps: [ { agent: code_reviewer, output: [模拟 LLM 输出] 已收到输入长度 64正在生成审查意见... } ] } }这一步跑通后你就拥有了一个“TOML 配置驱动 Webhook 触发”的 Agent 引擎雏形。接下来可以去替换agent.code_reviewer的模型调用代码让它真正调用大模型。6. Webhook 接收端的安全与可靠性设计Webhook 本质上就是把一个可被外部调用的 URL 暴露到了公网或内网。一旦 URL 被恶意调用轻则白跑资源重则数据泄露或触发危险操作。因此安全设计是不可省略的一环。6.1 签名校验最通用的做法是事件源在请求头带上签名引擎使用共享密钥对请求体计算 HMAC 签名并比对两者是否一致。expected hmac.new( WEBHOOK_SECRET.encode(), body, hashlib.sha256, ).hexdigest() if not hmac.compare_digest(signature, expected): return jsonify({error: invalid signature}), 401使用hmac.compare_digest而不是是为了防止时序攻击。同时要注意校验的对象必须是原始请求体request.get_data()而不是request.get_json()重新序列化后的字符串因为序列化可能改变空格和顺序导致签名对不上。6.2 幂等处理Webhook 事件在网络抖动时可能被事件源重发多次。如果每次收到事件都执行一次 Agent就可能重复扣费、重复回传结果。解决方式是给每个事件设置一个唯一 ID在引擎内部保存“已处理事件 ID”列表。收到事件后先查重如果已处理则直接返回旧结果。processed_events set() if event_id in processed_events: return jsonify({status: duplicate}), 200 processed_events.add(event_id)生产环境建议把 ID 存在 Redis 或数据库里并设置合理的过期时间。6.3 重试与超时引擎作为调用方在回调业务系统时也要考虑超时和重试。任何 HTTP 请求都可能失败所以回调客户端应配置连接超时比如 3 秒。读取超时比如 15 秒。重试策略对 5xx、网络超时等错误进行指数退避重试。最大重试次数比如 3 次避免无限重试。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session requests.Session() retries Retry(total3, backoff_factor1, status_forcelist[500, 502, 503, 504]) session.mount(https://, HTTPAdapter(max_retriesretries)) session.mount(http://, HTTPAdapter(max_retriesretries))7. 常见问题与排查思路在搭建和接入这套架构时下面几个问题最容易遇到。问题现象常见原因解决思路Webhook 请求返回 404trigger 中配置的 path 与服务器路由不一致检查 TOML 里的trigger.path确保与 Flask 路由一致签名校验失败事件源和引擎使用的密钥不一致或签名计算方式不一致对比双方签名算法、密钥、参与签名的字段事件能收到但工作流没有执行find_workflow没匹配上或步骤中 agent 名称拼写错误检查工作流名称、agent 名称、method 大小写回调业务系统超时回调地址不可达或者没有配置超时时间使用 curl 手动测试回调地址增加超时配置重复收到同一事件事件源重试机制导致增加幂等处理保存已处理事件 IDTOML 解析报错数组或内联表语法错误使用tomllib解析异常信息定位行列事件体里的字段取不到值模板 key 写错或事件体结构变化打印 event 原始结构核对{{ event.xxx }}补充一个排查技巧在 Webhook 服务里加一个“调试模式”把原始请求体、签名、命中工作流名称都写入日志。这样可以快速定位是网络层问题还是配置层问题还是引擎逻辑问题。import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(webhook) app.route(/webhooks/code-review, methods[POST]) def code_review(): body request.get_data() logger.info(receive webhook, path%s, body%s, request.path, body.decode(utf-8, errorsreplace)) ...8. 最佳实践与工程建议8.1 TOML 配置管理把 TOML 配置当作代码一样管理建议做到以下四点放入 Git 仓库通过 MR/PR 流程变更保证可审计。使用环境变量替换密钥和 URL不要直接把生产环境地址写在配置里。配置变更要有版本记录至少保留最近 N 个版本方便回滚。每次加载配置后做一次 schema 校验避免字段缺失或类型错误在运行阶段才暴露。def validate_config(config: dict): assert engine in config, 缺少 [engine] 段 assert agent in config, 缺少 [agent] 段 for name, agent in config[agent].items(): assert type in agent, fagent {name} 缺少 type8.2 日志与可观测性Agent 工作流相比普通接口调用更复杂。一次请求可能跨越多个步骤、多个外部系统因此建议为每次执行生成一个trace_id并从 Webhook 入口一路传递到后续的每一步。import uuid trace_id str(uuid.uuid4()) logger.info(workflow start, trace_id%s, workflow%s, trace_id, wf_name) # 每个步骤执行时都打印 trace_id step_index这样排查问题时可以按 trace_id 把所有相关日志串联起来快速定位是模型调用慢还是回调失败。8.3 异步执行与任务队列Webhook 请求应该快速返回耗时的 Agent 调用不适合直接阻塞在 HTTP 请求线程里。推荐的演进路径是Webhook 服务校验签名解析事件后把任务写入消息队列Redis Stream、RabbitMQ、Kafka 等。Worker 从队列里取出任务执行工作流。执行完成后回调业务系统。这样既降低了 Webhook 服务的负载也避免了事件源等待太久导致超时重发。8.4 安全边界对外暴露的 Webhook URL 必须有签名校验。不对外暴露引擎的管理接口配置修改走内网或运维平台。回调地址建议做白名单限制防止引擎被用来攻击内网其他服务SSRF。引擎在发起 HTTP 请求前应校验目标 URL 是否在允许列表内。日志中不要打印完整密钥、事件体中的敏感字段必要时脱敏后再记录。9. 总结与下一步学习方向到这里你已经掌握了“Agent Engine 不依赖 SDK只用 TOML 加 Webhook”的核心设计思路也亲手搭建了一个最小可运行的代码审查 Agent。回顾整条链路外部系统发送 Webhook 事件引擎根据 TOML 配置路由到对应工作流工作流按步骤调用 Agent最后把结果通过 Webhook 回调给业务系统。全程没有 SDK没有语言绑定没有复杂的连接管理。如果你打算把这个雏形应用到真实项目中建议按下面三步走第一步把模拟 LLM 输出替换成真实的大模型调用跑通第一个带真实业务价值的 Agent 场景。第二步加入消息队列和异步执行让 Webhook 服务可以秒回202让工作流在后台稳定执行。第三步完善配置校验、事件幂等、回调白名单、Trace 日志再接入配置中心实现线上动态更新。最后留一个值得思考的方向当工作流步骤变多之后顺序执行往往不够用你可能需要支持条件分支和并行步骤。到时候可以在 TOML 中增加if字段、for_each字段也可以在引擎层引入 DAG有向无环图来编排步骤。这会是这个“无 SDK”架构走向生产级的一个自然演进方向。如果你在接入 GitLab Webhook、配置签名校验或者设计 TOML 步骤结构时遇到了具体报错欢迎在评论区把报错信息和配置贴出来我们可以一起排查。如果这篇文章对你有帮助也可以收藏备用后面做 Agent 编排时会经常回来查。
无需SDK：用TOML和Webhook构建Agent工作流引擎
如果你想快速搭建一套 Agent 工作流又不想为每个语言环境分别维护 SDK那“只用 TOML 定义配置 通过 Webhook 通信”的设计会很值得参考。这个思路最早出现在 Show HN 的一条项目介绍上An agent engine with no SDK, just TOML and webhooks。它把 Agent 引擎做成了一个“配置驱动、事件驱动”的轻量层接入方不用安装任何 SDK只需要提交一个 TOML 描述文件然后在自己的系统里暴露一个 Webhook 接收通知即可。这篇文章会围绕这个设计理念展开先解释 Agent Engine、TOML、Webhook 三个核心概念再说明为什么要去掉 SDK接着用一套最小可运行的 Python 示例带你从零搭建一个“TOML 定义 Agent 工作流 Webhook 触发与回调”的引擎雏形。文章末尾还有常见问题排查和安全建议适合想自研轻量 Agent 编排平台或者对低代码化 AI 工作流感兴趣的同学。1. Agent Engine、TOML 与 Webhook 到底是做什么的1.1 Agent Engine把“智能体能力”变成可编排的工作流Agent Engine智能体引擎是一种运行环境负责接收一个任务拆解成步骤再调用不同类型的执行单元完成步骤最终返回结果。它和传统函数调用最大的区别在于执行单元往往不是写死在代码里的而是通过配置和协议来描述的。举个例子一个典型的 Agent 工作流可能包含接收一段用户提问。调用大模型生成回复。判断回复是否需要查数据库。如果需要执行 SQL 查询。把结果拼装成最终答案。如果把这些步骤全部通过硬编码写在一个类里系统会很难扩展。Agent Engine 的思路是把这些步骤抽成可配置的工作流Workflow每个步骤可以指向一个 Agent也可以指向一个外部动作比如发送 HTTP 请求、调用内部工具。1.2 TOML一种适合描述 Agent 配置的轻量格式TOMLToms Obvious Minimal Language是一种配置文件格式设计目标是“易于阅读、语义明确、能无歧义地映射为哈希表”。一个最简单的 TOML 文件长这样name demo-agent version 1.0.0 [agent] model demo-model max_tokens 1024TOML 在 Agent Engine 中的定位是“工作流和智能体的描述语言”。它不需要用户学习新的配置语法也不用写代码只需要声明“有哪些 Agent”“每个 Agent 用什么模型”“工作流有哪些步骤”即可。相比 JSONTOML 对人和 diff 更友好相比 YAMLTOML 的缩进和类型规则更严格不容易踩到“同一个 key 在不同引号下类型不一致”这类坑。因此在“配置复杂、希望减少解释成本”的场景里TOML 是一个合适的中立选择。1.3 Webhook让外部系统不装 SDK 也能与引擎协作Webhook 的含义是“反向 API”。通常我们调用 API 是主动向服务器发请求而 Webhook 是服务器在某些事件发生时通过 HTTP POST 请求通知第三方。它的本质是一个回调 URL。在 Agent Engine 中Webhook 承担了两个关键职责作为引擎的输入GitLab、GitHub、工单系统、支付系统等事件源把事件 POST 到引擎指定的 Webhook 地址触发对应工作流。作为引擎的输出工作流执行完成后引擎将结果 POST 回调用方提供的回调地址。这样一来接入方只需要维护两个 HTTP 地址不需要引入引擎的客户端 SDK也没有语言绑定、版本冲突、依赖升级问题。2. 为什么选择“无 SDK”架构2.1 SDK 集成方式带来的普遍问题SDKSoftware Development Kit本身是一个非常好的抽象它把复杂的网络通信、序列化、签名、失败重试等逻辑封装成函数。但在 Agent Engine 这类偏“平台化”的组件里SDK 集成模式会带来几个容易被低估的成本语言绑定成本引擎如果只提供 Java SDK那 Python、Go、Node.js 团队都要自己维护一个客户端。版本同步成本SDK 与引擎核心版本的兼容关系需要严格管理否则经常出现“SDK 更新了启动报 NoSuchMethodError”的问题可以参考许多 Android SDK、Vivado SDK 类工具链的兼容性痛点。升级推广成本业务方不升级 SDK就拿不到新能力升级 SDK又要重新回归测试。代码侵入成本业务系统需要引入依赖、初始化客户端、维护连接池使原本可以依靠配置完成的事情被迫进入了代码层。2.2 TOML Webhook 架构的优势让“无 SDK 架构”成立的核心是用 TOML 描述“做什么”用 Webhook 解决“怎么通信”两者组合起来就形成了一套事件驱动、配置驱动的轻量协议。这种架构有四个明显优势跨语言任何能发送 HTTP POST 请求、能解析 TOML 的语言都能接入前提是引擎没有隐藏依赖。最小化接入成本接入方只要写一个 TOML 文件、提供两个 URLSDK 和客户端库都不需要。方便可视化编排配置本身是纯文本可以被上层 UI 直接编辑和保存适合做低代码 Agent 工作流平台。故障边界清晰引擎只依赖 HTTP 协议调用方可以通过重试、超时、幂等等方式控制可靠性不再受制于某个 SDK 内部的连接管理。2.3 适用场景与边界这套设计不是万能的它更适合以下几类场景已有多个异构系统比如 Java 业务服务 Python AI 服务 Node.js 工单服务希望统一接入 Agent 能力。团队希望以“配置变更”而不是“代码发版”来调整 Agent 工作流。事件驱动型任务比如“代码变更 - 自动 review”“工单创建 - 生成摘要 - 回填业务系统”。如果是低延迟双向流式对话、强类型 RPC、需要复杂事务补偿的场景那纯 Webhook TOML 的模型就需要扩展比如叠加 WebSocket、消息队列或注册中心。它的价值在于简单和通用而不是替代所有中间件。3. 环境准备与项目结构3.1 运行环境为了演示一个最小可运行的 Agent Engine我会用 Python 来实现引擎主体因为 Python 3.11 之后内置了tomllib模块可以直接解析 TOML不需要额外安装解析库。# 建议环境 # Python 3.11 # 可选Flask 用于接收 Webhook pip install flask requests需要说明的是本文给出的代码是以“演示引擎设计思想”为目的并不是某个已发布项目的源码。实际项目中你可以用 Go、Java、Node.js 实现同样的解析和路由逻辑思路完全一致。3.2 推荐目录结构一个最小可运行的参考工程可以这样组织agent-engine-demo/ ├── engine.py # 引擎核心逻辑加载配置、路由、执行工作流 ├── webhook_server.py # Webhook 接收服务接收外部事件 ├── callbacks.py # 回调客户端向业务系统发送执行结果 ├── configs/ │ └── demo-agent.toml # Agent 与工作流配置 └── requirements.txt # Python 依赖这种结构的好处是配置、引擎逻辑、网络入口三者分离。你可以把configs/目录放到独立的配置中心或 Git 仓库后续做配置审计和版本回滚都很方便。3.3 依赖说明flask提供 Webhook 接收服务方便我们快速验证 HTTP POST 请求。requests用于引擎执行完成后向业务系统回传结果。内置模块tomllib解析 TOML、hmac签名校验、hashlib摘要算法。如果你使用的 Python 版本低于 3.11可以安装tomli作为兼容替代# Python 3.10 及以下 try: import tomllib except ModuleNotFoundError: import tomli as tomllib4. 核心配置与语法拆解4.1 用一个 TOML 文件描述整条 Agent 工作流下面是一份完整的demo-agent.toml你可以把它放到configs/目录下。# 文件路径configs/demo-agent.toml [engine] name demo-engine call_back_url https://business.example.com/callback [agent.code_reviewer] type llm model your-model-name system_prompt 你是一名资深代码审查专家请从可读性、安全性和性能三个角度 分析下方补丁并以 Markdown 格式输出审查意见。 max_tokens 2048 temperature 0.3 [workflow.code_review] description 收到代码仓库的 Merge Request 事件后执行代码审查 trigger { type webhook, path /webhooks/code-review, method POST } [[workflow.code_review.steps]] agent code_reviewer input {{ event.patch }} [[workflow.code_review.steps]] type webhook url {{ engine.call_back_url }} method POST payload { review_result: {{ steps[0].output }} }4.2 TOML 关键字段解释这段配置分为三个部分。[engine]定义引擎级信息name引擎名称只用于日志展示。call_back_url工作流执行完成后引擎需要把结果回调给哪个地址。这里用{{ engine.call_back_url }}引用避免在步骤里重复写 URL。[agent.code_reviewer]定义 Agenttype llm当前 Agent 类型是大模型调用。实际项目里还可能有tool、http、human_approval等类型。model模型名称示例中用了your-model-name占位需要替换成你实际能访问的模型。system_prompt系统提示词用三引号字符串保持多行格式。max_tokens/temperature调用大模型时控制输出长度和随机性。[workflow.code_review]定义工作流trigger表示该工作流被哪个 Webhook 事件触发。path是引擎接收事件的 URL 路径method指定 HTTP 方法。steps执行步骤列表使用 TOML 的数组格式按顺序执行。第一步agent code_reviewer表示调用名为code_reviewer的 Agent并把{{ event.patch }}作为输入。第二步type webhook表示向{{ engine.call_back_url }}发送结果。这里的{{ ... }}是模板语法不是 TOML 内置能力。引擎在运行时会把eventWebhook 事件体、steps步骤执行记录、engine引擎级配置注入模板上下文再进行字符串替换。这样配置里就可以相对自然地引用动态数据而不需要写代码。4.3 为什么选择“数组 内联表”描述步骤TOML 里表示步骤列表有两种常用方式[[workflow.code_review.steps]] agent code_reviewer input {{ event.patch }}和[workflow.code_review.steps] 1 { agent code_reviewer, input {{ event.patch }} } 2 { type webhook, url {{ engine.call_back_url }} }第一种方式适合步骤较多、字段较多的情况读起来像表格第二种适合快速定义一个简单顺序。推荐使用第一种因为后续每一步可能增加超时、重试、条件分支等字段二维表结构扩展性更好。5. 完整实战案例搭建一个 Webhook 触发的代码审查 Agent5.1 业务场景假设你的开发团队使用 GitLab希望在每次提交 Merge RequestMR时自动调用大模型进行代码审查审查结果再回传给业务系统。所有协调都通过 HTTP 完成GitLab 发送 MR 事件到引擎的/webhooks/code-review。引擎解析事件体渲染模板调用代码审查 Agent。引擎把审查结果 POST 到业务回调地址。接下来我会一步步写出可运行的最小实现。你可以直接复制到本地运行再根据实际场景调整。5.2 实现引擎核心逻辑文件路径engine.pyimport json import tomllib from pathlib import Path def load_config(path: str) - dict: 读取并解析 TOML 配置文件 with open(path, rb) as f: return tomllib.load(f) def find_workflow(config: dict, path: str, method: str): 根据 Webhook 的 path 和 method 找到对应的工作流 workflows config.get(workflow, {}) for name, wf in workflows.items(): trigger wf.get(trigger, {}) if trigger.get(path) path and trigger.get(method, POST).upper() method.upper(): return name, wf return None, None def render_template(template: str, context: dict) - str: 极简模板渲染把 {{ key.subkey }} 替换为上下文中的值 这里只做演示实际项目建议使用 Jinja2 result template while {{ in result and }} in result: start result.find({{) end result.find(}}, start) 2 expr result[start 2:end - 2].strip() value context for part in expr.split(.): if part.isdigit(): value value[int(part)] else: value value.get(part, ) result result[:start] str(value) result[end:] return resultfind_workflow是路由映射的关键函数。它遍历所有工作流的trigger如果发现path和method都匹配就把这条工作流返回给调用方。render_template是模板渲染的极简实现仅用于理解原理生产环境建议直接使用 Jinja2避免自己处理边界条件。继续补充步骤执行逻辑def run_workflow(config: dict, wf_name: str, wf: dict, event: dict) - dict: 按顺序执行工作流中的每一步 agents config.get(agent, {}) context { event: event, engine: config.get(engine, {}), steps: [], } for step in wf.get(steps, []): if agent in step: agent_conf agents.get(step[agent]) if not agent_conf: raise RuntimeError(fAgent not found: {step[agent]}) # 这里简化处理实际项目会在这里调用大模型服务 # 下面用一段字符串模拟 LLM 输出结果 prompt render_template(step.get(input, ), context) output f[模拟 LLM 输出] 已收到输入长度 {len(prompt)}正在生成审查意见... context[steps].append({agent: step[agent], output: output}) elif step.get(type) webhook: url render_template(step.get(url, ), context) payload_raw json.dumps(step.get(payload, {})) payload_str render_template(payload_raw, context) payload json.loads(payload_str) context[steps].append({webhook_url: url, payload: payload}) return context这段代码演示了一个极简的步骤执行器。真实引擎中agent步骤通常会封装对不同模型提供方的 HTTP 调用webhook步骤会把结果 POST 给目标系统。这里的重点是执行器本身不包含任何业务代码它只依据 TOML 一步步调度。5.3 实现 Webhook 接收服务文件路径webhook_server.pyimport hmac import hashlib from flask import Flask, request, jsonify import engine app Flask(__name__) # 配置一个 Webhook 签名密钥实际生产环境应该从环境变量或密钥管理系统读取 WEBHOOK_SECRET your-webhook-secret app.route(/) def index(): return {status: ok} app.route(/webhooks/code-review, methods[POST]) def code_review(): body request.get_data() # 1. 校验签名防止伪造事件 signature request.headers.get(X-Signature, ) expected hmac.new( WEBHOOK_SECRET.encode(), body, hashlib.sha256, ).hexdigest() if not hmac.compare_digest(signature, expected): return jsonify({error: invalid signature}), 401 # 2. 解析事件 event request.get_json(silentTrue) or {} # 3. 加载配置 config engine.load_config(configs/demo-agent.toml) # 4. 路由到工作流 wf_name, wf engine.find_workflow(config, /webhooks/code-review, POST) if not wf: return jsonify({error: workflow not found}), 404 # 5. 执行工作流 result engine.run_workflow(config, wf_name, wf, event) # 6. 在演示中直接返回执行结果生产环境可以先返回 202 再异步执行 return jsonify({workflow: wf_name, result: result}), 200 if __name__ __main__: app.run(host0.0.0.0, port8080)这个 Webhook 服务做了四件事校验签名、解析事件、找到对应工作流、执行工作流。需要注意演示里我直接在 HTTP 请求线程里执行了工作流这对长耗时任务不友好生产环境通常会把事件先放入消息队列再异步执行同时立即返回202 Accepted。5.4 运行与验证先把项目目录准备好pip install flask requests python webhook_server.py服务启动后另开一个终端发送测试请求curl -X POST http://127.0.0.1:8080/webhooks/code-review \ -H Content-Type: application/json \ -H X-Signature: 计算出的签名 \ -d {patch: diff --git a/src/main.py b/src/main.py\nprint(1)}签名可以用 Python 快速计算import hmac, hashlib body b{patch: diff --git a/src/main.py b/src/main.py\\nprint(1)} print(hmac.new(byour-webhook-secret, body, hashlib.sha256).hexdigest())把输出值替换进X-Signature请求头就能看到类似下面的返回{ workflow: code_review, result: { steps: [ { agent: code_reviewer, output: [模拟 LLM 输出] 已收到输入长度 64正在生成审查意见... } ] } }这一步跑通后你就拥有了一个“TOML 配置驱动 Webhook 触发”的 Agent 引擎雏形。接下来可以去替换agent.code_reviewer的模型调用代码让它真正调用大模型。6. Webhook 接收端的安全与可靠性设计Webhook 本质上就是把一个可被外部调用的 URL 暴露到了公网或内网。一旦 URL 被恶意调用轻则白跑资源重则数据泄露或触发危险操作。因此安全设计是不可省略的一环。6.1 签名校验最通用的做法是事件源在请求头带上签名引擎使用共享密钥对请求体计算 HMAC 签名并比对两者是否一致。expected hmac.new( WEBHOOK_SECRET.encode(), body, hashlib.sha256, ).hexdigest() if not hmac.compare_digest(signature, expected): return jsonify({error: invalid signature}), 401使用hmac.compare_digest而不是是为了防止时序攻击。同时要注意校验的对象必须是原始请求体request.get_data()而不是request.get_json()重新序列化后的字符串因为序列化可能改变空格和顺序导致签名对不上。6.2 幂等处理Webhook 事件在网络抖动时可能被事件源重发多次。如果每次收到事件都执行一次 Agent就可能重复扣费、重复回传结果。解决方式是给每个事件设置一个唯一 ID在引擎内部保存“已处理事件 ID”列表。收到事件后先查重如果已处理则直接返回旧结果。processed_events set() if event_id in processed_events: return jsonify({status: duplicate}), 200 processed_events.add(event_id)生产环境建议把 ID 存在 Redis 或数据库里并设置合理的过期时间。6.3 重试与超时引擎作为调用方在回调业务系统时也要考虑超时和重试。任何 HTTP 请求都可能失败所以回调客户端应配置连接超时比如 3 秒。读取超时比如 15 秒。重试策略对 5xx、网络超时等错误进行指数退避重试。最大重试次数比如 3 次避免无限重试。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session requests.Session() retries Retry(total3, backoff_factor1, status_forcelist[500, 502, 503, 504]) session.mount(https://, HTTPAdapter(max_retriesretries)) session.mount(http://, HTTPAdapter(max_retriesretries))7. 常见问题与排查思路在搭建和接入这套架构时下面几个问题最容易遇到。问题现象常见原因解决思路Webhook 请求返回 404trigger 中配置的 path 与服务器路由不一致检查 TOML 里的trigger.path确保与 Flask 路由一致签名校验失败事件源和引擎使用的密钥不一致或签名计算方式不一致对比双方签名算法、密钥、参与签名的字段事件能收到但工作流没有执行find_workflow没匹配上或步骤中 agent 名称拼写错误检查工作流名称、agent 名称、method 大小写回调业务系统超时回调地址不可达或者没有配置超时时间使用 curl 手动测试回调地址增加超时配置重复收到同一事件事件源重试机制导致增加幂等处理保存已处理事件 IDTOML 解析报错数组或内联表语法错误使用tomllib解析异常信息定位行列事件体里的字段取不到值模板 key 写错或事件体结构变化打印 event 原始结构核对{{ event.xxx }}补充一个排查技巧在 Webhook 服务里加一个“调试模式”把原始请求体、签名、命中工作流名称都写入日志。这样可以快速定位是网络层问题还是配置层问题还是引擎逻辑问题。import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(webhook) app.route(/webhooks/code-review, methods[POST]) def code_review(): body request.get_data() logger.info(receive webhook, path%s, body%s, request.path, body.decode(utf-8, errorsreplace)) ...8. 最佳实践与工程建议8.1 TOML 配置管理把 TOML 配置当作代码一样管理建议做到以下四点放入 Git 仓库通过 MR/PR 流程变更保证可审计。使用环境变量替换密钥和 URL不要直接把生产环境地址写在配置里。配置变更要有版本记录至少保留最近 N 个版本方便回滚。每次加载配置后做一次 schema 校验避免字段缺失或类型错误在运行阶段才暴露。def validate_config(config: dict): assert engine in config, 缺少 [engine] 段 assert agent in config, 缺少 [agent] 段 for name, agent in config[agent].items(): assert type in agent, fagent {name} 缺少 type8.2 日志与可观测性Agent 工作流相比普通接口调用更复杂。一次请求可能跨越多个步骤、多个外部系统因此建议为每次执行生成一个trace_id并从 Webhook 入口一路传递到后续的每一步。import uuid trace_id str(uuid.uuid4()) logger.info(workflow start, trace_id%s, workflow%s, trace_id, wf_name) # 每个步骤执行时都打印 trace_id step_index这样排查问题时可以按 trace_id 把所有相关日志串联起来快速定位是模型调用慢还是回调失败。8.3 异步执行与任务队列Webhook 请求应该快速返回耗时的 Agent 调用不适合直接阻塞在 HTTP 请求线程里。推荐的演进路径是Webhook 服务校验签名解析事件后把任务写入消息队列Redis Stream、RabbitMQ、Kafka 等。Worker 从队列里取出任务执行工作流。执行完成后回调业务系统。这样既降低了 Webhook 服务的负载也避免了事件源等待太久导致超时重发。8.4 安全边界对外暴露的 Webhook URL 必须有签名校验。不对外暴露引擎的管理接口配置修改走内网或运维平台。回调地址建议做白名单限制防止引擎被用来攻击内网其他服务SSRF。引擎在发起 HTTP 请求前应校验目标 URL 是否在允许列表内。日志中不要打印完整密钥、事件体中的敏感字段必要时脱敏后再记录。9. 总结与下一步学习方向到这里你已经掌握了“Agent Engine 不依赖 SDK只用 TOML 加 Webhook”的核心设计思路也亲手搭建了一个最小可运行的代码审查 Agent。回顾整条链路外部系统发送 Webhook 事件引擎根据 TOML 配置路由到对应工作流工作流按步骤调用 Agent最后把结果通过 Webhook 回调给业务系统。全程没有 SDK没有语言绑定没有复杂的连接管理。如果你打算把这个雏形应用到真实项目中建议按下面三步走第一步把模拟 LLM 输出替换成真实的大模型调用跑通第一个带真实业务价值的 Agent 场景。第二步加入消息队列和异步执行让 Webhook 服务可以秒回202让工作流在后台稳定执行。第三步完善配置校验、事件幂等、回调白名单、Trace 日志再接入配置中心实现线上动态更新。最后留一个值得思考的方向当工作流步骤变多之后顺序执行往往不够用你可能需要支持条件分支和并行步骤。到时候可以在 TOML 中增加if字段、for_each字段也可以在引擎层引入 DAG有向无环图来编排步骤。这会是这个“无 SDK”架构走向生产级的一个自然演进方向。如果你在接入 GitLab Webhook、配置签名校验或者设计 TOML 步骤结构时遇到了具体报错欢迎在评论区把报错信息和配置贴出来我们可以一起排查。如果这篇文章对你有帮助也可以收藏备用后面做 Agent 编排时会经常回来查。
