FaustBot 插件系统 API Reference¶

~/.faustbot/plugins/my-plugin/
├── plugin.json              # 插件元数据（必需）
├── impl.py                  # 入口文件（推荐）
├── data/                    # 插件数据目录（可选）
│   └── data.db
└── frontend/                # 前端资源（可选）
    ├── panel.js
    ├── app-hook.js
    └── panel.css

{
  "id": "my-plugin",
  "name": "My Plugin",
  "version": "1.0.0",
  "description": "插件功能描述",
  "author": "FaustBot",
  "homepage": "https://github.com/...",
  "enabled": true,
  "entry": "impl.py",
  "permissions": ["tool:myTool", "middleware:myMiddleware"],
  "priority": 500
}

# 创建触发器
ctx.trigger_create({
    "id": "my_event::1234567890",
    "type": "event",
    "event_name": "my_event",
    "payload": {"summary": "事件摘要"},
    "recall_description": "Agent 可读的描述",
    "lifespan": 7200,
})

# 列出所有触发器
triggers = ctx.trigger_list()

# 获取单个触发器
trigger = ctx.trigger_get("my_event::1234567890")

# 更新触发器
ctx.trigger_update("my_event::1234567890", {"payload": {"summary": "新摘要"}})

# 删除触发器
ctx.trigger_delete("my_event::1234567890")

# 注册配置 schema（list 格式，推荐）
ctx.register_config([
    {"key": "PUSH_THRESHOLD", "type": "int", "label": "推送阈值", "default": 3},
    {"key": "QUIET_START", "type": "str", "label": "静默开始", "default": "23:00"},
    {"key": "ENABLE_FEATURE", "type": "bool", "label": "启用功能", "default": True},
])

# 注册配置 schema（string 格式，旧版兼容）
ctx.register_config("""
PUSH_THRESHOLD:int:推送阈值=3
QUIET_START:str:静默开始=23:00
""")

# 读取配置
threshold = ctx.get_config("PUSH_THRESHOLD", 3)

# 写入配置
ctx.set_config("PUSH_THRESHOLD", 5)

# 列出所有配置
configs = ctx.list_configs()

# 读取文本
content = ctx.vfs_read_text("/plugins/my-plugin/data.md", default="")

# 写入文本
ctx.vfs_write("/plugins/my-plugin/index.md", "# My Plugin\n\n内容...")

# 写入动态内容（符号链接，每次读取时调用函数）
ctx.vfs_write_symbolic(
    "/plugins/my-plugin/dynamic.json",
    lambda path: json.dumps(get_data()),
    should_be_included_in_search=True,
)

# 删除文件
ctx.vfs_delete("/plugins/my-plugin/old.md")

# 列出目录
files = ctx.vfs_list("/plugins/my-plugin/")

@hookimpl
def plugin_loaded(self, ctx: PluginContext) -> None:
    global _PLUGIN
    _PLUGIN = self
    self.ctx = ctx
    self.store = MyStore(ctx.plugin_data_dir or (ctx.plugin_dir / "data"))

@hookimpl
def plugin_unloaded(self, ctx: PluginContext) -> None:
    global _PLUGIN
    if _PLUGIN is self:
        _PLUGIN = None

def startup(self, ctx: PluginContext) -> None:
    self.ctx = ctx
    ctx.register_config([...])
    ctx.vfs_write("/plugins/my-plugin.md", "...")

@hookimpl
def heartbeat(self, ctx: PluginContext) -> None:
    pending = self.store.count_pending()
    if pending >= self.threshold:
        self.push_digest()

@hookimpl
def health_check(self) -> dict | None:
    return {
        "status": "ok",
        "plugin": "my-plugin",
        "items": len(self.store.list_items()),
    }

@hookimpl
def register_frontend(self) -> list[dict]:
    return [
        {"type": "js", "path": "/faust/plugins/my-plugin/frontend/panel.js"},
        {"type": "js", "path": "/faust/plugins/my-plugin/frontend/app-hook.js"},
        {"type": "css", "path": "/faust/plugins/my-plugin/frontend/panel.css"},
    ]

@hookimpl
def communicate_handler(self, payload: dict, ctx: PluginContext) -> dict | None:
    action = str((payload or {}).get("action") or "").strip().lower()
    if action == "get_state":
        return {"status": "ok", "state": self.get_state()}
    return {"status": "error", "detail": f"unknown action: {action}"}

@hookimpl
def register_schedules(self) -> list[dict]:
    return [
        {
            "id": "my-plugin-fetch",
            "interval": 300,
            "callback": self.fetch_data,
            "description": "每5分钟拉取数据",
        },
    ]

