parse_csv

Instructions

Implementation Reference

• mcp_tool/tools/csv_tool.py:12-111 (handler)

  The CsvTool class with the execute() method that parses CSV files: validates file_path, auto-detects encoding (using chardet or default utf-8), reads with pandas, and returns file info, data preview, and descriptive statistics.

  @ToolRegistry.register
  class CsvTool(BaseTool):
      """
      CSV文件处理工具，用于解析CSV文件内容
      """
      
      name = "parse_csv"
      description = "解析CSV文件内容，支持各种编码格式"
      input_schema = {
          "type": "object",
          "required": ["file_path"],
          "properties": {
              "file_path": {
                  "type": "string",
                  "description": "CSV文件的本地路径，例如'/path/to/data.csv'",
              },
              "encoding": {
                  "type": "string",
                  "description": "文件编码格式，例如'utf-8'、'gbk'等，默认自动检测",
              }
          },
      }
      
      async def execute(self, arguments: Dict[str, Any]) -> List[types.TextContent]:
          """
          解析CSV文件内容
          
          Args:
              arguments: 参数字典，必须包含'file_path'键，可选'encoding'键
          
          Returns:
              解析结果列表
          """
          if "file_path" not in arguments:
              return [types.TextContent(
                  type="text",
                  text="错误: 缺少必要参数 'file_path'"
              )]
          
          file_path = arguments["file_path"]
          # 处理文件路径，支持挂载目录的转换
          file_path = self.process_file_path(file_path)
          
          if not os.path.exists(file_path):
              return [types.TextContent(
                  type="text",
                  text=f"错误: 文件不存在: {file_path}"
              )]
          
          try:
              # 尝试自动检测编码
              encoding = arguments.get("encoding", None)
              if encoding is None:
                  try:
                      import chardet
                      with open(file_path, 'rb') as f:
                          raw_data = f.read()
                          encoding = chardet.detect(raw_data)['encoding']
                  except ImportError:
                      encoding = 'utf-8'  # 如果没有chardet，默认使用utf-8
              
              # 读取CSV文件
              df = pd.read_csv(file_path, encoding=encoding)
              
              # 获取基本信息
              info = {
                  "文件名": os.path.basename(file_path),
                  "行数": len(df),
                  "列数": len(df.columns),
                  "列名": list(df.columns),
                  "数据预览": df.head().to_string()
              }
              
              # 生成描述性统计
              stats = df.describe().to_string()
              
              # 组合结果
              result = (
                  f"CSV文件解析结果:\n\n"
                  f"基本信息:\n"
                  f"- 文件名: {info['文件名']}\n"
                  f"- 行数: {info['行数']}\n"
                  f"- 列数: {info['列数']}\n"
                  f"- 列名: {', '.join(info['列名'])}\n\n"
                  f"数据预览:\n{info['数据预览']}\n\n"
                  f"描述性统计:\n{stats}"
              )
              
              return [types.TextContent(
                  type="text",
                  text=result
              )]
              
          except Exception as e:
              error_details = traceback.format_exc()
              return [types.TextContent(
                  type="text",
                  text=f"错误: 处理CSV文件时发生错误: {str(e)}\n{error_details}"
              )] 

• mcp_tool/tools/csv_tool.py:18-33 (schema)

  Input schema for parse_csv: requires 'file_path' (string), optional 'encoding' (string, auto-detected if not provided).

  name = "parse_csv"
  description = "解析CSV文件内容，支持各种编码格式"
  input_schema = {
      "type": "object",
      "required": ["file_path"],
      "properties": {
          "file_path": {
              "type": "string",
              "description": "CSV文件的本地路径，例如'/path/to/data.csv'",
          },
          "encoding": {
              "type": "string",
              "description": "文件编码格式，例如'utf-8'、'gbk'等，默认自动检测",
          }
      },
  }

• mcp_tool/tools/csv_tool.py:12-13 (registration)

  CsvTool is registered via the @ToolRegistry.register decorator, which adds it to ToolRegistry._tools dict with key = class.name ('parse_csv').

  @ToolRegistry.register
  class CsvTool(BaseTool):

• mcp_tool/tools/__init__.py:47-50 (registration)

  The ToolRegistry.register classmethod stores tool classes in _tools dict keyed by their name attribute.

  def register(cls, tool_class: Type[BaseTool]) -> Type[BaseTool]:
      """注册工具"""
      cls._tools[tool_class.name] = tool_class
      return tool_class

• mcp_tool/tools/__init__.py:6-38 (helper)

  BaseTool.process_file_path is a helper used by parse_csv's execute method to convert host-mounted file paths to container paths.

  class BaseTool:
      """所有工具的基类"""
      name: str = ""
      description: str = ""
      input_schema: dict = {}
      
      @classmethod
      def get_tool_definition(cls) -> types.Tool:
          """获取工具定义"""
          return types.Tool(
              name=cls.name,
              description=cls.description,
              inputSchema=cls.input_schema
          )
      
      async def execute(self, arguments: dict) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
          """执行工具逻辑，需要在子类中实现"""
          raise NotImplementedError("Tool implementation must override execute method")
      
      def process_file_path(self, file_path: str) -> str:
          """
          处理文件路径，支持挂载目录的转换
          
          如果路径以HOST_MOUNT_SOURCE环境变量开头，则将其转换为容器内的路径
          """
          host_mount_source = os.environ.get('HOST_MOUNT_SOURCE', '')
          host_mount_target = os.environ.get('HOST_MOUNT_TARGET', '/host_files')
          
          # 如果路径以挂载源目录开头，则替换为挂载目标目录
          if host_mount_source and file_path.startswith(host_mount_source):
              return file_path.replace(host_mount_source, host_mount_target, 1)
          
          return file_path