# Data Quality MCP Server

A Model Context Protocol (MCP) server that runs **data-quality checks** on a SQL
warehouse and exposes them to an AI assistant — so anyone can ask, in plain
English, *"were there any duplicate PAN numbers in last night's load?"* and get
a real answer from live data.

Built to demonstrate a pattern I care about as a Data QA engineer:
**write your validation logic once, then serve it through two front doors.**

## The core idea: one logic layer, two front doors

The checks live in `src/checks.py` as plain Python functions. Nothing about them
is AI-specific. That single layer is then reused by:

| Front door | File | Who uses it | When |
|---|---|---|---|
| **Pipeline** | `src/run_checks.py` | CI/CD, Airflow | Automated, scheduled. Exits non-zero on failure so a build breaks when data breaks. |
| **Conversational** | `src/server.py` | Claude Desktop / any MCP client | On demand, in natural language, for non-engineers. |

Same duplicate-PAN check runs in your nightly pipeline *and* answers a product
manager's ad-hoc question. Write once; run automatically; ask conversationally.

## Checks included

Classic QA — `row_count`, `check_nulls` (null/empty %), `check_duplicates`.
Profiling — `profile_column` (min/max/distinct), `check_freshness` (row age).
Discovery — `list_tables`, `describe_table`. Plus `run_all_checks` for the suite.

Every check returns a structured result with a `PASS` / `FAIL` / `FRESH` / `STALE`
status. Table and column names are validated against the live schema before any
SQL is built (prevents injection, gives clear errors).

## Project structure

```
data-quality-mcp/
├── src/
│   ├── checks.py       # core check logic (the shared layer)
│   ├── db.py           # SQLite connection + identifier validation
│   ├── server.py       # MCP tools (front door #1: conversational)
│   └── run_checks.py   # CLI report + CI exit code (front door #2: pipeline)
├── scripts/seed_db.py  # sample warehouse with intentional issues
├── tests/test_checks.py
├── requirements.txt
└── README.md
```

## Quick start

```bash
python -m venv venv
# Windows: venv\Scripts\activate   |   macOS/Linux: source venv/bin/activate
pip install -r requirements.txt

python scripts/seed_db.py     # build the sample DB
python -m src.run_checks      # pipeline front door — prints a report
pytest -q                     # run the tests
```

The sample data is seeded with deliberate problems (missing emails, a duplicate
PAN, a negative order amount) so every check has something real to catch.

## Front door #1 — connect to Claude Desktop

Add this to Claude Desktop's config (Settings → Developer → Edit config), using
absolute paths to your venv Python and the repo:

```json
{
  "mcpServers": {
    "data-quality": {
      "command": "/absolute/path/to/venv/bin/python",
      "args": ["-m", "src.server"],
      "cwd": "/absolute/path/to/data-quality-mcp"
    }
  }
}
```

Restart Claude Desktop, then ask: *"List the tables, then check the customers
table for duplicate PANs and tell me if any order amounts look wrong."*

## Front door #2 — use in CI/CD

`python -m src.run_checks` exits `1` if any check fails, so it drops straight
into a pipeline step or an Airflow task — the run fails loudly when data quality
regresses.

## Tech

Python · SQLite · Model Context Protocol (MCP) · pytest

---
*Built by Prashant (github.com/psat022) — Senior Data QA / ETL engineer, exploring
the overlap between data validation and AI tooling.*