第 2 章：接入业务系统：ERP 等系统的集成

2.1 为什么必须"接系统"：AI 的三层能力

很多企业上了 AI 后发现"它是挺聪明，但查不到我的库存、看不到我的订单、进不了我的 ERP——那就是个高级聊天机器人"。AI Agent 真正的业务价值不在"能聊天"，而在"能办事"。按需集成深度，AI 可分为三层，见表 2-1：

表 2-1　四个能力层级的集成需求

层级	能力	示例	是否需要集成
L1 对话层	回答问题、生成内容	"退货政策是什么？"	不需要
L2 查询层	读取业务数据	"我昨天的订单发货了吗？"	需要（只读 API）
L3 操作层	执行业务操作	"帮我把这个订单的地址改成 XX"	需要（读写 API）

系统集成的深度，决定了 AI 在企业里的价值上限。 绝大多数企业做 AI 只做到了 L1，天花板很低。

2.2 三种集成方式：API / 数据库 / MCP

方式一：API 对接（最推荐，最安全）

AI 通过 Function Calling 调用企业系统的 REST API 读写数据。权限可控制在 API 层，标准化、易维护。

工作流程：用户提问 → AI 识别意图 → 调用"订单查询"工具 → API 请求 ERP → 返回订单状态 → AI 翻译成自然语言回复。

适合：目标系统有开放 API（金蝶云、用友云、织信 ERP、SAP 云等）[16]。

方式二：数据库直连（Plan B）

目标系统没有 API 但有数据库时，用只读数据库账号直连查询。部署快，但安全性风险高。

⚠️ 数据库直连的强制安全措施（每条都是底线）：

1. 只给 AI 分配只读权限的账号，绝对不能有写权限；
2. 对 AI 生成的 SQL 做白名单校验——只允许 SELECT，禁止 DROP / UPDATE / DELETE；
3. 设置查询超时（如 5 秒），防止 AI 生成低效 SQL 拖垮数据库；
4. 敏感字段（手机号、身份证等）返回前脱敏。

# 安全的 SQL 查询方案：预定义模板 + 参数填充
SQL_TEMPLATES = {
    "order_status": "SELECT status, logistics FROM orders WHERE order_id = %s",
    "inventory":    "SELECT product_name, stock FROM inventory WHERE sku = %s",
}

def safe_query(template_name: str, params: tuple) -> list:
    if template_name not in SQL_TEMPLATES:
        raise ValueError(f"未知查询模板: {template_name}")
    cursor.execute(SQL_TEMPLATES[template_name], params)
    return cursor.fetchall()

方式三：MCP 协议（新趋势）

MCP（Model Context Protocol）是 Anthropic 发起的开放协议[15]，相当于"AI 世界的 USB 接口"。通过一个 MCP Server，多个 AI 客户端（Claude、Cursor、WorkBuddy、TRAE Work 等）都能标准化地访问同一数据源。

优点：标准化、可复用、社区生态发展快。缺点：协议仍在早期，需要一定的技术配置能力。

2.3 ERP 接入通用方法论：三层架构

针对不同成熟度的 ERP，推荐"API 为主、数据库只读为辅、老旧 ERP 用 RPA 兜底"的三层架构，见表 2-2：

表 2-2　三类 ERP 的集成方式

ERP 类型	集成方式	典型厂商
云端 ERP	REST API 对接	金蝶云、用友云、织信 ERP
本地部署 ERP	数据库只读 + VPN/内网穿透	K3 WISE、U9、SAP 本地版
老旧无 API 的 ERP	RPA 模拟操作	管家婆、速达、老版单机软件

三层架构选型流程（照着这个判断走，不纠结），如图 2-1 所示：

图 2-1　ERP 接入方式决策路径

flowchart TD
    A[要接的 ERP 是什么形态?] --> B{有没有开放 API?}
    B -->|有| C[REST API 对接<br>最安全、最推荐]
    B -->|没有| D{能否开放只读数据库?}
    D -->|能| E[数据库只读 + VPN/内网穿透]
    D -->|不能| F[RPA 模拟操作<br>兜底方案]
    C --> G[权限拆分: 查询/制单账号分离]
    E --> G
    F --> G

