老系统对接AI Agent有多难？遗留ERP/CRM改造为Agent Tool的3种实战方案

shanwei_spider

46人浏览 · 2026-06-22 07:09:21

shanwei_spider · 2026-06-22 07:09:21 发布

做过企业AI落地的人大概率都碰过这种硬骨头：想给销售做个CRM查询Agent，结果老CRM是十年前的C/S架构，连个HTTP接口都没有；想对接ERP查库存，原厂报个接口开发要几十万，周期三个月；硬闯数据库吧，几百张表没文档，动错一个字段就是生产事故。

新系统对接Agent是搭积木，老系统对接Agent是考古——你得在不敢碰、不能改、没人懂的三重约束下，把尘封的业务能力挖出来，变成Agent能调用的标准工具。这中间的坑，远不止技术本身。

本文基于十余个制造、流通类企业的落地经验，总结出三套经过生产验证的改造方案，从零侵入快速验证到标准化长期演进，覆盖从用友U8、金蝶K3到定制化老ERP的各类场景，讲清每种方案的适用边界、实现步骤、性能表现与踩坑底线，帮你在不折腾生产系统的前提下，把遗留系统快速接入AI Agent体系。

一、先搞懂：遗留系统为什么是Agent落地的拦路虎

十年以上的ERP、CRM、OA系统，普遍有三大死穴，决定了它不能像新系统那样直接接API：

1.1 接口能力几乎为零

大量C/S架构系统只有胖客户端，没有REST API，部分连WebService都没开放
少数有接口的系统，协议老旧（SOAP、DCOM、RFC），参数复杂，文档缺失
厂商锁死严重，二次开发报价高、周期长，部分厂商甚至已经倒闭

1.2 业务逻辑全在黑盒里

表结构无文档，一个业务状态可能分散在五六张表中，靠触发器、存储过程联动
直接改数据库会绕过应用层校验，导致数据不一致、流程断档、审计缺失
核心业务跑在上面，任何改造都把「不出事」放在第一位，宁可不做也不能出错

1.3 技术栈断层严重

Delphi、VB6、PowerBuilder、.NET Framework 2.0等老旧技术栈，现在能维护的人越来越少
部署环境复杂，有的跑在Windows Server 2008上，连Python都装不上
数据孤岛严重，系统间靠手工导Excel同步，本身就没有统一的数据入口

正因如此，遗留系统改造的核心原则永远是：稳定优先，侵入性越低越好；价值先行，先跑通核心场景再谈扩展。不要追求一步到位做完美架构，先让Agent能用、有用，比什么都重要。

二、方案一：数据层直连封装 —— 只读场景最快落地路径

这是成本最低、风险最小、见效最快的方案，也是绝大多数企业落地的第一站。核心思路很简单：绕开应用层，直接从数据库读取数据，封装成标准工具供Agent调用。完全不改老系统应用，只需要一个只读账号。

2.1 核心架构与适用场景

适用场景：

80%的查询类需求：客户信息查询、订单状态查询、库存余量查询、财务报表统计
老系统无开放接口，但数据库可访问（Oracle、SQL Server、MySQL均可）
项目初期快速验证价值，不想投入大量开发成本

绝对不适用： 写操作、需要触发业务流程的场景（比如创建订单、审批单据），直接改库会绕过所有业务校验。

2.2 实施步骤

数据资产盘点：反向梳理核心业务表，搞清楚客户、订单、库存等核心数据在哪张表、关键字段是什么、状态枚举值对应关系。优先梳理高频查询场景，不要一上来就全表盘点。
权限最小化开通：申请独立的只读账号，只开放必要的表和字段；有条件的一定要连从库或只读实例，绝对不碰主库。
业务SQL封装：把常用查询写成参数化SQL，内置业务逻辑映射（比如状态码转中文、金额单位换算），不要把原始表结构暴露给Agent。
工具标准化封装：包装成MCP Server或REST API，加上参数校验、数据脱敏、调用审计，供Agent调用。

2.3 核心代码实现

下面是一个典型的ERP库存查询MCP Server实现，底层直连SQL Server，上层暴露标准工具接口：

# erp_db_mcp_server.py
import os
import pymssql
from dotenv import load_dotenv
from mcp.server import Server
from mcp.types import TextContent
from mcp.server.stdio import stdio_server

load_dotenv()
server = Server("erp-inventory-server")

# 统一数据库连接管理
def get_db_conn():
    return pymssql.connect(
        server=os.getenv("ERP_DB_HOST"),
        user=os.getenv("ERP_DB_READ_USER"),
        password=os.getenv("ERP_DB_READ_PASS"),
        database="UFDATA_001",
        as_dict=True,
        charset="utf8"
    )

