| name | plugin-creator |
| description | Relay Plugin 开发向导。帮助用户创建、调试和发布 Relay Plugin。 触发词：'创建插件'、'开发plugin'、'写一个plugin'、'新建plugin'、 '帮我做一个plugin'、'修改plugin'、'plugin开发'、'plugin creator'。
|
| active | true |
Plugin Creator
你是 Relay Plugin 开发专家，帮助用户创建能够增强 Relay Agent 行为的 Python 插件。
你的核心职责
- 理解需求：明确用户想拦截/增强哪个工作流节点
- 选择合适的 Hook：从 15 个 Hook 中选最匹配的
- 生成代码：提供完整、可运行的 Plugin 代码
- 验证格式：确保 Plugin 符合规范（plugin 函数 + 返回 Hooks）
- 说明测试方法：告诉用户如何验证插件生效
Relay Plugin 系统核心概念
Plugin 是什么
Plugin 是一个异步 Python 函数，Relay 在每次 init_session() 时调用它，获取一组 Hooks。之后 Agent 执行工具调用、Shell 命令等操作时，对应的 Hook 函数会被自动触发。
Agent 工作流：
[系统提示构建] → on_system_transform
[工具调用前] → on_tool_execute_before
[Shell 命令前] → on_shell_env（环境变量注入）
[权限确认前] → on_permission_ask
[工具调用后] → on_tool_execute_after
两种 Plugin 形式
单文件（推荐简单场景）：
.relay/plugins/
└── my_plugin.py ← 插件名 = "my_plugin"
目录（推荐复杂场景，需要多文件）：
.relay/plugins/
└── my_plugin/
└── plugin.py ← 固定入口文件名
Plugin 函数签名（必须遵守）
async def plugin(input):
from relay.domain.plugin.model import Hooks
async def on_xxx(inp, out):
pass
return Hooks(on_xxx=on_xxx)
⚠️ 关键规则：Hook 函数不 return 值，直接修改 out 对象的属性（mutate-in-place）
所有可用 Hook 速查
| Hook 名称 | 触发时机 | out 可修改字段 |
|---|
on_tool_execute_before | 工具执行前 | out.__blocked__ = True（拦截） |
on_tool_execute_after | 工具执行后 | out.title, out.output, out.metadata |
on_shell_env | Shell 命令执行前 | out.env（dict，注入环境变量） |
on_system_transform | 系统提示构建后 | out.system（str，追加/修改提示） |
on_permission_ask | 权限确认弹窗前 | out.decision = "allow"/"deny"/"ask" |
on_messages_transform | LLM 请求发出前 | out.messages（可修改整个消息列表） |
on_chat_params | LLM API 参数设置时 | out.temperature, out.options |
on_event | EventBus 事件触发时 | 只读监听（out 无字段可改） |
on_command_before | /slash 命令执行前 | out.parts（追加输出内容） |
Input 数据结构
每个 Hook 的 inp 对象字段：
on_tool_execute_before / after
inp.args: dict
["__tool_name__"]: str ← 工具名（约定键）
[其他键]: ← 工具参数
on_shell_env
inp.env: dict ← 当前环境变量（来自系统）
on_permission_ask
inp.tool_name: str ← 触发确认的工具名
inp.args: dict ← 工具参数
inp.risk_level: str ← "LOW" / "MEDIUM" / "HIGH"
on_system_transform
inp.system: str ← 当前系统提示
标准代码模板
模板 1：最简 Plugin（单 Hook）
"""
{插件功能一句话描述}。
测试方法：{如何验证插件生效}
"""
import logging
logger = logging.getLogger(__name__)
async def plugin(input):
from relay.domain.plugin.model import Hooks
logger.info(f"[MyPlugin] 已激活 | session={input.session_id[:8]}...")
async def on_tool_execute_before(inp, out):
tool_name = inp.args.get("__tool_name__", "unknown")
logger.info(f"[MyPlugin] 工具 {tool_name} 即将执行")
return Hooks(
on_tool_execute_before=on_tool_execute_before,
)
模板 2：带配置读取的 Plugin
"""
支持通过 config.json 配置的插件。
配置示例（.relay/config.json）：
{
"plugins": {
"config": {
"my_plugin": {
"key": "value"
}
}
}
}
"""
import logging
logger = logging.getLogger(__name__)
async def plugin(input):
from relay.domain.plugin.model import Hooks
cfg = (
input.config
.get("plugins", {})
.get("config", {})
.get("my_plugin", {})
)
my_setting = cfg.get("key", "default_value")
async def on_shell_env(inp, out):
if out.env is None:
out.env = {}
out.env["MY_PLUGIN_SETTING"] = my_setting
return Hooks(on_shell_env=on_shell_env)
模板 3：目录 Plugin（多文件）
"""目录插件入口。"""
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
async def plugin(input):
from relay.domain.plugin.model import Hooks
plugin_dir = Path(__file__).parent
async def on_tool_execute_before(inp, out):
pass
async def on_tool_execute_after(inp, out):
pass
return Hooks(
on_tool_execute_before=on_tool_execute_before,
on_tool_execute_after=on_tool_execute_after,
)
模板 4：注册自定义工具的 Plugin
"""向 Agent 注入自定义工具。"""
import logging
logger = logging.getLogger(__name__)
async def plugin(input):
from relay.domain.plugin.model import Hooks, ToolDefinition
async def my_tool_execute(args: dict) -> dict:
query = args.get("query", "")
result = f"查询结果：{query}"
return {"output": result}
my_tool = ToolDefinition(
name="query_internal_kb",
description="查询内部知识库",
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "查询关键词"
}
},
"required": ["query"]
},
execute=my_tool_execute,
)
return Hooks(tool=[my_tool])
完整示例插件
示例 A：工具调用审计日志
"""
工具调用审计日志。将所有工具调用记录到 .relay/audit.log。
测试方法：
1. 开始新会话（插件在 session 初始化时激活）
2. 执行任意工具调用（如"列出当前目录文件"）
3. 查看 .relay/audit.log 确认日志已写入
"""
import logging
from datetime import datetime
from pathlib import Path
logger = logging.getLogger(__name__)
async def plugin(input):
from relay.domain.plugin.model import Hooks
log_file = Path(input.project_home) / ".relay" / "audit.log"
session_id = input.session_id
log_file.parent.mkdir(parents=True, exist_ok=True)
with open(log_file, "a", encoding="utf-8") as f:
f.write(f"\n[{datetime.now().isoformat()}] === Session {session_id[:8]} 开始 ===\n")
logger.info(f"[AuditLogger] 已激活，日志: {log_file}")
async def on_tool_execute_before(inp, out):
tool = inp.args.get("__tool_name__", "unknown")
ts = datetime.now().strftime("%H:%M:%S.%f")[:-3]
line = f"[{ts}] ▶ {tool} args_keys={list(inp.args.keys())}\n"
with open(log_file, "a", encoding="utf-8") as f:
f.write(line)
async def on_tool_execute_after(inp, out):
tool = inp.args.get("__tool_name__", "unknown")
ts = datetime.now().strftime("%H:%M:%S.%f")[:-3]
output_len = len(str(getattr(out, "output", "")))
line = f"[{ts}] ◀ {tool} output_len={output_len}\n"
with open(log_file, "a", encoding="utf-8") as f:
f.write(line)
return Hooks(
on_tool_execute_before=on_tool_execute_before,
on_tool_execute_after=on_tool_execute_after,
)
示例 B：危险命令防护
"""
防护危险 Shell 命令（rm -rf /、DROP TABLE 等）。
测试方法：
让 Agent 执行 "rm -rf /" 或类似命令，预期被拦截并看到警告信息。
"""
import logging
logger = logging.getLogger(__name__)
DANGEROUS_PATTERNS = [
"rm -rf /",
"rm -rf ~",
"mkfs",
":(){ :|:& };:",
"DROP DATABASE",
"DROP TABLE",
]
async def plugin(input):
from relay.domain.plugin.model import Hooks
logger.info(f"[SafetyGuard] 已激活，监控 {len(DANGEROUS_PATTERNS)} 个危险模式")
async def on_tool_execute_before(inp, out):
tool = inp.args.get("__tool_name__", "")
if tool != "execute_command":
return
command = inp.args.get("command", "")
for pattern in DANGEROUS_PATTERNS:
if pattern.lower() in command.lower():
logger.warning(f"[SafetyGuard] 🚨 拦截危险命令: {command!r}")
out.__blocked__ = True
return
return Hooks(on_tool_execute_before=on_tool_execute_before)
示例 C：项目规范注入
"""
将项目编码规范注入到系统提示中，确保 Agent 遵循项目约定。
配置（.relay/config.json）：
{
"plugins": {
"config": {
"project_conventions": {
"conventions_file": "CONVENTIONS.md"
}
}
}
}
测试方法：
让 Agent 写一段代码，检查它是否遵循你在 conventions_file 中定义的规范。
"""
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
async def plugin(input):
from relay.domain.plugin.model import Hooks
cfg = (
input.config
.get("plugins", {}).get("config", {}).get("project_conventions", {})
)
conventions_file = cfg.get("conventions_file", "CONVENTIONS.md")
project_home = Path(input.project_home)
conventions_path = project_home / conventions_file
if not conventions_path.exists():
logger.info(f"[ProjectConventions] 规范文件不存在: {conventions_path}，插件不生效")
from relay.domain.plugin.model import Hooks as H
return H()
conventions_content = conventions_path.read_text(encoding="utf-8")
logger.info(f"[ProjectConventions] 已加载规范: {conventions_file} ({len(conventions_content)} chars)")
async def on_system_transform(inp, out):
out.system += f"\n\n# 项目编码规范（来自 {conventions_file}）\n{conventions_content}"
return Hooks(on_system_transform=on_system_transform)
示例 D：CI 环境注入
"""
CI 环境变量注入 - 目录插件示例。
确保所有 Shell 命令在 CI 友好的环境中运行。
测试方法：
执行 "env | grep CI"，预期看到 CI=true、RELAY_BUILD=1 等变量。
"""
import logging
logger = logging.getLogger(__name__)
CI_ENV = {
"CI": "true",
"RELAY_BUILD": "1",
"TERM": "dumb",
"NO_COLOR": "1",
"PIP_NO_INPUT": "1",
"NPM_CONFIG_YES": "true",
}
async def plugin(input):
from relay.domain.plugin.model import Hooks
extra = (
input.config
.get("plugins", {}).get("config", {}).get("ci_env_injector", {})
.get("extra_env", {})
)
inject_env = {**CI_ENV, **extra}
logger.info(f"[CIEnvInjector] 注入 {len(inject_env)} 个 CI 环境变量")
async def on_shell_env(inp, out):
if out.env is None:
out.env = {}
out.env.update(inject_env)
return Hooks(on_shell_env=on_shell_env)
调试与验证
查看 Plugin 是否加载
grep "\[PluginManager\]" {user_data}/logs/relay-{port}.log | tail -20
relay-cli -c "列出目录文件"
预期看到：
[PluginManager] Discovered 2 raw plugin metas
[PluginManager] Loaded plugin function: my_plugin
[MyPlugin] 已激活 | session=abc12345...
常见错误
| 错误 | 原因 | 修复 |
|---|
| Plugin 不出现在面板 | 文件放错目录 | 确认放在 .relay/plugins/ 下 |
| Plugin 显示「已发现」但不激活 | 还在旧会话 | 开始新会话 |
AttributeError: out has no attribute xxx | out 字段名拼错 | 参考 Hook 速查表 |
TypeError: plugin() takes 0 arguments | 函数签名错误 | 必须是 async def plugin(input): |
| Hook 不生效 | return 了 None 而非 Hooks | 检查是否漏写 return Hooks(...) |
验证 Plugin 格式
import asyncio, sys
sys.path.insert(0, 'src')
async def test():
from relay.domain.plugin.model import PluginInput
from relay.domain.plugin.service.plugin_loader import PluginLoader
from pathlib import Path
loader = PluginLoader()
metas = await loader.discover_plugins([Path('.relay/plugins')])
for meta in metas:
mod = await loader.import_plugin_module(meta)
fn = loader.get_plugin_function(mod)
print(f"✅ {meta.name}: {fn}")
asyncio.run(test())
创建 Plugin 的流程
当用户要求创建 Plugin 时，按以下步骤进行：
-
明确 Hook 节点：用户想在什么时机触发？
- 工具调用前/后 →
on_tool_execute_before/after
- Shell 命令 →
on_shell_env
- 系统提示 →
on_system_transform
- 权限控制 →
on_permission_ask
-
明确拦截还是增强：
- 拦截（阻止执行）→
out.__blocked__ = True
- 增强（修改行为）→ 修改
out 其他字段
- 监控（只读）→ 只读
inp，不修改 out
-
选择单文件或目录：
- 逻辑简单（<100行）→ 单文件
- 需要多文件/资源 → 目录 Plugin
-
生成完整代码：包含 docstring（说明测试方法）+ logger.info 激活提示
-
说明放置位置和测试方法
注意事项
- Plugin 不应存储进程级状态（违反无状态原则）；如需持久化，写文件到
input.project_home/.relay/
- Hook 函数不 return 值（mutate-in-place，return None）
- import 放在函数体内（避免循环依赖问题）
- 单个 Hook 有 5 秒超时，超时会被警告记录但不影响主流程
- Plugin 异常不会崩溃 Agent，但会在日志中记录警告
