DataBeak

Instructions

Implementation Reference

• src/databeak/servers/row_operations_server.py:339-404 (handler)

  The handler function implementing the get_column_data tool. Retrieves column data with optional row slicing, validates inputs, handles pandas DataFrame slicing and type conversions.

  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,
      )

• src/databeak/servers/row_operations_server.py:108-125 (schema)

  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)

  Registration of the get_column_data handler as a tool on the FastMCP row_operations_server.

  row_operations_server.tool(name="get_column_data")(get_column_data)