内容
# Teradata 工作负载管理 (WLM) MCP 服务器
[](https://www.python.org/downloads/)
[](https://modelcontextprotocol.io)
Teradata 工作负载管理 (WLM) 的模型控制协议 (MCP) 服务器，提供全面的监控和管理功能，适用于 Teradata 系统。
## 功能
- **41 个管理工具**：28 个核心监控工具 + 13 个配置管理工具
- **39 个 MCP 资源**：参考数据、模板、规则集探索和工作流程指导
- **完整的 Teradata 连接性**：TD2、LDAP、Kerberos、TDNEGO、JWT 认证
- **TLS/SSL 支持**：ALLOW、PREFER、REQUIRE、VERIFY-CA、VERIFY-FULL 模式
- **每个工具的审计轨迹**：每个工具调用的 QueryBand 跟踪，用于工作负载管理
- **异步数据库操作**：所有数据库调用都在线程池中运行（非阻塞）
- **连接恢复力**：具有指数退避和抖动的智能重试
- **连接健康监测**：自动健康检查和恢复
- **OAuth 2.1 认证**：可选的 Keycloak 集成，基于范围的访问控制
- **模板驱动的配置**：常见 TDWM 配置的预建模式
- **通用错误消息**：不会向客户端泄露原始数据库错误
## 用例
### 性能故障排除
**场景**：识别查询为何运行缓慢或延迟
**使用的工具**：
1. `show_sessions` - 识别活动会话及其状态
2. `identify_blocking` - 查找导致阻塞的会话
3. `show_sql_text_for_session` - 查看正在运行的 SQL
4. `monitor_amp_load` - 检查 AMP 利用率
5. `display_delay_queue` - 查看查询是否延迟
6. `show_tasm_statistics` - 分析 TASM 工作负载分布
**资源探索**：
- `tdwm://summary` - 整体系统状态
- `tdwm://throttle-statistics` - 当前节流限制和延迟
### 紧急响应
**场景**：系统负载过高，需要快速限制工作负载
**工作流程**：
1. 使用 `tdwm://workflow/emergency-throttle` 资源获取逐步指南
2. 使用 `tdwm://template/throttle/application-basic` 获取快速节流模式
3. `create_system_throttle`，并发性限制较低
4. `enable_throttle`，激活
5. `activate_ruleset`，立即应用更改
6. 使用 `show_trottle_statistics` 监控
### 预定维护
**场景**：在维护窗口期间阻止用户访问
**工作流程**：
1. 使用 `tdwm://workflow/maintenance-window` 获取完整指南
2. 使用 `create_filter` 创建过滤器，针对所有用户或特定应用程序
3. `enable_filter` + `activate_ruleset`，阻止访问
4. 执行维护
5. `disable_filter` + `activate_ruleset`，恢复访问
## 安装
```bash
pip install tdwm-mcp
```
## 快速入门
```bash
# 1. 安装软件包
pip install tdwm-mcp
# 2. 配置数据库连接
export DATABASE_URI="teradata://username:password@hostname/database"
# 3. 启动 MCP 服务器
uv run tdwm-mcp
# 4. 使用简单的工具调用进行测试（通过您的 MCP 客户端）
# 工具：show_sessions
```
## 配置
所有配置都通过环境变量完成。有关完整参考，请参见 `.env.example`。
### 数据库连接
```bash
# 连接 URL（必需）
export DATABASE_URI="teradata://username:password@hostname/database"
```
### 认证机制
服务器通过 `DB_LOGMECH` 环境变量支持所有 Teradata 认证方法：
#### TD2（默认 - 用户名/密码）
```bash
export DATABASE_URI="teradata://username:password@hostname/database"
# DB_LOGMECH=TD2 是默认值，无需显式设置
```
#### LDAP
```bash
export DATABASE_URI="teradata://@hostname/database"
export DB_LOGMECH=LDAP
export DB_LOGDATA="authcid=user@domain.com password=ldap_password"
```
| 目录 | authcid 格式 |
|-----------|----------------|
| Active Directory (Simple) | `user@domain.com` |
| Active Directory (DIGEST-MD5) | `DOMAIN\username` |
| OpenLDAP / Sun DS | `username` |
#### Kerberos (KRB5)
```bash
export DATABASE_URI="teradata://@hostname/database"
export DB_LOGMECH=KRB5
# 需要有效的 Kerberos 票证（kinit）
```
#### TDNEGO（令牌协商）
```bash
export DB_LOGMECH=TDNEGO
# 在 TD2、LDAP、KRB5 之间自动协商
```
#### JWT（JSON Web Token）
```bash
export DATABASE_URI="teradata://@hostname/database"
export DB_LOGMECH=JWT
export DB_LOGDATA="authcid=token_issuer password=jwt_token_string"
```
### TLS/SSL 配置
```bash
# SSL 模式：ALLOW、PREFER、REQUIRE、VERIFY-CA、VERIFY-FULL
export DB_SSL_MODE=REQUIRE
# 传输加密（默认：true）
export DB_ENCRYPT_DATA=true
```
| 模式 | 描述 |
|------|-------------|
| `ALLOW` | 尝试 SSL，退回到纯文本 |
| `PREFER` | 优先使用 SSL，但允许纯文本 |
| `REQUIRE` | 需要 SSL，如果不可用则失败 |
| `VERIFY-CA` | 验证 CA 证书 |
| `VERIFY-FULL` | 验证 CA 和主机名 |
### 传输配置
```bash
# 传输：stdio（默认）、sse、streamable-http
export MCP_TRANSPORT=stdio
export MCP_HOST=0.0.0.0
export MCP_PORT=8000
export MCP_PATH=/mcp/
```
### CORS 配置
```bash
# 允许的来源（默认：* for all）
export CORS_ALLOWED_ORIGINS="https://your-app.example.com"
```
### 重试配置
```bash
# 工具级重试
export TOOL_MAX_RETRIES=2
export TOOL_RETRY_INITIAL_DELAY=0.5
export TOOL_MAX_RETRY_DELAY=2.0
# 连接级重试
export DB_MAX_RETRIES=3
export DB_INITIAL_BACKOFF=1.0
export DB_MAX_BACKOFF=30.0
```
### OAuth 2.1（可选）
```bash
export OAUTH_ENABLED=true
export KEYCLOAK_URL=https://your-keycloak-server.com
export KEYCLOAK_REALM=your-realm
export KEYCLOAK_CLIENT_ID=tdwm-mcp
export KEYCLOAK_CLIENT_SECRET=your-secret
export OAUTH_RESOURCE_SERVER_URL=https://your-mcp-server.com
export OAUTH_REQUIRED_SCOPES=tdwm:read,tdwm:monitor
export OAUTH_REQUIRE_HTTPS=true
```
**范围**：`tdwm:read`、`tdwm:write`、`tdwm:admin`、`tdwm:query`、`tdwm:monitor`、`tdwm:workload`
## Docker
所有 docker-compose 文件都使用环境变量扩展 - 没有硬编码的凭证。在运行之前，请从 `.env.example` 创建一个 `.env` 文件。
```bash
# 基本 SSE 传输
cp .env.example .env
# 使用您的凭证编辑 .env
docker compose up -d
# 使用 OAuth（Keycloak）
docker compose -f docker-compose.oauth.yml up -d
# 在备用端口上的 SSE
docker compose -f docker-compose.sse.yml up -d
```
## 连接恢复力
服务器包含用于处理 Teradata 连接故障的自动重试逻辑。
### 工作原理
所有工具都使用智能重试装饰器包装，该装饰器：
1. **检测连接错误** - 区分连接故障（可重试）和 SQL 错误（不可重试）
2. **智能重试逻辑** - 基于操作安全性：
- **读取操作**（查询、监控）：最多 2 次重试
- **写入操作**（创建、更新）：最多 1 次重试
- **危险操作**（删除、终止）：无自动重试
3. **指数退避** - 渐进式延迟（0.5s -> 1.0s -> 2.0s）带有抖动
4. **健康监测** - 每 5 分钟通过 `SELECT 1` 进行连接健康检查
### 检测到的 Teradata 错误
- 网络超时和断开连接
- 会话断开连接（错误 3126）
- 由于 TDWM 终止导致的事务中止（错误 2631）
- 会话限制超过（错误 8017）
- 通信链接故障
### 每个工具的 QueryBand
每个工具调用都为审计和工作负载管理设置事务级 QueryBand：
```sql
SET QUERY_BAND = 'ApplicationName=TDWM_MCP;ToolName=show_sessions;Transport=stdio;' FOR TRANSACTION
```
这使 Teradata 管理员能够跟踪哪个 MCP 工具启动了每个查询。
## 架构
```
+------------------------------------------------------------+
| MCP 客户端 |
| （Claude Desktop 等） |
+----------------------------+-------------------------------+
| MCP 协议
+----------------------------v-------------------------------+
| TDWM MCP 服务器 |
| +------------------------------------------------------+ |
| | server.py - FastMCP 应用程序 + 设置 | |
| +------------------------------------------------------+ |
| | |
| +------------------------------------------------------+ |
| | fnc_common.py | |
| | - 连接管理器 - QueryBand 帮助程序 | |
| | - get_connection() - _set_queryband() | |
| | - @with_connection_retry 装饰器 | |
| +------+-------------------+---------------------------+ |
| | | |
| +------v------+ +--------v---------+ +--------------+ |
| | fnc_tools | | fnc_tools_ | | fnc_ | |
| | .py | | priority1.py | | resources.py | |
| | | | | | | |
| | 28 个核心 | | 13 个配置 | | 39 个资源 | |
| | 监控工具 | | 管理工具 | | | |
| +------+------+ +--------+---------+ +------+-------+ |
| | | | |
| +------v------------------v--------------------v--------+ |
| | settings.py queryband.py retry_utils.py | |
| | connection_manager.py tdsql/tdsql.py | |
| +-------------------------------------------------------+ |
+----------------------------+---------------------------------+
| teradatasql（TD2/LDAP/KRB5/JWT）
+----------------------------v---------------------------------+
| Teradata 数据库 |
| （TDWM/TASM 系统） |
+--------------------------------------------------------------+
```
### 模块组织
| 模块 | 目的 |
|--------|---------|
| `settings.py` | 冻结设置数据类 - 所有环境变量在启动时读取一次 |
| `queryband.py` | 每个工具的 QueryBand 构建器，带有值净化 |
| `tdsql/tdsql.py` | Teradata 连接包装器，支持 LOGMECH/TLS |
| `connection_manager.py` | 连接健康监测、重试、重新连接 |
| `retry_utils.py` | `@with_connection_retry()` 装饰器、错误检测 |
| `fnc_common.py` | 共享实用程序、连接访问、QueryBand 帮助程序 |
| `fnc_tools.py` | 28 个核心监控工具（异步，带有 QueryBand） |
| `fnc_tools_priority1.py` | 13 个配置管理工具 |
| `fnc_resources.py` | 39 个 MCP 资源处理程序 |
| `resource_reference.py` | 参考数据资源（8） |
| `resource_templates.py` | 配置模板（13） |
| `resource_queries.py` | 规则集探索查询（8） |
| `server.py` | FastMCP 应用程序、传输设置、初始化 |
| `oauth_context.py` | OAuth 基于范围的授权 |
| `auth/` | OAuth 2.1 配置、中间件、端点、元数据 |
### 依赖图
```
settings.py, queryband.py, retry_utils.py（无内部依赖）
|
tdsql/tdsql.py（使用 settings）
|
connection_manager.py（使用 tdsql, queryband）
|
fnc_common.py（使用 connection_manager, queryband, retry_utils）
|
+-- fnc_tools_priority1.py（使用 fnc_common）
| |
+-- fnc_tools.py（使用 fnc_common + fnc_tools_priority1）
| |
+-- fnc_resources.py（使用 fnc_common）
|
server.py（使用上述所有 + settings + auth）
```
## 可用工具
### 会话管理
- **show_sessions** - 显示我的活动会话
- **show_sql_steps_for_session** - 显示特定会话的 SQL 执行步骤
- **show_sql_text_for_session** - 显示特定会话的 SQL 文本
- **abort_sessions_user** - 终止特定用户的所有会话
- **identify_blocking** - 识别导致阻塞情况的用户
### 查询带和监控
- **monitor_session_query_band** - 监控特定会话的查询带
- **list_query_band** - 按类型（TRANSACTION、PROFILE、SESSION 或 ALL）列出查询带
- **show_query_log** - 显示特定用户的查询日志
### 系统资源监控
- **show_physical_resources** - 监控系统物理资源
- **monitor_amp_load** - 监控 AMP（Access Module Processor）负载
- **monitor_awt** - 监控 AWT（AMP Worker Task）资源
- **monitor_config** - 监控虚拟配置
### 工作负载管理
- **list_active_WD** - 列出活动工作负载（WD）
- **list_WD** - 列出所有工作负载（WD）
- **show_tdwm_summary** - 显示工作负载汇总信息
### 延迟队列管理
- **list_delayed_request** - 列出所有延迟查询
- **abort_delayed_request** - 终止特定会话的延迟请求
- **display_delay_queue** - 按类型（WORKLOAD、SYSTEM、UTILITY 或 ALL）显示延迟队列详细信息
- **release_delay_queue** - 释放特定会话或用户的延迟请求
### 节流和性能
- **show_trottle_statistics** - 显示节流统计信息（ALL、QUERY、SESSION、WORKLOAD）
- **list_utility_stats** - 列出系统上实用程序使用的统计信息
### 系统信息
- **show_cod_limits** - 显示 COD（Capacity On Demand）限制
- **show_top_users** - 显示消耗最多资源的顶级用户
- **show_sw_event_log** - 显示系统软件事件日志（OPERATIONAL 或 ALL）
### TASM（Teradata Active System Management）
- **tdwm_list_clasification** - 列出工作负载（TASM）规则的分类类型
- **show_tasm_statistics** - 显示 TASM 性能统计信息
- **show_tasm_even_history** - 显示 TASM 事件历史记录
- **show_tasm_rule_history_red** - 显示系统进入 RED 状态的原因
### 配置管理
#### 节流管理
- **create_system_throttle** - 创建具有并发限制的系统级节流
- **modify_throttle_limit** - 动态调整节流并发限制
- **delete_throttle** - 删除节流定义
- **enable_throttle** - 激活节流规则
- **disable_throttle** - 停用节流规则
#### 过滤器管理
- **create_filter** - 创建过滤器以阻止/拒绝查询
- **delete_filter** - 删除过滤器定义
- **enable_filter** - 激活过滤器规则
- **disable_filter** - 停用过滤器规则
#### 规则管理
- **add_classification_to_rule** - 将分类标准添加到任何规则
- **add_subcriteria_to_target** - 添加子标准（例如 FTSCAN for TABLE）
- **activate_ruleset** - 应用所有待更改以使其生效
- **list_rulesets** - 列出所有可用的规则集
## 可用资源
### 参考数据资源
- `tdwm://reference/classification-types` - 所有 31 种分类类型及其类别
- `tdwm://reference/classification-types/{category}` - 按类别过滤
- `tdwm://reference/operators` - 分类运算符（I、O、IO）
- `tdwm://reference/subcriteria-types` - 子标准类型（FTSCAN、MINSTEPTIME 等）
- `tdwm://reference/actions` - 过滤器操作类型（E=Exception、A=Abort）
- `tdwm://reference/throttle-types` - 节流类型（DM、M）
- `tdwm://reference/states` - TASM 系统状态（GREEN、YELLOW、ORANGE、RED）
- `tdwm://reference/catalog` - 资源的综合目录
### 配置模板
- `tdwm://templates/throttle` - 列出所有节流模板
- `tdwm://template/throttle/{template_id}` - 特定节流模板
- `tdwm://templates/filter` - 列出所有过滤器模板
- `tdwm://template/filter/{template_id}` - 特定过滤器模板
- `tdwm://templates/catalog` - 模板目录
### 规则集探索
- `tdwm://rulesets` - 列出所有可用的规则集
- `tdwm://system/active-ruleset` - 获取当前活动的规则集名称
- `tdwm://ruleset/{ruleset_name}` - 详细的规则集信息
- `tdwm://ruleset/{name}/throttles` - 规则集中的节流列表
- `tdwm://ruleset/{name}/throttle/{throttle_name}` - 节流详细信息
- `tdwm://ruleset/{name}/filters` - 规则集中的过滤器列表
- `tdwm://ruleset/{name}/filter/{filter_name}` - 过滤器详细信息
- `tdwm://ruleset/{name}/pending-changes` - 检查待处理更改
### 工作流程模板
- `tdwm://workflows` - 列出所有可用的工作流程
- `tdwm://workflow/create-throttle` - 完整的节流创建工作流程
- `tdwm://workflow/create-filter` - 完整的过滤器创建工作流程
- `tdwm://workflow/maintenance-window` - 启用/禁用维护的过滤器
- `tdwm://workflow/emergency-throttle` - 快速紧急响应工作流程
- `tdwm://workflow/modify-existing-throttle` - 修改现有节流工作流程
### 旧版资源
- `tdwm://sessions`、`tdwm://workloads`、`tdwm://active-workloads`、`tdwm://summary`
- `tdwm://delayed-queries`、`tdwm://throttle-statistics`、`tdwm://physical-resources`
- `tdwm://amp-load`、`tdwm://classification-types`
## 使用示例
### 基本会话监控
```python
await call_tool("show_sessions")
await call_tool("show_sql_text_for_session", {"sessionNo": 1234})
await call_tool("show_sql_steps_for_session", {"sessionNo": 1234})
```
### 创建系统节流
```python
await call_tool("create_system_throttle", {
"ruleset_name": "MyFirstConfig",
"throttle_name": "ETL_THROTTLE",
"description": "限制 ETL 工作负载并发性",
"throttle_type": "DM",
"limit": 5,
"classification_criteria": [
{"description": "ETL 应用", "type": "APPL", "value": "ETL_APP", "operator": "I"}
]
})
```
### 创建维护窗口过滤器
```python
await call_tool("create_filter", {
"ruleset_name": "MyFirstConfig",
"filter_name": "MAINTENANCE_BLOCK",
"description": "在维护期间阻止非关键查询",
"classification_criteria": [
{"description": "报告用户", "type": "USER", "value": "reporting_user", "operator": "I"}
],
"action": "E"
})
```
驱动配置
```# 1. 读取模板 = await read_resource("tdwm://template/throttle/application-basic")
# 2. 获取活动规则集
active = await read_resource("tdwm://system/active-ruleset")
# 3. 使用模板结构创建限流
await call_tool("create_system_throttle", { ... })
# 4. 激活
await call_tool("activate_ruleset", {"ruleset_name": "MyFirstConfig"})
# 5. 验证
result = await read_resource("tdwm://ruleset/MyFirstConfig/throttle/MY_THROTTLE")
```
## 安全
### 凭证处理
- 数据库凭证仅通过环境变量传递（绝不硬编码）
- 通过 `obfuscate_password()` 在所有日志消息和错误输出中隐藏密码
- 面向客户端的错误是通用的（"请查看服务器日志以获取详细信息"）- 不泄露原始异常
- Docker 组合文件使用 `${VAR}` 扩展 - 版本控制中没有实际凭证
### SQL 安全
- 所有用户提供的参数都使用参数化查询（`?` 占位符）
- 通过 `CALL ... (?, ?, ?)` 模式调用 TDWM 存储过程
- 在 `MonitorSQLSteps` / `MonitorSQLText` 中使用前验证会话号为整数
### OAuth 2.1
- 使用 RS256/RS384/RS512/ES256/ES384/ES512 进行 JWT 验证
- 通过 Keycloak 进行令牌自省回退
- 每个工具的基于范围的访问控制（读/写/管理/查询/监视）
- 通过 `OAUTH_REQUIRE_HTTPS` 可配置 HTTPS 强制
### 异步安全
- 所有数据库操作都包装在 `asyncio.to_thread()` 中（事件循环从不阻塞）
- 连接管理器使用 `asyncio.Lock()` 防止并发重新连接竞争
- 每个工具的 QueryBand 设置为 `FOR TRANSACTION`（不是会话级，无跨工具泄漏）
## 开发
### 项目结构
```
tdwm-mcp/
+-- src/tdwm_mcp/
| +-- __init__.py
| +-- server.py # FastMCP 应用和初始化
| +-- settings.py # 冻结的 Settings 数据类
| +-- queryband.py # 每个工具的 QueryBand 生成器
| +-- fnc_common.py # 共享实用程序和连接管理
| +-- fnc_tools.py # 核心监控工具（28 个工具）
| +-- fnc_tools_priority1.py # 配置管理（13 个工具）
| +-- fnc_resources.py # 资源路由（39 个资源）
| +-- fnc_prompts.py # MCP 提示
| +-- retry_utils.py # 重试装饰器和错误检测
| +-- connection_manager.py # 连接健康管理
| +-- resource_reference.py # 参考数据资源
| +-- resource_templates.py # 配置模板
| +-- resource_queries.py # 规则集探索查询
| +-- tdwm_static.py # 静态参考数据
| +-- oauth_context.py # 基于范围的 OAuth 授权
| +-- tdsql/
| | +-- __init__.py
| | +-- tdsql.py # Teradata SQL 包装器（LOGMECH/TLS）
| +-- auth/
| +-- config.py # OAuth 配置
| +-- middleware.py # JWT 验证中间件
| +-- endpoints.py HTTP 端点
|-- metadata.py #资源元数据
+--.yml # 基本 SSE +-- docker-compose.oauth # 带有 Keycloak 集成
+-- docker-compose.sse # 在备用端口上的 SSE
+-- Dockerfile
+--.example # 环境变量参考
+-- pyproject.toml
uv.lock
+-- README```
### 添加新工具
```python
# 在 fnc 中
@with_connection_retry()
async def my_new_tool(param1) -> ResponseType:
描述工具的作用。"""
tdconn = await get_connection()
def _run():
_set_queryband(tdconn, "my_new cur = tdconn.cursor()
rows = cur.execute("SELECT * FROM my_table WHERE col1 = ?", [param1])
return format_text_response(list(rows.fetchall()))
try:
return await asyncio.to_thread(_run)
except ConnectionError:
raise
except Exception as e:
logger.error(f"my_new_tool 中出错：{e}")
return format_error_response("执行失败。请查看服务器日志以获取详细信息。")
# 在 handle_list_tools() 和 handle_tool_calls() 中注册
```
关键模式：
- 将数据库工作包装在 `asyncio.to_thread(_run)` 中（非阻塞）
- 调用 `_set_queryband(tdconn_name")`跟踪
- 重新引发 `ConnectionError`重试装饰器
- 返回消息（从不 `str(e)`）
### 测试
uv sync
export DATABASE_URI="teradata://user:pass@host/db"
uv run tdcp
# 使用 MCP测试
npx @model/inspector uv run tdcp
```
## 依赖项
- `teradatas20.0.0.30` - Teradata SQL 驱动程序
- `mcp]>=1.12. - 模型控制协议- `fastapi>=0.12` - SSE/可流式 HTTP 的 HTTP - `pydantic.11.4`验证
- `uvicorn>=0.34.2 ASGI 服务器
- `[crypto]>=2.0` - JWT 处理
- `auth1.2.0` - OAuth 2.1 - `aiohttp>=3.9.0` - 用于自省的异步 HTTP
- `httpx>=0.0` - HTTP 客户端
MCP 市场
