get_column_data
Extract data from specific columns with optional row range slicing for focused data analysis. Supports filtering by row indices to retrieve targeted column values with range metadata.
Instructions
Get data from specific column with optional row range slicing.
Supports row range filtering for focused analysis. Returns column values with range metadata.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| column | Yes | Column name to retrieve data from | |
| start_row | No | Starting row index (inclusive, 0-based) for data slice | |
| end_row | No | Ending row index (exclusive, 0-based) for data slice |
Implementation Reference
- The handler function that executes the get_column_data tool. It retrieves column data from the session's DataFrame, validates inputs, applies optional row slicing, handles data type conversions, and returns a structured result.def get_column_data( ctx: Annotated[Context, Field(description="FastMCP context for session access")], column: Annotated[str, Field(description="Column name to retrieve data from")], start_row: Annotated[ int | None, Field(description="Starting row index (inclusive, 0-based) for data slice"), ] = None, end_row: Annotated[ int | None, Field(description="Ending row index (exclusive, 0-based) for data slice"), ] = None, ) -> ColumnDataResult: """Get data from specific column with optional row range slicing. Supports row range filtering for focused analysis. Returns column values with range metadata. """ session_id = ctx.session_id _session, df = get_session_data(session_id) # Validate column exists if column not in df.columns: raise ColumnNotFoundError(column, list(df.columns)) # Validate and set row range if start_row is not None and start_row < 0: msg = "start_row" raise InvalidParameterError(msg, start_row, "must be non-negative") if end_row is not None and end_row < 0: msg = "end_row" raise InvalidParameterError(msg, end_row, "must be non-negative") if start_row is not None and start_row >= len(df): msg = f"start_row {start_row} out of range (0-{len(df) - 1})" raise ToolError(msg) if end_row is not None and end_row > len(df): msg = f"end_row {end_row} out of range (0-{len(df)})" raise ToolError(msg) if start_row is not None and end_row is not None and start_row >= end_row: msg = "start_row" raise InvalidParameterError(msg, start_row, "must be less than end_row") # Apply row range slicing if start_row is None and end_row is None: column_data = df[column] start_row = 0 end_row = len(df) - 1 elif start_row is None: column_data = df[column][:end_row] start_row = 0 elif end_row is None: column_data = df[column][start_row:] end_row = len(df) - 1 else: column_data = df[column][start_row:end_row] # Convert to list and handle pandas/numpy types values = convert_pandas_na_list(column_data.tolist()) # No longer recording operations (simplified MCP architecture) return ColumnDataResult( column=column, values=values, total_values=len(values), start_row=start_row, end_row=end_row, )
- Pydantic input schema model for the get_column_data tool parameters.class ColumnDataRequest(BaseModel): """Request parameters for column data retrieval.""" model_config = ConfigDict(extra="forbid") column: str = Field(description="Column name") start_row: int | None = Field(None, ge=0, description="Starting row index (inclusive)") end_row: int | None = Field(None, ge=0, description="Ending row index (exclusive)") @field_validator("start_row", "end_row") @classmethod def validate_row_indices(cls, v: int | None) -> int | None: """Validate row indices are non-negative.""" if v is not None and v < 0: msg = "Row indices must be non-negative" raise ValueError(msg) return v
- src/databeak/servers/row_operations_server.py:648-648 (registration)Registers the get_column_data handler as a tool named 'get_column_data' on the FastMCP row_operations_server.row_operations_server.tool(name="get_column_data")(get_column_data)
- Pydantic output schema model used by the get_column_data tool response.class ColumnDataResult(BaseToolResponse): """Response model for column data operations.""" column: str = Field(description="Column name") values: list[str | int | float | bool | None] = Field( description="Column values in specified range", ) total_values: int = Field(description="Number of values returned") start_row: int | None = Field( default=None, description="Starting row index used (None if from beginning)", ) end_row: int | None = Field(default=None, description="Ending row index used (None if to end)")