dbt

@us-all/dbt-mcp

dbt MCP 服务——通过一个 stdio MCP 提供 manifest.json、run_results.json、sources.json、catalog.json，以及 DQ 结果表（BigQuery / Postgres）。基于 @us-all/mcp-toolkit 构建。

为 LLM 客户端提供一个只读窗口，深入查看你的 dbt 项目。不触发 dbt run——仅支持深度检查、运行历史分析、源新鲜度、按列测试覆盖率、血缘关系遍历，以及（如果你有自定义 DQ 结果表）历史检查趋势和 Tier SLA 状态。

如需触发 DAG、查看运行历史或日志尾部，请同时安装配套的 @us-all/airflow-mcp。

• 3 个类别（dbt、quality、meta）中共 27 个工具——21 个基础工具 + 5 个聚合工具 + 1 个元工具
• 4 个用于分诊工作流的 MCP 提示词
• 5 个聚合工具，可替代 3–5 次“list / get / list”往返调用
• 高容量读取支持 extractFields 响应投影
• 默认只读
• 混合后端：BigQuery（默认）或 Postgres，用于 DQ 结果表——二者均通过对等依赖实现延迟导入

安装

# 1. add the MCP server
pnpm add -D @us-all/dbt-mcp
# 2. add the DQ backend you actually use (only if you query custom DQ tables):
pnpm add -D @google-cloud/bigquery   # OR
pnpm add -D pg

运行

DBT_PROJECT_DIR=/path/to/dbt-project \
DQ_RESULTS_TABLE=my-project.data_ops.quality_checks \
npx @us-all/dbt-mcp

该服务使用 MCP stdio 通信；可将其接入 Claude Desktop / Cursor / 任意 MCP 客户端。设置 MCP_TRANSPORT=http 可启用 Streamable HTTP 传输（Bearer 身份验证，/health 端点）。

类别

类别	工具	用途
dbt	15 + 3 个聚合工具	解析 manifest.json / run_results.json / sources.json / catalog.json
quality	6 + 2 个聚合工具	查询 quality_checks 和 quality_score_daily（BQ 或 PG）；通过 dq-tier-by-source 按层级汇总
meta	1 个（始终启用）	使用 search-tools 以自然语言发现工具

使用 DBT_TOOLS=dbt（允许列表）或 DBT_DISABLE=quality（拒绝列表）进行切换。

工具一览

dbt（15 + 3）

dbt-list-models、dbt-get-model、dbt-list-tests、dbt-get-test、dbt-list-sources、dbt-get-source、dbt-list-exposures、dbt-list-macros、dbt-get-macro、dbt-list-runs、dbt-get-run-results、dbt-failed-tests、dbt-slow-models、dbt-coverage、dbt-graph、freshness-status、incident-context、dbt-sla-status

quality（6 + 2）

dq-list-checks、dq-get-check-history、dq-failed-checks-by-dataset、dq-score-trend、dq-tier-status、dq-tier-by-source、failed-tests-summary、dq-score-snapshot

提示词

提示词	使用场景
investigate-failed-tests	“过去 24 小时内出了什么问题？”
freshness-degradation-triage	“有没有数据源过期？”（可选：重点关注 Tier 1）
dq-trend-report	“给我一份面向利益相关者的 DQ 趋势报告”
incident-triage	“对 <model | source> 进行分诊”——汇总所有信号

环境变量

环境变量	必填	说明
DBT_PROJECT_DIR	是	dbt 项目根目录（dbt_project.yml 所在位置）
DBT_TARGET_DIR	否	默认为 $DBT_PROJECT_DIR/target
DBT_RUN_HISTORY_DIR	否	用于存档 run_results.json 历史记录的可选目录
DQ_BACKEND	否	bigquery（默认）或 postgres
DQ_RESULTS_TABLE	否	检查表的 FQN；仅基于检查的质量工具需要此项
DQ_SCORE_TABLE	否	每日评分表的 FQN；仅评分工具需要此项
GOOGLE_APPLICATION_CREDENTIALS	否	用于 BigQuery 后端（支持 ADC 回退）
BQ_PROJECT_ID	否	显式指定 BQ 项目（否则从 ADC 推断）
PG_CONNECTION_STRING	否	当 DQ_BACKEND=postgres 时使用（密钥）
DQ_SCHEMA	否	generic（默认）或 us-all — quality 类别的基础架构预设
DQ_COL_*	否	在 DQ_SCHEMA 基础上覆盖各列（见下文）。覆盖值必须是简单的 SQL 标识符。
DQ_TIER1_TARGET_PCT	否	未配置 tier 列时，dq-tier-status 的 Tier 1 SLA 阈值（默认值为 99.5）。若同时设置了 DBT_SLA_CONFIG_PATH 中的 tier_sla.1，则以其为准。
DBT_SLA_CONFIG_PATH	否	可选 YAML 路径，包含 tier_sla 和 dbt_sla 配置块。用于设置 dq-tier-status 阈值和 dq-tier-by-source 的各 Tier 目标值。会缓存修改时间。
DBT_ALLOW_WRITE	否	为未来的写入工具保留（目前没有）
DBT_TOOLS / DBT_DISABLE	否	类别开关

