CockroachDB MCP Server

Instructions

Args:
    table: Table name (schema.table or just table).
    id_value: Primary key value.
    id_column: Name of the ID column (default: 'id').

Returns:
    Delete result.

Implementation Reference

• src/cockroachdb_mcp/server.py:304-324 (registration)

  Registration and handler for the 'delete_row' MCP tool. Decorated with @mcp.tool() and delegates to crud.delete_row.

  @mcp.tool()
  async def delete_row(
      table: str,
      id_value: str | int,
      id_column: str = "id",
  ) -> dict[str, Any]:
      """Delete a row by primary key.
  
      Args:
          table: Table name (schema.table or just table).
          id_value: Primary key value.
          id_column: Name of the ID column (default: 'id').
  
      Returns:
          Delete result.
      """
      try:
          return await crud.delete_row(table, id_value, id_column)
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/tools/crud.py:281-333 (handler)

  Core implementation of delete_row: constructs DELETE SQL query, executes it using connection_manager, handles read-only mode, validation, and returns detailed result.

  async def delete_row(
      table: str,
      id_value: str | int,
      id_column: str = "id",
  ) -> dict[str, Any]:
      """Delete a row by primary key.
  
      Args:
          table: Table name (schema.table or just table).
          id_value: Primary key value.
          id_column: Name of the ID column (default: 'id').
  
      Returns:
          Delete result.
      """
      # Check read-only mode
      if settings.read_only:
          return {"status": "error", "error": "Server is in read-only mode"}
  
      # Validate table name
      valid, error = _validate_table_name(table)
      if not valid:
          return {"status": "error", "error": error}
  
      schema, table_name = _parse_table_name(table)
  
      query = f"DELETE FROM {schema}.{table_name} WHERE {id_column} = %s"
  
      conn = await connection_manager.ensure_connected()
  
      try:
          async with conn.cursor() as cur:
              await cur.execute(query, (id_value,))
  
              if cur.rowcount == 0:
                  return {
                      "status": "warning",
                      "table": f"{schema}.{table_name}",
                      "action": "delete",
                      "id": id_value,
                      "message": "No row found with specified ID",
                      "rows_affected": 0,
                  }
  
              return {
                  "status": "success",
                  "table": f"{schema}.{table_name}",
                  "action": "deleted",
                  "id": id_value,
                  "rows_affected": cur.rowcount,
              }
      except Exception as e:
          return {"status": "error", "error": str(e)}