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

MCP tools (13)

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
preview_contract_checks	Preview SQL checks before deploy
deploy_contract_dq_job	Generate + deploy CDE Airflow job
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:

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.

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

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}

Monitor:

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

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

uv sync
uv run pytest
uv run ruff check src tests

License

Apache-2.0