MCP Sheet Parser

Instructions

Implementation Reference

• src/mcp_server/server.py:26-27 (registration)

  Registers all tools including parse_sheet by calling register_tools on the MCP Server instance.
  # 注册所有工具 register_tools(server)
• src/models/tools.py:64-101 (schema)

  Defines the input schema and description for the parse_sheet tool in the MCP tool list.
  Tool( name="parse_sheet", description="解析Excel/CSV文件为结构化JSON数据。默认返回文件概览信息（行数、列数、数据类型、前几行预览），避免上下文过载。LLM可通过参数控制是否获取完整数据、样式信息等。适合数据分析和处理，修改后可用apply_changes写回。", inputSchema={ "type": "object", "properties": { "file_path": { "type": "string", "description": "目标表格文件的绝对路径，支持 .csv, .xlsx, .xls, .xlsb, .xlsm 格式。" }, "sheet_name": { "type": "string", "description": "【可选】要解析的工作表名称。如果留空，使用第一个工作表。" }, "range_string": { "type": "string", "description": "【可选】单元格范围，如'A1:D10'。指定范围时会返回该范围的完整数据。" }, "include_full_data": { "type": "boolean", "description": "【可选，默认false】是否返回完整数据。false时只返回概览和预览，true时返回所有行数据。大文件建议先查看概览。" }, "include_styles": { "type": "boolean", "description": "【可选，默认false】是否包含样式信息（字体、颜色、边框等）。样式信息会显著增加数据量。" }, "preview_rows": { "type": "integer", "description": "【可选，默认5】预览行数。当include_full_data为false时，返回前N行作为数据预览。" }, "max_rows": { "type": "integer", "description": "【可选】最大返回行数。用于限制大文件的数据量，超出部分会被截断并提示。" } }, "required": ["file_path"] } ),
• src/models/tools.py:237-333 (handler)

  MCP tool handler for parse_sheet: validates arguments, calls CoreService.parse_sheet_optimized, formats response as JSON.
  async def _handle_parse_sheet(arguments: dict[str, Any], core_service: CoreService) -> list[TextContent]: """处理 parse_sheet 工具调用，避免上下文爆炸。""" try: # 获取和验证参数 file_path = arguments["file_path"] if not isinstance(file_path, str) or not file_path.strip(): raise ValueError("file_path必须是非空字符串") sheet_name = arguments.get("sheet_name") if sheet_name is not None and not isinstance(sheet_name, str): raise ValueError("sheet_name必须是字符串") range_string = arguments.get("range_string") if range_string is not None and not isinstance(range_string, str): raise ValueError("range_string必须是字符串") include_full_data = arguments.get("include_full_data", False) if not isinstance(include_full_data, bool): raise ValueError("include_full_data必须是布尔值") include_styles = arguments.get("include_styles", False) if not isinstance(include_styles, bool): raise ValueError("include_styles必须是布尔值") preview_rows = arguments.get("preview_rows", 5) if not isinstance(preview_rows, int) or preview_rows <= 0: raise ValueError("preview_rows必须是正整数") max_rows = arguments.get("max_rows") if max_rows is not None and (not isinstance(max_rows, int) or max_rows <= 0): raise ValueError("max_rows必须是正整数或None") result = core_service.parse_sheet_optimized( file_path=file_path, sheet_name=sheet_name, range_string=range_string, include_full_data=include_full_data, include_styles=include_styles, preview_rows=preview_rows, max_rows=max_rows ) # 为LLM添加使用指导 result["llm_guidance"] = { "current_mode": "overview" if not include_full_data else "full_data", "next_steps": _generate_next_steps_guidance(result, include_full_data, include_styles), "data_access": { "headers": "result['headers'] - 列标题", "preview": "result['preview_rows'] - 数据预览", "full_data": "设置 include_full_data=true 获取完整数据", "styles": "设置 include_styles=true 获取样式信息" } } response = { "success": True, "operation": "parse_sheet", "data": result } return [TextContent( type="text", text=json.dumps(response, ensure_ascii=False, indent=2) )] except FileNotFoundError as e: return [TextContent( type="text", text=json.dumps({ "success": False, "error_type": "file_not_found", "error_message": f"文件未找到: {str(e)}", "suggestion": "请检查文件路径是否正确。支持的格式: .xlsx, .xls, .xlsb, .xlsm, .csv" }, ensure_ascii=False, indent=2) )] except ValueError as e: return [TextContent( type="text", text=json.dumps({ "success": False, "error_type": "invalid_parameter", "error_message": f"参数错误: {str(e)}", "suggestion": "请检查sheet_name是否存在，range_string格式是否正确（如'A1:D10'）" }, ensure_ascii=False, indent=2) )] except Exception as e: return [TextContent( type="text", text=json.dumps({ "success": False, "error_type": "parsing_error", "error_message": f"解析失败: {str(e)}", "suggestion": "请检查文件是否损坏，或尝试指定具体的工作表名称和范围" }, ensure_ascii=False, indent=2) )]