2.3.1 前置准备（最容易翻车、最容易被跳过的环节）

原则：准备工作没做完，不做集成配置。 主数据不干净，后面所有自动化都是"垃圾进垃圾出"。

1. 云端 ERP：向厂商申请开放 REST API、开通独立集成 Token、配置 IP 白名单。Token 权限要拆清楚，不要拿管理员 Token 直接上。
2. 本地 ERP：DBA 新建只读数据库账号，仅开放业务库查询权限，禁止 update/delete。梳理核心业务表（库存、销售订单、采购订单、应收应付、生产订单、BOM），不必全库暴露。
3. 老旧 ERP：在专用虚拟机部署 ERP 客户端，RPA 固定账号登录，仅允许单据录入、报表导出，禁止反审核、删单操作。

权限拆分铁律：

• 查询账号：只读销售/采购/库存/财务/生产单据，无任何修改权限；

• 制单账号：仅允许新增订单、入库单、凭证草稿，禁止删除、过账、反审核。

2.4 具体示例一：金蝶云星空 MCP 接入

金蝶云星空（k3cloud）在国内中型制造企业铺得很广。通过社区开源的 kingdee-mcp 或 erp-mcp，可以让 Claude / Cursor / WorkBuddy 等直接查库存、建单、审批。

接入四步总览（本节 2.4 的四个步骤对应此图），如图 2-2 所示：

图 2-2　金蝶云星空 MCP 接入四步流程

flowchart LR
    A[1 安装 MCP 包<br>pip install kingdee-mcp] --> B[2 金蝶后台授权<br>创建集成用户/AppID/AppSecret]
    B --> C[3 配置 AI 客户端<br>写入 claude_desktop_config.json]
    C --> D[4 重启客户端<br>自然语言直接操作 ERP]

第 1 步：安装

pip install kingdee-mcp
# 或更全的工具集（51 个工具，覆盖金蝶云星空 + 用友）
pip install erp-mcp

第 2 步：金蝶后台授权

进入「系统管理 → 第三方系统登录授权 → 新增」，创建集成用户，获取 AppID 和 AppSecret（或按旧接口配置账套 ID / 用户名 / 密码）。

第 3 步：配置 AI 客户端（以 Claude Desktop 为例）

编辑 %APPDATA%\Claude\claude_desktop_config.json：

{
  "mcpServers": {
    "kingdee": {
      "command": "python",
      "args": ["-m", "kingdee_mcp.server"],
      "env": {
        "KINGDEE_BASE_URL": "https://your-org.ik3cloud.com/k3cloud",
        "KINGDEE_ACCT_ID": "你的账套ID",
        "KINGDEE_USERNAME": "集成用户名",
        "KINGDEE_APP_ID": "AppID",
        "KINGDEE_APP_SEC": "AppSecret"
      }
    }
  }
}

第 4 步：重启客户端，直接用自然语言操作

• "帮我看下金蝶里有多少个未审核的销售订单"

• "查一下物料编码 MAT001 的即时库存"

• "帮我新建一张采购订单，供应商 S001，物料 MAT001，数量 100"

• "审核这几张入库单：12345, 12346"

⚠️ 安全提醒：金蝶云星空的旧式认证直接把用户名密码 POST 过去拿 cookie，只适合内部使用，不要把 API 暴露到公网。

2.5 具体示例二：用友 YonSuite 接入

用友 YonSuite / YonBIP 通过 erp-mcp 的适配器层对接，环境变量配置：

{
  "mcpServers": {
    "erp": {
      "command": "python",
      "args": ["-m", "erp_mcp.server"],
      "env": {
        "ERP_TYPE": "yonsuite",
        "YONSUITE_BASE_URL": "https://yapi.yonyoucloud.com",
        "YONSUITE_APP_KEY": "...",
        "YONSUITE_APP_SECRET": "...",
        "YONSUITE_TENANT_ID": "..."
      }
    }
  }
}

erp-mcp 提供 51 个工具，绝大多数只读（销售/采购/库存/客户/应收应付/生产/经营看板/财务对账），写操作仅审批与自升级两个，均带二次确认门——这与第二部第 13 章"审批链"思想一致：AI 可以提，但关键操作必须人确认。

