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 (CDE deployment, CDW connection, SQL operators) and 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

Related MCP server: aegis-dq

MCP tools (14)

Configuration

Copy .env.example to .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
DEFAULT_DAG_SCHEDULE=0 3 * * *

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.

{
  "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. 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.

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:

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):

{
  "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:

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:

{
  "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 operators per Astronomer's SQL data quality guide:

Column check types (SQLColumnCheckOperator)

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

Table checks (SQLTableCheckOperator)

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

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 activates the DAG schedule (unpause) by default. Do not combine unpause with trigger_after_deploy=true — that starts two runs (one scheduled/backfill + one manual). Use either:

• default deploy → schedule enabled, waits for next cron tick

• trigger_after_deploy=true → one manual run, schedule stays paused

Differences from data-quality (WAP project)

Development

uv sync
uv run pytest
uv run ruff check src tests

License

Apache-2.0