DQ 结果表架构类型

quality 类别通过 DQ_SCHEMA 支持两种架构预设：

DQ_SCHEMA=generic（默认）

假定 DQ_RESULTS_TABLE 中包含以下列：run_at、check_name、check_type、dataset、table_name、status、severity、failure_count、message。

假定 DQ_SCORE_TABLE 中包含以下列：score_date、scope、tier、completeness_pct、freshness_pct、validity_pct、anomaly_free_pct、overall_score。

dq-tier-status 按各 scope 行汇总 Tier 1/2/3。

DQ_SCHEMA=us-all

us-all 使用的真实架构（Postgres data_ops 数据库）：

quality_checks：run_date、check_type、dimension、source、target_name、status、metric_value、threshold、details (JSONB)。

quality_score_daily：run_date、completeness_pct、freshness_pct、validity_pct、anomaly_free_pct、overall_score、total_checks、failed_checks。

此类型中，quality_score_daily 每天只有一行（不按 scope 汇总，也没有 tier 列）。dq-tier-status 会回退为将当天的 overall_score 与 DQ_TIER1_TARGET_PCT 比较（默认值为 99.5）。

dq-get-check-history 要求 checkName 的格式为 '<check_type>:<target_name>'，因为 us-all 没有原生的 check_name 列。

按列覆盖 — DQ_COL_*

如果你的 DQ 表与任一预设都不匹配，可以在 DQ_SCHEMA 基础上添加按列覆盖。设置任何 DQ_COL_* 环境变量后，该变量都会替换对应单列的预设值。未设置的变量保留预设默认值。

覆盖值会验证为简单 SQL 标识符，以避免通过环境变量注入原始 SQL。DQ_RESULTS_TABLE / DQ_SCORE_TABLE 中的表名也会经过验证，并针对配置的后端加上引号。

环境变量	逻辑概念	通用预设	us-all 预设
DQ_COL_RUN_AT	检查表中的时间戳/日期	run_at	run_date
DQ_COL_CHECK_TYPE	检查类型/维度类别	check_type	check_type
DQ_COL_STATUS	通过/失败/警告/错误	status	status
DQ_COL_DATASET	数据集/来源/架构	dataset	source
DQ_COL_TABLE_NAME	表名或目标名称	table_name	target_name
DQ_COL_SEVERITY	严重性/维度	severity	dimension
DQ_COL_FAILURE_COUNT	数值型失败计数/指标	failure_count	metric_value
DQ_COL_MESSAGE	自由文本或 JSON 消息	message	details::text
DQ_COL_CHECK_NAME	检查项的自然标识符	check_name	(无)
DQ_COL_SCORE_DATE	评分表中的日期列	score_date	run_date
DQ_COL_SCOPE	评分表中的范围/租户列	scope	(无)
DQ_COL_TIER	评分表中的层级列	tier	(无)

对于三个可空列（DQ_COL_CHECK_NAME、DQ_COL_SCOPE、DQ_COL_TIER），将值设为 none / null / -，即可声明“无原生列”：

• 没有 check_name → 工具会通过 check_type || ':' || table_name 合成一个值。此时，dq-get-check-history 要求 checkName 的格式为 '<check_type>:<table_name>'。
• 没有 scope → dq-score-trend 的 scope 筛选器会被忽略（有一项注意事项），且 dq-tier-status 会切换到单一 overall_score 路径，并与 DQ_TIER1_TARGET_PCT 比较。
• 没有 tier → 同样回退到单一 overall_score 路径。

示例 — 通用预设用于列名恰好不同的 Postgres 架构：