2.6 常见企业系统对接速查

表 2-3　常见企业系统对接速查

企业系统	集成方式	难度	周期	关键注意事项
企业微信/钉钉/飞书	官方 Bot API	低	1-2 周	作为 AI 交互入口，最先要接
OA 审批系统	API 对接	中	2-3 周	审批权限：AI 只能查自己发起的审批
ERP（用友/金蝶/SAP）	API 或数据库直连	高	3-6 周	接口老旧且文档不全，预留更多时间
CRM 系统	API 对接	中	2-3 周	主流 CRM 的 API 较完善
邮件系统	IMAP/SMTP	低	1-2 周	注意收发权限与发信频率限制

2.7 实现剖析：适配层契约、工厂与字段映射

2.3 节给出"API 为主、只读为辅、RPA 兜底"的三层架构。落到代码上，"接一个 ERP"被拆成三件事：统一契约（业务侧只依赖抽象接口）、工厂 + 延迟导入（未实现的厂商不阻断启动）、字段映射表（厂商字段 → 本地列）。本节从 prog/integrations/（4 个模块，共 216 行）摘取三段。

2.7.1 统一契约：业务侧只认 BaseERPAdapter

class BaseERPAdapter:
    """ERP 适配器抽象基类（契约）。

    实现类需覆盖 sync_orders / sync_inventory / sync_customers 三个同步入口；
    未覆盖的方法调用时抛 NotImplementedError，由调用方降级处理。
    """
    name: str = "base"            # 适配器标识（sap / yonyou / kingdee）
    # ★ 是否已具备真实实现（默认 False=骨架）；骨架适配器经 get_adapter 返回
    #   None，调用方自行降级，而不返回"方法即抛异常"的活实例。
    _implemented: bool = False

    def __init__(self, config: Optional[Dict[str, Any]] = None) -> None:
        self.config: Dict[str, Any] = dict(config or {})

    def sync_orders(self, **kwargs: Any) -> Any:
        """同步销售订单（ERP -> 本地 orders）。"""
        raise NotImplementedError("sync_orders 未实现")

    def sync_inventory(self, **kwargs: Any) -> Any:
        """同步库存（ERP -> 本地 inventory）。"""
        raise NotImplementedError("sync_inventory 未实现")

    def sync_customers(self, **kwargs: Any) -> Any:
        """同步客户主数据（ERP -> 本地 customers）。"""
        raise NotImplementedError("sync_customers 未实现")

这段代码印证 2.3 与 6.6 节：集成的复杂度被"抽象基类 + 三个同步入口"收口。业务侧只依赖 sync_orders / sync_inventory / sync_customers 三个动作，不感知 SAP 的 RFC、用友的 REST、金蝶的 WebAPI 差异；新增一个厂商只需新增一个子类。

_implemented 这个类属性是"诚实契约"的关键——骨架实现显式声明"我还没实现"，而不是留一个"看着能实例化、一调就抛异常"的活实例。

2.7.2 工厂 + 延迟导入：未实现或加载失败一律返回 None 降级

def get_adapter(name: str, config: Optional[Dict[str, Any]] = None) -> Optional[BaseERPAdapter]:
    """ERP 适配器工厂：按名称返回适配器实例（未实现/导入失败返回 None）。"""
    target = (name or "").strip().lower()
    try:
        if target == "sap":
            from prog.integrations.sap import SAPAdapter          # ★ 延迟导入
            cls = SAPAdapter
        elif target == "yonyou":
            from prog.integrations.yonyou import YonyouAdapter
            cls = YonyouAdapter
        elif target == "kingdee":
            from prog.integrations.kingdee import KingdeeAdapter
            cls = KingdeeAdapter
        else:
            _logger.warning("ERP 适配器 %r 未实现（可用：sap/yonyou/kingdee），返回 None", name)
            return None
        if not getattr(cls, "_implemented", False):
            _logger.warning("ERP 适配器 %r 当前为骨架未实现，返回 None，调用方请自行降级", name)
            return None
        return cls(config)
    except Exception as e:
        _logger.warning("ERP 适配器 %r 加载失败：%s，返回 None", name, e)
    return None

