MCP Data Pipeline Assistant

MCP数据管道助手

对Azure数据工厂管道和Delta Lake的自然语言控制——由Claude和模型上下文协议提供支持。

    ](https://codespaces.new/guna-thota/mcp-data-pipeline-assistant)

______________________________________________________________________

这有什么作用

该项目允许您直接通过Claude与他们交谈，而不是编写脚本或导航云控制台来管理数据管道：

You:    "Show me all failed pipelines from the last 24 hours"
Claude: ❌ healthcare_ingestion_pipeline — Schema mismatch on patient_id
        ❌ billing_reconciliation_pipeline — SQL Server connection timeout

You:    "What's the average failure rate per pipeline over the last 30 days?"
Claude: patient_records_etl          — 22.9% failure rate
        healthcare_ingestion_pipeline — 14.3% failure rate
        delta_lake_refresh_pipeline   — 8.6%  failure rate

You:    "Trigger the healthcare ingestion pipeline"
Claude: 🚀 Pipeline queued. Run ID: adf-run-ed368a9a. SLA: 30 minutes.

______________________________________________________________________

建筑

┌─────────────────────────────────────────┐
│           Claude (via MCP)              │
│   Natural language → tool selection     │
└────────────────┬────────────────────────┘
                 │ MCP Protocol (stdio)
┌────────────────▼────────────────────────┐
│         MCP Server (server.py)          │
│   5 tools exposed as MCP primitives     │
└──────┬──────────────┬───────────────────┘
       │              │
┌──────▼──────┐  ┌────▼──────────────────┐
│  ADF Layer  │  │   Delta Lake Layer     │
│             │  │                        │
│ • Pipeline  │  │ • DuckDB query engine  │
│   status    │  │ • Parquet storage      │
│ • Trigger   │  │ • 35 days of metrics   │
│   runs      │  │ • Preset SQL queries   │
│ • List all  │  │ • Custom SQL support   │
└─────────────┘  └────────────────────────┘
       │
  [Mock data]           [Production]
  JSON files    →    azure-mgmt-datafactory
  (dev/demo)         REST API calls

______________________________________________________________________

MCP工具

______________________________________________________________________

项目结构

mcp-data-pipeline-assistant/
├── server.py                        # MCP server — entry point
├── demo.py                          # Run all tools without Claude
├── requirements.txt
├── .env.example                     # Environment variable template
├── .gitignore
│
├── tools/
│   ├── pipeline_status.py           # get_pipeline_status tool
│   ├── trigger_pipeline.py          # trigger_pipeline tool
│   └── query_delta.py               # query_delta_table tool
│
├── mock_data/
│   ├── pipeline_runs.json           # 10 realistic ADF run records
│   ├── pipeline_definitions.json    # 4 pipeline configs with metadata
│   ├── generate_delta_data.py       # Generates 35 days of metrics
│   └── metrics_delta/               # Delta Lake table (Parquet + _delta_log)
│       ├── pipeline_metrics.parquet
│       └── _delta_log/
│           └── 00000000000000000000.json
│
├── tests/
│   └── test_tools.py                # 23 unit tests across all tools
│
├── .devcontainer/
│   └── devcontainer.json            # GitHub Codespaces config
│
└── .github/
    └── workflows/
        └── ci.yml                   # GitHub Actions CI pipeline

______________________________________________________________________

快速开始

选项1:GitHub代码空间（无需安装）

1. 点击 “在GitHub代码空间中打开” 上方徽章
2. 等待约60秒以构建环境
3. 在Codespace终端中：

python demo.py

所有10个演示场景都会立即运行。无需设置。

选项2：本地设置

# Clone the repo
git clone https://github.com/guna-thota/mcp-data-pipeline-assistant.git
cd mcp-data-pipeline-assistant

# Install dependencies
pip install -r requirements.txt

# Generate Delta Lake mock data
python mock_data/generate_delta_data.py

# Run the demo
python demo.py

# Run tests
python -m pytest tests/ -v

______________________________________________________________________

连接到克劳德桌面

要将其用作Claude Desktop的实时MCP服务器，请执行以下操作：

1.安装克劳德桌面 从 claude.ai/下载

2.查找您的配置文件：

• macOS： ~/Library/Application Support/Claude/claude_desktop_config.json
• 窗户： %APPDATA%\Claude\claude_desktop_config.json

3.添加MCP服务器配置：

{
  "mcpServers": {
    "data-pipeline-assistant": {
      "command": "python",
      "args": ["/absolute/path/to/mcp-data-pipeline-assistant/server.py"]
    }
  }
}

4.重新启动克劳德桌面。 您将在Claude界面中看到可用的工具。

5.尝试以下提示：

• *“显示过去48小时内的所有管道故障”*
• *“在过去的30天里，每条管道的故障率是多少？”*
• *“触发患者记录ETL管道”*
• *“哪个管道违反SLA的次数最多？”*
• *“按管道显示平均运行时间”*

______________________________________________________________________

Delta Lake预设查询

内置四个预设查询，可通过Claude或直接访问：

from tools.query_delta import get_preset_query

# Failure rate by pipeline
get_preset_query("failure_rate")

# SLA breach history
get_preset_query("sla_breaches")

# Average run duration
get_preset_query("avg_duration")

# Daily records processed
get_preset_query("daily_records")

自定义SQL也完全支持：

from tools.query_delta import query_delta_table

result = query_delta_table("""
    SELECT pipeline_name, SUM(records_processed) as total
    FROM pipeline_metrics
    WHERE run_date >= '2026-04-01'
    GROUP BY pipeline_name
    ORDER BY total DESC
""")
print(result["formatted_table"])

______________________________________________________________________

连接到真正的Azure数据工厂

mock层是对真正的ADF REST API的一个替换。切换到生产：

1.将Azure凭据添加到 .env:

cp .env.example .env
# Fill in your Azure values

2.替换模拟呼入 tools/pipeline_status.py:

# Replace this mock:
with open(DATA_PATH) as f:
    all_runs = json.load(f)

# With this real ADF call:
from azure.mgmt.datafactory import DataFactoryManagementClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = DataFactoryManagementClient(credential, os.getenv("AZURE_SUBSCRIPTION_ID"))

filter_params = RunFilterParameters(
    last_updated_after=cutoff,
    last_updated_before=datetime.utcnow()
)
runs = client.pipeline_runs.query_by_factory(
    os.getenv("AZURE_RESOURCE_GROUP"),
    os.getenv("AZURE_DATA_FACTORY_NAME"),
    filter_params
)

3.更换模拟触发器 tools/trigger_pipeline.py:

# Replace mock return with:
run_response = client.pipelines.create_run(
    os.getenv("AZURE_RESOURCE_GROUP"),
    os.getenv("AZURE_DATA_FACTORY_NAME"),
    pipeline_name,
    parameters=parameters or {}
)
return {"success": True, "run_id": run_response.run_id, ...}

不需要其他更改。MCP服务器和工具层保持不变。

______________________________________________________________________

运行测试

python -m pytest tests/ -v

23 passed in 1.50s

TestPipelineStatus   — 7 tests
TestTriggerPipeline  — 6 tests
TestQueryDelta       — 10 tests

______________________________________________________________________

技术栈

______________________________________________________________________

为什么这个项目

大多数数据工程团队手动检查Azure Portal仪表板或编写一次性脚本来调试管道故障。该项目展示了MCP如何作为生产数据栈和人工智能助手之间的结构化接口；在不牺牲准确性、可追溯性或控制性的情况下，实现自然语言管道的可观察性。

同样的架构可以扩展到任何数据平台：copula、Airflow、dbt、Redshift或Snowflake。

______________________________________________________________________

作者

Guna Durga Prashanth Thota 数据工程师——Azure | Rancher | PySpark | Delta Lake | MCP

领英 · · 投资组合

______________________________________________________________________

许可证

MIT许可证——见 许可证 了解详情。