@server.tool()
async def query_inventory(material_code: str, warehouse: str = None) -> list[TextContent]:
    """
    查询ERP物料库存数量
    Args:
        material_code: 物料编码，必填
        warehouse: 仓库编码，可选，不填则查询所有仓库
    """
    # 参数合法性校验
    if not material_code or len(material_code) > 20:
        return [TextContent(type="text", text="参数非法：物料编码格式不正确")]
    
    conn = None
    try:
        conn = get_db_conn()
        with conn.cursor() as cursor:
            sql = """
                SELECT cInvCode as 物料编码, cInvName as 物料名称,
                       cWhCode as 仓库, iQuantity as 现存量, iAvailableQty as 可用量
                FROM CurrentStock
                WHERE cInvCode = %s
            """
            params = [material_code]
            if warehouse:
                sql += " AND cWhCode = %s"
                params.append(warehouse)
            
            cursor.execute(sql, params)
            rows = cursor.fetchall()
            
            if not rows:
                return [TextContent(type="text", text=f"未查询到物料 {material_code} 的库存记录")]
            
            # 格式化返回，屏蔽原始字段名
            result_lines = ["库存查询结果："]
            for row in rows:
                result_lines.append(
                    f"【{row['仓库']}】{row['物料名称']} 现存量:{row['现存量']} 可用量:{row['可用量']}"
                )
            return [TextContent(type="text", text="\n".join(result_lines))]
    except Exception as e:
        return [TextContent(type="text", text=f"查询失败：{str(e)}")]
    finally:
        if conn:
            conn.close()

async def main():
    async with stdio_server() as (r, w):
        await server.run(r, w, server.name)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

2.4 优缺点与踩坑指南

优势：

开发极快，单个场景1~2天就能上线
零侵入，完全不改动老系统应用代码
性能好，直接查库比走应用层快得多
成本极低，不需要原厂配合

劣势：

只能读不能写，无法触发业务流程
复杂业务逻辑还原难度大，比如订单状态可能是多表关联计算的结果
依赖对数据结构的理解，表结构错了查出来的数就不准

必踩的坑：

千万别用主库：我见过有人直接在生产主库跑大查询，把ERP卡了半小时，全公司停摆。必须用从库或只读实例，账号只给SELECT权限。
禁止万能查询：不要让Agent写自由SQL，必须封装成固定参数的工具，否则分分钟给你查出全表数据，拖垮数据库还泄露数据。
业务口径对齐：数据库里的「现存量」和业务上的「可用库存」往往不是一回事，一定要和业务人员确认口径，不能看着字段名瞎猜。
敏感数据脱敏：客户手机号、金额、成本价等字段，必须在封装层自动打码，不能直接返回给Agent。

三、方案二：UI自动化代理 —— 零侵入的终极兜底方案

如果老系统连数据库都不敢碰，或者只有客户端程序、没有数据库访问权限，UI自动化就是最后的兜底方案。原理很简单：让程序像人一样操作客户端界面，点击按钮、输入内容、读取表格数据，完全复刻人工操作流程。

3.1 核心架构与适用场景

适用场景：

完全没有API、数据库也无法开放的古董级C/S系统
必须走应用层流程的写操作（比如创建工单、提交审批、单据审核）
厂商完全不配合，没有任何技术支持的场景

不推荐场景： 高频调用、高并发、对速度要求高的场景。UI自动化天生慢且脆弱。

3.2 技术选型

B/S架构老系统：优先用Playwright，稳定、速度快、元素定位精准
Windows桌面客户端：WinAppDriver（微软官方）、PyAutoGUI、商业RPA工具（影刀、UiPath）
极端难搞的界面：结合CV+OCR识别界面文字，不依赖控件ID

3.3 核心实现思路

以Windows桌面端老CRM为例，封装一个「创建客户档案」的工具，核心流程是：打开系统 → 切换到客户管理 → 点击新增 → 填入信息 → 保存 → 读取客户编号。

# crm_ui_agent.py
import time
from pywinauto import Application
from mcp.server import Server
from mcp.types import TextContent

server = Server("crm-ui-server")

def open_crm_client():
    """启动并连接CRM客户端"""
    app = Application(backend="uia").start("C:\\CRM\\Client.exe")
    time.sleep(3)
    # 登录逻辑（生产环境用加密配置存密码）
    dlg = app.window(title="系统登录")
    dlg["用户名Edit"].type_keys("agent_user")
    dlg["密码Edit"].type_keys("******")
    dlg["登录Button"].click()
    time.sleep(5)
    return app

