get_row_data
Retrieve specific row data from CSV files with optional column filtering. Converts pandas types for JSON serialization in DataBeak's MCP server.
Instructions
Get data from specific row with optional column filtering.
Returns complete row data or filtered by column list. Converts pandas types for JSON serialization.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| row_index | Yes | Row index (0-based) to retrieve data from | |
| columns | No | Optional list of column names to retrieve (all columns if None) |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| data | Yes | Row data as column name to value mapping | |
| columns | Yes | List of column names included in data | |
| success | No | Whether operation completed successfully | |
| row_index | Yes | Row index (0-based) |
Implementation Reference
- The core handler function for the 'get_row_data' tool. Retrieves data from a specific row in the session's DataFrame, supports optional column filtering, validates row index and columns, converts pandas/numpy types to JSON-serializable Python types (handling NaN as None), and returns a structured RowDataResult.
def get_row_data( ctx: Annotated[Context, Field(description="FastMCP context for session access")], row_index: Annotated[int, Field(description="Row index (0-based) to retrieve data from")], columns: Annotated[ list[str] | None, Field(description="Optional list of column names to retrieve (all columns if None)"), ] = None, ) -> RowDataResult: """Get data from specific row with optional column filtering. Returns complete row data or filtered by column list. Converts pandas types for JSON serialization. """ session_id = ctx.session_id _session, df = get_session_data(session_id) # Validate row index if row_index < 0 or row_index >= len(df): msg = f"Row index {row_index} out of range (0-{len(df) - 1})" raise ToolError(msg) # Handle column filtering if columns is None: selected_columns = list(df.columns) row_data = df.iloc[row_index].to_dict() else: # Validate all columns exist missing_columns = [col for col in columns if col not in df.columns] if missing_columns: raise ColumnNotFoundError(missing_columns[0], list(df.columns)) selected_columns = columns row_data = df.iloc[row_index][columns].to_dict() # Handle pandas/numpy types for JSON serialization for key, value in row_data.items(): if pd.isna(value): row_data[key] = None elif hasattr(value, "item"): # numpy scalar row_data[key] = value.item() # No longer recording operations (simplified MCP architecture) return RowDataResult( row_index=row_index, data=row_data, columns=selected_columns, ) - src/databeak/servers/row_operations_server.py:647-647 (registration)Registers the get_row_data handler function as an MCP tool named 'get_row_data' on the row_operations_server FastMCP instance.
row_operations_server.tool(name="get_row_data")(get_row_data)