DQ_SCHEMA=generic
DQ_COL_RUN_AT=checked_at
DQ_COL_DATASET=schema_name
DQ_COL_TABLE_NAME=tbl
DQ_COL_FAILURE_COUNT=fail_n
DQ_COL_CHECK_NAME=none      # synthesize from check_type+tbl
DQ_COL_SCOPE=none           # no per-team rollup
DQ_COL_TIER=none            # use DQ_TIER1_TARGET_PCT instead

SLA 配置（可选）— DBT_SLA_CONFIG_PATH

将 DBT_SLA_CONFIG_PATH 设为 YAML 文件，以便质量工具读取项目定义的层级目标和 DBT SLA。架构如下（忽略额外键）：

dbt_sla:
  test_pass_pct: 99.0          # consumed by dbt-sla-status (test pass rate threshold)
  freshness_pass_pct: 99.5     # consumed by dbt-sla-status (source freshness pass rate threshold)

tier_sla:
  1: 99.5                      # tier-1 overall_score / per-source pass-rate target
  2: 99.0
  3: 95.0

设置后，tier_sla 映射会驱动：

• dq-tier-status — 按层级汇总时，会将每一行的 overall_score 与对应目标进行比较。未提供此文件时，使用硬编码值 {1: 99.5, 2: 99.0, 3: 95.0}。
• dq-tier-by-source — 将每个来源的通过率与该来源层级的目标进行比较（从 dbt sources.yml meta.tier 解析）。
• dq-tier-status 无层级列路径（us-all 预设 / DQ_COL_TIER=none）— 使用 tier_sla.1 作为唯一目标。未设置 SLA 文件时，DQ_TIER1_TARGET_PCT 环境变量仍可作为回退值。

dbt_sla 区块会驱动：

• dbt-sla-status — 根据最新的 run_results.json 计算测试通过率，并根据 sources.json 计算新鲜度通过率，然后将各项与 dbt_sla.test_pass_pct / dbt_sla.freshness_pass_pct 进行比较。每项都会返回 passPct、target、meeting，并在字段或文件缺失时提供注意事项。

该文件会按修改时间缓存；工具调用之间的编辑会自动生效。

从 quality_checks 按层级汇总 — dq-tier-by-source

对于 quality_score_daily 每天只有一行（没有按范围/层级细分）的架构，dq-tier-by-source 会根据原始 quality_checks 行重建按层级划分的概况。有两种模式：

mode: "source"（默认）— 按来源／数据集列分组

适用于 quality_checks 中的每一行代表对一个来源组的检查，且数据集／来源列直接包含 dbt 来源组名称的情况。

1. 从 dbt 清单的 sources.<source>.<table>.meta.tier 构建 source_name -> tier 映射（每个来源组使用第一个表的 tier）。
2. 按数据集／来源列对 quality_checks 行分组，并计算日期或 sinceHours 时间范围内每个来源的通过率。
3. 查找每个来源的 tier 和目标值（来自 SLA 配置或默认值），并按 tier 报告达标／缺失情况。

mode: "table" — 按 table_name 列分组

适用于数据集／来源列是类别（bq / dbt / airflow），而实际 dbt 来源表标识符位于 table_name / target_name 列中，格式为 <source_group>.<table> 的情况。这在整合来自不同后端的信号的检查表中很常见。

1. 使用清单中每个来源条目的 source_name + name + meta.tier 构建 <source_group>.<table> -> tier 映射——自然地获取表级 tier 覆盖值。
2. 按 table_name 列对 quality_checks 行分组。当只有部分类别会生成可解析的目标名称时，可使用 sourceFilter（例如 sourceFilter: "bq"）进行预筛选。
3. 每个汇总键都会被解析为 <source_group>.<table>；不含 . 或键不在清单中的行会被放入 caveats[]。

未设置 tier 的行（没有清单 meta.tier）和无法解析的行始终会出现在 caveats[] 中，方便你为它们设置 tier，或接受这一缺口。

测试过的架构

• dbt 清单架构 v11 / v12 — 当前最高版本。dbt 1.7 会生成 v11；dbt 1.8 到 1.12 均会生成 v12（架构会在原有基础上以增量方式演进）。较新／未知版本仍可解析，但会有一条 caveats 提示标记这些版本。

配套服务

如需执行 Airflow DAG 操作（列表、运行、任务实例、日志尾部、触发、清除），请将 @us-all/airflow-mcp 与此服务一同安装。

构建

pnpm install
pnpm run build      # tsc → dist/
pnpm test           # vitest
pnpm run smoke      # spawns dist/index.js, calls initialize + tools/list (set env first)

许可证

MIT — 参见 LICENSE。