@server.tool()
async def create_customer(customer_name: str, contact: str, phone: str) -> list[TextContent]:
    """
    在CRM系统中创建新客户档案
    Args:
        customer_name: 客户名称
        contact: 联系人
        phone: 联系电话
    """
    try:
        app = open_crm_client()
        main_win = app.window(title_re=".*客户关系管理系统.*")
        
        # 导航到客户管理
        main_win.menu_select("基础档案->客户管理")
        time.sleep(2)
        
        # 点击新增按钮
        main_win["新增Button"].click()
        time.sleep(1)
        
        # 填写表单
        form = main_win.window(title="客户信息编辑")
        form["客户名称Edit"].type_keys(customer_name)
        form["联系人Edit"].type_keys(contact)
        form["联系电话Edit"].type_keys(phone)
        
        # 保存
        form["保存Button"].click()
        time.sleep(2)
        
        # 读取返回的客户编号
        result_text = form["提示Static"].window_text()
        
        return [TextContent(type="text", text=f"客户创建完成：{result_text}")]
    except Exception as e:
        return [TextContent(type="text", text=f"操作失败：{str(e)}")]

3.4 优缺点与踩坑指南

优势：

真正的零侵入，老系统一行代码都不用改
理论上什么系统都能接，只要人能操作就能自动化
读写都支持，能完整触发应用层的业务流程和校验

劣势：

速度慢，单次操作几秒到几十秒不等
稳定性依赖界面，客户端升级、分辨率变化、弹窗干扰都可能失效
并发能力差，基本只能串行操作
维护成本高，界面一变就要改脚本

必踩的坑：

不要用坐标定位：纯靠鼠标坐标点击，换个分辨率就全崩。尽量用控件ID、控件名定位，实在不行再用图像识别。
弹窗地狱：老系统各种提示框、报错框、更新提示，随时可能弹出来打断流程。必须把常见弹窗都做识别和处理逻辑。
环境固化：运行环境要固定死——分辨率、系统字体、客户端版本、显示缩放比例，任何一个变了都可能导致识别失败。
不要高频调用：UI自动化是替代人工操作，不是给Agent高频查询用的。查询类需求优先走方案一，UI自动化只用来做低频写操作。

四、方案三：服务化适配层 —— 长期演进的标准化架构

如果企业有多个Agent要复用老系统能力，需要统一管控权限、审计、限流，那就不能东一榔头西一棒子地写脚本，需要一套标准化的适配层架构。这也是从「能用」到「好用」的必经之路。

核心思路是：在老系统外面包一层适配+服务化中间层，向下通过数据库、API、UI自动化等多种方式对接遗留系统，向上通过MCP协议暴露统一的标准工具，所有Agent共用一套能力。

4.1 整体分层架构

4.2 各层核心职责

多源适配层：对接不同形态的遗留系统，抹平底层差异。有API的走API，没API的走数据库，写操作走UI自动化。上层完全感知不到底层差异。
服务封装层：按业务领域封装能力，做参数校验、业务逻辑编排、幂等控制、权限判断、审计日志。这一层是业务规则的守护者。
统一协议层：以MCP标准对外暴露工具，所有Agent按统一规范调用，自动获得服务发现、权限管控、限流熔断、调用审计能力。

4.3 设计要点

防腐层设计：适配层隔离老系统的脏乱差，上层永远只看干净的业务模型。老系统换了、改了，只动适配层，上层业务不用改。
读写分离：读操作走数据库适配器，保证性能；写操作走API或UI自动化，保证业务流程完整。
幂等设计：所有写操作必须有幂等键，防止Agent重试导致重复下单、重复创建客户。
统一审计：所有调用全量留痕，谁、什么时候、调用了什么工具、参数是什么、返回什么，全部可追溯。

4.4 优缺点与踩坑指南

优势：

标准化，一次封装全公司Agent复用
可管控，权限、审计、限流、脱敏统一处理
可演进，底层老系统替换、升级，上层无感知
架构清晰，维护成本随规模增长平缓

劣势：

前期投入大，需要整体设计和规划
开发周期长，不适合快速验证的小项目
需要专门团队维护，小公司投入不起

必踩的坑：

不要追求一步到位：别一开始就想把所有系统、所有功能都封装进去。先封装3~5个高频读工具，跑起来再慢慢加。
适配层要足够厚：不要把老系统的字段名、状态码直接透传上去。一定要做业务语义转换，否则上层还是和老系统耦合。
做好降级预案：老系统不稳定是常态，适配层要做超时、重试、熔断、降级，不能老系统一挂，所有Agent全崩。

五、三种方案横向对比与选型决策

5.1 核心指标对比

对比维度	方案一：数据层直连	方案二：UI自动化	方案三：服务化适配层
系统侵入性	极低（只读账号）	零侵入	低（外部包裹）
开发周期	1~2周/场景	2~3周/场景	1~3个月
改造成本	极低	中等	较高
读写支持	仅读	读写都支持	读写都支持
执行速度	极快（毫秒级）	慢（秒~十秒级）	较快（百毫秒级）
并发能力	高	极低（基本串行）	中高
稳定性	高（依赖数据库）	低（依赖界面）	高（有容错设计）
维护成本	低	高	中
适合阶段	验证期、快速落地	兜底补充、低频写操作	成熟期、规模化复用