AI导哈

dbt项目管理多智能体系统

基于AWS Bedrock Agent Core构建的多智能体系统，提供智能化的dbt项目管理和分析功能，包括编译、模型分析和远程MCP服务器连接。

  • by StephenR-DBT
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 StephenR-DBT/dbt-aws-agent-starter

使用说明

dbt AWS代理核心多代理

使用AWS Bedrock agent Core构建的多代理系统，提供智能dbt项目管理和分析功能。

建筑

该项目使用三个专用工具实现了多代理架构：

  1. dbt编译工具 -本地dbt编译功能
  2. dbt模型分析器 -数据模型分析和建议
  3. dbt MCP服务器工具 -远程dbt MCP服务器连接

📋 先决条件

  • Python 3.10+
  • dbt CLI已安装并配置
  • dbt Fusion已安装
  • AWS代理核心设置

🛠️ 安装

  1. 克隆仓库:

    git clone <repository-url>
cd dbt-aws-agent

  2. 创建虚拟环境:

    python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

  3. 安装依赖项:

    pip install -r requirements.txt

  4. 设置环境变量:

    cp .env.example .env
# Edit .env with your configuration

  5. 跑:

    cd dbt_data_scientist
python agent.py

项目结构

dbt-aws-agent/
├── dbt_data_scientist/            # Main application package
│   ├── __init__.py               # Package initialization
│   ├── agent.py                  # Main agent with Bedrock Agent Core integration
│   ├── prompts.py                # Agent prompts and instructions
│   ├── test_all_tools.py         # Comprehensive test suite
│   ├── quick_mcp_test.py         # Quick MCP connectivity test
│   └── tools/                    # Tool implementations
│       ├── __init__.py
│       ├── dbt_compile.py        # Local dbt compilation tool
│       ├── dbt_mcp.py            # Remote dbt MCP server tool (translated from Google ADK)
│       └── dbt_model_analyzer.py # Data model analysis tool
├── requirements.txt              # Python dependencies
├── env.example                   # Environment configuration template
└── README.md                     # This documentation

工具概述

1.dbt编译工具(dbt_compile.py)

  • 目的：本地dbt项目编译和故障排除
  • 特性:
    • 跑 dbt compile --log-format json 本地
    • 解析JSON日志以进行结构化分析
    • 提供编译错误分析和建议
    • 路由到专门的dbt编译代理以进行智能响应

2.dbt模型分析工具(dbt_model_analyzer.py)

  • 目的：数据模型分析和建议
  • 特性:
    • 分析模型结构和依赖关系
    • 评估数据质量模式和测试覆盖率
    • 审查dbt最佳实践的遵守情况
    • 提供优化建议
    • 生成模型文档建议

3.dbt MCP服务器工具(dbt_mcp.py)

  • 目的：使用AWS Bedrock代理核心进行远程dbt MCP服务器连接
  • 特性:
    • 使用可流式传输的HTTP客户端连接到远程dbt MCP服务器
    • 支持带标头的dbt Cloud身份验证
    • 动态列出可用的MCP工具
    • 执行dbt MCP工具功能
    • 为适当的工具提供智能查询路由
    • 内置连接测试和错误处理

### 3. Test the Setup

Before running the full application, test that everything is working:

```bash
# Quick MCP test
python dbt_data_scientist/quick_mcp_test.py

# Full test suite
python dbt_data_scientist/test_all_tools.py

4.运行应用程序

对于AWS基岩代理核心：

python -m dbt_data_scientist.agent

对于本地测试：

python -m dbt_data_scientist.agent

使用示例

dbt编译工具

> "Compile my dbt project and find any issues"
> "What's wrong with my models in the staging folder?"

dbt模型分析工具

> "Analyze my data modeling approach for best practices"
> "Review the dependencies in my dbt project"
> "Check the data quality patterns in my models"

dbt MCP服务器工具

> "List all available dbt MCP tools"
> "Show me the catalog from dbt Cloud"
> "Run my models in dbt Cloud"
> "What tests are available in my dbt project?"

测试

该项目包括全面的测试功能，以验证所有组件是否正常工作。

快速测试

仅测试MCP连接

python dbt_data_scientist/quick_mcp_test.py
  • 快速、最小限度地测试MCP连接
  • 验证环境变量和连接
  • 列出可用的MCP工具

直接测试MCP工具

python dbt_data_scientist/tools/dbt_mcp.py
  • 直接测试MCP模块
  • 内置连接测试
  • 显示详细的错误消息

综合测试

完整测试套件

python dbt_data_scientist/test_all_tools.py
  • 单独测试所有工具
  • 验证代理初始化
  • 测试工具集成
  • 全面的错误报告

哪些测试可以验证

  1. 环境变量 -所有必需的变量都已设置
  2. 工具进口 -所有工具都可以成功导入
  3. 代理初始化 -代理加载所有工具
  4. 单个工具测试 -每个工具都能正确执行
  5. 代理集成 -工具在代理中协同工作
  6. MCP连接 -远程MCP服务器连接正常

测试输出示例

🚀 Complete Tool and Agent Test Suite
==================================================
🔧 Testing Environment Setup
------------------------------
  ✅ DBT_MCP_URL: https://your-mcp-server.com
  ✅ DBT_TOKEN: ****************
  ✅ DBT_USER_ID: your_user_id
  ✅ DBT_PROD_ENV_ID: your_env_id
  ✅ Environment setup complete!

📦 Testing Tool Imports
------------------------------
  ✅ All tools imported successfully
  ✅ dbt_compile is callable
  ✅ dbt_mcp_tool is callable
  ✅ dbt_model_analyzer_agent is callable

... (more tests)

🎉 All tests passed! Your agent and tools are working correctly.

主要特点

🔄 智能路由

主代理根据关键字和上下文自动将查询路由到适当的专用工具。

🌐 MCP服务器集成

通过适当的身份验证和错误处理与远程dbt MCP服务器无缝连接。

📊 综合分析

多方面分析，包括汇编、建模最佳实践和数据质量评估。

⚡ 异步支持

对MCP操作的完全异步/等待支持，同时保持与Bedrock Agent Core的兼容性。

🛡️ 错误处理

所有工具操作的强大错误处理和回退机制。

发展

添加新工具

  1. 在中创建新的工具文件 dbt_data_scientist/tools/
  2. 使用 @tool 线绳装饰师
  3. 将该工具添加到中的主代理工具列表中 agent.py
  4. 更新主代理系统提示中的路由逻辑

故障排除

测试优先

在排除故障之前，运行测试套件以识别问题：

# Quick test for MCP issues
python dbt_data_scientist/quick_mcp_test.py

# Comprehensive test for all issues
python dbt_data_scientist/test_all_tools.py

常见问题

  1. MCP连接失败

    • 跑 python dbt_data_scientist/quick_mcp_test.py 诊断
    • 验证 DBT_MCP_URL 是正确的
    • 检查身份验证标头
    • 确保dbt MCP服务器可访问
    • 检查网络连接

  2. dbt编译错误

    • 验证 DBT_PROJECT_LOCATION 路径存在
    • 检查 DBT_EXECUTABLE 在PATH中
    • 确保dbt项目有效
    • 跑 dbt compile 手动测试

  3. 环境变量问题

    • 复制 env.example 到 .env
    • 验证是否设置了所有必需的变量
    • 检查变量值是否正确
    • 使用测试套件验证配置

  4. 代理初始化问题

    • 检查是否可以导入所有工具
    • 验证MCP服务器是否可访问
    • 确保安装了所有依赖项
    • 运行单个工具测试

调试模式

要进行详细的调试，您可以运行单个组件：

# Test MCP tool directly
python dbt_data_scientist/tools/dbt_mcp.py

# Test individual tools
python -c "from dbt_data_scientist.tools import dbt_compile; print(dbt_compile('test'))"

贡献

  1. 复刻仓库
  2. 创建要素分支
  3. 进行更改
  4. 添加新功能的测试
  5. 提交拉取请求

许可证

此项目根据MIT许可证获得许可-有关详细信息，请参阅许可证文件。

相关 MCP