async def fetch_data(self):
    # 异步任务
    pass

@hookimpl
def register_pip_deps(self) -> list[str]:
    return ["pandas>=2.0", "httpx"]

from langchain.tools import tool
from faust_backend.plugin_system import ToolSpec

@tool
def my_search(query: str) -> str:
    """搜索指定内容。"""
    return f"搜索结果: {query}"

@hookimpl
def register_tools(self, ctx: PluginContext) -> list:
    return [
        ToolSpec(
            name="mySearch",
            tool=my_search,
            enabled_by_default=True,
            description="自定义搜索工具",
        ),
    ]

from faust_backend.plugin_system import MiddlewareSpec

@hookimpl
def register_middlewares(self, ctx: PluginContext) -> list:
    return [
        MiddlewareSpec(
            name="my_middleware",
            middleware=MyMiddlewareClass(),
            priority=260,
            enabled_by_default=True,
            description="上下文裁剪中间件",
        ),
    ]

@hookimpl
def tool_call_pre(self, name: str, args: dict, ctx: PluginContext) -> dict | None:
    if name == "execute" and "dangerous" in args.get("command", ""):
        return None  # 阻止危险命令
    return args

@hookimpl
def tool_call_post(self, name: str, args: dict, result: Any, ctx: PluginContext) -> Any:
    if name == "read":
        return f"[已缓存] {result}"
    return None  # 不修改

@hookimpl
def message_received(self, msg: Any, history: list, ctx: PluginContext) -> str | None:
    # 记录用户活动时间
    self.last_user_activity_ts = time.time()
    return None  # 不修改消息

@hookimpl
def agent_event_sent(self, event: dict, current_history: list, ctx: PluginContext) -> dict | None:
    # 抑制 EmotionInvoke 工具事件
    if event.get("tool_name") == "EmotionInvoke":
        return None
    return None  # 不修改

@hookimpl
def memory_read_pre(self, query: str, filters: dict | None, ctx: PluginContext) -> str | None:
    # 追加关键词
    return f"{query} 最近更新"

@hookimpl
def memory_read_post(self, query: str, results: list, ctx: PluginContext) -> list | None:
    # 只保留前5条
    return results[:5]

@hookimpl
def memory_write_pre(self, content: str, metadata: dict | None, ctx: PluginContext) -> str | None:
    # 追加标签
    if metadata is None:
        metadata = {}
    metadata["tags"] = ["my-plugin"]
    return content

@hookimpl
def memory_write_post(self, content: str, metadata: dict | None, id: str, ctx: PluginContext) -> None:
    log.info("记忆已写入: %s", id)

@hookimpl
def trigger_append(self, payload: dict, ctx: PluginContext) -> dict | None:
    # 过滤特定类型
    if payload.get("type") == "blocked":
        return None
    return payload

@hookimpl
def trigger_fire(self, payload: dict, ctx: PluginContext) -> dict | None:
    # 修改触发内容
    payload["payload"]["summary"] = f"[已审核] {payload['payload'].get('summary', '')}"
    return payload

@hookimpl
def register_prompt_suffix(self) -> list[str]:
    return [
        "\n[My Plugin]\n"
        "本插件提供 XXX 功能。\n"
        "当用户请求 XXX 时，请使用 mySearch 工具。\n"
    ]

@hookimpl
def config_changed(self, key: str, old: Any, new: Any, ctx: PluginContext) -> None:
    if key == "PUSH_THRESHOLD":
        self.threshold = int(new or 3)

@dataclass
class ToolSpec:
    name: str                    # 工具名称
    tool: Any                    # 工具函数或 StructuredTool
    enabled_by_default: bool = True
    description: str = ""

@dataclass
class MiddlewareSpec:
    name: str                    # 中间件名称
    middleware: Any              # 中间件实例
    priority: int = 100          # 优先级（越小越先执行）
    enabled_by_default: bool = True
    description: str = ""

@dataclass
class PluginManifest:
    plugin_id: str
    name: str
    version: str = "0.1.0"
    description: str = ""
    author: str = ""
    homepage: str = ""
    enabled: bool = True
    entry: str = "main.py"
    permissions: list[str] = field(default_factory=list)
    priority: int = 100

{"status": "ok", ...}

{"status": "error", "detail": "错误描述"}

const result = await window.pluginUI.communicate('my-plugin', {
  action: 'get_state'
});

const result = await window.faustAppUI.communicate('my-plugin', {
  action: 'get_context'
});