Python module for building data pipelines.
## Luigi MCP Server: Python Workflow Pipelines The **Luigi MCP Server** integrates Spotify's Luigi workflow engine into Google Antigravity, enabling developers to build complex batch processing pipelines with AI-assisted task management and dependency resolution. ### Why Luigi MCP? - **Dependency Management**: Automatically resolve and execute task dependencies in correct order - **Failure Recovery**: Resume failed pipelines from the last successful checkpoint - **Central Scheduler**: Coordinate tasks across multiple workers and machines - **Visualization**: View pipeline DAGs and execution status in real-time - **Extensible Framework**: Create custom task types for any data processing need ### Key Features #### 1. Pipeline Task Definition ```python import luigi class ExtractData(luigi.Task): date = luigi.DateParameter() def output(self): return luigi.LocalTarget(f"data/raw/{self.date}.json") def run(self): data = fetch_daily_data(self.date) with self.output().open("w") as f: json.dump(data, f) class TransformData(luigi.Task): date = luigi.DateParameter() def requires(self): return ExtractData(date=self.date) def output(self): return luigi.LocalTarget(f"data/processed/{self.date}.parquet") def run(self): with self.input().open() as f: raw_data = json.load(f) df = transform_pipeline(raw_data) df.to_parquet(self.output().path) ``` #### 2. Pipeline Execution and Monitoring ```python # Execute pipeline via MCP result = await luigi.runPipeline({ "task": "LoadToWarehouse", "params": { "date": "2024-12-15" }, "workers": 4, "local_scheduler": False }) # Check pipeline status status = await luigi.getPipelineStatus("LoadToWarehouse") print("Completed tasks:", status.completed) print("Pending tasks:", status.pending) print("Failed tasks:", status.failed) ``` ### Configuration ```json { "mcpServers": { "luigi": { "command": "npx", "args": ["-y", "@anthropic/mcp-luigi"], "env": { "LUIGI_CONFIG_PATH": "/path/to/luigi.cfg", "LUIGI_SCHEDULER_URL": "http://localhost:8082" } } } } ``` ### Use Cases **Data Warehousing**: Build ETL pipelines that extract from multiple sources, transform data, and load into data warehouses. **ML Feature Engineering**: Create reproducible feature engineering pipelines with proper dependency tracking. **Report Generation**: Automate complex report generation workflows with multiple data preparation steps. The Luigi MCP Server brings battle-tested workflow orchestration to AI-assisted Python development.
{
"mcpServers": {
"luigi": {}
}
}