# AI Data Engineer Plugin

> 🎥 Full tutorial: [https://www.youtube.com/@agenticdatalab](https://www.youtube.com/@agenticdatalab)

A portable AI agent plugin built to the **Agent Plugins 1.0** spec — one set of skills and MCP servers that runs across Cursor, Codex, VS Code, and Claude Code.

**The demo scenario:** A production `customer_orders_daily` pipeline failed at 2:16 AM. A developer changed `customer_id` from `int` to `"CUST-{id}"` format. The warehouse target column expects `BIGINT`. PostgreSQL raises `invalid input syntax for type bigint: "CUST-1009"`. The plugin investigates this incident, identifies the root cause, and tells you whether it is safe to rerun.

---

## Plugin structure (Agent Plugins 1.0)

```
ai-data-engineer-plugin/
│
├── plugin.json                    ← root manifest (Agent Plugins 1.0)
├── mcp.json                       ← MCP server declarations
│
├── skills/
│   ├── debug-pipeline/SKILL.md    ← investigate a failed job
│   ├── investigate-data/SKILL.md  ← data quality failures
│   ├── generate-rca/SKILL.md      ← formal RCA report
│   ├── analyze-sql/SKILL.md       ← query performance
│   └── validate-rerun/SKILL.md    ← rerun safety check
│
├── mcp-servers/
│   ├── job-telemetry/server.py    ← reads SQLite job run records
│   ├── sql-metadata/server.py     ← PostgreSQL schema + query tools
│   └── observability/server.py    ← reads real pipeline log files
│
├── pipeline/
│   ├── customer_orders_daily.py   ← the broken pipeline (real failure)
│   ├── data/seed_data.sql         ← PostgreSQL schema + seed data
│   └── logs/                      ← real log files written by the pipeline
│
├── adapters/
│   └── claude-code/plugin.json    ← Claude Code native adapter
│
└── docker-compose.yml             ← PostgreSQL on port 5433
```

---

## Quick start

**1. Prerequisites:** Docker, Python 3.11+, Node.js

**2. Setup:**
```bash
cd ai-data-engineer-plugin
python3 -m venv venv && source venv/bin/activate
pip install -r requirements.txt
cp .env.example .env
# Edit .env — add your GITHUB_TOKEN
```

**3. Start PostgreSQL:**
```bash
docker compose up -d postgres
```

**4. Run the broken pipeline (generates real failure logs):**
```bash
export $(cat .env | xargs)
python pipeline/customer_orders_daily.py
```

Expected output: `[LOAD] DataError: invalid input syntax for type bigint — customer_id received VARCHAR "CUST-1009"`

**5. Verify real data exists:**
```bash
# Check job telemetry (SQLite)
sqlite3 pipeline/job_runs.db "SELECT job_run_id, status, error_message FROM job_runs ORDER BY start_time DESC LIMIT 3;"

# Check log files
ls pipeline/logs/

# Check PostgreSQL tables
docker compose exec postgres psql -U pipeline_user -d pipeline_db -c "\dt"
```

---

## MCP server tools

| Server | Tool | What it returns |
|--------|------|----------------|
| job-telemetry | `get_latest_job_run` | Status, error message, log file path |
| job-telemetry | `list_job_runs` | Last N runs with timestamps |
| job-telemetry | `get_job_logs` | Full log file content for a run |
| sql-metadata | `get_table_schema` | Column names and data types |
| sql-metadata | `run_query` | Execute SELECT queries |
| sql-metadata | `check_type_mismatch` | Compare column types across tables |
| sql-metadata | `explain_query` | EXPLAIN ANALYZE output |
| observability | `get_error_logs` | All ERROR lines from recent logs |
| observability | `search_logs` | Grep-style search with context |
| observability | `get_pipeline_metrics` | Success rate, last success time |
| github | `list_commits` | Recent commits with diffs |
| github | `get_file_contents` | Source code of pipeline files |

---

## The five experiments (video)

1. **Portability test** — same plugin, same incident query, three different agent clients
2. **Skill discovery** — natural language incident → agent picks the right skill automatically
3. **Failure isolation** — break one MCP server, verify others still work independently
4. **Claude Code compatibility** — load the Agent Plugins 1.0 package into Claude Code, compare the wrapper formats, demonstrate the adapter
5. **Production reality** — what Agent Plugins 1.0 does not yet solve (versioning, distribution, trust)

---

## The conceptual distinction (teach this first)

```
SKILL  =  HOW the agent should do something
           "Follow these 8 steps when debugging a failed pipeline"

MCP    =  WHAT external actions the agent can take
           get_job_run(), read_logs(), query_table(), get_git_diff()

PLUGIN =  HOW we package related Skills and MCP servers together
```

---

## Environment variables

| Variable | Default | Description |
|----------|---------|-------------|
| `DB_HOST` | `127.0.0.1` | PostgreSQL host |
| `DB_PORT` | `5433` | PostgreSQL port (Docker) |
| `DB_NAME` | `pipeline_db` | Database name |
| `DB_USER` | `pipeline_user` | Database user |
| `DB_PASSWORD` | `pipeline_pass` | Database password |
| `JOB_DB_PATH` | `./pipeline/job_runs.db` | SQLite job telemetry file |
| `LOG_DIR` | `./pipeline/logs` | Pipeline log directory |
| `GITHUB_TOKEN` | — | GitHub personal access token (repo read scope) |