内容
# Teradata ETL MCP 扩展
统一的模型上下文协议（MCP）服务器，用于全面的 ELT/ETL 操作，集成 Teradata、Airbyte、Apache Airflow 和 dbt，用于端到端的数据管道管理。
## 目录
- [快速入门](#快速入门)
- [功能](#功能)
- [架构](#架构)
- [安装](#安装)
- [SSH 设置](#ssh-设置双向)
- [配置](#配置)
- [连接配置文件](#连接配置文件)
- [用法](#用法)
- [工具目录](#工具目录)
- [开发](#开发)
- [项目结构](#项目结构)
- [故障排除](#故障排除)
- [贡献](#贡献)
- [文档](#文档)
- [许可证](#许可证)
---
## 快速入门
```bash
# 1. 克隆并安装
git clone https://github.com/Teradata/teradata-etl-mcp-server.git
cd teradata-etl-mcp-server
pip install -e ".[dev,all]"
# 2. 在源代码仓库外部创建工作空间文件夹
mkdir ../teradata-etl-mcp-test && cd ../teradata-etl-mcp-test
cp ../teradata-etl-mcp-server/.env.example .env
cp ../teradata-etl-mcp-server/connections.yaml.example connections.yaml
# 3. 编辑 .env — 仅需 Teradata 凭证即可启动
# 必需：TERADATA_HOST, TERADATA_USERNAME, TERADATA_PASSWORD
# 4. 编辑 connections.yaml — 更新源的 hosts/凭证
# 5. 配置 MCP 客户端 (.vscode/mcp.json 或 claude_desktop_config.json)
# { "servers": { "etl-mcp": { "command": "etl-mcp-server",
# "args": ["--env-file", "/absolute/path/to/teradata-etl-mcp-test/.env"] } } }
# 6. 启动服务器
etl-mcp-server --env-file .env
```
> **最低要求**：Python 3.10+，Teradata 主机和 MCP 客户端（Claude Desktop 或 VS Code with Copilot 扩展）。Airflow、Airbyte 和 dbt 均为可选。
---
## 功能
服务器暴露了 **22 个 MCP 工具**，涵盖 7 个类别。每个工具都是一个路由器，接受一个 `action` 或 `method` 参数 — 一个工具名称涵盖多个操作。
| 类别 | 工具 | 描述 |
|----------|-------|-------------|
| 管道管理 | 5 | 部署 DAG，控制调度，管理 Airflow 连接，验证 DAG |
| 流程和执行 | 3 | 触发 DAG 运行，监控状态，重试任务，获取日志 |
| 数据移动 | 5 | Airbyte 管道，同步，流选择，TdLoad/CSV DAG 生成 |
| dbt 管理 | 5 | 运行/测试/构建模型，生成文档，生成模型元数据 |
| 元数据发现 | 2 | 发现表，描述模式，数据分析，比较结构 |
| 连接配置文件 | 1 | 列出和重新加载凭证配置文件（LLM 永远不会看到密钥） |
| TTU 管理 | 1 | 通过 teradatasql 执行 DDL，加载数据通过 tdload，运行 BTEQ 脚本，检查 TTU 安装 |
### 安全：凭证隔离
LLM **永远不会** 看到密码、令牌或 API 密钥。所有凭证都在服务器端从 `connections.yaml` 配置文件中解析。LLM 只引用配置文件名称：
```
用户：“从 Postgres 到 Teradata 的每日 ELT 管道，用于 customers 表”
LLM 调用：create_intelligent_airbyte_pipeline(
source_profile="my_postgres", # 只是名称
destination_profile="prod_teradata", # 只是名称
...
)
服务器：从 connections.yaml 解析凭证，创建管道
响应：清理 — LLM 看到成功状态但没有密码
```
---
## 架构
```
+---------------------------------------------------------------+
| MCP 服务器层 |
| +----------------------------------------------------------+ |
| | 22 个工具（7 个类别）通过 FastMCP | |
| +----------------------------------------------------------+ |
+---------------------------------------------------------------+
|
+---------------------------------------------------------------+
| 管道编排器 |
| +-------------------+ +-------------+ +--------------------+ |
| | 凭证 | | 智能引擎 | | 代码生成器 | |
| | 解析器 | | | | （DAG，dbt，TPT， | |
| | （connections.yaml）| | | | BTEQ，TdLoad） | |
| +-------------------+ +-------------+ +--------------------+ |
| +-----------------+ +---------------+ +--------------------+ |
| | 响应 | | 验证器 | | 元数据存储 | |
| | 清理器 | | 和实用工具 | | （SQLite/JSON） | |
| +-----------------+ +---------------+ +--------------------+ |
+---------------------------------------------------------------+
|
+---------------------------------------------------------------+
| 客户端层 |
| +----------+ +----------+ +----------+ +----------+ |
| | Teradata | | Airflow | | Airbyte | | dbt | |
| | 客户端 | | 客户端 | | 客户端 | | 客户端 | |
| +----------+ +----------+ +----------+ +----------+ |
+---------------------------------------------------------------+
|
+---------------------------------------------------------------+
| 外部系统 |
| +----------+ +----------+ +----------+ +----------+ |
| | Teradata | | Airflow | | Airbyte | | dbt | |
| | 数据库 | | 服务器 | | 服务器 | | 项目 | |
| +----------+ +----------+ +----------+ +----------+ |
+---------------------------------------------------------------+
```
### 关键组件
| 组件 | 描述 |
|-----------|-------------|
| **FastMCP 服务器** | MCP 协议层，向 LLM 客户端公开所有工具 |
| **管道编排器** | 中央协调器；延迟加载客户端通过 `@property` |
| **凭证解析器** | 加载 `connections.yaml`，解析 `${ENV_VAR}` 插值，按名称提供配置文件 |
| **响应清理器** | 深度克隆并屏蔽所有工具响应中的敏感键（密码，令牌，密钥等） |
| **智能引擎** | AI 驱动的传输方法选择（Airbyte vs TPT） |
| **代码生成器** | 基于 Jinja2 的 Airflow DAG，dbt 模型，TPT 脚本，BTEQ 查询生成器 |
| **客户端** | Teradata，Airflow，Airbyte 和 dbt API 的抽象层 |
| **元数据存储** | 执行历史的可选持久性（SQLite 或 JSON） |
| **插件管理器** | 自定义运算符和验证器的可扩展性框架 |
---
## 安装
> **受众**：希望运行 MCP 服务器并将其与 LLM 客户端（Copilot，Claude Desktop，Claude Code 等）一起使用的终端用户。
### 先决条件
| 要求 | 版本 | 注意 |
|-------------|---------|-------|
| Python | 3.10 -- 3.13 | 必需 |
| Teradata 数据库 | 任何支持的版本 | Teradata 操作必需 |
| Teradata 工具和实用程序（TTU） | 17.20+ | MCP 客户端上 BTEQ/TdLoad/TPT 执行必需 |
| OpenSSH 客户端 | 任意 | MCP 客户端上 DAG 部署到 Airflow 必需 |
| OpenSSH 服务器 | 任意 | 如果 Airflow 通过 SSH 在远程执行 BTEQ/TdLoad，则 MCP 客户端必需 |
| Apache Airflow | 2.x | 可选 — DAG 编排需要 |
| Airbyte | OSS | 可选 — 数据复制需要 |
| dbt + dbt-teradata | >=1.7,<2.0 + 0.19.0+ | 可选 — 转换需要 |
### 步骤
```bash
# 1. 克隆仓库
git clone https://github.com/Teradata/teradata-etl-mcp-server.git
cd teradata-etl-mcp-server
# 2. 创建并激活虚拟环境
python -m venv .venv
# Windows
.venv\Scripts\activate
# Linux/macOS
source .venv/bin/activate
# 3. 安装带有所有额外功能的软件包（包括所有可选依赖项）
pip install -e ".[dev,all]"
```
### 安装后设置
> 在单独的工作空间文件夹中创建 **配置文件** — 不要将 `.env` 或 `connections.yaml` 放在源代码仓库中（它受到 pre-commit 钩子的保护，阻止 `.env` 提交）。
```bash
# 5. 在源代码仓库外部创建专用工作空间文件夹
mkdir ../teradata-etl-mcp-test
cd ../teradata-etl-mcp-test
# 6. 从源代码仓库复制模板
cp ../teradata-etl-mcp-server/.env.example .env
cp ../teradata-etl-mcp-server/connections.yaml.example connections.yaml
# 7. 编辑 .env 文件，包含 Teradata、Airflow、Airbyte 和 dbt 设置
# 8. 编辑 connections.yaml 文件，包含连接配置文件（参见连接配置文件部分）
```
### 验证安装
```bash
# 启动服务器（stdio 传输，默认）
python -m elt_mcp_server
```
### SSH 设置（双向）
对于 DAG 部署到 Airflow 和运行时 BTEQ/TdLoad/TPT 执行，需要在 MCP 客户端和 Airflow 服务器之间建立双向 SSH 连接。
**完整的 SSH 设置指南：** 参见 [SSH-SETUP.md](SSH-SETUP.md)，获取详细的平台特定说明（Windows、Linux、macOS）。
**涵盖的关键部分：**
- 安装 SSH 客户端和服务器
- 生成 Ed25519 密钥对（双向）
- 在两台机器上授权密钥
- 文件权限配置
- SSH 配置文件设置
- 验证检查清单
**快速参考 - 环境变量：**
```bash
# MCP 客户端 → Airflow（DAG 部署）
AIRFLOW_REMOTE_HOST=<airflow-host>
AIRFLOW_REMOTE_USER=airflow
AIRFLOW_REMOTE_SSH_KEY=~/.ssh/id_ed25519_airflow
# Airflow → MCP 客户端（运行时执行）
MCP_CLIENT_SSH_HOST=<your-machine-ip>
MCP_CLIENT_SSH_USER=<your-username>
MCP_CLIENT_SSH_KEY_PATH=~/.ssh/id_ed25519_mcp
```
**主机密钥验证：** DAG 部署工具通过 `strict_host_key_checking` 参数支持严格的 SSH 主机密钥验证。参见 [SSH-SETUP.md § SSH 主机密钥验证](SSH-SETUP.md)，获取安全最佳实践。
---
## 配置
> **SSH 主机密钥验证** 用于 DAG 部署，通过调用中的 `strict_host_key_checking` 工具参数控制（默认为 `False`，并在每次连接时记录警告）。参见 [SSH-SETUP.md § SSH 主机密钥验证](SSH-SETUP.md)，获取威胁模型和升级步骤。
### 环境变量（`.env`）
复制模板并填写您的值：
```bash
cp .env.example .env
```
`.env.example` 中的关键部分：
| 部分 | 变量 | 描述 | 是否必需 |
|---------|----------|-------------|-----------|
| **环境** | `ENVIRONMENT` | 运行时环境：`development`，`staging`，`production` | 否（默认为 `development`） |
| **Teradata** | `TERADATA_HOST` | Teradata 数据库主机或 IP 地址 | 是 |
| | `TERADATA_USERNAME` | Teradata 登录用户名 | 是 |
| | `TERADATA_PASSWORD` | Teradata 登录密码 | 是 |
| | `TERADATA_DATABASE` | 默认数据库/模式 | 否 |
| | `TERADATA_PORT` | 数据库端口 | 否（默认为 `1025`） |
| | `TERADATA_LOGMECH` | 认证机制：`TD2`，`LDAP`，`JWT`，`BEARER`，`SECRET` | 否（默认为 `TD2`） |
| **Teradata 到 Teradata** | `TERADATA_SOURCE_HOST` | 源 Teradata 主机（用于跨系统传输） | 否 |
| | `TERADATA_SOURCE_USERNAME` | 源 Teradata 用户名 | 否 |
| | `TERADATA_SOURCE_PASSWORD` | 源 Teradata 密码 | 否 |
| | `TERADATA_SOURCE_DATABASE` | 源 Teradata 数据库 | 否 |
| | `TERADATA_TARGET_HOST` | 目标 Teradata 主机 | 否 |
| | `TERADATA_TARGET_USERNAME` | 目标 Teradata 用户名 | 否 |
| | `TERADATA_TARGET_PASSWORD` | 目标 Teradata 密码 | 否 |
| | `TERADATA_TARGET_DATABASE` | 目标 Teradata 数据库 | 否 |
| **Airflow API** | `AIRFLOW_BASE_URL` | Airflow REST API URL（例如 `http://localhost:8080`） | 用于编排 |
| | `AIRFLOW_USERNAME` | Airflow API 用户名 | 用于编排 |
| | `AIRFLOW_PASSWORD` | Airflow API 密码 | 用于编排 |
| | `AIRFLOW_TOKEN_ENDPOINT` | JWT 令牌端点 | 否（默认为 `/auth/token`） |
| | `AIRFLOW_ACCESS_TOKEN` | 预配置的 Bearer 令牌 | 否 |
| **Airflow DAG 部署** | `AIRFLOW_REMOTE_HOST` | 用于 SSH DAG 部署的 Airflow 服务器主机名 | 用于 DAG 部署 |
| | `AIRFLOW_REMOTE_USER` | Airflow 服务器上的 SSH 用户名 | 用于 DAG 部署 |
| | `AIRFLOW_REMOTE_SSH_KEY` | SSH 私钥路径（在该机器上） | 用于 DAG 部署 |
| | `AIRFLOW_REMOTE_PASSWORD` | SSH 密码（如果不使用密钥身份验证） | 否 |
| | `AIRFLOW_REMOTE_PORT` | Airflow 服务器上的 SSH 端口 | 否（默认为 `22`） |
| | `AIRFLOW_REMOTE_SSH_KEY_PASSPHRASE` | SSH 密钥的通行短语 | 否 |
| | `AIRFLOW_DAG_FOLDER` | Airflow 服务器上的远程 DAG 文件夹路径 | 否（默认为 `/opt/airflow/dags`） |
| **MCP 客户端 SSH** | `MCP_CLIENT_SSH_HOST` | 此机器的主机名/IP（Airflow SSH 返回此处运行时） | 用于运行时 SSH |
| | `MCP_CLIENT_SSH_USER` | 此机器上的 SSH 用户名 | 用于运行时 SSH |
| | `MCP_CLIENT_SSH_PORT` | 此机器上的 SSH 端口 | 否（默认为 `22`） |
| | `MCP_CLIENT_SSH_KEY_PATH` | Airflow 工作节点上的 SSH 私钥路径 | 用于运行时 SSH |
| **Airbyte** | `AIRBYTE_ENABLED` | 启用 Airbyte 集成 | 否（默认为 `false`） |
| | `AIRBYTE_BASE_URL` | Airbyte API 基本 URL | 当 Airbyte 启用时 |
| | `AIRBYTE_CLIENT_ID` | OAuth2 客户端 ID（来自 Airbyte 设置 > 应用程序） | 否 |
| | `AIRBYTE_CLIENT_SECRET` | OAuth2 客户端密钥 | 否 |
| | `AIRBYTE_TOKEN_URL` | OAuth2 令牌端点 | 否 |
| | `AIRBYTE_WORKSPACE_ID` | 默认工作空间 ID（如果省略则自动检测） | 否 |
| | `AIRBYTE_DEFAULT_NAMESPACE` | 连接的默认命名空间 | 否（默认为 `default`） |
| **dbt** | `DBT_PROJECT_DIR` | dbt 项目目录路径 | 用于 dbt |
| | `DBT_PROFILES_DIR` | dbt 配置文件路径 | 否（默认为 `~/.dbt`） |
| | `DBT_TARGET` | dbt 目标环境 | 否（默认为 `dev`） |
| | `DBT_THREADS` | dbt 执行的线程数 | 否（默认为 `4`） |
| **管道** | `PIPELINE_DAGS_OUTPUT_DIR` | 生成的 DAG 文件目录 | 否（默认为 `./airflow_dags`） |
| | `PIPELINE_DEFAULT_SCHEDULE_INTERVAL` | 生成的 DAG 的默认计划 | 否（默认为 `@daily`） |
| **MCP 服务器** | `MCP_LOG_LEVEL` | 日志级别：`DEBUG`，`INFO`，`WARNING`，`ERROR`，`CRITICAL` | 否（默认为 `INFO`） |
| | `MCP_LOG_FILE` | 日志文件路径 | 否（默认为 `./logs/etl-mcp-server.log`) |
| | `MCP_FAIL_FAST_ON_STARTUP` | 启动时连接失败时崩溃 | 否（默认为 `false`） |
| **TTU** | `TTU_ENABLED` | 启用本地 TPT/BTEQ/TdLoad 执行 | 否（默认为 `false`） |
| | `TTU_TTU_VERSION` | TTU 版本（例如 `17.20`）；如果未设置则自动检测 | 否 |
| | `TTU_TPT_BINARY_PATH` | `tbuild` 二进制路径（从版本自动检测） | 否 |
| | `TTU_BTEQ_BINARY_PATH` | `bteq` 二进制路径（从版本自动检测） | 否 |
| | `TTU_TDLOAD_BINARY_PATH` | `tdload` 二进制路径（从版本自动检测） | 否 |
| | `TTU_SCRIPTS_DIR` | 生成的 TTU 脚本目录 | 否（默认为 `./ttu_scripts`） |
| | `TTU_COMMAND_TIMEOUT` | 子进程超时（秒） | 否（默认为 `600`） |
| **安全** | `SECURITY_CONNECTIONS_FILE` | `connections.yaml` 的路径，用于凭证配置文件 | 否（自动发现） |
---
## 连接配置文件
连接配置文件将凭证与 LLM 交互分离。LLM 按 **名称** 引用配置文件；服务器在运行时解析实际凭证。
### 设置
```bash
cp connections.yaml.example connections.yaml
# 编辑 connections.yaml 以包含实际凭证
```
### 文件位置（按顺序搜索）
1. 通过 `CONNECTIONS_FILE` 环境变量设置的路径
2. 当前工作目录中的 `connections.yaml`
3. `settings.security.connections_file`（如果在服务器设置中配置）
### 格式
```yaml
version: "1"
profiles:
postgres_prod:
host: "pg-host.example.com"
port: 5432
database: "testdb"
username: "testuser"
password: "${POSTGRES_PASSWORD}" # 环境变量插值
schemas:
- "public"
description: "生产 Postgres 数据库"
teradata_prod:
host: "td-host.example.com"
port: 1025
username: "dbc"
password: "${TERADATA_PASSWORD}"
default_schema: "analytics_raw"
description: "生产 Teradata 目标"
airflow_ssh:
host: ${MCP_CLIENT_SSH_HOST}
port: ${MCP_CLIENT_SSH_PORT}
username: ${MCP_CLIENT_SSH_USER}
key_file: ${MCP_CLIENT_SSH_KEY_PATH}
description: "MCP 客户端机器 — Airflow SSH 返回此处"
aliases:
source: "postgres_prod"
teradata: "teradata_prod"
ssh: "airflow_ssh"
```
### 关键行为
- **`${ENV_VAR}`** 值在加载时进行插值
- **`description`** 暴露给 LLM；所有其他字段都隐藏
- **别名** 允许您编写 `source` 而不是 `postgres_prod`
- **`connection_profiles(action="list")`** 返回名称和描述 — 不包含机密
- **`connection_profiles(action="reload")`** 拾取文件更改，无需重启服务器
### 工具如何使用配置文件
| 工具参数 | 示例值 | 描述 |
|----------------|--------------|-------------|
| `source_profile` | `"my_postgres"` | Airbyte 源凭证 |
| `destination_profile` | `"prod_teradata"` | Airbyte 目标凭证 |
| `source_teradata_profile` | `"td_source"` | TdLoad 源 Teradata 连接 |
| `target_teradata_profile` | `"prod_teradata"` | TdLoad 目标 Teradata 连接 |
| `teradata_profile` | `"prod_teradata"` | Airflow Teradata 连接 |
| `ssh_profile` | `"airflow_ssh"` | Airflow SSH 连接 |
| `connection_profile` | `"my_postgres"` | 环境/机密连接 |
---
## 用法
### 启动服务器
```bash
# 启动服务器（stdio 传输 — 适用于任何 MCP 客户端）
python -m elt_mcp_server
# 或使用控制台脚本
etl-mcp-server
```
### 使用 Claude Desktop
将以下内容添加到您的 Claude Desktop 配置文件 (`claude_desktop_config.json`) 中：
```json
{
"mcpServers": {
"etl-mcp": {
"command": "etl-mcp-server",
"args": ["--env-file", "/absolute/path/to/teradata-etl-mcp-test/.env"]
}
}
}
```
> 使用 `--env-file` 选项可以将所有凭证保存在 `.env` 文件中，而不将敏感信息暴露在 `claude_desktop_config.json` 中。
### 使用 VS Code
将以下内容添加到您的 VS Code MCP 配置文件（工作空间中的 `.vscode/mcp.json` 文件）：
```json
{
"servers": {
"etl-mcp": {
"command": "etl-mcp-server",
"args": ["--env-file", "/absolute/path/to/teradata-etl-mcp-test/.env"]
}
}
}
```
> 请使用绝对路径指向 `.env` 文件。在 Windows 上，可以使用正斜杠或转义的反斜杠：`C:/Users/you/teradata-etl-mcp-test/.env`。
### 示例：创建 Airbyte Pipeline
```
用户：“创建一个每日 ELT 管道，从 Postgres 到 Teradata，用于 customers 和 orders 表，增量同步，每天 02:00 UTC”
LLM 将执行：
1. 调用 list_connection_profiles() 以发现可用的配置文件
2. 调用 airbyte_pipeline(
action="create",
source_name="postgres_source",
source_type="Postgres",
source_profile="source",
destination_name="teradata_dest",
destination_type="Teradata",
destination_profile="target",
streams=[{"name": "customers"}, {"name": "orders"}],
schedule_type="cron",
schedule_cron="0 2 * * *"
)
```
### 示例：生成 TdLoad DAG
```
用户：“创建一个从 staging 到 production 的表转移 DAG，用于 sales_data 表，每天凌晨 3 点”
LLM 将调用：
airflow_teradata_load(
method="table_transfer",
dag_id="transfer_sales_data",
source_teradata_profile="td_source",
target_teradata_profile="teradata_prod",
source_database="staging_db",
source_table="sales_data",
target_database="prod_db",
target_table="sales_data",
schedule="0 3 * * *"
)
```
---
## 工具目录
所有工具都遵循 **路由器模式**：单个工具名称接受一个 `action` 或 `method` 参数，该参数选择操作。这样可以保持 MCP 工具列表简洁，同时保留完整的 功能。
### 管道管理（5 个工具）
| 工具 | 主要操作/方法 | 描述 |
|------|-----------------------|-------------|
| `pipeline_status` | `dag`, `task`, `log` | 查询 DAG 运行状态、任务状态和任务日志 |
| `pipeline_control` | `list`, `pause`, `resume`, `delete`, `update_schedule` | 列出、暂停、恢复、删除 DAG，或更改其计划 |
| `pipeline_deploy` | `deploy_dags`, `deploy_complete`, `create_sync_dag` | 通过 SSH/SFTP 将 DAG 文件或完整管道工件部署到 Airflow |
| `pipeline_validate` | `dag`, `directory`, `files` | 在部署前验证 DAG 语法和配置 |
| `airflow_connections` | `list`, `create_teradata`, `create_airbyte`, `create_ssh` | 创建和列出 Airflow 连接（Teradata、Airbyte、SSH） |
### 流程和执行（3 个工具）
| 工具 | 主要操作/方法 | 描述 |
|------|-----------------------|-------------|
| `dag_trigger` | `run`, `idempotent`, `backfill` | 立即触发 DAG 运行，具有去重功能或作为补丁 |
| `dag_monitor` | `status`, `history`, `logs`, `metrics` | 查询 DAG 运行状态、历史记录、任务日志和性能指标 |
| `airflow_admin` | `health`, `reset_circuit_breaker` | Airflow 健康检查和断路器管理 |
### 数据移动（5 个工具）
| 工具 | 主要操作/方法 | 描述 |
|------|-----------------------|-------------|
| `airbyte_pipeline` | `create`, `update`, `preview`, `check_health` | 端到端的 Airbyte 管道，具备智能流选择和调度功能 |
| `airbyte_sync` | `trigger`, `status`, `cancel` | 触发和监控 Airbyte 同步作业 |
| `airbyte_inventory` | `list_connectors`, `list_workspaces`, `get_schema` | 浏览连接器定义、工作区和源模式 |
| `airbyte_manage` | `create_source`, `create_destination`, `create_connection`, `select_streams`, `build_catalog`, `delete_*` | 创建、配置和删除 Airbyte 源、目标和连接 |
| `airflow_teradata_load` | `csv_dag`, `table_transfer`, `csv_complete` | 为 CSV 加载或表传输生成 Airflow DAG，借助 TdLoad/TPT |
### dbt 管理（5 个工具）
| 工具 | 主要操作/方法 | 描述 |
|------|-----------------------|-------------|
| `dbt_execute` | `run`, `test`, `build`, `compile`, `snapshot`, `seed`, `clean`, `debug`, `deps`, `parse` | 执行任何 dbt 命令，支持模型选择和变量支持 |
| `dbt_docs` | `generate`, `generate_schema` | 生成 dbt 文档（返回用于本地服务的 shell 命令）和模式 YAML |
| `dbt_info` | `list_models`, `list_sources`, `list_tests`, `project_info` | 检查项目结构、模型、源和测试 |
| `dbt_generate_model` | *(位置参数：表名)* | 从 Teradata 表元数据生成 dbt 模型 SQL |
| `dbt_project` | `init`, `clean`, `debug`, `deps` | 项目级生命周期操作 |
### 元数据发现（2 个工具）
| 工具 | 主要操作/方法 | 描述 |
|------|-----------------------|-------------|
| `teradata_discover` | `find`, `describe`, `profile`, `preview`, `compare`, `list` | 查找、描述、配置文件、预览和比较 Teradata 表 |
| `teradata_analyze` | `column`, `size`, `lineage`, `search` | 列统计、大小估计、血缘和元数据搜索 |
### 连接配置文件（1 个工具）
| 工具 | 主要操作 | 描述 |
|------|-------------|-------------|
| `connection_profiles` | `list`, `reload` | 列出可用配置文件（无秘密）或在编辑后从 `connections.yaml` 重新加载 |
### TTU 管理（1 个工具）
| 工具 | 主要操作 | 描述 |
|------|-------------|-------------|
| `ttu_execute` | `execute_ddl`, `load_data`, `execute_bteq`, `check_installation` | 通过 teradatasql 执行 DDL，通过 tdload 加载数据，运行 BTEQ 脚本（带有 teradatasql 回退），检查 TTU 安装 |
---
## 开发
> **受众**：希望修改、测试或扩展代码库的贡献者。
### 开发设置
```bash
# 克隆并使用开发依赖项安装
git clone https://github.com/Teradata/teradata-etl-mcp-server.git
cd teradata-etl-mcp-server
python -m venv .venv
.venv\Scripts\activate # Windows
# source .venv/bin/activate # Linux/macOS
pip install -e ".[dev,all]"
# 安装 pre-commit 钩子（每次 git 提交时运行检查）
pre-commit install
# 复制配置模板
cp .env.example .env
cp connections.yaml.example connections.yaml
```
### 运行 Lint 工具
```bash
# Lint 并自动修复
ruff check src tests --fix
# 格式化代码
ruff format src tests
# 类型检查
mypy src
# 安全扫描
bandit -c pyproject.toml -r src
```
### 代码模式
- **工具注册**：所有工具都是异步函数，通过 `register_*_tools(orchestrator)` 注册，返回 `Dict[str, Callable]`。
- **流程编排器**：`PipelineOrchestrator` 通过 `@property` 装饰器延迟加载客户端。
- **凭证解析**：工具接受 `*_profile` 字符串参数。服务器调用 `orchestrator.credential_resolver.resolve_profile(name)` 获取实际凭证。
- **响应净化**：所有工具响应通过 `sanitize_response()`，该函数深度克隆并屏蔽敏感键（`password`, `secret`, `token`, `api_key`, `credential`, `connection_configuration`）。
### 测试
#### 运行测试
```bash
# 运行所有测试
pytest
# 运行详细输出
pytest -v
# 运行特定测试文件
pytest tests/unit/test_airbyte_client.py
# 运行特定测试类或方法
pytest tests/unit/test_airbyte_client.py::TestCreateAirbyteSource
pytest tests/unit/test_airbyte_client.py::TestCreateAirbyteSource::test_success
# 运行匹配关键字的测试
pytest -k "intelligent_pipeline"
# 仅运行单元测试 / 跳过慢速测试
pytest -m unit
pytest -m "not slow"
```
#### 覆盖率
覆盖率配置在 `pyproject.toml` 中，并随 pytest 自动运行。报告生成为：
- **终端**：term-missing（内联 pytest 输出）
- **HTML**：`htmlcov/index.html`
- **XML**：`coverage.xml`
#### 测试文件
| 测试文件 | 涵盖 |
|-----------|--------|
| `test_airbyte_client.py` | Airbyte 客户端 + 所有数据移动工具（288 个测试） |
| `test_credential_resolver.py` | 配置文件解析、环境变量插值、别名（13 个测试） |
| `test_response_sanitizer.py` | 工具响应中的敏感键屏蔽（18 个测试） |
| `test_connection_profile_tools.py` | 配置文件列表/重载工具（5 个测试） |
| `test_airflow_client.py` | Airflow REST API 客户端 |
| `test_teradata_client.py` | Teradata 数据库客户端 |
| `test_dbt_client.py` | dbt CLI 包装器 |
| `test_config.py` | 设置和配置加载 |
| `test_orchestrator.py` | 管道编排器 |
| `test_pipeline_management_tools.py` | 管道管理 MCP 工具 |
| `test_metadata_discovery_tools.py` | 元数据发现 MCP 工具 |
| `test_airflow_dag_generator.py` | Airflow DAG 代码生成 |
| `test_airflow_tdload_dag_generator.py` | TdLoad DAG 代码生成 |
| `test_csv_analyzer.py` | CSV 文件分析 |
| `test_dbt_generator.py` | dbt 模型代码生成 |
| `test_bteq_generator.py` | BTEQ 脚本生成 |
| `test_tpt_generator.py` | TPT 脚本生成 |
| `test_intelligence_engine.py` | 传输方法推荐 |
| `test_metrics_collector.py` | 指标收集 |
| `test_metadata_store.py` | 元数据持久性 |
| `test_plugin_manager.py` | 插件系统 |
| `test_validators.py` | 输入验证实用程序 |
#### 编写测试
- 使用 `unittest.mock.Mock()` / `AsyncMock()` 模拟 `PipelineOrchestrator` 及其客户端。
- 始终在编排器模拟上包含一个 `credential_resolver` 模拟：
```python
orch = Mock()
resolver = Mock()
resolver.resolve_profile.return_value = {"host": "localhost", "username": "user", "password": "pw"}
orch.credential_resolver = resolver
```
- 通过 `register_*_tools(orchestrator)` 返回的工具字典测试内部闭包。
- HTTP 响应模拟必须包含 `resp.headers = {"Content-Type": "application/json"}`。
### Pre-commit 钩子
该项目使用 13 个 Pre-commit 钩子，在每次 `git commit` 时运行：
| 钩子 | 描述 |
|------|-------------|
| `ruff-lint` | Python Linting，具有自动修复（pycodestyle, pyflakes, bugbear, security, etc.） |
| `ruff-format` | 代码格式化检查 |
| `bandit` | 安全漏洞扫描（仅源文件） |
| `check-ast` | Python 语法验证 |
| `no-debug-statements` | 检测 `print()`, `breakpoint()`, `pdb` 在源代码中 |
| `no-private-keys` | 检测任何文件中的私钥 |
| `no-env-files` | 防止提交 `.env` 文件 |
| `no-hardcoded-secrets` | 在源代码中检测硬编码密码/令牌 |
| `trailing-whitespace` | 删除尾随空格 |
| `check-yaml` | YAML 语法验证 |
| `check-toml` | TOML 语法验证 |
| `check-merge-conflict` | 检测合并冲突标记 |
| `no-large-files` | 拒绝 > 500 KB 的文件 |
```bash
# 设置（一次性）
pre-commit install
# 在暂存文件上手动运行
pre-commit run
# 在所有文件上运行
pre-commit run --all-files
# 运行特定钩子
pre-commit run ruff-lint
pre-commit run bandit
```
**处理失败：**
- **ruff-lint**：自动修复应用于。审查更改，重新暂存并再次提交。
- **ruff-format**：运行 `ruff format src tests` 进行修复，然后重新暂存。
- **bandit**：为假阳性添加 `# nosec BXXX` 内联注释。为假阳性添加规则到 `pyproject.toml` 进行项目范围的抑制。
---
## 项目结构
```
teradata-etl-mcp-server/
|-- src/
| |-- elt_mcp_server/
| |-- __init__.py
| |-- __main__.py # 控制台脚本入口
| |-- main.py # CLI（argparse, 信号处理, 异步）
| |-- server.py # FastMCP 服务器，工具注册
| |-- orchestrator.py # PipelineOrchestrator（延迟加载客户端）
| |-- config.py # Pydantic 设置（环境变量，.env, YAML）
| |-- credential_resolver.py # 连接配置文件解析
| |-- response_sanitizer.py # 屏蔽响应中的敏感键
| |-- intelligence.py # 传输方法推荐引擎
| |
| |-- clients/
| | |-- airbyte_client.py # Airbyte Public API v1 客户端
| | |-- airflow_client.py # Airflow REST API 客户端
| | |-- teradata_client.py # Teradata SQL 客户端
| | |-- dbt_client.py # dbt CLI 包装器
| |
| |-- tools/
| | |-- pipeline_management.py # 20 个管道 CRUD + Airflow 连接工具
| | |-- orchestration_execution.py # 6 个 DAG 运行 + 监控工具
| | |-- data_movement.py # 21 个 Airbyte + TdLoad + CSV 工具
| | |-- dbt_management.py # 27 个 dbt 操作工具
| | |-- governance_observability.py # 5 个血缘 + 审计 + 质量工具
| | |-- metadata_discovery.py # 10 个表发现 + 配置文件工具
| | |-- connection_profiles.py # 2 个配置文件列表/重载工具
| | |-- environment_secrets.py # 6 个连接 + 环境变量工具
| | |-- extensibility.py # 插件管理工具
| | |-- deployment_validator.py # 部署验证实用程序
| |
| |-- generators/
| | |-- airflow_dag_generator.py # Airflow DAG Jinja2 模板
| | |-- airflow_tdload_dag_generator.py # TdLoad DAG 生成
| | |-- bteq_generator.py # BTEQ 脚本生成
| | |-- dbt_generator.py # dbt 模型生成
| | |-- tpt_generator.py # TPT 脚本生成
| |
| |-- monitoring/
| | |-- metrics_collector.py # Prometheus 格式的指标
| |
| |-- plugins/
| | |-- plugin_manager.py # 插件发现和生命周期
| |
| |-- storage/
| | |-- metadata_store.py # SQLite/JSON 元数据持久性
| |
| |-- utils/
| |-- csv_analyzer.py # CSV 文件分析
| |-- file_operations.py # 文件 I/O 实用程序
| |-- validators.py # 输入验证
|
|-- tests/
| |-- unit/ # 27 个测试文件，324+ 测试 |
|
|-- scripts/ # 用于手动测试的实用脚本
|-- airflow_dags/ # 生成的 DAG 输出目录
|-- .env.example # 环境变量模板
|-- connections.yaml.example # 连接配置文件模板
|-- .pre-commit-config.yaml # Pre-commit 钩子配置
|-- pyproject.toml # 构建配置，工具设置，依赖项
|-- DESIGN.md # 高级架构设计文档
|-- README.md # 入门和一般用法（此文件）
---
## 故障排除
### 常见问题
**Pre-commit 钩子在设置后第一次提交失败：```bash
pre-commit install
pre-commit run --all-files # 首先修复所有现有问题）
```
**Bandit 假阳性：**
B608（硬编码 SQL 表达式）在 `pyproject.toml` 中全局跳过，因为项目从内部配置值构造 BTEQ/验证 SQL，而不是用户输入。对于其他 bandit 发现，使用 `# nosec BXXX` 内联注释。对于假阳性，请添加规则到 `pyproject.toml` 进行项目范围的抑制。
**测试失败时 `TypeError: unexpected keyword argument`**:
确保测试模拟包含编排器模拟上的 `credential_resolver` 模拟：
```python
resolver = Mock()
resolver.resolve_profile.return_value = {"host": "localhost", "username": "user", "password": "pw"}
orch.credential_resolver = resolver
```
**找不到 `connections.yaml`：**
解析器按以下顺序搜索这些位置：
1. 环境变量 `CONNECTIONS_FILE`
2. 当前工作目录中的 `connections.yaml`
3. `settings.security.connections_file`（如果在服务器设置中配置）
**Airbyte API 返回意外格式：**
Airbyte 公共 API v1 将列表响应包装在 `{"data": [...]}` 中。客户端在内部通过 `resp.get("data", [])` 处理此问题。
---
## 贡献
我们欢迎贡献！请参阅 [CONTRIBUTING.md](CONTRIBUTING.md) 了解以下指南：
- 开发设置
- 代码风格和标准
- 测试要求
- 发布更新
---
## 文档
| 文档 | 用途 |
|----------|---------|
| [README.md](README.md) | 入门和一般用法（此文件） |
| [SSH-SETUP.md](SSH-SETUP.md) | 双向 SSH 配置，用于 Airflow 集成 |
| [DESIGN.md](DESIGN.md) | 高级架构和系统设计 |
| [CONTRIBUTING.md](CONTRIBUTING.md) | 开发指南和工作流程 |
---
## 许可证
该项目根据 Apache License 2.0 许可。有关详细信息，请参阅 [LICENSE](LICENSE)。
MCP 配置
以下是此 MCP Server 的配置信息，可直接复制到 Cursor 或其他 MCP 客户端中使用。
mcp.json
连接信息
猜你喜欢
暂无推荐服务器