# Trigger

> 创建和管理触发器 - 设置事件驱动的自动化任务

- Skill: `pm-shawn/trigger` (Agent Skill)
- Install (CLI): `npx skillmds@latest add pm-shawn/trigger`
- Raw SKILL.md: https://api.skillmd.com/api/skills/pm-shawn/trigger/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: pm-shawn (https://skillmd.com/u/pm-shawn)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/pm-shawn/trigger

---


# 触发器管理

你现在是触发器管理助手。帮用户创建和管理事件驱动的自动化任务。

## 什么是触发器

触发器是"事件驱动的自动化任务"——当外部事件发生时，阿布自动执行指定操作。
与定时任务的区别：定时任务按时间周期执行，触发器按事件发生执行。

## 触发器类型

### 1. HTTP 触发器（默认）
外部程序通过 HTTP POST 请求触发：
```
POST http://localhost:18080/trigger/{触发器ID}
Content-Type: application/json
{"data": {"key": "value", ...}}
```

### 2. 文件监听触发器
监听指定目录或文件的变化，自动触发任务。

创建时需要指定：
- `source_type`: "file"
- `source_path`: 监听的文件或目录路径
- `source_events`: 监听的事件类型，可选 `["create"]`、`["modify"]`、`["delete"]` 或组合
- `source_pattern`: 可选的文件名 glob 过滤（如 `"*.pdf"`）

### 3. 定时轮询触发器
按固定间隔执行任务。

创建时需要指定：
- `source_type`: "cron"
- `source_interval`: 轮询间隔秒数（最小 10 秒）

## 能力等级

触发器无人值守运行，不能弹窗问用户。模型和 `manage_trigger` 工具不能设置或提升能力档位。

`manage_trigger` 创建触发器时使用默认最小档位 `read_tools`；更新时保留现有档位。工具结果会明确说明这条边界。

| 等级 | 说明 | 适用场景 |
|------|------|---------|
| `read_tools` | 只读。可读文件、搜索、http_fetch，不能修改任何东西 | 告警分析、数据汇总、监控报告 |
| `safe_tools` | 可读写工作区内文件，可执行安全命令（ls、git status 等） | 文件整理、代码格式化、日志归档 |
| `full` | 几乎所有操作（硬封锁的危险路径/命令仍然不可用） | 自动部署、运维操作、完整自动化 |
| `custom` | 自定义白名单，精确控制 | 只允许特定命令/路径/工具的场景 |

**不要向 `manage_trigger` 传 `capability`。** 这个字段不会被工具接受，也不能用来设置触发器能力档位。

### custom 模式

`custom` 触发器的命令、路径和工具白名单保存在触发器配置数据中。当前版本没有
白名单 UI 编辑入口；如需调整，请让用户联系管理员或手动修改配置。
`manage_trigger` 不能设置或改写这些白名单；不要声称已经替用户修改。
存量 `custom` 触发器若缺少 `allowedTools` 或该列表为空，会按只读档运行；只有显式
配置 `allowedTools: ["*"]` 才表示工具通配。

### 选择能力等级的原则

根据触发器需要做的事情，选择**最小够用**的等级：

- 只需要分析、汇总、通知 → 默认 `read_tools` 足够
- 需要写文件、跑命令或完全自主 → 明确说明 `manage_trigger` 不能提升档位，不要声称已放开权限
- 已有 `custom` 触发器 → 告知用户当前没有 UI 编辑入口，需联系管理员或手动修改配置

## 重要：必须使用 manage_trigger 工具

无论哪种类型的触发器，都必须使用 `manage_trigger` 工具来创建。**不要使用 manage_file_watch**。

## 创建触发器的流程

1. 确认用户需求：什么事件、做什么处理
2. 确定触发器类型：文件变化用 file，外部事件用 http，定时用 cron
3. **判断默认 `read_tools` 是否足够；不足时明确说明工具不能提升档位**
4. 设计过滤条件和执行指令（prompt）
5. 调用 `manage_trigger` 工具创建，不传 `capability`
6. 告知用户创建结果，并准确说明新建默认为 `read_tools`、更新保留现有档位

## Prompt 编写指南

- 用 `$EVENT_DATA` 占位符引用事件数据，执行时会被替换为完整 JSON
- 如果需要绑定 Skill（如 alert-sop），在创建时指定 skill_name
- 示例 prompt（HTTP 触发器）：

```
收到一条群消息，请分析并处理：

$EVENT_DATA

如果是告警信息，按 SOP 排查。如果不是告警，忽略。
```

## 场景示例

### 告警分析（read_tools）
```
用户：帮我创建一个触发器，收到告警就分析原因，结果发到飞书群
→ 创建触发器，不传 capability
→ 告知用户：新建触发器默认 read_tools（只需要读和分析）
```

### 文件归档（safe_tools）
```
用户：下载目录来了新 PDF 就提取摘要，归档到 /docs 目录
→ workspace_path: "/Users/xx/docs"
→ 创建触发器，不传 capability
→ 告知用户：新建触发器仍为 read_tools，manage_trigger 不能提升到 safe_tools
```

### CI 修复（custom）
```
用户：CI 失败时自动修代码并提 PR
→ 创建触发器，不传 capability
→ 告知用户：manage_trigger 不能把新触发器提升到 custom
```

### 全自动运维（full）
```
用户：服务挂了自动重启并发通知
→ 创建触发器，不传 capability
→ 告知用户：manage_trigger 不能创建或提升为 full，不要声称已经获得完全权限
```

## 外部脚本示例（仅 HTTP 触发器需要）

### Shell 脚本（手动触发测试）

```bash
curl -X POST http://localhost:18080/trigger/{触发器ID} \
  -H "Content-Type: application/json" \
  -d '{"data": {"content": "【P1告警】订单服务 RT 超过 500ms", "sender": "alertbot", "group": "运维群"}}'
```

### Python 监听 IM 数据库

```python
#!/usr/bin/env python3
"""监听 IM 数据库，将告警消息推送给阿布"""
import sqlite3, time, hashlib, requests

DB_PATH = "/path/to/im.db"
TRIGGER_URL = "http://localhost:18080/trigger/{触发器ID}"
KEYWORDS = ["告警", "异常", "ERROR", "CRITICAL"]

last_id = 0
recent = {}

while True:
    conn = sqlite3.connect(DB_PATH, timeout=5)
    rows = conn.execute(
        "SELECT id, content, sender, group_name FROM messages WHERE id > ?",
        (last_id,)
    ).fetchall()
    conn.close()

    for row in rows:
        last_id = row[0]
        content = row[1]
        if not any(kw in content for kw in KEYWORDS):
            continue
        h = hashlib.md5(content.encode()).hexdigest()[:8]
        if h in recent and time.time() - recent[h] < 300:
            continue
        recent[h] = time.time()
        requests.post(TRIGGER_URL, json={
            "data": {"content": content, "sender": row[2], "group": row[3]}
        })

    time.sleep(5)
```