从零手写MCP Server:让AI直接操作Excel实现数据清洗与合并 1. 为什么我要自己动手写一个 MCP1.1 从一次崩溃的 Excel 合并说起上个月帮朋友处理一批销售数据二十多个 Excel 文件每个文件结构还不完全一样有的表头在第一行有的在第三行有的列名叫“销售额”有的叫“成交金额”。我一开始想用 Python 脚本批量处理写了大概一百多行跑完发现有三张表的数据对不上——原因是某几个文件的日期列里混了文本格式pandas 读进来直接变成了字符串后续的求和全部失效。那天晚上我盯着屏幕想了很久这种“看一眼就知道怎么处理但写脚本要调半天”的活儿能不能让 AI 直接帮我干不是那种把数据贴进对话框、让它给我一段代码的方式而是让 AI 真正“看到”我的 Excel 文件理解我要做什么然后直接操作文件、返回结果。这就是我接触 MCP 的起点。MCP 全称 Model Context Protocol翻译过来叫“模型上下文协议”。你可以把它理解成一套标准接口让 AI 模型能够调用外部工具——读文件、查数据库、发请求、执行代码。它本身不是某个具体软件而是一种约定只要你的工具按照这个约定暴露能力AI 就能发现它、调用它。我决定动手写自己的第一个 MCP Server目标很明确让 AI 能够直接读写 Excel 文件完成数据清洗、合并、统计、格式转换这些日常操作。不需要我每次手动写脚本也不需要把数据传来传去。1.2 这个 MCP 能做什么适合谁看具体来说我做的这个 Excel MCP Server 提供以下几类能力读取与探查打开一个 Excel 文件返回所有 sheet 名称、每个 sheet 的行列数、表头内容、前几行数据预览。AI 拿到这些信息后就能判断数据长什么样。数据清洗处理合并单元格、去除空行空列、统一日期格式、修正数值列中的文本型数字。多表合并把多个结构相似的文件按行拼接或按列关联自动对齐列名。条件统计比如“统计同一列中包含某个关键词的行的数值总和”这在 Excel 里用 SUMIF 能做但跨文件、跨 sheet 就麻烦了。导出与转换把处理结果写回 Excel或者转成 CSV、Markdown 表格。这套东西适合什么人如果你经常跟 Excel 打交道又不想每次都手动写 Python 脚本或者你已经在用 AI 辅助工作但苦于 AI 没法直接碰你的文件那这个思路值得参考。哪怕你 Python 只是入门水平跟着走一遍也能跑起来。提示MCP 是一个开放协议不同 AI 客户端对它的支持程度不一样。我用的客户端支持本地 stdio 方式启动 MCP Server这也是最通用的方式。如果你用的客户端只支持远程连接思路是一样的只是传输层换一下。2. 整体设计为什么这样搭架子2.1 技术选型背后的取舍做这个 MCP Server我面临几个选择用什么语言写、用什么库处理 Excel、怎么组织工具接口。语言选 Python理由很直接pandas 和 openpyxl 是处理 Excel 最成熟的组合生态里现成的轮子多遇到问题搜一下就有答案。而且 MCP 官方提供了 Python SDK接入成本低。如果你更熟悉 Node.js 或 Go也有对应的 SDK逻辑是通的。Excel 处理库选 openpyxl pandas。openpyxl 负责底层读写能精确控制单元格、样式、合并区域pandas 负责数据分析和转换处理表格逻辑更顺手。两者配合既能读原始格式又能做复杂运算。有人会问为什么不只用 pandas——因为 pandas 读 Excel 时会丢失一些格式信息比如合并单元格的原始范围、单元格颜色而这些在清洗阶段可能有用。工具接口设计遵循“单一职责”。我没有做一个万能的process_excel工具而是拆成read_excel、clean_excel、merge_excel、query_excel、write_excel等独立工具。这样做的好处是 AI 调用时意图明确出错也容易定位。比如 AI 先调read_excel看数据结构再决定调clean_excel传什么参数最后调write_excel保存。每一步都可控。2.2 MCP Server 的基本结构一个 MCP Server 的核心就三件事声明工具列表、接收调用请求、返回执行结果。用 Python SDK 写大致长这样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, description读取Excel文件返回sheet列表和预览数据, inputSchema{ type: object, properties: { file_path: {type: string, description: Excel文件路径} }, required: [file_path] } ), # ... 其他工具 ] app.call_tool() async def call_tool(name: str, arguments: dict): if name read_excel: result handle_read_excel(arguments[file_path]) return [TextContent(typetext, textresult)] # ... 其他工具分发 async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options())这段代码的关键点在于inputSchema。它用 JSON Schema 描述每个工具需要什么参数、参数类型是什么、哪些必填。AI 客户端读到这个 schema就知道该怎么调用。比如read_excel需要一个字符串类型的file_pathAI 就会从对话里提取文件路径填进去。call_tool是总入口根据工具名分发到具体处理函数。返回的TextContent就是 AI 看到的执行结果。这里我选择返回文本因为 AI 对文本的理解最直接。如果返回的是结构化数据也可以序列化成 JSON 字符串再包进TextContent。2.3 为什么不用现成的 Excel 插件有人可能想Excel 本身就有 VBA、Power Query为什么还要绕一圈用 MCP我的体会是VBA 和 Power Query 解决的是“在 Excel 内部自动化”的问题而 MCP 解决的是“让 AI 理解并操作 Excel”的问题。前者需要你懂 Excel 的脚本语言后者你只需要用自然语言描述需求。比如我说“把这三个文件里销售额大于一千的记录合并按日期排序”AI 通过 MCP 工具就能一步步完成我不需要写任何公式或脚本。另一个原因是跨工具协作。MCP 不只连 Excel还能连数据库、浏览器、文件系统。当 AI 同时拥有这些工具时它能做的事情就超出了单个软件的边界。比如从 Excel 读数据调另一个 MCP 查数据库做比对再把结果写回 Excel。这种组合能力是传统插件做不到的。3. 核心工具的实现细节3.1 read_excel让 AI 先“看见”数据read_excel是整个流程的起点。AI 在动手处理之前必须先了解文件里有什么。这个工具我返回三部分信息sheet 名称列表告诉 AI 有哪些工作表。每个 sheet 的维度行数、列数让 AI 判断数据量级。前 5 行数据预览以 Markdown 表格形式返回AI 能直接看到表头和样例数据。实现上用 pandas 的ExcelFile打开文件遍历sheet_names对每个 sheet 用read_excel(nrows5)读前几行。注意这里要处理异常——文件可能被占用、可能不是有效的 xlsx、可能加密。我用 try-except 包住返回友好的错误信息而不是堆栈。import pandas as pd def handle_read_excel(file_path: str) - str: try: xl pd.ExcelFile(file_path) lines [f文件: {file_path}, fSheet列表: {xl.sheet_names}, ] for sheet in xl.sheet_names: df xl.parse(sheet, nrows5) full_df xl.parse(sheet, nrows0) lines.append(f--- Sheet: {sheet} ---) lines.append(f列名: {list(df.columns)}) lines.append(f预览:) lines.append(df.to_markdown(indexFalse)) lines.append() return \n.join(lines) except Exception as e: return f读取失败: {str(e)}注意to_markdown需要tabulate库记得pip install tabulate。如果数据里有中文Markdown 表格可能对齐不好看但 AI 读没问题。这里有个细节我只读前 5 行做预览不读全量。因为全量数据可能很大塞给 AI 会浪费上下文窗口。AI 看完预览后如果需要具体某列的数据可以再调query_excel做针对性查询。3.2 clean_excel处理那些“脏”数据数据清洗是最耗时的环节。我总结了几类常见问题在clean_excel里逐一处理合并单元格openpyxl 读合并区域时只有左上角单元格有值其他是 None。我的处理方式是取消合并然后把左上角的值填充到整个区域。这样后续按行处理时不会丢数据。空行空列用df.dropna(howall)去掉全空行df.dropna(axis1, howall)去掉全空列。但要注意有时候空列是有意义的占位所以这个操作我做成可选参数。日期格式混乱有的单元格是 datetime 对象有的是字符串“2024/1/1”有的是“2024-01-01”。统一用pd.to_datetime转换errorscoerce让无法转换的变成 NaT再决定是删除还是保留。文本型数字Excel 里经常有数字被存成文本导致求和为 0。我用pd.to_numeric尝试转换errorscoerce处理异常值。def handle_clean_excel(file_path: str, sheet: str, drop_empty: bool True, parse_dates: list None, numeric_cols: list None) - str: df pd.read_excel(file_path, sheet_namesheet) original_shape df.shape if drop_empty: df df.dropna(howall).dropna(axis1, howall) if parse_dates: for col in parse_dates: if col in df.columns: df[col] pd.to_datetime(df[col], errorscoerce) if numeric_cols: for col in numeric_cols: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) # 保存清洗后的结果到临时文件 output_path file_path.replace(.xlsx, _cleaned.xlsx) df.to_excel(output_path, indexFalse) return f清洗完成。原始形状: {original_shape}清洗后: {df.shape}。已保存到: {output_path}实操心得清洗前一定先备份原文件。我有一次直接覆盖了源文件结果发现某列被误删只能从回收站找。现在我的习惯是输出文件名统一加后缀绝不覆盖原始数据。3.3 merge_excel多文件合并的坑与解法合并多个 Excel 文件听起来简单实际坑很多。最大的问题是列名不一致。比如 A 文件叫“姓名”B 文件叫“客户姓名”C 文件叫“name”。如果直接pd.concatpandas 会按列名对齐对不上的列会变成 NaN。我的解法是提供一个column_mapping参数让 AI 或用户指定映射关系。比如{客户姓名: 姓名, name: 姓名}先把各文件的列名统一再合并。另一个坑是表头位置不同。有的文件第一行就是表头有的前面有几行标题。我加一个header_row参数指定从第几行开始读。def handle_merge_excel(file_paths: list, output_path: str, column_mapping: dict None, header_row: int 0) - str: dfs [] for path in file_paths: df pd.read_excel(path, headerheader_row) if column_mapping: df df.rename(columnscolumn_mapping) df[_source_file] path # 标记来源方便追溯 dfs.append(df) merged pd.concat(dfs, ignore_indexTrue, sortFalse) merged.to_excel(output_path, indexFalse) return f合并完成。共 {len(file_paths)} 个文件合并后 {len(merged)} 行{len(merged.columns)} 列。已保存到: {output_path}加_source_file列是我的个人习惯。合并后如果发现某行数据有问题能立刻知道来自哪个文件。这个列在最终交付前可以删掉但排查阶段非常有用。3.4 query_excel条件统计的灵活实现“统计同一列中包含关键词的对应数据求和”——这是热词里出现的一个具体需求。在 Excel 里用 SUMIF 加通配符能做但跨 sheet、跨文件就麻烦。我在query_excel里实现了一个通用的查询接口。设计上我让 AI 传入一个查询描述而不是直接传 SQL 或复杂表达式。因为 AI 擅长把自然语言转成结构化参数但不一定擅长写 pandas 查询。所以我定义了几个简单参数filter_column要筛选的列filter_keyword包含的关键词sum_column要求和的列group_by可选的分组列def handle_query_excel(file_path: str, sheet: str, filter_column: str, filter_keyword: str, sum_column: str None, group_by: str None) - str: df pd.read_excel(file_path, sheet_namesheet) mask df[filter_column].astype(str).str.contains(filter_keyword, naFalse) filtered df[mask] if sum_column: if group_by: result filtered.groupby(group_by)[sum_column].sum() return f按 {group_by} 分组{sum_column} 求和结果:\n{result.to_markdown()} else: total filtered[sum_column].sum() return f包含 {filter_keyword} 的行中{sum_column} 总和为: {total} else: return f包含 {filter_keyword} 的行数: {len(filtered)}\n{filtered.head(10).to_markdown(indexFalse)}这个接口的好处是 AI 容易调用。它只需要从用户的话里提取“哪一列”“什么关键词”“求和哪一列”填进参数就行。比如用户说“帮我算一下销售区域里包含‘华东’的订单金额总和”AI 就能自动填filter_column销售区域、filter_keyword华东、sum_column订单金额。3.5 write_excel安全落盘写文件看起来简单但有几个注意点。第一不要覆盖源文件除非用户明确要求。第二处理索引pandas 默认会把 index 写进去通常不需要所以indexFalse。第三大文件写入可能慢如果数据超过几万行考虑用 openpyxl 的 write_only 模式。def handle_write_excel(data: list, output_path: str, columns: list None) - str: df pd.DataFrame(data, columnscolumns) df.to_excel(output_path, indexFalse) return f写入完成。{len(df)} 行{len(df.columns)} 列。已保存到: {output_path}这里data参数接收的是列表的列表AI 从对话里提取数据后传进来。如果数据量大这种方式不太合适更好的做法是让 AI 调用其他工具生成中间文件再调write_excel做格式转换。4. 把 MCP Server 跑起来完整实操流程4.1 环境准备与依赖安装先把环境搭好。我假设你用的是 Windows 或 macOSPython 版本 3.10 以上。# 创建虚拟环境 python -m venv mcp-env # 激活Windows mcp-env\Scripts\activate # 激活macOS/Linux source mcp-env/bin/activate # 安装依赖 pip install mcp pandas openpyxl tabulatemcp是官方 SDKpandas和openpyxl处理 Exceltabulate用于 Markdown 表格输出。版本方面我实测mcp1.0.0、pandas2.0.0、openpyxl3.1.0比较稳定。注意如果你之前装过旧版mcp先pip uninstall mcp再装避免 API 不兼容。我有一次没卸载干净list_tools装饰器报错排查了半天。4.2 编写 Server 主文件新建excel_mcp_server.py把前面几节的代码整合进去。完整结构如下import asyncio import pandas as pd 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, description读取Excel文件返回sheet列表、列名和前5行预览, inputSchema{ type: object, properties: { file_path: {type: string, description: Excel文件的绝对路径} }, required: [file_path] } ), Tool( nameclean_excel, description清洗Excel数据去空行空列、统一日期和数值格式, inputSchema{ type: object, properties: { file_path: {type: string}, sheet: {type: string}, drop_empty: {type: boolean, default: True}, parse_dates: {type: array, items: {type: string}}, numeric_cols: {type: array, items: {type: string}} }, required: [file_path, sheet] } ), Tool( namemerge_excel, description合并多个Excel文件支持列名映射, inputSchema{ type: object, properties: { file_paths: {type: array, items: {type: string}}, output_path: {type: string}, column_mapping: {type: object}, header_row: {type: integer, default: 0} }, required: [file_paths, output_path] } ), Tool( namequery_excel, description按关键词筛选并求和支持分组, inputSchema{ type: object, properties: { file_path: {type: string}, sheet: {type: string}, filter_column: {type: string}, filter_keyword: {type: string}, sum_column: {type: string}, group_by: {type: string} }, required: [file_path, sheet, filter_column, filter_keyword] } ), Tool( namewrite_excel, description将数据写入Excel文件, inputSchema{ type: object, properties: { data: {type: array, items: {type: array}}, output_path: {type: string}, columns: {type: array, items: {type: string}} }, required: [data, output_path] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): try: if name read_excel: result handle_read_excel(arguments[file_path]) elif name clean_excel: result handle_clean_excel( arguments[file_path], arguments[sheet], arguments.get(drop_empty, True), arguments.get(parse_dates), arguments.get(numeric_cols) ) elif name merge_excel: result handle_merge_excel( arguments[file_paths], arguments[output_path], arguments.get(column_mapping), arguments.get(header_row, 0) ) elif name query_excel: result handle_query_excel( arguments[file_path], arguments[sheet], arguments[filter_column], arguments[filter_keyword], arguments.get(sum_column), arguments.get(group_by) ) elif name write_excel: result handle_write_excel( arguments[data], arguments[output_path], arguments.get(columns) ) else: result f未知工具: {name} return [TextContent(typetext, textresult)] except Exception as e: return [TextContent(typetext, textf执行出错: {str(e)})] async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: asyncio.run(main())把前面写的handle_*函数也放进同一个文件或者单独放一个handlers.py再 import。我习惯放同一个文件方便调试。4.3 在 AI 客户端里配置 MCP Server不同客户端的配置方式不一样但核心都是告诉客户端用什么命令启动这个 Server。以支持 stdio 的客户端为例配置通常是一个 JSON{ mcpServers: { excel-mcp: { command: python, args: [/absolute/path/to/excel_mcp_server.py], env: {} } } }关键点是command和args。command是 Python 解释器路径如果你用了虚拟环境要指向虚拟环境里的 python。args是脚本的绝对路径。配置好后重启客户端它就会在启动时拉起这个 Server并通过 stdio 通信。实操心得路径一定要用绝对路径。我一开始用了相对路径客户端工作目录不对一直报“找不到文件”。改成绝对路径后一次成功。另外如果客户端有日志功能打开看 Server 的 stderr 输出调试很有用。4.4 一次完整的对话流程演示配置好后我在客户端里输入帮我看看 D:\data\sales.xlsx 里有什么AI 会调用read_excel返回文件: D:\data\sales.xlsx Sheet列表: [Sheet1, Sheet2] --- Sheet: Sheet1 --- 列名: [日期, 销售区域, 订单金额, 客户姓名] 预览: | 日期 | 销售区域 | 订单金额 | 客户姓名 | |------------|----------|----------|----------| | 2024-01-01 | 华东 | 1200 | 张三 | | 2024-01-02 | 华南 | 800 | 李四 | ...然后我继续说帮我统计销售区域包含“华东”的订单金额总和AI 调用query_excel参数filter_column销售区域、filter_keyword华东、sum_column订单金额返回包含 华东 的行中订单金额总和为: 45600整个过程我不需要写任何代码AI 自动完成了工具调用和参数提取。这就是 MCP 的价值——把 AI 的语言理解能力和程序的数据处理能力接在一起。5. 踩过的坑与排查技巧5.1 常见问题速查表问题现象可能原因排查方法解决方案客户端启动后看不到工具Server 启动失败查看客户端日志中 Server 的 stderr手动运行python excel_mcp_server.py看报错调用工具返回“执行出错”文件路径不对或文件被占用检查路径是否存在、文件是否在 Excel 中打开关闭 Excel用绝对路径读取中文列名乱码文件编码问题用 openpyxl 直接打开看确保文件是 xlsx 而非 csvcsv 需指定 encoding求和结果为 0数值列是文本格式用df.dtypes查看列类型调clean_excel的numeric_cols参数转换合并后数据错位列名不一致或表头行不同对比各文件列名用column_mapping统一列名用header_row指定表头日期变成数字Excel 日期序列号未转换查看原始单元格格式用pd.to_datetime转换或指定parse_dates写入文件后打不开数据中有非法字符检查数据内容过滤控制字符确保输出路径有写权限5.2 三个让我印象深刻的坑第一个坑合并单元格的隐形数据。有一次处理一张考勤表姓名列有合并单元格openpyxl 读出来只有第一行有名字后面全是 None。我一开始用dropna把空行删了结果丢了一半数据。后来改成先检测合并区域用 openpyxl 的merged_cells.ranges拿到范围手动填充值。这个逻辑我封装进了clean_excel现在遇到合并单元格会自动处理。第二个坑pandas 的read_excel默认把第一行当表头。有个文件的表头在第三行前两行是标题和空行。我没注意读进来列名变成了“Unnamed: 0”“Unnamed: 1”。AI 看到这些列名也懵了不知道该操作哪一列。后来加了header_row参数让 AI 可以先读预览、判断表头位置再传正确的行号。第三个坑AI 调用工具时参数类型不对。我在 schema 里定义file_paths是数组但 AI 有时候传一个字符串进来。pd.concat收到字符串会报错。后来我在处理函数里加了类型检查如果是字符串就转成单元素列表。这种防御性编程在 MCP Server 里很有必要因为 AI 的输出不完全可控。5.3 性能与安全方面的注意事项性能如果 Excel 文件超过 10 万行pandas 读取会明显变慢。我的做法是先用 openpyxl 的read_only模式快速获取行数如果超过阈值提示 AI 分批处理或改用其他方案。另外to_markdown在大数据量下很慢预览时只取前几行。安全MCP Server 运行在本地能访问文件系统。我限制了工具只能操作指定目录下的文件避免 AI 误删其他文件。具体做法是在处理函数里检查file_path是否在允许的根目录下不在就拒绝。这个检查用os.path.abspath和os.path.commonpath实现。import os ALLOWED_ROOT os.path.abspath(D:/mcp_data) def is_path_allowed(file_path: str) - bool: abs_path os.path.abspath(file_path) return abs_path.startswith(ALLOWED_ROOT)提示这个限制很重要。AI 有时候会“自作主张”去读其他路径的文件加上白名单能避免很多意外。如果你信任 AI 的操作也可以放开但生产环境建议限制。6. 还能怎么扩展这个 Excel MCP Server 目前覆盖了读写、清洗、合并、查询这几类核心操作。实际用下来我发现还有几个方向可以继续加格式导出把 Excel 转成 Markdown 表格、CSV、JSON方便贴到文档或传给其他工具。热词里提到的“markdown表格转换excel”反过来也成立Excel 转 Markdown 同样有用。图表生成用 openpyxl 的 chart 功能让 AI 根据数据自动生成柱状图、折线图插入到 Excel 里。这个我还在试验主要是图表类型和数据的匹配逻辑需要调。公式注入不是写死计算结果而是把公式写进单元格比如SUM(B2:B100)。这样用户在 Excel 里打开后还能继续编辑。openpyxl 支持写公式但要注意公式里的中文和特殊字符转义。与数据库联动再写一个数据库 MCP Server让 AI 能把 Excel 数据和数据库记录做比对、同步。两个 Server 配合AI 就能完成“从 Excel 读数据、查数据库验证、把差异写回 Excel”这样的跨系统流程。批量处理现在一次处理一个文件如果目录下有几十个文件可以让 AI 遍历目录、逐个调用工具。这个逻辑可以放在 Server 里做一个batch_process工具也可以让 AI 自己循环调用。我个人的体会是MCP 最大的价值不是替代某个具体工具而是给 AI 装上了“手”。以前 AI 只能动嘴给建议现在它能真正操作文件、执行代码、返回结果。Excel 只是一个切入点同样的思路可以套到 PDF 处理、图片批处理、日志分析上。关键是先把一个场景跑通理解 MCP 的通信机制和工具设计方法后面扩展就是复制粘贴改参数的事。最后分享一个小技巧调试 MCP Server 时可以在call_tool里加一行日志把工具名和参数写到文件里。这样即使客户端不显示详细日志你也能知道 AI 到底传了什么进来。我用的就是最简单的with open(mcp_debug.log, a) as f: f.write(f{name}: {arguments}\n)排查参数问题时特别管用。