从手工搬砖到智能流水线：用Python和MCP协议重构Excel自动化工作流

1. 从手工搬砖到智能流水线我为什么要自己写一个 MCP做数据运营和报表自动化的朋友大概率都有过这种体验每天打开电脑第一件事就是收邮件、下附件、把十几个 Excel 表里的数据合并、清洗、透视再贴回另一个模板里发出去。这套动作我干了快三年熟练到闭着眼睛都能点鼠标但心里清楚这就是纯体力活。后来我陆续用 Python 写脚本把其中一部分自动化了效率确实上来了可新的问题又冒出来——脚本越写越多每个脚本只认自己那套固定格式业务方稍微改个表头、加一列脚本就报错维护成本反而更高。真正的转折点是我开始接触MCPModel Context Protocol模型上下文协议。简单说它是一套让 AI 模型能够安全、标准化地调用外部工具和数据的协议。你可以把它理解成给 AI 装了一个万能插座以前 AI 只能跟你聊天现在通过 MCP它能真正去读你的文件、跑你的代码、操作你的表格。这跟传统的写死逻辑的脚本完全是两个思路——脚本是我预判所有情况然后写死分支MCP 是我把能力暴露给 AI让 AI 根据当前情况自己决定怎么调。这篇内容我想完整复盘一下我开发第一个 MCP 服务的全过程怎么把一个原本靠 Python 脚本硬扛的 Excel 处理工作流重构成一个 AI 可以自主调用的 MCP 工具集。涉及的技术栈主要是Python MCP 协议 Excel 处理库适合有一定 Python 基础、被重复性表格工作折磨过、想了解 AI Agent 和 MCP 到底怎么落地的人。哪怕你只是刚学 Python 不久我也会把每一步的意图和踩坑点讲清楚照着做能跑通。先说清楚这个 MCP 到底解决什么问题。我原来的工作流是这样的业务方发来一堆格式各异的 Excel我要合并、去重、按规则计算、生成汇总表、再导出。用脚本做每次格式一变就得改代码。做成 MCP 之后我把读表合并筛选透视写回这些原子能力封装成工具AI 拿到用户的自然语言指令比如把这三张表按订单号合并去掉重复行算一下每个客户的总额自己决定调用哪几个工具、按什么顺序调。格式变了AI 自己看表头就能适配我不用再改一行代码。这就是 MCP 相对传统脚本的核心价值把写死的流程变成可组合的能力。2. 动手之前把 MCP 和 Excel 处理这两件事想明白2.1 MCP 到底是什么别被协议两个字吓到很多人一听协议就觉得是高深的东西其实 MCP 的本质非常朴素。它规定了三件事工具怎么描述、AI 怎么请求调用、服务怎么返回结果。你可以把它类比成餐厅点菜——菜单工具描述上写清楚每道菜是什么、需要什么配料参数顾客AI看着菜单点菜发起调用厨房MCP 服务做好端上来返回结果。中间不需要顾客进厨房也不需要厨师猜顾客想吃什么。MCP 服务通常以两种方式运行一种是本地进程通过标准输入输出通信另一种是网络服务通过特定的连接地址对外提供能力。对于 Excel 处理这种涉及本地文件的场景本地进程方式最直接文件不出本机安全可控。我第一个版本就用的本地方式跑起来之后 AI 客户端能直接发现我注册的工具。这里有个概念要澄清MCP 不是某个具体软件也不是某个 AI 产品专属的功能它是一个开放的协议标准。这意味着你写一次 MCP 服务理论上任何支持这个协议的 AI 客户端都能用。这一点很关键它避免了我为 A 平台写一套、为 B 平台再写一套的重复劳动。2.2 为什么选 Python 而不是别的语言选 Python 做 MCP 服务理由很实在。第一Excel 处理的生态在 Python 里最成熟openpyxl、pandas、xlrd这些库覆盖了从读取、计算到写出的全流程遇到问题网上资料也最多。第二MCP 的官方 SDK 对 Python 支持完善注册工具、定义参数 schema 都是几行代码的事。第三调试方便我可以先用普通 Python 脚本把 Excel 逻辑跑通确认没问题了再包一层 MCP 的外壳风险可控。如果你之前只写过一点点 Python也不用慌。这个项目用到的语法基本就是函数定义、字典、列表操作加上几个库的 API 调用。真正需要花时间理解的是 MCP 的工具注册机制而不是 Python 本身。2.3 环境准备把地基打牢环境这块我踩过坑所以单独拎出来说。首先是 Python 版本建议 3.10 及以上因为 MCP SDK 用到了较新的类型注解特性。装 Python 的时候记得勾选Add to PATH不然后面命令行里敲python会提示找不到命令这个坑新手几乎必踩。依赖库我列一下清单都是实测稳定的组合库名用途安装命令mcpMCP 协议官方 SDKpip install mcpopenpyxl读写 xlsx 文件pip install openpyxlpandas数据合并、透视、清洗pip install pandaspydantic参数校验MCP 依赖它随 mcp 自动安装提示强烈建议用虚拟环境别把依赖装到全局。python -m venv venv建环境激活后再装库这样以后换项目不会互相污染。装完之后验证一下敲python -c import mcp, openpyxl, pandas; print(ok)能打印 ok 就说明地基没问题。这一步看着简单但很多人卡在依赖冲突上提前验证能省后面大量排查时间。3. 核心设计把 Excel 工作流拆成 AI 能调用的原子能力3.1 拆解原则一个工具只干一件事设计 MCP 工具最容易犯的错就是把整个流程塞进一个大工具里比如搞一个process_excel函数参数传一堆内部把合并、清洗、计算全干了。这样做的后果是 AI 根本没法灵活组合等于又回到了写死脚本的老路。我的拆解原则是一个工具只做一件语义清晰的事参数尽量少而明确。最后我拆出了这么几个核心工具read_excel_sheet读取指定 Excel 文件的某个工作表返回表头和前若干行数据让 AI 先看一眼数据长什么样。merge_excel_files把多个 Excel 文件按指定列合并成一张表。filter_rows按条件筛选行比如金额大于 1000。pivot_summary按某列分组做聚合比如按客户汇总金额。write_excel把处理好的数据写回一个新的 Excel 文件。这五个工具组合起来几乎能覆盖我日常 90% 的表格处理需求。关键在于AI 拿到合并三张表再按客户汇总这种指令时会自己规划成先 merge再 pivot最后 write的调用链我不需要预先写死这个顺序。3.2 参数设计让 AI 看得懂、填得对工具的参数设计直接决定 AI 能不能用对。我的经验是参数名要自解释类型要明确能不给选择就不给选择。举个例子filter_rows我最初设计成传一个复杂的条件表达式字符串结果 AI 经常拼错语法。后来改成传三个参数列名、操作符大于/小于/等于、比较值AI 填起来准确率高了一大截。再比如read_excel_sheet我加了一个max_rows参数默认返回 20 行。为什么要限制因为如果表有几万行全返回给 AI 会撑爆上下文既慢又贵。让 AI 先看 20 行了解结构需要细节时再针对性读取这是很实用的一个设计。每个工具我都写了详细的描述文档docstring说明它做什么、参数什么含义、返回什么。别小看这个描述AI 就是靠它来判断该不该调用这个工具的。描述写得含糊AI 就会乱调或者不调。3.3 数据流转中间结果放哪这里有个容易被忽略的问题工具之间怎么传递数据比如merge_excel_files合并完的结果怎么交给pivot_summary有两种方案一是每个工具都读写文件中间结果落盘二是把数据放在内存里用一个会话级的缓存。我选的是混合方案小数据量走内存缓存大数据量落临时文件。原因是纯内存方案在数据量大时会吃爆内存纯落盘方案又会产生一堆临时文件难清理。具体实现上我用一个模块级的字典做缓存key 是数据集的标识value 是 DataFrame。工具返回结果时带上这个标识下一个工具用标识去取数据。这个设计让整个调用链既高效又可控。4. 实操落地从零写出第一个能跑的 MCP 服务4.1 搭骨架最小可运行结构先写一个最小的 MCP 服务骨架确认能跑起来再往里加工具。这是我一贯的做法——先让Hello World跑通再堆功能避免一上来就写几百行然后不知道哪出错。from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app Server(excel-mcp) app.list_tools() async def list_tools(): return [ Tool( nameread_excel_sheet, description读取Excel文件的指定工作表返回表头和前若干行数据, inputSchema{ type: object, properties: { file_path: {type: string, description: Excel文件路径}, sheet_name: {type: string, description: 工作表名称}, max_rows: {type: integer, description: 最多返回行数, default: 20} }, required: [file_path, sheet_name] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name read_excel_sheet: result read_excel_sheet(**arguments) return [TextContent(typetext, textresult)] async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这段代码是整个服务的骨架。list_tools负责告诉 AI我有哪些工具call_tool负责实际执行。注意inputSchema用的是 JSON Schema 格式这是 MCP 规定的标准AI 靠它理解参数结构。4.2 实现读取工具让 AI 先看清数据读取工具是所有流程的起点实现要点是稳和快。稳是指各种异常都要处理——文件不存在、工作表名写错、文件被占用都得给出清晰的错误信息而不是抛一堆堆栈让 AI 懵圈。快是指别一次性读全表用openpyxl的只读模式或者pandas的nrows参数限制读取量。import pandas as pd def read_excel_sheet(file_path: str, sheet_name: str, max_rows: int 20) - str: try: df pd.read_excel(file_path, sheet_namesheet_name, nrowsmax_rows) except FileNotFoundError: return f错误找不到文件 {file_path} except ValueError as e: return f错误工作表 {sheet_name} 不存在请检查名称 header | .join(str(c) for c in df.columns) rows \n.join( | .join(str(v) for v in row) for row in df.values) return f表头{header}\n数据前{len(df)}行\n{rows}这里返回的是纯文本格式用竖线分隔。为什么不用 JSON因为 AI 读文本表格更直观而且省 token。实测下来同样的数据用文本格式返回AI 理解准确率更高响应也更快。4.3 实现合并与汇总把 pandas 能力包进来合并和汇总是核心价值所在。merge_excel_files接收一个文件路径列表和一个合并键内部用pandas.concat或merge完成。这里有个关键决策用concat纵向堆叠还是merge横向关联我把它做成参数让 AI 选因为这两种语义完全不同AI 根据用户指令能判断出来。def merge_excel_files(file_paths: list, key_column: str None, how: str concat) - str: frames [pd.read_excel(p) for p in file_paths] if how concat: merged pd.concat(frames, ignore_indexTrue) else: merged frames[0] for f in frames[1:]: merged merged.merge(f, onkey_column, howouter) cache_id fds_{len(_cache)} _cache[cache_id] merged return f合并完成共 {len(merged)} 行数据集标识{cache_id}注意最后返回的是数据集标识而不是数据本身。这是前面说的缓存机制——大数据不往 AI 上下文里塞只给个取货凭证后续工具凭这个标识操作。pivot_summary同理接收数据集标识、分组列、聚合列、聚合方式用groupby完成结果同样存缓存返回标识。这样整条链路的数据都在服务端流转AI 只负责决策不搬运数据。4.4 实现写出工具把结果落盘最后一步是把处理好的数据写成 Excel。这里要注意的是文件路径的处理——不能让 AI 随便往系统目录写我加了一个输出目录的限制所有文件只能写到指定目录下避免误操作。import os OUTPUT_DIR ./output def write_excel(cache_id: str, output_name: str) - str: if cache_id not in _cache: return f错误数据集 {cache_id} 不存在 os.makedirs(OUTPUT_DIR, exist_okTrue) path os.path.join(OUTPUT_DIR, output_name) _cache[cache_id].to_excel(path, indexFalse) return f已写出到 {path}共 {len(_cache[cache_id])} 行写完这一步整个 MCP 服务的闭环就通了读 → 合并 → 筛选/汇总 → 写出。AI 拿到用户指令后会自动规划调用顺序我只需要在客户端里把服务配置好就行。4.5 配置与联调让 AI 真正用起来服务写完了得让 AI 客户端知道它的存在。不同客户端的配置方式略有差异但核心都是告诉它启动这个命令来运行 MCP 服务。配置里一般要填命令比如python、参数服务脚本路径、工作目录。配好之后重启客户端如果工具列表里能看到我注册的那几个工具就说明连接成功了。联调阶段我建议先用最简单的指令测试比如读一下 output 目录下的 test.xlsx确认单个工具能调通再测试组合指令。我第一