get_row_data
Retrieve specific row data from datasets with optional column filtering. Extracts complete row information or selected columns while converting pandas types for JSON serialization.
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) |
Implementation Reference
- The primary handler function implementing the get_row_data tool. Retrieves row data from the session's pandas DataFrame, supports optional column selection, validates bounds, handles NaN and numpy types for serialization, and returns 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, )
- Pydantic model defining the output response schema for the get_row_data tool, including row_index, data (dict of column values), and columns list.class RowDataResult(BaseToolResponse): """Response model for row data operations.""" row_index: int = Field(description="Row index (0-based)") data: dict[str, str | int | float | bool | None] = Field( description="Row data as column name to value mapping", ) columns: list[str] = Field(description="List of column names included in data")
- 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 FastMCP row_operations_server.row_operations_server.tool(name="get_row_data")(get_row_data)