Auto DBT generation Pipeline

[projectxr]

Smart Data Preprocessing for MXCP

LLM-Powered Automatic Data Transformation Inference

Core Architecture

Raw Data → Smart Analyzer → Auto-Generated dbt Models → DuckDB → MXCP Tools

Phase 1: Data Discovery & Analysis

class SmartDataPreprocessor:
    def analyze_data_source(self, file_path):
        # 1. READ AND SAMPLE DATA
        sample_data = read_sample(file_path, max_rows=100)
        headers = extract_headers(sample_data)
        
        # 2. LLM HEADER ANALYSIS
        llm_analysis = llm_call(f"""
        Analyze these column headers and sample data:
        Headers: {headers}
        Sample rows: {sample_data[:5]}
        
        Tasks:
        1. Group related columns (e.g., customer/Customer/customer_name)
        2. Identify data types for each column
        3. Detect data quality issues (nulls, formatting inconsistencies)
        4. Suggest cleaning transformations needed
        
        Return JSON format:
        {{
            "column_groups": [
                {{
                    "concept": "customer_identity",
                    "columns": ["customer", "Customer", "customer_name"],
                    "best_column": "customer_name",
                    "data_type": "string",
                    "cleaning_needed": ["trim", "title_case"]
                }}
            ],
            "transformations": [...],
            "primary_metrics": ["amount", "sales_total"],
            "dimensions": ["region", "product", "date"]
        }}
        """)
        
        return parse_json(llm_analysis)

Phase 2: Auto-Generate dbt Models

    def generate_dbt_model(self, analysis_result, table_name):
        # 3. BUILD SQL TRANSFORMATION
        sql_parts = []
        
        for group in analysis_result["column_groups"]:
            column_name = group["concept"]
            best_col = group["best_column"]
            data_type = group["data_type"]
            cleaning = group["cleaning_needed"]
            
            # Generate SQL based on analysis
            sql_part = build_sql_transformation(
                column=best_col,
                target_name=column_name,
                data_type=data_type,
                cleaning_rules=cleaning
            )
            sql_parts.append(sql_part)
        
        # 4. GENERATE COMPLETE dbt MODEL
        dbt_model = f"""
        {{{{ config(materialized='table') }}}}
        
        SELECT
            {',\n    '.join(sql_parts)}
        FROM {{{{ source('raw', '{table_name}') }}}}
        WHERE {generate_quality_filters(analysis_result)}
        """
        
        return dbt_model

    def build_sql_transformation(self, column, target_name, data_type, cleaning_rules):
        # Smart SQL generation based on LLM analysis
        sql = column
        
        if "trim" in cleaning_rules:
            sql = f"TRIM({sql})"
        if "upper_case" in cleaning_rules:
            sql = f"UPPER({sql})"
        if "remove_currency" in cleaning_rules:
            sql = f"CAST(REPLACE({sql}, '$', '') as DECIMAL)"
        if data_type == "date":
            sql = f"{sql}::date"
            
        return f"{sql} as {target_name}"

Phase 3: Auto-Generate MCP Tools

    def generate_mcp_tools(self, analysis_result, clean_table_name):
        # 5. CREATE SMART TOOLS BASED ON DATA STRUCTURE
        tools = []
        
        metrics = analysis_result["primary_metrics"]
        dimensions = analysis_result["dimensions"]
        
        # Auto-generate common analytical tools
        for metric in metrics:
            for dimension in dimensions:
                tool = {
                    "name": f"get_{metric}_by_{dimension}",
                    "description": f"Analyze {metric} broken down by {dimension}",
                    "parameters": [
                        {"name": dimension, "type": "string", "description": f"Filter by {dimension}"}
                    ],
                    "source": {
                        "code": f"""
                        SELECT 
                            {dimension},
                            SUM({metric}) as total_{metric},
                            AVG({metric}) as avg_{metric},
                            COUNT(*) as record_count
                        FROM {clean_table_name}
                        WHERE {dimension} = ${dimension}
                        GROUP BY {dimension}
                        ORDER BY total_{metric} DESC
                        """
                    }
                }
                tools.append(tool)
        
        return tools

Phase 4: Integration with MXCP Pipeline

class MXCPSmartInit:
    def smart_bootstrap(self, data_files):
        processor = SmartDataPreprocessor()
        
        for file_path in data_files:
            # 1. ANALYZE DATA
            analysis = processor.analyze_data_source(file_path)
            table_name = extract_table_name(file_path)
            
            # 2. GENERATE dbt MODEL
            dbt_model = processor.generate_dbt_model(analysis, table_name)
            write_file(f"models/{table_name}_clean.sql", dbt_model)
            
            # 3. GENERATE MCP TOOLS
            tools = processor.generate_mcp_tools(analysis, f"{table_name}_clean")
            for tool in tools:
                write_yaml(f"tools/{tool['name']}.yml", tool)
            
            # 4. UPDATE MXCP CONFIG
            update_mxcp_config(table_name, analysis)
        
        print("✅ Smart preprocessing complete!")
        print("Run: dbt run && mxcp serve")

Enhanced MXCP CLI Integration

# New smart commands
mxcp smart-init ./data/sales.csv ./data/customers.xlsx
# → Analyzes files, generates dbt models + MCP tools automatically

mxcp smart-update 
# → Re-analyzes data, detects schema drift, updates models

mxcp smart-validate
# → Compares LLM inferences against actual data patterns

Key Benefits Over Manual MXCP

1. Zero SQL Required: Developers just point at data files
2. Automatic Tool Generation: Creates sensible analytical endpoints
3. Schema Evolution: Re-runs analysis when data changes
4. Quality Validation: LLM can spot data quality issues humans miss
5. Documentation: Auto-generates descriptions for all transformations

Error Handling & Validation

    def validate_llm_inferences(self, analysis, sample_data):
        # Test LLM assumptions against actual data
        validation_results = []
        
        for group in analysis["column_groups"]:
            # Check if suggested transformations actually work
            test_sql = build_test_query(group)
            result = execute_test(test_sql, sample_data)
            
            if result.has_errors():
                # Fallback or re-prompt LLM with error context
                corrected_analysis = self.fix_analysis(group, result.errors)
                validation_results.append(corrected_analysis)
        
        return validation_results

Integration Points with Existing MXCP

• Extends mxcp init with smart analysis
• Compatible with existing dbt/DuckDB pipeline
• Enhances tool generation (vs manual YAML writing)
• Preserves all security/audit/policy features
• Adds automatic schema drift detection

Had worked on a pipeline where i did something like above, let me know if it makes sense/poses challenges! Sorry for the AI generated explainer above!

TLDR: iterate over column groups to generate dbt using slms/llms!

[miguelbranco80]

Thanks for the idea; it makes a lot of sense.
Happy to report that this is now supported using Claude Skills.
Take a look at https://github.com/raw-labs/claude-code-marketplace
If you install the MXCP plugin from our marketplace, it effectively implements a similar concept to this. Our goal is to improve skills overtime to support more cases/best practices.