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许可证-有关详细信息，请参阅许可证文件。