1. 从 Demo 到生产:Agent 落地最容易被低估的那道坎
做过 Agent 项目的人大概都有过这种体验:本地跑一个 ReAct 循环,接上大模型,调几个工具函数,演示效果惊艳,老板看完拍板“上生产”。然后真正开始对接企业系统的时候,才发现事情完全不是那么回事。模型选型、Prompt 调优、推理链路这些在 Demo 阶段最显眼的问题,到了生产环境反而变成了相对可控的变量。真正让人头疼的,是那些看起来“没什么技术含量”的活儿——把 OA、ERP 这些企业里跑了好多年的老系统接进来。
这个项目标题里提到的 FDE MCP Blade,本质上就是在解决这个问题。FDE 是 Forward Deployed Engineer 的缩写,这个角色在 AI Agent 落地场景里越来越常见,核心职责就是扎到客户的真实业务环境里,把 Agent 和客户已有的系统打通。MCP 是 Model Context Protocol,一个让模型能够标准化调用外部工具和数据的协议层。Blade 则是这套方案里负责“切割”和“桥接”的那把刀——把 OA、ERP 里那些盘根错节的接口、表单、审批流,切成 Agent 能理解的原子能力。
说白了,这套东西要回答一个问题:当你的 Agent 需要帮员工查一个报销单状态、发起一个采购申请、或者从 ERP 里拉一份库存报表的时候,它到底该怎么跟这些系统说话。这个问题听起来简单,做起来能让人掉一层皮。因为 OA 和 ERP 这类系统,往往有着自己的认证体系、自己的数据模型、自己的接口风格,有些甚至压根没有对外接口,只能靠页面自动化去操作。
这篇文章适合两类人看。一类是正在做 Agent 项目、马上要对接企业系统的开发者,另一类是 FDE 工程师或者准备往这个方向转的人。我会从整体设计思路讲到具体实操细节,把 OA、ERP 对接过程中那些坑一个个拆开来说。内容会比较长,但都是实打实的东西,不是那种看完就忘的概述。
2. 整体设计思路:为什么不能直接让模型去调 OA 接口
2.1 企业系统对接的三个现实约束
在动手写代码之前,得先把企业系统对接的现实约束想清楚。我见过不少团队一上来就想让模型直接调 OA 的 REST API,结果卡在认证环节就动不了了。这里面的约束主要有三个层面。
第一个约束是认证体系的碎片化。OA 系统常见的认证方式包括 Session-based 认证、CAS 单点登录、OAuth2、以及各种自定义的 Token 机制。ERP 系统更复杂,有些用数据库直连做权限校验,有些用中间件做统一认证,还有些老系统干脆就是 IP 白名单加固定账号。你不可能让模型去理解这些认证细节,它也不应该理解。认证这件事必须在 Agent 和业务系统之间有一个专门的层来处理。
第二个约束是数据模型的语义鸿沟。OA 里的“流程实例”和 ERP 里的“订单”在数据库层面可能是完全不同的结构,但在业务语义上它们可能有关联。模型需要的是语义化的工具描述,而不是数据库字段。比如模型应该看到的是“查询某员工的待办审批”,而不是“SELECT * FROM workflow_task WHERE assignee_id = ?”。这个语义转换层是 MCP Server 要做的核心工作之一。
第三个约束是操作的安全边界。让模型直接操作生产系统的数据库或者调用写接口,风险极高。一个幻觉可能导致错误的审批、错误的数据修改。所以 MCP 层必须做权限收敛和操作确认。读操作可以相对放开,写操作必须有明确的确认机制和回滚方案。
2.2 MCP 协议在中间层扮演的角色
MCP 协议的核心价值在于它定义了一套标准化的工具描述和调用格式。模型不需要知道背后是 OA 还是 ERP,它只需要知道“有一个叫 query_leave_balance 的工具,输入是员工工号,输出是剩余年假天数”。这个抽象层让 Agent 的代码和具体业务系统解耦。
从架构上看,MCP Server 位于 Agent 和业务系统之间。Agent 通过 MCP Client 发送工具调用请求,MCP Server 接收到请求后,负责认证、参数校验、协议转换、实际调用业务系统、再把结果格式化返回。这个过程中,MCP Server 可以做一些很关键的事情:比如缓存频繁查询的数据、对敏感字段做脱敏、对写操作做二次确认、记录完整的审计日志。
Blade 在这个架构里的定位,我理解是 MCP Server 的一个实现框架或者工具集。它可能提供了一些预置的适配器,用来对接常见的 OA 和 ERP 系统,比如泛微、通达、金蝶、用友这些。也可能提供了一套 DSL 或者配置方式,让 FDE 工程师能够快速定义新的工具。从热词里出现的“泛微 OA 建模引擎”“通达 OA CAS”这些来看,Blade 应该是针对国内企业系统生态做了不少适配工作。
2.3 为什么选择 MCP 而不是自定义 Function Calling
有人可能会问,直接用 OpenAI 的 Function Calling 或者自己定义一套 HTTP 接口不就行了吗,为什么要用 MCP。这个问题我在实际项目里也纠结过,后来发现 MCP 有几个实际的好处。
首先是工具描述的标准性。MCP 定义了 tools/list 和 tools/call 的标准格式,不同模型、不同 Agent 框架都能兼容。你不需要为每个模型单独写一套工具描述。其次是传输层的灵活性。MCP 支持 stdio 和 SSE 两种传输方式,本地工具和远程工具可以用同一套协议。再者是生态兼容性。现在越来越多的工具和平台开始支持 MCP,比如各种 IDE 插件、浏览器自动化工具,这意味着你可以复用已有的 MCP Server。
当然 MCP 也不是没有缺点。它的调试相对麻烦一些,尤其是 SSE 模式下出问题的时候,排查链路比较长。另外 MCP 的权限模型还比较粗糙,细粒度的权限控制需要自己在 Server 层实现。但总体来看,对于需要对接多个企业系统的 Agent 项目,MCP 带来的标准化收益是值得的。
3. 核心细节解析:OA 与 ERP 对接的实操要点
3.1 认证环节的三种典型场景与处理方式
认证是对接的第一道坎,也是最多人踩坑的地方。我按实际遇到的情况分成三类来说。
第一类是标准 CAS 单点登录。泛微、通达这些 OA 系统都支持 CAS。处理方式是在 MCP Server 里维护一个 CAS Client,用服务账号登录后拿到 Ticket,再用 Ticket 换取 Session。这里的关键点是 Session 的保活和复用。OA 的 Session 通常有超时时间,比如 30 分钟不操作就失效。MCP Server 需要有一个后台线程定期刷新 Session,或者在每次调用前检查 Session 有效性。我一般会做一个 Session 池,维护几个有效 Session 轮询使用,避免并发调用时互相踢下线。
第二类是 Token 认证。有些 ERP 系统提供 REST API,用 Bearer Token 或者 API Key 认证。这种相对简单,但要注意 Token 的存储安全。不要把 Token 硬编码在代码里,也不要以明文放在配置文件里。可以用环境变量加启动时解密的方式,或者接入密钥管理服务。另外要处理 Token 过期和刷新的逻辑,尤其是 Refresh Token 的并发刷新问题,多个请求同时发现 Token 过期时,应该只有一个去刷新,其他的等待刷新结果。
第三类是页面自动化。这是最麻烦的情况,有些老系统没有 API,只能通过模拟浏览器操作来对接。这时候就需要用到 Playwright 或者类似的工具。MCP 生态里已经有 Playwright MCP 和 Browser Use MCP 这样的方案。页面自动化的核心难点在于登录态维持和页面元素定位的稳定性。我的经验是尽量用语义化的选择器,比如按文本内容定位按钮,而不是依赖容易变化的 CSS 类名。另外要处理好验证码的情况,如果系统有验证码,可能需要人工介入或者接入打码服务,但打码服务有合规风险,需要谨慎评估。
3.2 数据模型映射:从业务语义到工具描述
MCP 工具的描述质量直接决定了模型能不能正确调用。我见过很多项目把工具描述写得很技术化,比如“调用 workflow API 的 getTaskList 方法”,模型根本不知道这是什么意思。好的工具描述应该是业务语义化的。
举个例子,OA 里的待办查询。技术层面的接口可能是/workflow/task/list?assignee=xxx&status=pending。但给模型的工具描述应该是这样的:工具名query_pending_approvals,描述是“查询指定员工的待办审批事项,返回审批标题、发起人、发起时间、当前节点”。参数是员工工号或者姓名。这样模型在用户说“帮我看看有什么要审批的”的时候,就能正确选择这个工具。
数据映射的另一个重点是字段的语义化。OA 数据库里的字段名往往是拼音缩写或者无意义的编码,比如lcid、spr、hj。这些字段在返回给模型之前,必须映射成可读的字段名。我一般会在 MCP Server 里维护一个映射表,把数据库字段映射成业务字段。这个映射表可以配置化,方便不同客户环境适配。
还有一个容易忽略的点是数据格式的统一。OA 返回的日期格式可能是20240101,ERP 可能是2024-01-01 00:00:00,还有可能是时间戳。MCP Server 应该统一转成 ISO 8601 格式再返回给模型,减少模型解析的负担。
3.3 写操作的安全设计:确认、幂等与回滚
读操作相对安全,写操作必须谨慎。我的原则是:任何写操作都要有明确的确认机制,任何写操作都要考虑幂等性,任何写操作都要有回滚或者补偿方案。
确认机制的设计有两种思路。一种是在 MCP 工具层面做确认,工具调用时返回一个“待确认”状态,需要 Agent 再次调用确认工具才真正执行。另一种是在 Agent 层面做确认,Agent 在调用写工具前先向用户确认。我倾向于两者结合:MCP 层做技术性的二次确认,防止模型误调用;Agent 层做业务性的用户确认,确保操作符合用户意图。
幂等性的处理方式是在 MCP Server 里为每个写操作生成一个唯一的事务 ID,业务系统如果支持幂等键就传入,不支持的话就在 Server 层做去重。比如发起审批这个操作,同一个用户、同一个审批类型、同样的内容,在短时间内重复提交应该被识别为重复操作。
回滚方案要看具体业务系统。有些系统支持撤销操作,比如 OA 的审批可以撤回。有些系统不支持,那就需要做补偿操作,比如发起一个反向的流程来抵消。最坏的情况是只能人工介入,这时候 MCP Server 应该记录完整的操作日志,包括操作时间、操作人、操作参数、返回结果,方便后续追溯。
4. 实操过程:从零搭建一个 OA 对接的 MCP Server
4.1 环境准备与依赖安装
假设我们要对接一个泛微 OA 系统,实现待办查询和审批发起两个功能。先来准备环境。
基础环境需要 Python 3.10 以上,Node.js 18 以上(如果要用 Playwright 做页面自动化)。Python 这边主要用到 mcp 这个包,以及 requests、lxml 这些做 HTTP 请求和 HTML 解析的库。如果要做 CAS 认证,还需要 cas-client 或者自己实现 CAS 协议。
pip install mcp requests lxml python-dateutil如果是用 TypeScript 开发,对应的包是 @modelcontextprotocol/sdk。
npm install @modelcontextprotocol/sdk axios cheerio我个人的选择是 Python 做 MCP Server,因为数据处理和文本解析的库更丰富。但如果团队主要是前端背景,TypeScript 也是很好的选择,类型系统在定义工具 schema 的时候很有帮助。
4.2 定义工具 Schema 与实现查询逻辑
先定义待办查询工具。MCP 的工具定义包括 name、description、inputSchema 三个核心部分。
from mcp.server import Server from mcp.types import Tool, TextContent import httpx from datetime import datetime app = Server("oa-mcp-server") @app.list_tools() async def list_tools(): return [ Tool( name="query_pending_approvals", description="查询指定员工的待办审批事项。返回审批标题、发起人、发起时间、当前节点和流程实例ID。", inputSchema={ "type": "object", "properties": { "employee_id": { "type": "string", "description": "员工工号,例如 E10086" }, "limit": { "type": "integer", "description": "返回的最大条数,默认10", "default": 10 } }, "required": ["employee_id"] } ) ]工具描述里我特意写了返回字段和示例工号格式,这些细节能显著提升模型调用的准确率。很多开发者只写一句“查询待办”,模型经常搞不清楚参数该传什么。
查询逻辑的实现要考虑几个点。首先是 Session 的获取和复用,其次是分页处理,再者是错误处理。
class OASessionManager: def __init__(self, base_url, username, password): self.base_url = base_url self.username = username self.password = password self.session = None self.last_active = None async def get_session(self): if self.session is None or self._is_expired(): await self._login() return self.session def _is_expired(self): if self.last_active is None: return True elapsed = (datetime.now() - self.last_active).seconds return elapsed > 1500 # 25分钟,留5分钟余量 async def _login(self): async with httpx.AsyncClient() as client: resp = await client.post( f"{self.base_url}/api/login", json={"username": self.username, "password": self.password} ) resp.raise_for_status() self.session = resp.json()["sessionId"] self.last_active = datetime.now()这里我把 Session 超时设成 25 分钟,比 OA 默认的 30 分钟短一点,避免边界情况。实际项目中这个值要根据 OA 的配置调整。
查询待办的实现:
@app.call_tool() async def call_tool(name: str, arguments: dict): if name == "query_pending_approvals": employee_id = arguments["employee_id"] limit = arguments.get("limit", 10) session = await session_manager.get_session() async with httpx.AsyncClient() as client: resp = await client.get( f"{base_url}/api/workflow/pending", params={"assignee": employee_id, "pageSize": limit}, headers={"Cookie": f"JSESSIONID={session}"} ) data = resp.json() results = [] for item in data.get("rows", []): results.append({ "title": item.get("requestName"), "creator": item.get("creatorName"), "create_time": format_date(item.get("createTime")), "current_node": item.get("currentNodeName"), "request_id": item.get("requestId") }) return [TextContent( type="text", text=json.dumps(results, ensure_ascii=False, indent=2) )]返回结果用 JSON 格式,字段名用英文,值保持中文。这样模型解析起来最方便。
4.3 审批发起工具的实现与确认机制
审批发起是写操作,需要更谨慎的设计。我的做法是分两步:先调用 prepare_approval 工具生成一个待确认的审批草稿,返回草稿 ID 和摘要信息;用户确认后再调用 submit_approval 工具真正提交。
@app.list_tools() async def list_tools(): return [ # ... 前面的查询工具 Tool( name="prepare_approval", description="准备一个审批申请草稿。返回草稿ID和审批内容摘要,需要用户确认后再调用submit_approval提交。", inputSchema={ "type": "object", "properties": { "employee_id": {"type": "string", "description": "申请人工号"}, "approval_type": {"type": "string", "description": "审批类型,如leave(请假)、expense(报销)、purchase(采购)"}, "form_data": {"type": "object", "description": "审批表单数据,不同审批类型字段不同"} }, "required": ["employee_id", "approval_type", "form_data"] } ), Tool( name="submit_approval", description="提交之前准备好的审批草稿。需要传入prepare_approval返回的草稿ID。", inputSchema={ "type": "object", "properties": { "draft_id": {"type": "string", "description": "prepare_approval返回的草稿ID"} }, "required": ["draft_id"] } ) ]草稿存在内存或者 Redis 里,设置 10 分钟过期。submit 的时候检查草稿是否存在、是否过期、是否已经被提交过。这样能有效防止重复提交和过期提交。
审批类型的表单字段映射是个细致活。请假审批需要开始时间、结束时间、请假类型、事由;报销审批需要费用明细、金额、发票信息。这些字段在 OA 的接口里可能有不同的名称和格式要求。我一般会为每种审批类型写一个适配器,把统一的 form_data 转换成 OA 需要的格式。
4.4 用 Playwright MCP 处理无 API 的老系统
有些老系统确实没有 API,只能走页面自动化。这时候可以用 Playwright MCP 作为补充。Playwright MCP 提供了 browser_navigate、browser_click、browser_type、browser_snapshot 这些工具,模型可以通过这些原子操作来完成页面交互。
但直接让模型操作页面效率很低,而且容易出错。更好的做法是在 Playwright MCP 之上再封装一层业务工具。比如“查询库存”这个业务操作,底层可能是导航到库存页面、输入商品编码、点击查询、解析结果表格这一系列操作。这系列操作封装成一个 MCP 工具,模型只需要调用一次。
async def query_inventory(product_code: str): # 通过 Playwright MCP 的接口执行页面操作 await playwright_mcp.call("browser_navigate", {"url": f"{erp_url}/inventory"}) await playwright_mcp.call("browser_type", { "selector": "#productCode", "text": product_code }) await playwright_mcp.call("browser_click", {"selector": "#searchBtn"}) await asyncio.sleep(2) # 等待结果加载 snapshot = await playwright_mcp.call("browser_snapshot", {}) # 解析 snapshot 中的表格数据 return parse_inventory_table(snapshot)这里的关键是页面元素选择器的稳定性。我踩过的坑是用了自动生成的 CSS 类名,系统一升级就失效。后来改成用文本内容或者稳定的 ID 来定位。另外等待时间不要写死,最好用 wait_for_selector 这样的条件等待。
5. 常见问题与排查技巧实录
5.1 认证类问题速查
认证问题是对接过程中最高频的故障来源。我整理了一个速查表,覆盖大部分场景。
| 现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| 401 Unauthorized | Session 过期或无效 | 检查 Session 最后刷新时间 | 重新登录获取 Session |
| 403 Forbidden | 账号权限不足 | 用该账号直接登录系统验证 | 申请对应权限或换服务账号 |
| CAS 重定向循环 | Ticket 验证失败 | 检查 CAS Server 地址和 Service 地址配置 | 确保 Service 地址与注册一致 |
| 登录成功但接口返回空 | Session 未正确传递 | 抓包检查 Cookie 或 Header | 确认 Session 传递方式 |
| 并发调用时部分失败 | Session 被踢下线 | 检查是否多线程共用 Session | 使用 Session 池 |
CAS 重定向循环这个问题特别常见。原因是 Service 地址在 CAS Server 注册的和实际请求的不一致。比如注册的是http://oa.example.com,但实际请求走的是http://10.0.0.1,CAS Server 会认为 Service 不合法。解决方法是确保两边完全一致,包括协议、域名、端口。
5.2 数据解析类问题
OA 和 ERP 返回的数据格式往往不规范。我遇到过返回 JSON 里混着 HTML 标签的,也遇到过日期字段返回空字符串的,还有返回的编码是 GBK 而不是 UTF-8 的。
编码问题最隐蔽。有些老系统默认用 GBK 编码,requests 库如果没指定编码,解析出来就是乱码。解决方法是在请求时显式指定编码,或者在响应头里找 charset。
resp = requests.get(url) resp.encoding = 'gbk' # 显式指定 data = resp.json()日期格式的处理我建议统一用 dateutil 来解析,它能自动识别多种格式。
from dateutil import parser def format_date(date_str): if not date_str: return None try: dt = parser.parse(str(date_str)) return dt.isoformat() except: return str(date_str)5.3 模型调用工具时的典型错误
模型调用工具出错的情况主要有几种。一种是参数格式不对,比如该传字符串的传了数字,该传数组的传了单个值。这种要在工具 schema 里把类型定义清楚,description 里给示例。另一种是工具选择错误,用户问待办,模型调了查询库存。这种要在工具描述里把适用场景写清楚,必要时在系统 Prompt 里加引导。
还有一种比较隐蔽的问题是模型编造参数值。比如用户说“查一下我的待办”,模型不知道工号,就编了一个。这种情况要在工具描述里明确说明“如果不知道工号,先向用户询问”,或者在 Agent 层做参数校验,发现工号格式不对就返回错误提示。
我在实际项目里会在 MCP Server 层加一层参数校验,用 Pydantic 或者 JSON Schema 验证。校验失败时返回明确的错误信息,模型看到错误信息后通常会重新尝试或者向用户询问。
5.4 性能与并发问题
Agent 在生产环境面临的并发压力往往被低估。一个热门时段的并发调用可能达到几十甚至上百 QPS。MCP Server 如果每次调用都新建连接、重新登录,性能会非常差。
优化手段主要有几个。连接池是基础,HTTP 连接要复用。Session 池也很关键,维护一组已登录的 Session 轮询使用。对于读多写少的场景,可以加缓存,比如组织架构、审批类型这些变化不频繁的数据,缓存几分钟能显著降低后端压力。
还有一个容易忽略的点是超时设置。OA 和 ERP 的接口响应时间可能很长,尤其是复杂查询。MCP Server 的超时时间要设置合理,太短会导致大量超时失败,太长会拖垮整个 Agent 的响应。我一般设置连接超时 5 秒,读取超时 30 秒,对于特别慢的接口单独配置。
6. FDE 视角下的经验总结与踩坑记录
做 FDE 这几年,对接过的 OA 和 ERP 系统少说也有十几种。有些经验是通用的,有些是特定系统才有的坑。这里挑几个印象深刻的说说。
泛微 OA 的建模引擎是个双刃剑。它允许你自定义表单和流程,灵活性很高,但接口的规范性就差一些。不同客户环境里同一个业务对象的字段名可能完全不同。我的做法是在 MCP Server 里做一层配置化的字段映射,每个客户环境一份配置,代码逻辑不变。这样新客户上线的时候只需要改配置,不用改代码。
通达 OA 的 CAS 集成有个坑是登录时长设置。默认的 Session 超时时间比较短,而且有些版本不支持通过接口延长。解决方法是定期发一个轻量的心跳请求保持 Session 活跃。但心跳频率不能太高,否则会被当成异常流量。我一般设置 10 分钟一次心跳。
ERP 系统里金蝶和用友的接口风格差异很大。金蝶的接口相对规范,有完整的 API 文档和 SDK。用友的接口有些是 SOAP 的,有些是自定义的 HTTP 接口,文档也不够完整。对接用友的时候,我经常需要抓包分析实际请求格式。这里提醒一句,抓包分析要在测试环境做,并且要获得客户授权。
还有一个通用的经验是:永远不要相信业务系统返回的数据格式是稳定的。我遇到过一次 OA 升级后,日期字段从字符串变成了时间戳,导致解析全部失败。后来我在解析层加了兼容逻辑,同时监控解析失败率,一旦异常升高就告警。
关于 FDE 这个角色本身,我的体会是技术能力只是一部分,更重要的是沟通能力和业务理解能力。你需要跟客户的 IT 部门沟通接口权限,跟业务部门沟通流程细节,跟管理层沟通项目进度。很多时候对接不顺利不是因为技术难,而是因为权限没开通、流程没确认、数据没准备好。把这些非技术问题提前解决,技术对接会顺利很多。
最后分享一个实用技巧:在 MCP Server 里加一个 debug 模式,把每次工具调用的完整请求和响应都记录下来。生产环境默认关闭,排查问题时可以临时开启。这个日志在定位问题时非常有用,尤其是模型调用参数错误或者业务系统返回异常的时候。日志要注意脱敏,密码、Token 这些敏感信息不能记录。
这套 FDE MCP Blade 的思路,核心就是把企业系统对接这件事标准化、配置化、可复用化。模型能力在快速进步,但企业系统的复杂性不会自动消失。把这一层做扎实,Agent 才能真正在生产环境跑起来。