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 
   cd dbt-aws-agent

1. 创建虚拟环境:

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

1. 安装依赖项:

   pip install -r requirements.txt

1. 设置环境变量:

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

1. 跑:

   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:

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. **环境变量** -所有必需的变量都已设置
1. **工具进口** -所有工具都可以成功导入
1. **代理初始化** -代理加载所有工具
1. **单个工具测试** -每个工具都能正确执行
1. **代理集成** -工具在代理中协同工作
1. **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/`
1. 使用 `@tool` 线绳装饰师
1. 将该工具添加到中的主代理工具列表中 `agent.py`
1. 更新主代理系统提示中的路由逻辑

## 故障排除

### 测试优先

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

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服务器可访问
   - 检查网络连接

1. **dbt编译错误**

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

1. **环境变量问题**

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

1. **代理初始化问题**

   - 检查是否可以导入所有工具
   - 验证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. 复刻仓库
1. 创建要素分支
1. 进行更改
1. 添加新功能的测试
1. 提交拉取请求

## 许可证

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