load_csv_from_content
Parse CSV data from string content with validation, returning session ID and data preview for further analysis operations in DataBeak.
Instructions
Load CSV data from string content into DataBeak session.
Parses CSV data directly from string with validation. Returns session ID and data preview for further operations.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| content | Yes | CSV data as string content | |
| delimiter | No | Column delimiter character (comma, tab, semicolon, pipe) | , |
| header_config | No | Header detection configuration |
Implementation Reference
- The core handler function that parses the provided CSV content string using pandas.read_csv on StringIO, handles errors, validates DataFrame size, loads data into the session, and returns a LoadResult with preview and metadata.async def load_csv_from_content( ctx: Annotated[Context, Field(description="FastMCP context for session access")], content: Annotated[str, Field(description="CSV data as string content")], delimiter: Annotated[ str, Field(description="Column delimiter character (comma, tab, semicolon, pipe)") ] = ",", *, header_config: Annotated[ HeaderConfigUnion | None, Field(default=None, description="Header detection configuration"), ] = None, ) -> LoadResult: """Load CSV data from string content into DataBeak session. Parses CSV data directly from string with validation. Returns session ID and data preview for further operations. """ # Get session_id from FastMCP context session_id = ctx.session_id await ctx.info("Loading CSV from content string") # Handle default header configuration if header_config is None: header_config = AutoDetectHeader() if not content or not content.strip(): msg = "Content cannot be empty" raise ToolError(msg) # Parse CSV from string using StringIO try: df = pd.read_csv( StringIO(content), delimiter=delimiter, header=resolve_header_param(header_config), ) except pd.errors.EmptyDataError as e: msg = "CSV content is empty or contains no data" raise ToolError(msg) from e except pd.errors.ParserError as e: msg = f"CSV parsing error: {e}" raise ToolError(msg) from e if df.empty: msg = "Parsed CSV contains no data rows" raise ToolError(msg) # Validate DataFrame size against limits validate_dataframe_size(df) # Get or create session session_manager = get_session_manager() session = session_manager.get_or_create_session(session_id) session.load_data(df, None) await ctx.info(f"Loaded {len(df)} rows and {len(df.columns)} columns from content") return create_load_result(df)
- src/databeak/servers/io_server.py:470-473 (registration)Registers the load_csv_from_content handler as an MCP tool named 'load_csv_from_content' on the io_server FastMCP instance.# Register the logic functions directly as MCP tools (no wrapper functions needed) io_server.tool(name="load_csv_from_url")(load_csv_from_url) io_server.tool(name="load_csv_from_content")(load_csv_from_content) io_server.tool(name="get_session_info")(get_session_info)
- Pydantic model defining the output schema of the load_csv_from_content tool, including rows affected, columns, data preview, and memory usage.class LoadResult(BaseToolResponse): """Response model for data loading operations.""" rows_affected: int = Field(description="Number of rows loaded") columns_affected: list[str] = Field(description="List of column names detected") data: DataPreview | None = Field(None, description="Sample of loaded data") memory_usage_mb: float | None = Field(None, description="Memory usage in megabytes")
- src/databeak/server.py:103-113 (registration)Mounts the io_server (containing load_csv_from_content tool) into the main DataBeak FastMCP server instance, making the tool available.# Mount specialized servers mcp.mount(system_server) mcp.mount(io_server) mcp.mount(row_operations_server) mcp.mount(statistics_server) mcp.mount(discovery_server) mcp.mount(validation_server) mcp.mount(transformation_server) mcp.mount(column_server) mcp.mount(column_text_server)
- Helper function called by the handler to construct the LoadResult response from the loaded DataFrame, generating data preview and computing metadata.def create_load_result(df: pd.DataFrame) -> LoadResult: """Create LoadResult from a DataFrame. Args: df: Loaded DataFrame Returns: LoadResult with data preview and metadata """ # Create data preview with indices preview_data = create_data_preview_with_indices(df, 5) data_preview = DataPreview( rows=preview_data["records"], row_count=preview_data["total_rows"], column_count=preview_data["total_columns"], truncated=preview_data["preview_rows"] < preview_data["total_rows"], ) return LoadResult( rows_affected=len(df), columns_affected=[str(col) for col in df.columns], data=data_preview, memory_usage_mb=df.memory_usage(deep=True).sum() / (1024 * 1024), )