AI导哈

DBT清单分析服务器

一个用于分析DBT清单的FastMCP服务器，支持自动模式版本检测和血缘追踪，适用于数据工程和数据分析场景。

  • by dabblefish-solutions
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 dabblefish-solutions/dabblefish__mcp__dbt-core-tools

使用说明

DBT清单MCP服务器

用于分析DBT清单的FastMCP服务器，具有自动模式版本检测和沿袭跟踪功能。

作者：大卫·B 公司：浅水鱼有限责任公司 许可证：MIT

特性

  • 自动模式版本检测：支持DBT清单架构版本v0-v12
  • 版本自适应解析：与旧版清单格式向后兼容
  • SQLite数据库存储：高效的查询和数据持久性
  • 谱系分析：上游和下游依赖性跟踪
  • 模型信息：详细的模型元数据和编译代码访问
  • 符合PEP-8标准：专业Python包结构

安装

来自PyPI（发布时）

pip install dbt-manifest-mcp

来源

git clone https://github.com/dabblefish/dbt-manifest-mcp.git
cd dbt-manifest-mcp
pip install -e .

开发安装

git clone https://github.com/dabblefish/dbt-manifest-mcp.git
cd dbt-manifest-mcp
pip install -e ".[dev]"

用法

运行服务器

# Using the installed command
dbt-manifest-mcp

# Or using Python module
python -m dbt_manifest_mcp.server

环境变量

  • DBT_MANIFEST_PATH：DBT manifest.json文件的路径（必填）
  • DBT_DB_PATH：SQLite数据库文件的路径（可选，默认为./dbt_manifest.db）

示例

export DBT_MANIFEST_PATH="/path/to/your/manifest.json"
export DBT_DB_PATH="./dbt_manifest.db"
dbt-manifest-mcp

可用工具

1.刷新_修改

通过解析和存储在SQLite数据库中来刷新DBT清单数据。

参数：

  • manifest_path （可选）：DBT manifest.json文件的路径

退货： 带有统计信息的成功消息

2.get_upstream_lineage

获取DBT模型的上游沿袭。

参数：

  • model_id：DBT模型的唯一ID（例如，“model.my_project.my_model”）

退货： 包含model_id、upstream_models列表和计数的字典

3.get_downstream_lineage

获取DBT模型的下游沿袭。

参数：

  • model_id：DBT模型的唯一ID（例如，“model.my_project.my_model”）

退货： 包含model_id、downstream_models列表和计数的字典

4.获取模式信息

获取DBT模型的详细信息，包括父/子计数和编译代码。

参数：

  • model_id：DBT模型的唯一ID（例如，“model.my_project.my_model”）

退货： 包含详细模型信息的词典

5.get_schema_info

获取有关加载的DBT清单架构版本、支持的功能和数据库统计信息的信息。

退货： 包含版本信息、功能和统计信息的词典

架构版本支持

服务器会自动检测并适应不同的DBT清单模式版本：

  • v0-v3：具有基本节点结构的传统格式
  • v4+：具有parent_map和child_map的现代格式
  • v12：具有增强元数据的最新格式

版本特定功能

版本 父映射 子映射 节点结构 元数据位置
v0-v3 ❌ （基于依赖关系构建） ❌ （基于依赖关系构建） 遗留 根
v4-v11 ✅ ✅ 现代 元数据
v12 ✅ ✅ 现代 元数据

数据库模式

服务器创建以下SQLite表：

  • metadata：架构版本和清单元数据
  • nodes：DBT模型、测试和其他节点
  • sources：DBT源定义
  • macros：DBT宏定义
  • parent_map：亲子关系
  • child_map：亲子关系

示例用法

# After starting the server, you can use the tools via MCP client

# Refresh manifest data
refresh_manifest("/path/to/manifest.json")

# Get upstream dependencies
upstream = get_upstream_lineage("model.my_project.customer_orders")

# Get downstream dependencies
downstream = get_downstream_lineage("model.my_project.raw_customers")

# Get detailed model information
model_info = get_model_info("model.my_project.customer_summary")

# Get schema version information
schema_info = get_schema_info()

错误处理

该服务器包括全面的错误处理功能，用于：

  • 清单文件缺失或无效
  • 不支持的架构版本
  • 数据库连接问题
  • 型号ID无效

许可证

MIT许可证-有关详细信息，请参阅许可证文件。

相关 MCP