CSV Editor

Instructions

Implementation Reference

• src/csv_editor/tools/io_operations.py:21-114 (handler)

  Core handler function that implements the load_csv tool logic: validates file path, reads CSV with pandas.read_csv, loads into session manager, provides progress updates via ctx, and returns detailed OperationResult.

  async def load_csv(
      file_path: str,
      encoding: str = "utf-8",
      delimiter: str = ",",
      session_id: Optional[str] = None,
      header: Optional[int] = 0,
      na_values: Optional[List[str]] = None,
      parse_dates: Optional[List[str]] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """Load a CSV file into a session.
      
      Args:
          file_path: Path to the CSV file
          encoding: File encoding (default: utf-8)
          delimiter: Column delimiter (default: comma)
          session_id: Optional existing session ID to use
          header: Row number to use as header (default: 0)
          na_values: Additional strings to recognize as NA/NaN
          parse_dates: Columns to parse as dates
          ctx: FastMCP context
      
      Returns:
          Operation result with session ID and data info
      """
      try:
          # Validate file path
          is_valid, validated_path = validate_file_path(file_path)
          if not is_valid:
              return OperationResult(
                  success=False,
                  message=f"Invalid file path: {validated_path}",
                  error=validated_path
              ).model_dump()
          
          if ctx:
              await ctx.info(f"Loading CSV file: {validated_path}")
              await ctx.report_progress(0.1, "Validating file...")
          
          # Get or create session
          session_manager = get_session_manager()
          session = session_manager.get_or_create_session(session_id)
          
          if ctx:
              await ctx.report_progress(0.3, "Reading file...")
          
          # Read CSV with pandas
          read_params = {
              "filepath_or_buffer": validated_path,
              "encoding": encoding,
              "delimiter": delimiter,
              "header": header,
          }
          
          if na_values:
              read_params["na_values"] = na_values
          if parse_dates:
              read_params["parse_dates"] = parse_dates
              
          df = pd.read_csv(**read_params)
          
          if ctx:
              await ctx.report_progress(0.8, "Processing data...")
          
          # Load into session
          session.load_data(df, validated_path)
          
          if ctx:
              await ctx.report_progress(1.0, "Complete!")
              await ctx.info(f"Loaded {len(df)} rows and {len(df.columns)} columns")
          
          return OperationResult(
              success=True,
              message=f"Successfully loaded CSV file",
              session_id=session.session_id,
              rows_affected=len(df),
              columns_affected=df.columns.tolist(),
              data={
                  "shape": df.shape,
                  "dtypes": {col: str(dtype) for col, dtype in df.dtypes.items()},
                  "memory_usage_mb": df.memory_usage(deep=True).sum() / (1024 * 1024),
                  "preview": df.to_dict('records')
              }
          ).model_dump()
          
      except Exception as e:
          if ctx:
              await ctx.error(f"Failed to load CSV: {str(e)}")
          return OperationResult(
              success=False,
              message="Failed to load CSV file",
              error=str(e)
          ).model_dump()

• src/csv_editor/server.py:109-118 (registration)

  MCP tool registration for 'load_csv' using @mcp.tool decorator. This is the entry point handler that delegates to the core implementation in io_operations.py.

  @mcp.tool
  async def load_csv(
      file_path: str,
      encoding: str = "utf-8",
      delimiter: str = ",",
      session_id: Optional[str] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """Load a CSV file into a session."""
      return await _load_csv(file_path, encoding, delimiter, session_id, ctx=ctx)