load_csv
Import a CSV file into a session for processing and analysis. Specify file path, encoding, and delimiter to prepare data for manipulation or validation.
Instructions
Load a CSV file into a session.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| delimiter | No | , | |
| encoding | No | utf-8 | |
| file_path | Yes | ||
| session_id | No |
Implementation Reference
- Core handler function that implements the load_csv tool logic: validates file path, reads CSV with pandas.read_csv, loads into session manager, returns OperationResult with data preview and stats.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-119 (registration)FastMCP tool registration decorator (@mcp.tool) for load_csv, which proxies arguments to the core implementation imported as _load_csv from 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)