Kimi K2 Heavy Processor MCP

   

Heavy computation and data processing MCP for Claude Desktop. Handle complex SQL operations, large-scale data transformations, and resilient batch processing with automatic retry mechanisms.

🌟 Features

• SQL Processing: Full SQLite support with complex queries
• Batch Operations: Process millions of records efficiently
• Resilient Execution: Automatic retry with exponential backoff
• Data Pipelines: ETL operations with streaming support
• Memory Management: Smart chunking for large datasets
• Progress Tracking: Real-time status updates
• Error Recovery: Checkpoint-based resumption

🚀 Core Capabilities

SQL Operations

• Complex JOIN operations across multiple tables
• Window functions and CTEs
• Bulk inserts and updates
• Transaction management
• Index optimization

Data Processing

• CSV/JSON/XML parsing and generation
• Data validation and cleansing
• Format conversions
• Aggregation pipelines
• Statistical computations

Resilience Features

• Automatic retry on failure (3 attempts)
• Exponential backoff (1s, 2s, 4s)
• Transaction rollback on error
• Progress checkpointing
• Partial result recovery

📦 Installation

Via NPM (Recommended)

npm install -g kimi-k2-heavy-processor-mcp

Manual Installation

git clone https://github.com/justmy2satoshis/kimi-k2-heavy-processor-mcp.git
cd kimi-k2-heavy-processor-mcp
pip install -r requirements.txt

🔧 Configuration

Add to your Claude Desktop configuration file:

Windows: %APPDATA%\Claude\claude_desktop_config.json macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

{
  "mcpServers": {
    "kimi-k2-heavy-processor": {
      "command": "python",
      "args": ["C:\\path\\to\\kimi-k2-heavy-processor-mcp\\src\\server.py"],
      "env": {
        "DB_PATH": "C:\\Users\\username\\AppData\\Local\\kimi-k2\\data.db",
        "MAX_MEMORY_MB": "2048",
        "CHUNK_SIZE": "10000"
      }
    }
  }
}

📖 Usage Examples

Execute SQL Query

result = await execute_sql({
  "query": "SELECT * FROM users WHERE created_at > ?",
  "params": ["2024-01-01"],
  "database": "main.db"
})

Batch Data Processing

processed = await process_batch({
  "input_file": "data.csv",
  "operations": [
    {"type": "filter", "condition": "amount > 100"},
    {"type": "transform", "mapping": "amount * 1.1"},
    {"type": "aggregate", "group_by": "category"}
  ],
  "output_format": "json"
})

Resilient Operation

result = await resilient_execute({
  "operation": "complex_etl",
  "source": "raw_data.csv",
  "max_retries": 3,
  "checkpoint_interval": 1000
})

Data Pipeline

pipeline = await create_pipeline({
  "stages": [
    {"name": "extract", "source": "database"},
    {"name": "transform", "rules": "business_logic.json"},
    {"name": "load", "target": "warehouse.db"}
  ],
  "parallel": true
})

💡 Use Cases

Data Analysis

• Large CSV file processing
• Statistical computations
• Data aggregation and grouping
• Time series analysis

ETL Operations

• Database migrations
• Data warehouse loading
• Format conversions
• Data cleansing pipelines

Batch Processing

• Bulk email processing
• Log file analysis
• Report generation
• Data validation

SQL Operations

• Complex reporting queries
• Database maintenance
• Index optimization
• Performance analysis

🏗️ Architecture

kimi-k2-heavy-processor-mcp/
├── src/
│   ├── server.py           # Main MCP server
│   ├── sql_processor.py    # SQL execution engine
│   ├── batch_processor.py  # Batch operations
│   ├── resilient.py        # Retry mechanisms
│   └── pipeline.py         # Data pipelines
├── examples/               # Usage examples
├── tests/                  # Test suite
└── requirements.txt

📊 Performance Metrics

| Operation | Records/Second | Memory Usage | |-----------|---------------|--------------| | CSV Read | 100,000 | <500MB | | SQL INSERT | 50,000 | <200MB | | JOIN Query | 1M rows/sec | <1GB | | Aggregation | 500,000 | <300MB | | Transform | 75,000 | <400MB |

🧪 Testing

pytest tests/

Tests cover:

• SQL operation accuracy
• Retry mechanism validation
• Memory management
• Performance benchmarks
• Error recovery

🤝 Contributing

Contributions welcome! See CONTRIBUTING.md for guidelines.

Priority Areas

1. Additional data formats
2. Performance optimizations
3. New SQL functions
4. Pipeline templates

🔒 Security

• SQL injection prevention
• Input sanitization
• Secure file operations
• Memory limit enforcement
• Process isolation

📝 License

MIT License - see LICENSE file for details

🙏 Acknowledgments

• Anthropic for Model Context Protocol
• SQLite team for embedded database
• Python community for data tools
• Contributors and testers

📧 Support

• Issues: GitHub Issues
• Discussions: GitHub Discussions

🚦 Status

• ✅ Production Ready
• ✅ Resilient execution
• ✅ Large-scale processing
• ✅ Comprehensive testing
• ✅ Claude Desktop compatible

⚡ Quick Start

# 1. Load CSV data
await load_csv("sales_data.csv", "sales_table")

# 2. Process with SQL
await execute_sql("""
  SELECT
    category,
    SUM(amount) as total,
    AVG(amount) as average
  FROM sales_table
  GROUP BY category
  HAVING total > 10000
""")

# 3. Export results
await export_results("summary.json", format="json")

---

Note: Requires Claude Desktop with MCP support enabled.

Built with ❤️ for data engineers and analysts