• src/core_service.py:133-190 (handler)

  Core optimized parsing logic invoked by the tool handler, handles file parsing, sheet selection, range extraction, preview/full data, styles.
  def parse_sheet_optimized(self, file_path: str, sheet_name: str | None = None, range_string: str | None = None, include_full_data: bool = False, include_styles: bool = False, preview_rows: int = 5, max_rows: int | None = None) -> dict[str, Any]: """ 参数： file_path: 文件路径 sheet_name: 工作表名称（可选） range_string: 单元格范围（可选） include_full_data: 是否返回完整数据（默认False，只返回概览） include_styles: 是否包含样式信息（默认False） preview_rows: 预览行数（默认5行） max_rows: 最大返回行数（可选） 返回： 优化后的JSON数据 """ try: # 验证文件输入 validated_path, _ = validate_file_input(file_path) # 获取解析器 parser = self.parser_factory.get_parser(str(validated_path)) # 解析文件 sheets = parser.parse(str(validated_path)) # 选择目标工作表 if sheet_name: target_sheet = next((s for s in sheets if s.name == sheet_name), None) if not target_sheet: available_sheets = [s.name for s in sheets] raise ValueError(f"工作表 '{sheet_name}' 不存在。可用工作表: {available_sheets}") else: if not sheets: raise ValueError("文件中没有找到任何工作表。") target_sheet = sheets[0] # 处理范围选择 if range_string: try: start_row, start_col, end_row, end_col = parse_range_string(range_string) return self._extract_range_data(target_sheet, start_row, start_col, end_row, end_col, include_styles) except ValueError as e: raise ValueError(f"范围格式错误: {e}") # 根据参数返回不同级别的数据 return self._extract_optimized_data( target_sheet, include_full_data=include_full_data, include_styles=include_styles, preview_rows=preview_rows, max_rows=max_rows ) except Exception as e: logger.error(f"优化解析失败: {e}") raise
• src/core_service.py:36-132 (helper)

  Primary parse_sheet method with caching, streaming decision, and fallback to optimized parsing.
  def parse_sheet(self, file_path: str, sheet_name: str | None = None, range_string: str | None = None, enable_streaming: bool = True, streaming_threshold: int | None = None) -> dict[str, Any]: """ 解析表格文件为标准化的JSON格式。 参数: file_path: 文件路径 sheet_name: 工作表名称（可选） range_string: 单元格范围（可选） enable_streaming: 是否启用自动流式读取（默认True） streaming_threshold: 流式读取的单元格数量阈值，None时使用配置默认值 返回: 标准化的TableModel JSON """ # 获取当前配置 current_config = get_config() if streaming_threshold is None: streaming_threshold = current_config.streaming_threshold_cells try: # 验证文件输入 validated_path, _ = validate_file_input(file_path) # 尝试从缓存获取数据 cache_manager = get_cache_manager() cached_data = cache_manager.get(file_path, range_string, sheet_name) if cached_data is not None: logger.info(f"从缓存获取数据: {file_path}") return cached_data['data'] # 获取解析器 parser = self.parser_factory.get_parser(str(validated_path)) # 检查是否应该使用流式读取 if enable_streaming and self._should_use_streaming(str(validated_path), streaming_threshold): json_data = self._parse_sheet_streaming(str(validated_path), sheet_name, range_string) else: # 使用传统方法 sheets = parser.parse(str(validated_path)) # 如果指定了工作表名称，则选择对应的工作表 if sheet_name: target_sheet = next((s for s in sheets if s.name == sheet_name), None) if not target_sheet: raise ValueError(f"工作表 '{sheet_name}' 不存在。") # 否则，默认使用第一个工作表 else: if not sheets: raise ValueError("文件中没有找到任何工作表。") target_sheet = sheets[0] # 检查工作表是否为空 if not target_sheet: logger.warning("目标工作表为None") return { "sheet_name": "Empty", "headers": [], "rows": [], "total_rows": 0, "total_columns": 0, "size_info": { "total_cells": 0, "processing_mode": "empty", "recommendation": "工作表为空，无数据可显示" } } if not target_sheet.rows: logger.warning(f"工作表 '{target_sheet.name}' 为空") return { "sheet_name": target_sheet.name, "headers": [], "rows": [], "total_rows": 0, "total_columns": 0, "size_info": { "total_cells": 0, "processing_mode": "empty", "recommendation": "工作表为空，无数据可显示" } } # 转换为标准化JSON格式 json_data = self._sheet_to_json(target_sheet, range_string) # 缓存解析结果 cache_manager.set(file_path, json_data, range_string, sheet_name) logger.debug(f"数据已缓存: {file_path}") return json_data except Exception as e: logger.error(f"解析表格失败: {e}") raise