# Data Quality CDE MCP Server

MCP server that generates and deploys **Cloudera Data Engineering (CDE) Airflow jobs** from **ODCS data contracts**. It validates **live table data** in CDW against schema and data-quality rules — **without Write-Audit-Publish (WAP)** staging branches.

Built on patterns from [data-quality](https://github.com/frothkoetter/data-quality) (CDE deployment, CDW connection, SQL operators) and [data-contract-mcp-server](https://github.com/frothkoetter/data-contract-mcp-server) (ODCS contract format, MCP tool design).

## What it does

1. **Parse** an ODCS-compatible data contract (`schema_objects` + `quality` rules)
2. **Generate** an Airflow DAG that runs daily (default `0 3 * * *`) using:
   - `SQLExecuteQueryOperator` — audit table setup
   - `SQLCheckOperator` — schema + quality checks against the **live** table
   - `BranchPythonOperator` — pass/fail gate with audit logging
3. **Deploy** to CDE — upload DAG + `lib/` + `sql/` resource, create Airflow job
4. **Monitor** CDE job runs and Airflow execution status

## MCP tools (14)

| Tool | Description |
|------|-------------|
| `get_cde_config` | Show current CDE/CDW settings |
| `diagnose_cde_connectivity` | Probe CDE CLI endpoint + list jobs |
| `generate_dag_from_contract` | Generate DAG Python from contract JSON |
| `generate_spark_dag_from_contract` | Generate Spark orchestrator DAG + bundle preview |
| `preview_contract_checks` | Preview SQL checks before deploy |
| `deploy_contract_dq_job` | Generate + deploy CDE job (`airflow_sql` or `spark`) |
| `list_cde_jobs` | List CDE jobs |
| `describe_cde_job` | Job details |
| `list_cde_resources` | List file resources |
| `trigger_cde_job` | Manual job run |
| `list_cde_job_runs` | List runs for a job |
| `get_cde_job_run_status` | Run status by id |
| `get_airflow_dag_run_status` | Latest Airflow run for a CDE job |
| `delete_cde_job` | Delete CDE job |

## Configuration

Copy `.env.example` to `.env`:

```bash
CDE_USER=your_cde_workload_user
CDE_PASSWORD=your_cde_workload_password
CDE_ENDPOINT_URL=https://your-cde-service.cloudera.site
CDW_CONNECTION_ID=CDW
DEFAULT_DAG_SCHEDULE=0 3 * * *
```

| Parameter | Description |
|-----------|-------------|
| `CDE_USER` | CDE workload username |
| `CDE_PASSWORD` | CDE workload password |
| `CDE_ENDPOINT_URL` | **Jobs API URL** from CDE cluster details (`…/dex/api/v1`) |
| `CDE_AUTH_URL` | Optional Knox token URL (auto-derived if unset) |
| `CDE_TRANSPORT` | `api` (default, Agent Studio) \| `cli` \| `auto` |
| `CDE_BIN` | Path to `cde` CLI (only for `CDE_TRANSPORT=cli`) |
| `CDW_CONNECTION_ID` | Airflow connection id for Hive/Impala in CDW (default: `CDW`) |
| `DEFAULT_DAG_SCHEDULE` | Cron schedule (default: daily at 03:00 UTC) |

**Agent Studio:** The CDE CLI is **not** installed in CrewAI/Agent Studio. Use `CDE_TRANSPORT=api` (default) with the **Jobs API URL** — not the Grafana or service-only URL.

**Local laptop:** Use `CDE_TRANSPORT=cli` if you have the `cde` binary configured, or keep `api` everywhere for consistency.

**Prerequisite:** For deploy tools, generated DAG bundles are uploaded via REST. The Airflow `CDW` connection must exist in your CDE environment.

## Cloudera Agent Studio (recommended)

Agent Studio only supports **stdio** MCP servers launched with **`uvx`** (Python) or **`npx`** (Node.js). Use a **git URL** so the runtime can install the package; do **not** use `uv run` unless the repo is checked out on the same machine.

```json
{
  "mcpServers": {
    "data-quality-cde-mcp-server": {
      "command": "uvx",
      "args": [
        "--from",
        "git+https://github.com/frothkoetter/data-quality-cde-mcp-server.git@master",
        "run-server"
      ],
      "env": {
        "CDE_USER": "<your_cde_workload_user>",
        "CDE_PASSWORD": "<your_cde_workload_password>",
        "CDE_ENDPOINT_URL": "https://<your-cde-service>.cloudera.site",
        "CDW_CONNECTION_ID": "CDW"
      }
    }
  }
}
```

See also `conf.yaml.example` for YAML-style Agent Studio registration.

**Registration tips**

1. Push this repo to GitHub first — `uvx --from git+https://...` cannot resolve a local-only repo.
2. Use placeholder CDE credentials during catalog registration; provide real values when attaching the MCP server to a workflow agent.
3. If you see *"We could not figure out the tools offered by the MCP server"*, the server may still work in workflows — [Agent Studio documents occasional tool-discovery failures](https://github.com/cloudera/CAI_STUDIO_AGENT/blob/main/docs/user_guide/mcp.md). Add the MCP server to your agent manually and select tools there.
4. Avoid writing to **stdout** from wrapper scripts — stdio transport uses stdout for JSON-RPC.
5. Ensure the Agent Studio environment has `uvx` on PATH (`which uvx`) and can reach `CDE_ENDPOINT_URL`.

See [Cloudera MCP registration docs](https://docs.cloudera.com/machine-learning/cloud/use-ai-studios/topics/ml-register-mcp-server.html).

## Troubleshooting MCP server load failures

If CrewAI Agent Studio, Cursor, or another MCP host fails to connect to this server via `uvx`, check the following.

### Wrong git branch in `uvx --from`

This repo's default branch is **`master`** (not `main`). Agent Studio registration fails if the git URL points at a non-existent branch:

```
git+https://github.com/frothkoetter/data-quality-cde-mcp-server.git@master
```

### `FastMCP` / `mcp` import error

`uvx` installs the latest `mcp` package (currently 2.x). MCP Python SDK 2.0 renamed `FastMCP` to `MCPServer`. Older server builds crash on startup with:

```
ImportError: cannot import name 'FastMCP' from 'mcp.server'
```

This server includes the dual-import fix (`MCPServer` + `FastMCP` fallback). Ensure you use a build from `main` after the MCP SDK 2.x compatibility update.

Verify locally:

```bash
CDE_USER="test" CDE_PASSWORD="test" CDE_ENDPOINT_URL="https://example.com" \
uv run python scripts/test_mcp_stdio.py uvx --from . run-server
```

Expected output: `OK initialize` and `OK tools/list: 13 tools`.

### Missing CDE credentials

CDE credentials are **not** required at startup (tool discovery works without them). They are required when calling deploy/monitor tools such as `deploy_contract_dq_job` or `list_cde_jobs`. Set `CDE_USER`, `CDE_PASSWORD`, and `CDE_ENDPOINT_URL` in the MCP host `env` block before using those tools.

### `FileNotFoundError: cde` in Agent Studio

CrewAI Agent Studio does **not** install the CDE CLI. The server now uses the **CDE REST API** by default (`CDE_TRANSPORT=api`). Ensure:

1. `CDE_ENDPOINT_URL` is the **Jobs API URL** from your virtual cluster details (`https://<vc>.<host>/dex/api/v1`) — not a generic service URL.
2. `CDE_TRANSPORT=api` (default) — do **not** set `cli` in Agent Studio.
3. `CDE_USER` / `CDE_PASSWORD` are valid CDP workload credentials.

Test connectivity with the `diagnose_cde_connectivity` MCP tool after re-registering.

### Agent Studio / CrewAI config

Ensure `uvx` is on the Agent Studio host PATH (`which uvx`).

## Cursor / MCP setup

Add to your MCP config (see `mcp.json.example`):

```json
{
  "mcpServers": {
    "data-quality-cde-mcp-server": {
      "command": "uvx",
      "args": ["--from", "git+https://github.com/frothkoetter/data-quality-cde-mcp-server.git@master", "run-server"],
      "env": {
        "CDE_USER": "<user>",
        "CDE_PASSWORD": "<password>",
        "CDE_ENDPOINT_URL": "https://<cde-host>",
        "CDW_CONNECTION_ID": "CDW"
      }
    }
  }
}
```

Local development:

```bash
cd data-quality-cde-mcp-server
uv sync
uv run run-server
```

## Example contract

See `examples/flights_contract.json` — based on the `airlinedata.flights` table from the data-quality project:

```json
{
  "contract_id": "flights-contract",
  "version": "1.0",
  "schema_objects": [{
    "name": "flights",
    "physical_name": "airlinedata.flights",
    "properties": [
      {"name": "uniquecarrier", "isRequired": true},
      {"name": "origin", "isRequired": true}
    ]
  }],
  "quality": [
    {"name": "carrier_completeness", "rule_type": "completeness", "element": "uniquecarrier"},
    {"name": "delay_range", "rule_type": "range", "element": "arrdelay", "threshold": "600"}
  ]
}
```

## Generated DAG flow

```
ensure_audit_table → column/table/custom checks → evaluate_contract_gate
  ├─ dq_success → persist_audit
  └─ contract_violation_alert (AirflowException + audit log)
```

Uses [Airflow Common SQL provider](https://airflow.apache.org/docs/apache-airflow-providers-common-sql/) operators per [Astronomer's SQL data quality guide](https://www.astronomer.io/docs/learn/airflow-sql-data-quality):

| Operator | Used for | Contract mapping |
|----------|----------|------------------|
| `SQLColumnCheckOperator` | Null, unique, min, max, distinct per column | `isRequired`, `primaryKey`, `completeness`, `uniqueness`, `range`, `min_value`, `distinct_count_min` |
| `SQLTableCheckOperator` | Row count, freshness, composite PK | Default `COUNT(*) >= 1`, `timeliness`/`freshness`, multi-column PK |
| `SQLCheckOperator` | Custom SQL | Rules with a `query` field, or unmapped complex checks |

### Column check types (SQLColumnCheckOperator)

| Check type | SQL abstraction | Example contract rule |
|------------|-----------------|----------------------|
| `null_check` | `SUM(CASE WHEN col IS NULL …)` | `isRequired: true`, `completeness` |
| `unique_check` | `COUNT - COUNT(DISTINCT)` | `primaryKey: true`, `uniqueness` |
| `distinct_check` | `COUNT(DISTINCT col)` | `distinct_count_min` |
| `min` / `max` | `MIN(col)` / `MAX(col)` | `min_value`, `range` / `max_value` |

Qualifiers: `equal_to`, `greater_than`, `geq_to`, `less_than`, `leq_to` (with optional `tolerance`).

### Table checks (SQLTableCheckOperator)

| Check | Example |
|-------|---------|
| Row count | `{"row_count_check": {"check_statement": "COUNT(*) >= 1"}}` |
| Freshness | `{"freshness_col": {"check_statement": "MAX(ts) >= CURRENT_TIMESTAMP - INTERVAL '24' HOUR"}}` |
| Composite PK | Custom `check_statement` with subquery |

### Custom checks (SQLCheckOperator)

Any rule with a `query` field, or checks spanning multiple tables. Returns a single row; task fails if any value is Python-falsy (e.g. `0`).

### Supported quality rule types

| Rule type | Generated SQL |
|-----------|---------------|
| `completeness` / `not_null` | Null count = 0 |
| `uniqueness` | No duplicate values |
| `range` / `max_value` | Column <= threshold |
| `min_value` | Column >= threshold |
| `timeliness` / `freshness` | Max timestamp within N hours |
| `distinct_count_min` | COUNT(DISTINCT col) >= threshold |
| Custom | Use `query` field — fails when query returns rows |

Schema checks always include: min row count, required-column null checks, primary-key uniqueness.

## Deploy from MCP

```
deploy_contract_dq_job(
  contract_json=<ODCS contract>,
  cde_user=...,
  cde_password=...,
  cde_endpoint_url=...,
  cdw_connection_id="CDW",
  trigger_after_deploy=true
)
```

This creates:

- CDE resource: `{contract-id}-dq-resource` (DAG + lib/ + sql/)
- CDE Airflow job: `{contract-id}-dq-job`
- DAG id: `dag_contract_dq_{contract_id}`

### Spark 3.4 orchestration (`execution_mode="spark"`)

Use when checks should run in Spark instead of Airflow SQL operators:

```
deploy_contract_dq_job(
  contract_json=<ODCS contract>,
  execution_mode="spark",
  spark_runtime_image="dex-spark-runtime",  # optional Spark 3.4 runtime resource
  schedule="0 * * * *",
  trigger_after_deploy=true
)
```

This creates:

- Same CDE resource with `contract_dq_spark.py`, `contract_config.json`, and orchestrator DAG
- **Spark child job**: `{contract-id}-spark-dq-job` (triggered by Airflow, not scheduled)
- **Airflow orchestrator job**: `{contract-id}-dq-job` with schedule; DAG ensures Iceberg
  `dq_audit_log` / `dq_audit_metric` tables then calls `CDEJobRunOperator`
- Spark job runs contract checks and writes audit log + Griffin metrics rows

Optional env: `CDE_SPARK_RUNTIME_IMAGE`, `CDE_SPARK_NUM_EXECUTORS`, `CDE_SPARK_EXECUTOR_MEMORY`,
`CDE_SPARK_DRIVER_MEMORY`, `CDE_SPARK_CONF` (JSON).

Monitor:

```
list_cde_job_runs(job_name="flights-contract-dq-job")
get_airflow_dag_run_status(job_name="flights-contract-dq-job")
```

**Schedule / duplicate runs:** Deploy leaves the CDE schedule **paused** by default.
Unpausing can immediately enqueue a missed-interval (backfill) run. Never combine
unpause with a manual trigger in the same step — that starts two runs within seconds.

- default deploy → schedule paused; unpause manually or pass `activate_schedule=true`
- `trigger_after_deploy=true` → one isolated manual run; schedule re-paused after trigger
- `trigger_cde_job` → same isolated manual run (schedule stays paused)
- All generated DAGs set `max_active_runs=1` to serialize overlapping DAG runs

## Differences from data-quality (WAP project)

| Feature | data-quality | This MCP server |
|---------|--------------|---------------|
| WAP / Iceberg branches | Yes | **No** |
| DataSketches baseline | Yes | No (contract-driven SQL) |
| Data ingest staging | Yes | No — checks live table |
| Deployment | Shell script | MCP tools + CDE CLI |
| Contract source | Airflow Variables | ODCS JSON / Atlas contracts |
| Schedule | `0 3 * * *` | Default `0 3 * * *` (configurable) |

## Development

```bash
uv sync
uv run pytest
uv run ruff check src tests
```

## License

Apache-2.0