cdq mcp
轻量级MCP服务器，使用FastMCP进行HTTP传输。运行SQL，添加规则，并创建DQ作业（100%REST API）。
工具
| 工具 | 输入 | 描述 |
|---|---|---|
| run_dq_job | 数据集（str）、run_id（str）和sql（str） | 注册数据集并运行dq作业 |
| get_recent_runs | none | 获取最近运行的DQ作业 |
| get_jobs_in_queue | 限制（int，默认值10），状态（str） | 获取当前在队列中的作业 |
| get_hoot_results | 数据集（str），run_id（str） | 获取特定运行的DQ结果 |
| run_sql | sql（str） | 执行sql查询 |
| search_catalog | search_string（str），limit（int，默认值50） | 在目录中搜索数据集 |
| get_rules_by_dataset | 数据集（str） | 获取数据集的DQ规则 |
| save_rule | 数据集、rule_name、rule_sql、点（1）、perc（1） | 创建新的DQ规则 |
| get_datat_definition | 数据集（str） | 获取数据集配置 |
| get_alerts | 数据集（str） | 获取数据集的警报 |
| save_alert | 数据集、alert_name、条件、电子邮件、消息 | 创建警报 |
| get_todays_date | none | 获取今天的日期（用于运行ID） |
示例用法
CLI（双子座，克劳德）
OpenAI（通过隧道）
IDE（光标、风帆）
私人模特
设置
创建 .env 文件中包含您的Collibra DQ凭据：
DQ_URL=https://your-collibra-dq-instance.com
DQ_USERNAME=your_username
DQ_PASSWORD=your_password
DQ_ISS=your_tenant
DQ_CXN=BIGQUERY创建虚拟环境并安装依赖关系：
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt虚拟环境
# Activate
source venv/bin/activate
# Deactivate
deactivate服务器
# Start (foreground)
./start.sh
# Or manually
source venv/bin/activate
python server.py
# Start (background)
./start.sh &
# Stop
pkill -f "python server.py"服务器运行于 http://0.0.0.0:8765/mcp
码头工人
塑造形象：
docker build -t cdq-mcp .使用env文件运行：
docker run --rm -p 8765:8765 --env-file .env cdq-mcp或者直接传递env变量：
docker run --rm -p 8765:8765 \
-e DQ_URL=https://your-server \
-e DQ_USERNAME=user \
-e DQ_PASSWORD=pass \
-e DQ_ISS=your-iss \
-e DQ_CXN=BIGQUERY \
cdq-mcp在后台运行：
docker run -d --name cdq-mcp -p 8765:8765 --env-file .env cdq-mcp
# Stop
docker stop cdq-mcp && docker rm cdq-mcp预制集装箱（2026-01-28）
docker pull brianmearns162/cdq-mcp:latest
docker run --rm -p 8765:8765 --env-file .env brianmearns162/cdq-mcp:latest与检查员一起测试（建议）
使用MCP检查器与工具交互，检查输入和输出。
npx @modelcontextprotocol/inspector
卷曲测试
MCP使用基于会话的协议。按照以下步骤测试服务器。
步骤1：初始化会话并获取会话ID
curl -i -X POST http://localhost:8765/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"curl-test","version":"1.0"}}}'寻找 mcp-session-id 在响应标头中：
mcp-session-id: abc123def456...步骤2：列出可用工具
使用步骤1中的会话ID：
curl -X POST http://localhost:8765/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-H "mcp-session-id: YOUR_SESSION_ID" \
-d '{"jsonrpc":"2.0","id":2,"method":"tools/list"}'步骤3：调用工具
curl -X POST http://localhost:8765/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-H "mcp-session-id: YOUR_SESSION_ID" \
-d '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"get_recent_runs","arguments":{}}}'快速健康检查（无需会话）
验证服务器是否响应MCP协议：
curl -s http://localhost:8765/mcp \
-H "Accept: text/event-stream"
# Should return: {"jsonrpc":"2.0",...,"message":"Bad Request: Missing session ID"}MCP客户端配置
光标
添加到 ~/.cursor/mcp.json:
{
"mcpServers": {
"cdq-mcp": {
"url": "http://localhost:8765/mcp",
"name": "Collibra Data Quality MCP"
}
}
}Claude CLI
添加到 ~/.claude/settings.json:
{
"mcpServers": {
"cdq-mcp": {
"url": "http://localhost:8765/mcp"
}
}
}克劳德桌面版
添加到 ~/Library/Application Support/Claude/claude_desktop_config.json （macOS）或 %APPDATA%\Claude\claude_desktop_config.json （Windows）：
{
"mcpServers": {
"cdq-mcp": {
"url": "http://localhost:8765/mcp"
}
}
}添加工具
@mcp.tool()
def my_tool(param: str) -> str:
"""Tool description"""
return f"Result: {param}"SQL交互
SQL交互使用配置连接的本机SQL方言。当与 下推 连接类型，其中查询直接在数据源上执行，而不是将数据拉入DQ引擎。
已知限制
- 可能需要进行调整，代码可能因用例和实例配置而异
- 用于便携式/个人MCP交互
- 这是一个非官方的参考服务器
- 目前一次支持1个连接（cxn）
- 规则类型仅限于自由形式（SQLF）
- 规则保存为“绝对”评分（不是百分比/阈值）
- SQL查询将运行与Web配置相同的限制
- 旨在作为参考框架 *经典* CDQ