Dify 插件开发实验（04）：企业系统对接工具——如何用插件对接企业 ERP/CRM？

Dify 实验系列 · 插件开发 04/12 | 实验编号：DIFY-106-04
基于 Dify 1.16.1 实测（2026-08）

1. 业务场景

先讲一个我们实际遇到的场景。

客服工单 SaaS 的客服处理工单时，经常要查企业内部的 ERP 和 CRM：用户问「我的订单 DTF-ORD-001 发货了没」，客服要去 ERP 查订单状态、金额、物流；用户问「我这个客户的等级是什么」，客服要去 CRM 查客户等级、联系人和历史工单数。这些查询散落在客服的日常工作里，每次都要切系统、手动查、再回来答复。

我们第一次接这类需求时，第一反应也是「写个 HTTP 请求调接口不就行了」。真正动手才发现——企业系统对接的难，从来不在「调通」，而在「怎么调得稳、换得动」：真实系统在内网不可直连，联调连个环境都没有；每家客户的网关地址还不一样，写死一个就换不了客户；mock 联调时返回的字段和真实系统对不上，上线才发现解析全错。

这不是个例。任何「客服/业务系统要对接企业 ERP、CRM、内部系统」的场景都是这个模式：订单查询、客户查询、库存查询、发票查询——真实环境里这些系统往往不可直连、地址各不相同、接口契约还随时可能变。

2. 场景痛点

这个流程的痛点，在对账查询时体现得最直接：

• 系统不可直连：企业 ERP/CRM 在内网，外部应用不能直接访问，联调时根本没有真实环境可用。
• 每个客户环境地址各异：这家客户的网关是 http://10.0.1.5，那家是 https://gw.xxx.com——地址写死在代码里，换一家客户就要改代码重新打包。
• 契约不一致：mock 联调时返回的字段和真实系统不一样，下游解析直接错位，上线才发现。
• 错误码一团乱：上游返回 404、500、超时，工具层不映射就直接抛给下游，业务无法区分「没有这个订单」和「系统挂了」。

本质上，企业系统对接的难点不在「调通一个接口」，而在「契约与环境的可管理性」——地址要能配置、返回要能对齐、错误要能分层。

3. 方案：为什么是工具插件

选工具插件，我们实际对比过：

• mock/真实双模式：本地 uvicorn mock 出与真实 API 返回结构一致的契约，开发期不依赖真实系统；
• base_url 配置化：服务地址放 credentials，客户环境改凭证即可，插件代码零改动——mock↔真实切换只改一个配置；
• 契约表 + 错误码映射落地：订单/客户各 4 个返回字段、四类错误码统一，下游消费有据可依。

这篇文章我们就用它搭一个对接企业系统的工具插件：order_query（按订单号查 ERP 订单）+ customer_query（按客户 ID 查 CRM 客户），跑通契约设计、mock 联调与配置化切换的全流程。

4. 整体架构

链路很清晰：入口收单号/客户号 → 工具查 ERP/CRM → 错误码映射 → 汇总输出。关键设计是「契约先行」——返回字段和错误码在动手写代码前先定成契约表，mock 与真实系统都按契约实现，切换才不出错。

5. 模块设计

5.1 多工具插件结构（provider yaml 的 tools 列表）

credentials_for_provider:
  api_key:
    type: secret-input
    required: true
  base_url:
    type: text-input
    default: http://host.docker.internal:8003   # 宿主机 mock 服务地址
    required: false
tools:
  - tools/order_query.yaml
  - tools/customer_query.yaml

一个 provider 声明多个工具：各自 tools/*.yaml + *.py；公共逻辑放 tools/common.py（官方 dify_extractor 有 helpers.py 先例，打包验证通过）。

5.2 公共请求模块（tools/common.py）

def request_json(base_url: str, api_key: str, path: str):
    """GET 请求，返回 (成功数据, 错误 JSON 字符串)。成功时错误为 None。"""
    try:
        resp = requests.get(f"{base_url}{path}",
                            headers={"X-API-Key": api_key}, timeout=10)
    except requests.exceptions.RequestException as e:
        return None, err(_ERR_UPSTREAM, f"enterprise system unreachable: {type(e).__name__}")
    if resp.status_code == 401:
        return None, err(_ERR_AUTH, "authentication failed, check api_key")
    if resp.status_code == 404:
        return None, err(_ERR_NOT_FOUND, f"resource not found: {path}")
    if resp.status_code != 200:
        return None, err(_ERR_UPSTREAM, f"enterprise system returned HTTP {resp.status_code}")
    try:
        return resp.json(), None
    except ValueError:
        return None, err(_ERR_UPSTREAM, "invalid JSON response from enterprise system")

两个工具复用同一 request_json，参数校验（ORD-/CUS- 正则）各自在工具代码内做，字段缺失用防御性 get 兜底。

5.3 base_url 配置化（切换零改代码）

base_url 放 credentials 而非写死在代码：客户环境把凭证里的地址改成真实 ERP/CRM 网关即可，插件代码零改动。mock↔真实切换实测：base_url 改错误地址 → upstream_error；改回 → 恢复。

5.4 插件网络路径（与 http 节点不同，实测结论）

工作流 http 节点走 ssrf_proxy（squid + 172.16.0.0/12 白名单）；插件内 requests 直连出口，不经代理——daemon 容器访问宿主机用 host.docker.internal；内网地址访问无 SSRF 限制，需在插件代码层自行控制，生产环境靠企业网络策略收口。

6. 运行验证

输入	预期	结果
mock 服务 curl 两端点	订单/客户契约结构正确	通过
order_query（正常订单）	订单详情结构正确	通过
customer_query（正常客户）	客户信息结构正确	通过
不存在的订单/客户	not_found	通过
错误 api_key	auth_failed	通过
mock 服务停止	upstream_error（timeout=10 生效）	通过
workflow 集成	双工具串联，订单错误不影响客户查询	通过
切换 base_url	错误地址 → upstream_error；改回 → 恢复	通过（配置化生效零改代码）

7. 实战坑

坑	现象	修复
契约不一致	mock 与真实返回字段不同 → 下游解析错	契约表落地（订单/客户各 4 字段）+ 字段缺失兜底（防御性 get）（实测）
插件出口网络（误以为受限）	以为与 http 节点一样走 ssrf_proxy 白名单	实测：插件 requests 直连不经 ssrf_proxy，无白名单限制——访问宿主机用 host.docker.internal，生产靠企业网络策略控制（实测，修正预期）
超时无兜底	上游卡死拖垮流程	requests timeout=10，超时归 upstream_error（与 02 四 code 统一）（实测）
base_url 写死	切换环境要改代码重新打包	credentials 配置化，mock↔真实切换零改代码（实测）
多工具公共逻辑重复	每个工具文件重复凭证/请求/错误代码	tools/common.py 收敛 request_json/err（实测，打包验证通过）

8. 实验文档及源码获取

• 实验文档（完整操作步骤）：DIFY-106-04：企业系统对接工具.md
• 源码（可直接导入）：dify106_04_验证应用.yml
• 插件包（签名安装包，控制台上传用）：dify106_04_enterprise_tool.signed.difypkg
• 全部源码目录：dify-106/dsl | 插件包目录：dify-106/plugins

文章聚焦核心配置与采坑点；实验的完整分步操作（节点搭建/参数表/调试指引）见实验文档原文。

---

下一篇：Dify 插件开发实验（05）：有状态与幂等——插件如何安全地保持状态和处理重复调用？

💬 你在这个实验的场景里踩过什么坑？欢迎评论区分享你的实战经验。