这段代码印证 2.3 与 6.6.3 节："未接入"与"接入失败"走同一条降级路径。三种情形——厂商名不认识、骨架未实现、导入或构造抛错——统一收敛为"打 warning 日志 + 返回 None"，由调用方决定降级行为（如提示"该 ERP 尚未接入，请改用只读查询通道"）。

这与 6.6.3 节数据库注入点是同一条纪律：外部依赖的缺失是默认路径而非异常分支，因此不会因为"某个 ERP 没接"而让整套系统起不来。

2.7.3 字段映射表：把厂商字段差异关进一张声明表

# 金蝶 K3Cloud（WebAPI 返回体）→ 本地列
KINGDEE_FIELD_MAPPINGS = {
    "orders": {
        "FBillNo": "order_no",           # 单据编号
        "FCustId": "#ref:customer_id",   # 客户内码（需映射客户编码）
        "FAmount": "total_amount",
        "FDate": "order_date",
        "FDocumentStatus": "#enum:status",   # 状态（需映射本地状态枚举）
    },
    "inventory": {
        "FMaterialId": "#ref:material_code",
        "FStockId": "#ref:warehouse_code",
        "FBaseQty": "quantity",
        "FBaseUnitId": "#ref:unit",
    },
    "customers": {
        "FCustNumber": "customer_id",
        "FCustName": "customer_name",
    },
}

# 统一注册表：厂商 -> 实体 -> 映射
FIELD_MAPPINGS = {
    "sap": SAP_FIELD_MAPPINGS,
    "yonyou": YONYOU_FIELD_MAPPINGS,
    "kingdee": KINGDEE_FIELD_MAPPINGS,
}

def get_mapping(vendor: str, entity: str) -> dict:
    """读取指定厂商/实体的字段映射（缺失返回空 dict）。"""
    return FIELD_MAPPINGS.get((vendor or "").lower(), {}).get(entity, {})

这段代码印证 2.3 与 14.3 节：字段映射不是散落在代码里的 if 分支，而是一张可按厂商/实体查询的声明表。值上的 # 前缀标记出三类需要转换的字段——#date:（日期格式）、#enum:（枚举映射）、#ref:（外键引用，如客户内码需查客户档案换成编码）。

把"差异"集中到一张表里，带来两个直接收益：① 换 ERP 时改表不改逻辑；② 映射本身成为可审计对象（哪几个字段做了枚举转换一目了然），与第 14 章"配置可训练、变更走审批"的治理思路一致。

2.8 与第二部框架的呼应：规则引擎兜底

接入系统后，AI 直接触达业务数据，风险也随之上升——这正是第二部"绳墨 Runtime"解决的核心问题。落地建议（呼应各章）：

• 只读查询（L2）：用规则引擎校验参数（如订单号格式、库存查询的物料是否存在），呼应第 11 章；

• 写操作（L3）：接入七层审核链，AI 生成"待确认"操作，人确认后才落库，呼应第 13 章；

• 权限：按角色/属性控制 AI 能看到的数据范围，呼应第 12 章；

• 审计：AI 的每次操作留痕、可追溯，呼应第 12 章七层审核链与 WORM 审计。

结论：工具（第 1 章）负责"把 AI 用起来"，接入（本章）负责"把 AI 接进系统"，而第二部原理负责"让 AI 接入后不闯祸"。三者缺一不可。

2.9 本章小结

• 集成方式按安全性与成熟度分三层：API 对接（首选）、数据库只读（Plan B）、RPA（兜底），见图 2-1；

• 数据库直连的四条底线：只读账号、SQL 白名单、查询超时、敏感字段脱敏——每条都是底线，不是可选项；

• MCP 把集成标准化为"AI 世界的 USB 接口"，一次开发、多端复用，是当前增长最快的接入方式；

• 接入的终点不是"能查到"，而是"写操作也受控"：参数校验（第 11 章）、审核链（第 13 章）、权限与审计（第 12 章）缺一不可；

• 一句话记住本章：工具负责"用起来"，接入负责"接进系统"，第二部负责"接进去不闯祸"。

---