无需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 配置的轻量格式

TOML（Tom's Obvious Minimal Language）是一种配置文件格式，设计目标是“易于阅读、语义明确、能无歧义地映射为哈希表”。一个最简单的 TOML 文件长这样：

name = "demo-agent" version = "1.0.0" [agent] model = "demo-model" max_tokens = 1024

TOML 在 Agent Engine 中的定位是“工作流和智能体的描述语言”。它不需要用户学习新的配置语法，也不用写代码，只需要声明“有哪些 Agent”“每个 Agent 用什么模型”“工作流有哪些步骤”即可。

相比 JSON，TOML 对人和 diff 更友好；相比 YAML，TOML 的缩进和类型规则更严格，不容易踩到“同一个 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 集成方式带来的普遍问题

SDK（Software 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 文件、提供两个 URL，SDK 和客户端库都不需要。
• 方便可视化编排：配置本身是纯文本，可以被上层 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 tomllib

4. 核心配置与语法拆解

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]定义 Agent：

• type = "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 内置能力。引擎在运行时会把event（Webhook 事件体）、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 触发的代码审查 Agent

5.1 业务场景

假设你的开发团队使用 GitLab，希望在每次提交 Merge Request（MR）时，自动调用大模型进行代码审查，审查结果再回传给业务系统。所有协调都通过 HTTP 完成：

1. GitLab 发送 MR 事件到引擎的/webhooks/code-review。
2. 引擎解析事件体，渲染模板，调用代码审查 Agent。
3. 引擎把审查结果 POST 到业务回调地址。

接下来我会一步步写出可运行的最小实现。你可以直接复制到本地运行，再根据实际场景调整。

5.2 实现引擎核心逻辑

文件路径：engine.py

import 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 result

find_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(f"Agent 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.py

import 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(silent=True) 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(host="0.0.0.0", port=8080)

这个 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\n+print(1)"}'

签名可以用 Python 快速计算：

import hmac, hashlib body = b'{"patch": "diff --git a/src/main.py b/src/main.py\\n+print(1)"}' print(hmac.new(b"your-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(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504]) session.mount("https://", HTTPAdapter(max_retries=retries)) session.mount("http://", HTTPAdapter(max_retries=retries))

7. 常见问题与排查思路

在搭建和接入这套架构时，下面几个问题最容易遇到。

问题现象	常见原因	解决思路
Webhook 请求返回 404	trigger 中配置的 path 与服务器路由不一致	检查 TOML 里的trigger.path，确保与 Flask 路由一致
签名校验失败	事件源和引擎使用的密钥不一致，或签名计算方式不一致	对比双方签名算法、密钥、参与签名的字段
事件能收到，但工作流没有执行	find_workflow没匹配上，或步骤中 agent 名称拼写错误	检查工作流名称、agent 名称、method 大小写
回调业务系统超时	回调地址不可达，或者没有配置超时时间	使用 curl 手动测试回调地址，增加超时配置
重复收到同一事件	事件源重试机制导致	增加幂等处理，保存已处理事件 ID
TOML 解析报错	数组或内联表语法错误	使用tomllib解析异常信息定位行列
事件体里的字段取不到值	模板 key 写错，或事件体结构变化	打印 event 原始结构，核对{{ event.xxx }}

补充一个排查技巧：在 Webhook 服务里加一个“调试模式”，把原始请求体、签名、命中工作流名称都写入日志。这样可以快速定位是网络层问题，还是配置层问题，还是引擎逻辑问题。

import logging logging.basicConfig(level=logging.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", errors="replace")) ...

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, f"agent {name} 缺少 type"

8.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 请求线程里。推荐的演进路径是：

1. Webhook 服务校验签名，解析事件后，把任务写入消息队列（Redis Stream、RabbitMQ、Kafka 等）。
2. Worker 从队列里取出任务，执行工作流。
3. 执行完成后，回调业务系统。

这样既降低了 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 编排时会经常回来查。