☰
从手工搬砖到智能流水线:用Python和MCP协议重构Excel自动化工作流
2026/11/3 12:12:15 网站建设 项目流程

1. 从手工搬砖到智能流水线:我为什么要自己写一个 MCP

做数据运营和报表自动化的朋友大概率都有过这种体验:每天打开电脑第一件事,就是收邮件、下附件、把十几个 Excel 表里的数据合并、清洗、透视,再贴回另一个模板里发出去。这套动作我干了快三年,熟练到闭着眼睛都能点鼠标,但心里清楚这就是纯体力活。后来我陆续用 Python 写脚本把其中一部分自动化了,效率确实上来了,可新的问题又冒出来——脚本越写越多,每个脚本只认自己那套固定格式,业务方稍微改个表头、加一列,脚本就报错,维护成本反而更高。

真正的转折点是我开始接触MCP(Model 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 mcp
openpyxl读写 xlsx 文件pip install openpyxl
pandas数据合并、透视、清洗pip install pandas
pydantic参数校验(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( name="read_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(type="text", text=result)] 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_name=sheet_name, nrows=max_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_index=True) else: merged = frames[0] for f in frames[1:]: merged = merged.merge(f, on=key_column, how="outer") cache_id = f"ds_{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_ok=True) path = os.path.join(OUTPUT_DIR, output_name) _cache[cache_id].to_excel(path, index=False) return f"已写出到 {path},共 {len(_cache[cache_id])} 行"

写完这一步,整个 MCP 服务的闭环就通了:读 → 合并 → 筛选/汇总 → 写出。AI 拿到用户指令后,会自动规划调用顺序,我只需要在客户端里把服务配置好就行。

4.5 配置与联调:让 AI 真正用起来

服务写完了,得让 AI 客户端知道它的存在。不同客户端的配置方式略有差异,但核心都是告诉它"启动这个命令来运行 MCP 服务"。配置里一般要填命令(比如python)、参数(服务脚本路径)、工作目录。配好之后重启客户端,如果工具列表里能看到我注册的那几个工具,就说明连接成功了。

联调阶段我建议先用最简单的指令测试,比如"读一下 output 目录下的 test.xlsx",确认单个工具能调通,再测试组合指令。我第一

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询