1. 从写死流程到智能编排MCP 协议下 AI Agent 与 RPA 的业务编排到底解决什么问题如果你做过企业自动化大概率遇到过这种场景一个订单处理流程要跨 5 个后台系统每个系统页面结构不同还三天两头改版。传统 RPA 靠录制和固定 XPath 定位元素页面一改整条流程直接崩。更头疼的是分支逻辑——库存不足要切供应商、客户加急要插队处理这些判断用 if-else 硬编码代码量爆炸维护成本高得离谱。MCP 协议驱动的 AI Agent 与 RPA 多步骤业务编排方案核心思路是把决策和执行拆开AI Agent 负责理解业务语义、分解任务、判断分支RPA 负责动手操作页面、填表、点击、取数中间用 MCP 协议做标准化通信。适合谁适合正在做企业级自动化、被传统 RPA 维护成本折磨、想接入大模型能力但又不想重写整套流程的开发者。我试过纯 RPA 方案也试过直接调 LLM API 让模型分析页面再执行前者死在改版上后者死在上下文丢失和调试困难上。MCP 协议的价值在于它把工具调用标准化了——一个 MCP Server 封装好 RPA 操作Claude、GPT、DeepSeek 都能调换模型不用重写集成代码。下面从接入配置到端到端验证一步步跑通。2. TaoToken 统一 Key 接入MCP 服务端与 Agent 的模型通道前置准备在搭建 Agent RPA 编排之前先解决模型调用通道的问题。MCP 协议本身只管工具通信但 Agent 的推理决策需要调 LLM。如果你用多个模型比如中文场景用 DeepSeek英文场景用 GPT每个模型单独配 Key、单独管额度维护起来很烦。TaoToken 提供统一 Key/API 通道一个 Key 走多个模型对 MCP 编排场景很实用。2.1 获取统一 Key 与模型 ID先到 TaoToken 控制台创建 API Key。地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 登录后点创建密钥复制保存。注意 Key 只在创建时显示一次丢了要重新生成。模型 ID 在文档页可以查到常用的是deepseek-v4、gpt-4、kimi这类。MCP Agent 编排场景建议选工具调用能力强的模型中文业务优先 DeepSeek 系列。文档地址https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。2.2 为什么 MCP 编排场景需要统一通道MCP 的架构是 Host宿主应用比如你的 Agent 系统通过 Client 连接多个 Server工具提供方。Agent 在 ReAct 循环里每轮都要调 LLM 做推理如果模型通道不稳定或者切换成本高整个编排链路就卡住了。统一 Key 的好处是Agent 代码里只配一个 Base URL 和一个 Key换模型只改 Model ID 参数不用动集成逻辑。另外MCP Server 如果用 stdio 本地启动RPA 操作完全在本地跑数据不出本机。但 LLM 推理必须走远程 API这时候通道的稳定性和合规性就很重要。TaoToken 的 API 地址是 https://taotoken.net/api 配置时 Base URL 填这个不要加多余路径。2.3 环境变量配置在项目根目录建.env文件把 Key 和 Base URL 写进去避免硬编码# .env TAOTOKEN_API_KEYsk-你的密钥 TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELdeepseek-v4然后在 Python 代码里用os.getenv读取。这样本地开发和部署到客户环境时只改.env就行代码不用动。3. 可复制配置MCP Server 封装 RPA 操作 Agent 编排层完整代码这一节是核心直接给可复制的配置和代码。分三块MCP Server 封装 RPA 工具、Agent 编排层接入 TaoToken 通道、以及 Claude Code / Cline 这类客户端的 MCP 配置片段。3.1 MCP Server把 RPA 操作封装成标准工具先装依赖pip install mcp playwright python-dotenv然后写rpa_mcp_server.py。这个 Server 暴露 5 个核心工具导航、提取数据、填表、点击、截图。每个工具用app.tool()装饰FastMCP 自动生成 JSON SchemaLLM 据此理解输入输出。# rpa_mcp_server.py import os import asyncio from mcp.server.fastmcp import FastMCP from playwright.async_api import async_playwright from dotenv import load_dotenv load_dotenv() app FastMCP(rpa-web-automation) # 全局浏览器实例避免每次调用都重启 _browser None _page None async def get_page(): global _browser, _page if _page is None: p await async_playwright().start() _browser await p.chromium.launch(headlessFalse) _page await _browser.new_page() return _page app.tool() async def navigate_to_url(url: str) - str: 打开指定网页返回页面标题 page await get_page() await page.goto(url, wait_untildomcontentloaded) title await page.title() return f已打开 {url}页面标题{title} app.tool() async def extract_data(selector: str, attribute: str textContent) - str: 根据 CSS 选择器提取网页元素数据attribute 可选 textContent/innerHTML/href page await get_page() elements await page.query_selector_all(selector) results [] for el in elements[:20]: # 限制数量避免返回过大 if attribute href: val await el.get_attribute(href) else: val await el.evaluate(fel el.{attribute}) results.append(val) return f提取到 {len(results)} 条数据{results} app.tool() async def fill_form(selector: str, value: str) - str: 在指定输入框填写内容 page await get_page() await page.fill(selector, value) return f已在 {selector} 填入 {value} app.tool() async def click_element(selector: str) - str: 点击指定元素 page await get_page() await page.click(selector) return f已点击 {selector} app.tool() async def take_screenshot() - str: 截取当前页面截图保存到本地返回文件路径 page await get_page() path screenshot.png await page.screenshot(pathpath, full_pageTrue) return f截图已保存到 {path} if __name__ __main__: app.run(transportstdio)这个 Server 用 stdio 传输Agent 通过子进程启动它。生产环境可以把 Playwright 换成你现有的 RPA 工具比如蓝印 RPA 的 API只要保持工具签名一致Agent 侧不用改。3.2 Agent 编排层ReAct MCP Client TaoToken 通道装依赖pip install langchain langchain-openai mcp写agent_orchestrator.py。关键点LLM 用 TaoToken 的 Base URL 和 KeyMCP Client 连接上面的 ServerReAct Agent 自动分解任务。# agent_orchestrator.py import os import asyncio from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain import hub from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client load_dotenv() class MCPRPAAgent: def __init__(self): # 关键用 TaoToken 统一通道一个 Key 走多模型 self.llm ChatOpenAI( modelos.getenv(TAOTOKEN_MODEL, deepseek-v4), api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), temperature0 ) self.mcp_tools [] self.session None async def connect_mcp_server(self): 连接 RPA 的 MCP Server动态发现工具 server_params StdioServerParameters( commandpython, args[rpa_mcp_server.py] ) self._stdio_ctx stdio_client(server_params) read, write await self._stdio_ctx.__aenter__() self._session_ctx ClientSession(read, write) self.session await self._session_ctx.__aenter__() await self.session.initialize() tools await self.session.list_tools() self.mcp_tools [ { name: t.name, description: t.description, input_schema: t.inputSchema } for t in tools.tools ] return self.mcp_tools def _build_langchain_tools(self): 把 MCP 工具转成 LangChain Tool实际调用走 MCP session lc_tools [] for t in self.mcp_tools: def make_func(tool_name): def func(input_str: str): # LangChain ReAct 传进来的是字符串这里简化处理 # 生产环境建议用 structured tool 或解析 JSON import json try: args json.loads(input_str) if input_str.strip().startswith({) else {} except Exception: args {} result asyncio.run_coroutine_threadsafe( self.session.call_tool(tool_name, args), asyncio.get_event_loop() ).result() return str(result.content) return func lc_tools.append(Tool( namet[name], descriptiont[description], funcmake_func(t[name]) )) return lc_tools def create_agent(self): prompt hub.pull(hwchase17/react) lc_tools self._build_langchain_tools() agent create_react_agent(self.llm, lc_tools, prompt) return AgentExecutor(agentagent, toolslc_tools, verboseTrue, max_iterations15) async def execute_task(self, task_description: str): await self.connect_mcp_server() executor self.create_agent() result await executor.ainvoke({input: task_description}) return result async def close(self): if self.session: await self._session_ctx.__aexit__(None, None, None) if hasattr(self, _stdio_ctx): await self._stdio_ctx.__aexit__(None, None, None) async def main(): agent MCPRPAAgent() task 帮我处理今天的电商订单 1. 打开 https://seller.example.com/orders 2. 提取所有待发货订单的订单号 3. 对每个订单检查库存系统是否有货 4. 有货的生成发货单没货的记录缺货 5. 给有货订单的客户发送发货通知 try: result await agent.execute_task(task) print(编排结果, result) finally: await agent.close() if __name__ __main__: asyncio.run(main())这段代码里ChatOpenAI的base_url指向 TaoToken 的 API 地址api_key用统一 Keymodel从环境变量读。换模型只改.env里的TAOTOKEN_MODELAgent 代码不动。3.3 Claude Code / Cline 的 MCP 配置片段如果你用 Claude Code 或 Cline 作为 MCP Host配置方式不一样。Claude Code 的 MCP 配置在~/.claude/claude_desktop_config.json或项目级.mcp.json格式如下{ mcpServers: { rpa-automation: { command: python, args: [/absolute/path/to/rpa_mcp_server.py], env: { TAOTOKEN_API_KEY: sk-你的密钥, TAOTOKEN_BASE_URL: https://taotoken.net/api, TAOTOKEN_MODEL: deepseek-v4 } } } }Cline 的 MCP 配置在 VS Code 设置里格式类似关键是command、args、env三件套。注意 Base URL 填https://taotoken.net/apiKey 填统一 KeyModel ID 填deepseek-v4或你选的模型。这三件套缺一不可否则 Agent 调不通。4. 验证请求一次端到端多步骤编排的完整跑通过程配置写完怎么确认整条链路通了分三步验证MCP Server 单独测、Agent 调 LLM 测、端到端编排测。4.1 验证 MCP Server 工具列表先单独启动 Server用 MCP Inspector 或写个简单脚本列出工具# test_mcp_tools.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters(commandpython, args[rpa_mcp_server.py]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() for t in tools.tools: print(f工具名{t.name}) print(f描述{t.description}) print(f参数{t.inputSchema}) print(---) asyncio.run(main())跑通后应该看到 5 个工具每个都有 name、description、inputSchema。如果报ModuleNotFoundError检查mcp包是否装了如果 Server 启动就退出检查rpa_mcp_server.py里app.run(transportstdio)是否在__main__里。4.2 验证 TaoToken 通道连通性单独测 LLM 调用确认 Key 和 Base URL 没问题# test_llm.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelos.getenv(TAOTOKEN_MODEL), api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) resp llm.invoke(用一句话说明 MCP 协议的作用) print(resp.content)如果返回 401说明 Key 错了或没生效如果返回Connection error检查 Base URL 是不是https://taotoken.net/api不要多加/v1或斜杠。如果返回model not found去文档页确认 Model ID 拼写。4.3 端到端编排验证跑agent_orchestrator.py观察 ReAct 循环的 verbose 输出。正常流程是第一轮LLM 推理需要先打开订单页面调用navigate_to_url观察返回页面标题。第二轮LLM 推理需要提取订单号调用extract_data观察返回数据条数。第三轮LLM 根据提取结果判断哪些有货继续调用后续工具。每轮都有 Thought、Action、Observation 三段输出。如果 Agent 卡在某一轮反复调同一个工具说明工具描述不够清晰LLM 没理解返回结果。解决办法是在工具 docstring 里写清楚返回格式比如extract_data返回的是列表字符串LLM 需要知道怎么解析。验证成功的标志Agent 在 max_iterations 内完成任务输出包含发货单已生成或类似结果且没有抛异常。你可以把max_iterations设成 15如果 15 轮还没完说明任务分解有问题需要优化 prompt 或工具粒度。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 报错对照实际跑的时候报错集中在几个地方。下面按报错信息对照排查。5.1 401 Unauthorized最常见。原因通常是 Key 没读到、Key 失效、或者 Base URL 配错导致请求发到了错误端点。排查步骤先print(os.getenv(TAOTOKEN_API_KEY))确认环境变量加载了再确认.env文件和代码在同一目录load_dotenv()在 import 之后调用最后确认 Base URL 是https://taotoken.net/api不是https://taotoken.net/api/v1。如果 Key 是从控制台复制的注意不要带空格。5.2 local proxy failed / connection refused这个报错通常出现在 MCP Client 启动 Server 子进程时。原因可能是command路径不对或者 Python 环境不对。排查把command改成绝对路径比如/usr/bin/python3或虚拟环境的venv/bin/python确认args里的脚本路径是绝对路径在终端手动跑python rpa_mcp_server.py看能不能启动。如果手动能跑但 MCP 里报错多半是环境变量没传进去在StdioServerParameters里加env参数。5.3 reading choices / model response parse error这个报错说明 LLM 返回的格式 Agent 解析不了。常见于模型不支持 Function Calling 或者返回了非标准 JSON。排查确认选的模型支持工具调用DeepSeek V4、GPT-4 都支持检查temperature是不是设太高设成 0如果用的是 LangChain 的 ReAct prompt确认 prompt 模板和模型匹配有些模型对 ReAct 格式支持不好可以换成create_tool_calling_agent。另外如果返回内容被截断检查max_tokens设置。5.4 OAuth / authentication failed如果 MCP Server 本身需要 OAuth比如连接企业内部系统报错会出现在 Server 初始化阶段。排查确认 OAuth token 没过期确认 Server 的env里传了必要的认证变量如果是 Claude Code 的 MCP 配置确认env字段写对了Claude Code 不会自动继承 shell 环境变量。对于 TaoToken 通道不需要 OAuth只要 API Key 就行如果报 OAuth 相关错误说明请求发到了错误的端点检查 Base URL。5.5 工具调用参数错误Agent 调工具时传的参数和 Schema 不匹配报Invalid arguments。排查在rpa_mcp_server.py里给每个工具的参数加类型注解和默认值在 Agent 侧如果用的是字符串传参确保 LLM 输出的是合法 JSON可以在工具函数里加参数校验返回清晰的错误信息帮助 LLM 下一轮修正。6. 语义一致 CTA从跑通到长期编排的下一步跑通上面的端到端流程后你已经有了一套可工作的 MCP Agent RPA 编排框架。接下来如果要长期用建议把模型通道固定下来避免每次换模型都改代码。TaoToken 的统一 Key 通道在这里的价值就体现出来了Agent 代码里只配一个 Base URL 和一个 Key换模型只改 Model ID。如果你还在调试接入阶段先去 API Keys 页面确认 Key 状态https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。接入文档里有各语言的配置示例https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。想先验证模型对话效果可以用模型对话页面直接测https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。如果这套编排要跑在生产环境、长期处理编码和 Agent 任务Coding Plan 更适合https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。最后提醒一点MCP Server 的工具描述要写清楚LLM 靠这个理解工具能力描述模糊是编排失败的头号原因。