Merge pull request #1125 from EstrellaXD/refactor/4.0-p5-events-api

feat(4.0-P5): 系统事件上总线、插件路由 / MCP 工具 / 通知模板

CHANGELOG.md

@@ -6,7 +6,10 @@
 - **新增（插件扩展点）**：插件可以提供下载器、通知渠道、LLM 解析提供商、搜索站点和定时任务（`ab_sdk.points`）。内置的 qB / aria2 和各通知渠道走同一套扩展注册表。设置页新增「插件」卡片，可查看状态、启停插件，并通过插件配置模型自动生成的表单修改配置；`secret_field` 声明的秘密字段在读取接口中一律掩码。下载器类型和通知渠道下拉框会列出插件提供的选项。新增插件开发文档（开发者指南 → 插件开发）
 - **新增（ingest 扩展点）**：RSS 下载流程开放 `torrent.filter`（种子过滤）、`title.parsed`（修正解析结果）、`torrent.adding`（修改保存路径 / 分类 / 标签）、`http.request`（为私有站点添加 Cookie 等请求头）四个钩子，以及 `metadata_provider` 元数据源扩展点：内置 mikan / tmdb 以 `core` 身份登记，插件可按 RSS 订阅的「解析器」取值提供新的元数据源，添加订阅时的解析器下拉框会列出插件选项。插件 SDK 升至 0.3.0，新增 `ab_sdk.ingest`。未安装插件时行为不变
 - **新增（种子过滤）**：新增默认启用的内置插件「种子过滤」（`ingest-filters`）。在 设置 → 插件 中填写「包含过滤」后，只下载名称匹配任一表达式的种子（对所有订阅生效）；留空时不过滤
+- **新增（事件与外部接口）**：系统事件（RSS 订阅异常、种子添加失败、下载器不可用、新版本等）在写入通知中心后发布到插件事件总线，插件可用 `@subscribe("rss_failure")` 等订阅；事件类移入 `ab_sdk.events`（`SystemEvent` 基类，新增 `i18n()` 返回前端 i18n key 与参数）。插件可以提供 REST 路由（挂载于 `/api/v1/plugins/<id>/`，强制登录鉴权，随插件启停出现和消失）、MCP 工具（工具名 `<插件 id>__<id>`）与 MCP 资源（`autobangumi://plugins/<插件 id>/<id>`），并可通过 `message_template` 钩子按事件类型和渠道改写系统通知的标题与正文。`ab_sdk` 升至 0.3.0
+- **优化（通知中心推送）**：SSE 的 `notification` 帧改为订阅事件总线，通知中心有新消息、已读或删除时立即推送（原先每 3 秒比较一次修订号），推送格式不变
 - **修复（搜索）**：`GET /api/v1/search/provider` 读取的是导入时的站点快照，新保存的搜索站点要重启后才会出现在列表中；现在实时读取
+- **修复（MCP）**：MCP 客户端每次断开 SSE 连接后，日志都会出现 `AssertionError: Unexpected message http.response.start`（3.x 起存在）。原因是 `/sse` 端点在 SSE 响应结束后又返回了一个空 `Response`，现在改为裸 ASGI 端点
 - **破坏性变更（升级路径）**：只支持从 3.3.x 升级。检测到更早版本的数据（`config/version.info` 低于 3.3，或残留 2.x 的 `data/data.json`）时拒绝启动，并提示先升级到最新 3.3.x 启动一次完成迁移
 - **破坏性变更（API）**：移除 3.2 兼容的 GET 控制端点（`/api/v1/restart`、`/start`、`/stop`、`/shutdown`）和 `GET /api/v1/auth/refresh_token`，请改用 POST
 - **破坏性变更（配置）**：移除旧版 `experimental_openai` 配置节和通知的单 provider 旧字段（`type` / `token` / `chat_id`），3.3 已把它们迁移到 `llm` 与 `notification.providers`；移除废弃的 `normal` 重命名方式，已有配置自动改为语义相同的 `none`

backend/src/ab_sdk/events.py

@@ -27,3 +27,310 @@ class PluginDisabled(Event):
     kind: ClassVar[str] = "plugin.disabled"
     plugin_id: str
     reason: str
+
+
+# --- P5：系统事件 -------------------------------------------------------------
+#
+# 以下事件由宿主经 ``NotificationManager.send_event`` 发出：先写入站内通知中心，
+# 再发布到事件总线（插件可 ``@subscribe(kind)``），最后推送到外部通知渠道。
+# ``kind`` 沿用 3.x 的取值（站内通知中心按 kind 存储与翻译），不带 ``.`` 前缀。
+
+
+@dataclass(frozen=True, slots=True)
+class SystemEvent(Event):
+    """可通知事件的基类：同时服务外部推送、站内通知中心与事件总线。
+
+    - ``describe()`` 返回默认中文 (标题, 正文)，用于外部推送与通知中心兜底展示；
+    - ``i18n()`` 返回 (i18n key, 参数)，前端据此按当前语言渲染，key 形如
+      ``notifications.kind.<kind>``，其下有 ``title`` / ``body`` 两条文案；
+    - ``severity`` 为 ``info`` / ``warning`` / ``error``；
+    - ``once=True`` 表示同 ``dedup_key()`` 终生只入库一次（如「新版本可用」）。
+
+    插件也可以继承它定义自己的可通知事件（``kind`` 须以 ``<plugin-id>.`` 开头）。
+    """
+
+    kind: ClassVar[str] = "system"
+    severity: ClassVar[str] = "info"
+    once: ClassVar[bool] = False
+
+    def dedup_key(self) -> str | None:
+        """通知中心的去重键；None 表示每次都新建一条。"""
+        return None
+
+    def payload(self) -> dict:
+        """结构化字段，入库为 JSON，也是 i18n 文案的插值参数。"""
+        from dataclasses import fields
+
+        return {f.name: getattr(self, f.name) for f in fields(self)}
+
+    def describe(self) -> tuple[str, str]:
+        """默认 (标题, 正文)。"""
+        return (self.kind, "")
+
+    def i18n(self) -> tuple[str, dict]:
+        """(i18n key, 插值参数)。默认 key 为 ``notifications.kind.<kind>``。"""
+        return f"notifications.kind.{self.kind}", self.payload()
+
+
+@dataclass(frozen=True, slots=True)
+class RssFailureEvent(SystemEvent):
+    """RSS 订阅连接状态从正常变为异常（仅状态翻转时触发一次，而非每个 tick）。"""
+
+    kind: ClassVar[str] = "rss_failure"
+    severity: ClassVar[str] = "error"
+    once: ClassVar[bool] = False
+
+    rss_name: str
+    rss_url: str
+    error: str
+
+    def dedup_key(self) -> str | None:
+        return f"rss_failure:{self.rss_url}"
+
+    def payload(self) -> dict:
+        return {"rss_name": self.rss_name, "rss_url": self.rss_url, "error": self.error}
+
+    def describe(self) -> tuple[str, str]:
+        """返回该事件的默认 (标题, 正文)。"""
+        return (
+            "RSS 订阅连接异常",
+            f"订阅：{self.rss_name}\n地址：{self.rss_url}\n错误：{self.error}",
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class DownloadFailureEvent(SystemEvent):
+    """匹配到的种子在（内置 HTTP 重试后）仍添加下载器失败。"""
+
+    kind: ClassVar[str] = "download_failure"
+    severity: ClassVar[str] = "error"
+    once: ClassVar[bool] = False
+
+    official_title: str
+    torrent_name: str
+
+    def dedup_key(self) -> str | None:
+        return f"download_failure:{self.official_title}"
+
+    def payload(self) -> dict:
+        return {
+            "official_title": self.official_title,
+            "torrent_name": self.torrent_name,
+        }
+
+    def describe(self) -> tuple[str, str]:
+        return (
+            "种子添加失败",
+            f"番剧：{self.official_title}\n种子：{self.torrent_name}\n"
+            "重试后仍添加失败，请检查下载器连接。",
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class OffsetReviewEvent(SystemEvent):
+    """番剧被标记为需要人工确认季度/集数偏移。"""
+
+    kind: ClassVar[str] = "offset_review"
+    severity: ClassVar[str] = "warning"
+    once: ClassVar[bool] = False
+
+    official_title: str
+    reason: str
+
+    def dedup_key(self) -> str | None:
+        return f"offset_review:{self.official_title}"
+
+    def payload(self) -> dict:
+        return {"official_title": self.official_title, "reason": self.reason}
+
+    def describe(self) -> tuple[str, str]:
+        return (
+            "集数偏移待确认",
+            f"番剧：{self.official_title}\n原因：{self.reason}\n请前往设置页确认偏移量。",
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class DownloaderUnavailableEvent(SystemEvent):
+    """下载器不可用：连不上、用户名/密码错误或 IP 被封禁。
+
+    ``reason`` 只进 payload 不进 dedup_key——同一台下载器从 unreachable 翻转
+    到 credentials 时合并为一条（内容以最新为准），不产生两行。
+    """
+
+    kind: ClassVar[str] = "downloader_unavailable"
+    severity: ClassVar[str] = "error"
+    once: ClassVar[bool] = False
+
+    host: str
+    reason: str  # unreachable | credentials | banned
+
+    def dedup_key(self) -> str | None:
+        return f"downloader:{self.host}"
+
+    def payload(self) -> dict:
+        return {"host": self.host, "reason": self.reason}
+
+    def describe(self) -> tuple[str, str]:
+        details = {
+            "credentials": "用户名或密码错误，请在设置中检查下载器凭据。",
+            "banned": "IP 已被下载器封禁，请在下载器 WebUI 中解封或重启下载器。",
+            "unreachable": "无法连接下载器，请检查地址、端口和网络。",
+        }
+        detail = details.get(self.reason, details["unreachable"])
+        return ("下载器连接异常", f"下载器：{self.host}\n{detail}")
+
+
+@dataclass(frozen=True, slots=True)
+class UpdateAvailableEvent(SystemEvent):
+    """检查到可用的新版本。"""
+
+    kind: ClassVar[str] = "update_available"
+    severity: ClassVar[str] = "info"
+    once: ClassVar[bool] = True
+
+    current: str
+    latest: str
+    channel: str
+    notes: str = ""
+
+    def dedup_key(self) -> str | None:
+        return f"update_available:{self.latest}"
+
+    def payload(self) -> dict:
+        return {
+            "current": self.current,
+            "latest": self.latest,
+            "channel": self.channel,
+            "notes": self.notes,
+        }
+
+    def describe(self) -> tuple[str, str]:
+        return (
+            "发现新版本",
+            f"当前版本：{self.current}\n最新版本：{self.latest}（{self.channel} 频道）\n"
+            "可前往 设置 → 软件更新 升级。",
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class UpdateAppliedEvent(SystemEvent):
+    """更新应用结果（成功或失败）。每次结果都单独入库，不做去重。"""
+
+    once: ClassVar[bool] = False
+
+    version: str
+    success: bool
+    message: str = ""
+
+    # kind / severity 随结果变化，因此用属性覆盖基类的类变量
+    @property
+    def kind(self) -> str:  # type: ignore[override]
+        return "update_applied" if self.success else "update_failed"
+
+    @property
+    def severity(self) -> str:  # type: ignore[override]
+        return "info" if self.success else "error"
+
+    def dedup_key(self) -> str | None:
+        return None
+
+    def payload(self) -> dict:
+        return {"version": self.version, "message": self.message}
+
+    def describe(self) -> tuple[str, str]:
+        if self.success:
+            return ("程序更新完成", f"已更新到 {self.version}，重启后生效。")
+        return ("程序更新失败", f"版本：{self.version}\n原因：{self.message}")
+
+
+@dataclass(frozen=True, slots=True)
+class LLMAuthFailureEvent(SystemEvent):
+    """订阅类 LLM 提供商凭据失效（刷新失败），需要用户重新连接。"""
+
+    kind: ClassVar[str] = "llm_auth_failure"
+    severity: ClassVar[str] = "error"
+    once: ClassVar[bool] = False
+
+    provider_id: str
+    account_label: str = ""
+    message: str = ""
+
+    def dedup_key(self) -> str | None:
+        return f"llm_auth:{self.provider_id}"
+
+    def payload(self) -> dict:
+        return {
+            "provider_id": self.provider_id,
+            "account_label": self.account_label,
+            "message": self.message,
+        }
+
+    def describe(self) -> tuple[str, str]:
+        account = f"（{self.account_label}）" if self.account_label else ""
+        return (
+            "LLM 提供商凭据失效",
+            f"提供商：{self.provider_id}{account}\n{self.message}\n"
+            "请前往 设置 → LLM 解析器 重新连接。",
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class LLMPluginInstallFailedEvent(SystemEvent):
+    """LLM 提供商插件安装失败（下载/签名校验/兼容性）。"""
+
+    kind: ClassVar[str] = "llm_plugin_install_failed"
+    severity: ClassVar[str] = "error"
+    once: ClassVar[bool] = False
+
+    plugin_id: str
+    version: str = ""
+    message: str = ""
+
+    def dedup_key(self) -> str | None:
+        return f"llm_plugin_install:{self.plugin_id}"
+
+    def payload(self) -> dict:
+        return {
+            "plugin_id": self.plugin_id,
+            "version": self.version,
+            "message": self.message,
+        }
+
+    def describe(self) -> tuple[str, str]:
+        return (
+            "LLM 插件安装失败",
+            f"插件：{self.plugin_id} {self.version}\n原因：{self.message}",
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class RenameConflictEvent(SystemEvent):
+    """媒体文件重命名遇到无法自动解决的目标路径冲突。"""
+
+    kind: ClassVar[str] = "rename_conflict"
+    severity: ClassVar[str] = "warning"
+    once: ClassVar[bool] = False
+
+    task_id: str
+    torrent_name: str
+    target_path: str
+    reason: str
+
+    def dedup_key(self) -> str | None:
+        return f"rename_conflict:{self.task_id}:{self.target_path}"
+
+    def payload(self) -> dict:
+        return {
+            "task_id": self.task_id,
+            "torrent_name": self.torrent_name,
+            "target_path": self.target_path,
+            "reason": self.reason,
+        }
+
+    def describe(self) -> tuple[str, str]:
+        return (
+            "媒体文件重命名冲突",
+            f"种子：{self.torrent_name}\n目标：{self.target_path}\n"
+            f"原因：{self.reason}",
+        )

backend/src/ab_sdk/mcp.py

@@ -0,0 +1,52 @@
+"""MCP 工具与资源契约。
+
+插件用 ``@provider(points.MCP_TOOL, id="search")`` 返回 :class:`McpTool`，
+用 ``@provider(points.MCP_RESOURCE, id="stats")`` 返回 :class:`McpResource`。
+``id`` 只需在插件内唯一，宿主对外暴露时加上插件 id 前缀以避免冲突：
+
+- 工具名：``<plugin-id>__<id>``（双下划线。不用 ``.``，因为不少 MCP 客户端
+  所接的 LLM API 只接受 ``^[a-zA-Z0-9_-]{1,64}$`` 的工具名）
+- 资源 URI：``autobangumi://plugins/<plugin-id>/<id>``
+
+MCP 端点（``/mcp``）沿用宿主的访问控制（IP 白名单 / ``scope=mcp`` 令牌），
+插件无需自行鉴权。
+"""
+
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from typing import Any
+
+DEFAULT_TOOL_TIMEOUT = 60.0
+
+
+@dataclass(frozen=True)
+class McpTool:
+    """一个 MCP 工具。
+
+    ``handler`` 接收客户端传入的参数字典（已按 ``input_schema`` 校验），返回
+    可 JSON 序列化的结果；宿主把结果序列化后作为文本内容返回给客户端。抛出的
+    异常会以 ``{"error": "..."}`` 返回，并计入插件熔断。
+    """
+
+    description: str
+    handler: Callable[[dict[str, Any]], Awaitable[Any]]
+    # JSON Schema；默认不接受参数
+    input_schema: dict[str, Any] = field(
+        default_factory=lambda: {"type": "object", "properties": {}}
+    )
+    # 执行超时（秒）
+    timeout: float = DEFAULT_TOOL_TIMEOUT
+
+
+@dataclass(frozen=True)
+class McpResource:
+    """一个只读的 MCP 资源。
+
+    ``handler`` 不接受参数，返回字符串（原样返回）或可 JSON 序列化的对象
+    （宿主序列化为 JSON）。
+    """
+
+    name: str
+    handler: Callable[[], Awaitable[Any]]
+    description: str = ""
+    mime_type: str = "application/json"

backend/src/ab_sdk/notify.py

@@ -46,3 +46,15 @@ class Notifier(Protocol):
 
 
 NotifierFactory = Callable[[NotifierSettings], Notifier]
+
+
+@dataclass(frozen=True)
+class RenderedMessage:
+    """``message_template`` 钩子处理的消息：系统事件推送前的标题与正文。
+
+    初始值来自事件的 ``describe()``；钩子返回新的 RenderedMessage 即可改写，
+    返回 None 表示不修改。多个钩子按优先级依次处理，后一个拿到前一个的结果。
+    """
+
+    title: str
+    body: str

backend/src/ab_sdk/points.py

@@ -32,3 +32,25 @@ TORRENT_FILTER = "torrent.filter"
 TITLE_PARSED = "title.parsed"
 TORRENT_ADDING = "torrent.adding"
 HTTP_REQUEST = "http.request"
+
+# --- P5：外部接口与通知模板 ---
+#
+# Provider 扩展点（id 只需在插件内唯一，宿主对外暴露时加插件 id 前缀）：
+#
+# - ``API_ROUTER``：返回 ``fastapi.APIRouter``，挂载于
+#   ``/api/v1/plugins/<plugin-id>/``，强制登录鉴权
+# - ``MCP_TOOL``：返回 :class:`ab_sdk.mcp.McpTool`，MCP 工具名为
+#   ``<plugin-id>__<id>``
+# - ``MCP_RESOURCE``：返回 :class:`ab_sdk.mcp.McpResource`，URI 为
+#   ``autobangumi://plugins/<plugin-id>/<id>``
+#
+# Transform 钩子（``@hook``）：
+#
+# - ``MESSAGE_TEMPLATE``：签名 ``(message: RenderedMessage, event: SystemEvent,
+#   channel: str) -> RenderedMessage | None``，在系统事件推送到外部通知渠道前
+#   改写标题与正文；``channel`` 为渠道类型（如 ``telegram``）
+
+API_ROUTER = "api_router"
+MCP_TOOL = "mcp_tool"
+MCP_RESOURCE = "mcp_resource"
+MESSAGE_TEMPLATE = "message_template"

backend/src/module/api/__init__.py

@@ -10,6 +10,7 @@ from .log import router as log_router
 from .movie import router as movie_router
 from .notification import router as notification_router
 from .passkey import router as passkey_router
+from .plugin_routes import router as plugin_routes_router
 from .plugins import router as plugins_router
 from .program import router as program_router
 from .rss import router as rss_router
@@ -41,3 +42,5 @@ v1.include_router(notification_router)
 v1.include_router(update_router)
 v1.include_router(llm_router)
 v1.include_router(plugins_router)
+# 插件路由分发须排在插件管理路由之后（后者的路径更具体）
+v1.include_router(plugin_routes_router)

backend/src/module/api/events.py

@@ -1,8 +1,13 @@
 """SSE 端点：合并推送 status/downloader/log 更新，取代前端的三个轮询循环
 (hooks/useAppInfo.ts:24、pages/index/downloader.vue:28、store/log.ts:25)。
 
-单个连接按 tick 节拍分别推送三种事件，节拍与原轮询间隔保持一致，
-前端订阅失败时回退到原有轮询逻辑。
+单个连接按 tick 节拍分别推送 status/downloader/log/update，节拍与原轮询间隔
+保持一致，前端订阅失败时回退到原有轮询逻辑。这几类是「状态快照」而非事件，
+仍按节拍采样。
+
+notification（通知中心未读数）由事件驱动：每个连接在插件事件总线上订阅
+``inbox.changed``（通知中心写入/已读/删除时发布），收到后立即查库推送，
+不再按节拍比较修订号。连接建立时推送一帧作为初始状态。
 """
 
 import asyncio
@@ -18,7 +23,8 @@ from module.conf import LOG_PATH, VERSION
 from module.core import AppContext
 from module.database import Database
 from module.downloader import DownloadClient
-from module.notification.inbox import inbox_revision
+from module.notification.inbox import InboxChanged, inbox_revision
+from module.plugin.host import get_bus
 from module.security.api import get_current_user
 from module.update import get_update_progress
 
@@ -32,7 +38,6 @@ _TICK_SECONDS = 1
 _STATUS_EVERY = 3
 _DOWNLOADER_EVERY = 5
 _LOG_EVERY = 10
-_NOTIFICATION_EVERY = 3
 
 # 下载器查询的超时上限（秒）。下载器不可达时 qB/aria2 的认证重试可能阻塞
 # 20-30s，而 SSE 是单连接串行推送——不设上限会把 status/log 事件一起卡住。
@@ -87,57 +92,112 @@ async def _log_payload() -> str | None:
     return data.decode("utf-8", errors="replace")
 
 
+class _InboxWatch:
+    """一个 SSE 连接对通知中心变化的订阅。
+
+    总线回调只置位 ``wake``；``pending`` 表示还欠前端一帧（初始帧、或上次查库
+    失败需要重试）。帧的 ``revision`` 取推送时的修订号，与旧版轮询一致。
+    """
+
+    def __init__(self) -> None:
+        self.wake = asyncio.Event()
+        self.pending = True
+        bus = get_bus()
+        self._unsubscribe = (
+            bus.subscribe(InboxChanged.kind, self._on_change)
+            if bus is not None
+            else None
+        )
+
+    def _on_change(self, _event) -> None:
+        self.wake.set()
+
+    def close(self) -> None:
+        if self._unsubscribe is not None:
+            self._unsubscribe()
+            self._unsubscribe = None
+
+    async def frame(self) -> dict | None:
+        """有待推送的变化时查库并返回 SSE 帧；没有或查库失败时返回 None。"""
+        if self.wake.is_set():
+            self.wake.clear()
+            self.pending = True
+        if not self.pending:
+            return None
+        rev = inbox_revision()
+        try:
+            payload = await _notification_payload()
+        except Exception:
+            logger.debug("SSE: notification payload unavailable", exc_info=True)
+            return None
+        self.pending = False
+        payload["revision"] = rev
+        return {"event": "notification", "data": json.dumps(payload)}
+
+    async def wait(self, deadline: float) -> bool:
+        """等到 ``deadline``（事件循环时间）或通知中心有变化；有变化返回 True。"""
+        remaining = deadline - asyncio.get_running_loop().time()
+        if remaining <= 0:
+            return False
+        try:
+            await asyncio.wait_for(self.wake.wait(), remaining)
+        except TimeoutError:
+            return False
+        return True
+
+
 async def _event_generator(request: Request, ctx: AppContext):
-    """按各自节拍推送 status/downloader/log 事件的核心循环。"""
+    """按各自节拍推送 status/downloader/log 事件、按需推送 notification 的核心循环。"""
+    loop = asyncio.get_running_loop()
+    inbox = _InboxWatch()
     tick = 0
     last_update: str | None = None
-    last_inbox_rev: int | None = None
-    while True:
-        if await request.is_disconnected():
-            break
+    try:
+        while True:
+            if await request.is_disconnected():
+                break
 
-        # 更新进度：仅在有进行中的更新（phase != idle）且负载变化时推送，
-        # 避免重复帧刷屏；每 tick 检查一次以保证下载进度足够实时。
-        progress = get_update_progress()
-        if progress.get("phase") != "idle":
-            payload = json.dumps(progress)
-            if payload != last_update:
-                last_update = payload
-                yield {"event": "update", "data": payload}
+            # 更新进度：仅在有进行中的更新（phase != idle）且负载变化时推送，
+            # 避免重复帧刷屏；每 tick 检查一次以保证下载进度足够实时。
+            progress = get_update_progress()
+            if progress.get("phase") != "idle":
+                payload = json.dumps(progress)
+                if payload != last_update:
+                    last_update = payload
+                    yield {"event": "update", "data": payload}
 
-        if tick % _STATUS_EVERY == 0:
-            yield {"event": "status", "data": json.dumps(_status_payload(ctx))}
+            if tick % _STATUS_EVERY == 0:
+                yield {"event": "status", "data": json.dumps(_status_payload(ctx))}
 
-        if tick % _DOWNLOADER_EVERY == 0:
-            # 不可用时推送 null 而非跳过，前端据此感知下载器降级状态
-            # （store/downloader.ts 对 null 保留旧数据，不会崩溃）。
-            torrents = await _downloader_payload()
-            yield {"event": "downloader", "data": json.dumps(torrents)}
+            if tick % _DOWNLOADER_EVERY == 0:
+                # 不可用时推送 null 而非跳过，前端据此感知下载器降级状态
+                # （store/downloader.ts 对 null 保留旧数据，不会崩溃）。
+                torrents = await _downloader_payload()
+                yield {"event": "downloader", "data": json.dumps(torrents)}
 
-        if tick % _LOG_EVERY == 0:
-            log_text = await _log_payload()
-            if log_text is not None:
-                yield {"event": "log", "data": log_text}
+            if tick % _LOG_EVERY == 0:
+                log_text = await _log_payload()
+                if log_text is not None:
+                    yield {"event": "log", "data": log_text}
 
-        # 通知中心：只在修订号变化时查库推送（写入/已读/删除都会 bump），
-        # 平时每 3 tick 只做一次整数比较，不产生数据库查询。
-        if tick % _NOTIFICATION_EVERY == 0:
-            rev = inbox_revision()
-            if rev != last_inbox_rev:
-                try:
-                    notif_payload = await _notification_payload()
-                except Exception:
-                    logger.debug("SSE: notification payload unavailable", exc_info=True)
-                else:
-                    last_inbox_rev = rev
-                    notif_payload["revision"] = rev
-                    yield {"event": "notification", "data": json.dumps(notif_payload)}
+            # 通知中心：连接建立时的初始帧，以及此前查库失败后的重试
+            frame = await inbox.frame()
+            if frame is not None:
+                yield frame
 
-        await asyncio.sleep(_TICK_SECONDS)
-        tick += _TICK_SECONDS
+            # 等待下一个 tick；其间 inbox.changed 到达时立即推送
+            deadline = loop.time() + _TICK_SECONDS
+            while await inbox.wait(deadline):
+                frame = await inbox.frame()
+                if frame is not None:
+                    yield frame
+            tick += 1
+    finally:
+        inbox.close()
 
 
 @router.get("/stream", dependencies=[Depends(get_current_user)])
 async def event_stream(request: Request, ctx: AppContext = Depends(get_context)):
-    """单一 SSE 连接，按各自节拍推送 status/downloader/log 事件。"""
+    """单一 SSE 连接，按各自节拍推送 status/downloader/log 事件，
+    通知中心有变化时推送 notification 事件。"""
     return EventSourceResponse(_event_generator(request, ctx))

backend/src/module/api/plugin_routes.py

@@ -0,0 +1,69 @@
+"""插件 REST 路由的分发端点：``/api/v1/plugins/{plugin_id}/{path:path}``。
+
+鉴权在这里以路由依赖的形式强制（``get_current_user``），插件路由无法绕过，
+也就无法注册匿名端点。请求转发给 ``AppContext.plugin_routes`` 中该插件的
+路由；插件未启用或未提供路由时返回 404。
+
+该路由至少需要 ``/plugins/<id>/`` 之后的一段（可为空），因此不会遮蔽
+``module/api/plugins.py`` 中的 ``GET /plugins``、``PUT /plugins/settings``、
+``GET /plugins/providers`` 与 ``PUT /plugins/{id}``。
+"""
+
+from fastapi import APIRouter, Depends, HTTPException
+from starlette.responses import Response
+from starlette.types import Receive, Scope, Send
+
+from module.core import AppContext
+from module.core.plugin_routes import PluginRoutes
+from module.security.api import get_current_user
+
+from .deps import get_context
+
+router = APIRouter(prefix="/plugins", tags=["plugins"])
+
+_METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]
+
+
+class _Forward(Response):
+    """把当前请求原样交给插件路由，由插件路由自己发送响应。
+
+    请求体未被读取（端点不声明 body 参数），插件路由可以正常读取。
+    """
+
+    def __init__(self, routes: PluginRoutes, plugin_id: str, sub_path: str) -> None:
+        super().__init__()
+        self._routes = routes
+        self._plugin_id = plugin_id
+        self._sub_path = sub_path
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        path: str = scope["path"]
+        # 与 Starlette Mount 一致：path 保持完整，root_path 指向挂载前缀，
+        # 插件路由按 ``path - root_path`` 匹配（即 /api/v1/plugins/<id> 之后的部分）
+        child = {
+            k: v
+            for k, v in scope.items()
+            if k not in ("route", "endpoint", "path_params", "router")
+        }
+        child["root_path"] = path[: len(path) - len(self._sub_path)]
+        handled = await self._routes.dispatch(self._plugin_id, child, receive, send)
+        if not handled:
+            raise HTTPException(status_code=404, detail="Not Found")
+
+
+@router.api_route(
+    "/{plugin_id}/{path:path}",
+    methods=_METHODS,
+    dependencies=[Depends(get_current_user)],
+    include_in_schema=False,
+)
+async def plugin_route(
+    plugin_id: str, path: str, ctx: AppContext = Depends(get_context)
+):
+    """转发到插件 ``plugin_id`` 通过 ``points.API_ROUTER`` 提供的路由。"""
+    routes = ctx.plugin_routes
+    if plugin_id not in routes.plugin_ids():
+        raise HTTPException(
+            status_code=404, detail=f"Unknown plugin route: {plugin_id}"
+        )
+    return _Forward(routes, plugin_id, "/" + path)

backend/src/module/core/context.py

@@ -20,8 +20,7 @@ from module.parser.analyser.mikan_parser import reset_cache as reset_mikan_cache
 from module.parser.analyser.tmdb_parser import reset_cache as reset_tmdb_cache
 from module.parser.title_parser import reset_cache as reset_llm_parser
 from module.plugin import PluginManager
-from module.plugin import host as plugin_host
-from module.plugin.host import get_registry
+from module.plugin.host import get_registry, set_bus, set_runner
 from module.rss import RSSAnalyser
 from module.searcher.searcher import reset_cache as reset_poster_cache
 from module.update import (
@@ -38,6 +37,7 @@ from .loops import (
     rss_tick,
     update_check_tick,
 )
+from .plugin_routes import PluginRoutes
 from .plugin_tasks import PluginTasks
 from .scheduler import PeriodicTask, Scheduler
 
@@ -90,10 +90,12 @@ class AppContext:
         self._plugin_tasks = PluginTasks(
             scheduler, self.plugins.registry, self.plugins.breaker
         )
+        self.plugin_routes = PluginRoutes(self.plugins.registry, self.plugins.breaker)
         self.plugins.on_change = self._on_plugins_changed
-        # 模块级代码（RSS 引擎、下载门面、网络层）没有 ctx，经进程级访问器
-        # 拿到 runner 执行 ingest 钩子
-        plugin_host.set_runner(self.plugins.runner)
+        # 宿主各处（通知管理器、站内通知中心、流水线）经 module.plugin.host 发布
+        # 事件与执行钩子，统一指向本进程的插件管理器
+        set_bus(self.plugins.bus)
+        set_runner(self.plugins.runner)
         # Downloader-status TTL cache (was ProgramStatus.check_downloader_status).
         self._downloader_status = False
         self._downloader_reason: str | None = None
@@ -112,6 +114,7 @@ class AppContext:
     async def _on_plugins_changed(self) -> None:
         """插件加载/重载/停用后，重建依赖插件 Provider 的派生状态。"""
         await self._plugin_tasks.sync()
+        self.plugin_routes.sync()
         # 插件提供的通知渠道在插件加载后才可解析；LLM 解析器可能缓存了旧的
         # 插件适配器。下载器客户端缓存键包含 Provider 登记项，会自动换新。
         self.notifier.rebuild()

backend/src/module/core/plugin_routes.py

@@ -0,0 +1,92 @@
+"""插件 REST 路由（points.API_ROUTER）的挂载表。
+
+FastAPI 不支持在运行时卸载路由，因此宿主只注册一个固定的分发路由
+``/api/v1/plugins/{plugin_id}/{path:path}``（见 ``module/api/plugin_routes.py``，
+鉴权在那里强制），再按 plugin_id 转发给这里维护的、由插件 ``APIRouter`` 组成的
+ASGI 应用。插件加载/重载/停用后 :meth:`PluginRoutes.sync` 重建该表，路由随插件
+出现和消失。
+"""
+
+import logging
+
+from fastapi import APIRouter
+from starlette.exceptions import HTTPException
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+from ab_sdk import points
+from module.plugin.registry import ExtensionRegistry, ProviderEntry
+from module.plugin.runner import CircuitBreaker
+
+logger = logging.getLogger(__name__)
+
+
+class PluginRoutes:
+    def __init__(self, registry: ExtensionRegistry, breaker: CircuitBreaker) -> None:
+        self._registry = registry
+        self._breaker = breaker
+        # plugin_id -> (构建时的 Provider 登记项, 合并后的路由)
+        self._apps: dict[str, tuple[tuple[ProviderEntry, ...], APIRouter]] = {}
+
+    def sync(self) -> None:
+        """按注册表重建：插件重载后登记项是新对象，据此替换为新实例的路由。"""
+        grouped: dict[str, list[ProviderEntry]] = {}
+        for entry in self._registry.providers(points.API_ROUTER).values():
+            grouped.setdefault(entry.plugin_id, []).append(entry)
+        apps: dict[str, tuple[tuple[ProviderEntry, ...], APIRouter]] = {}
+        for plugin_id, entries in grouped.items():
+            key = tuple(sorted(entries, key=lambda e: e.id))
+            current = self._apps.get(plugin_id)
+            if current is not None and current[0] == key:
+                apps[plugin_id] = current
+                continue
+            router = self._build(plugin_id, key)
+            if router is not None:
+                apps[plugin_id] = (key, router)
+        self._apps = apps
+
+    def plugin_ids(self) -> list[str]:
+        return sorted(self._apps)
+
+    def _build(
+        self, plugin_id: str, entries: tuple[ProviderEntry, ...]
+    ) -> APIRouter | None:
+        combined = APIRouter()
+        for entry in entries:
+            try:
+                router = entry.factory()
+                if not isinstance(router, APIRouter):
+                    raise TypeError(f"应返回 fastapi.APIRouter，收到 {type(router)}")
+                combined.include_router(router)
+            except Exception as e:
+                reason = f"{points.API_ROUTER} {entry.id}: {type(e).__name__}: {e}"
+                logger.warning("[Plugin:%s] 路由创建失败：%s", plugin_id, reason)
+                self._breaker.record_failure(plugin_id, reason)
+                return None
+        logger.debug("[Plugin:%s] 已挂载 %d 条路由", plugin_id, len(combined.routes))
+        return combined
+
+    async def dispatch(
+        self, plugin_id: str, scope: Scope, receive: Receive, send: Send
+    ) -> bool:
+        """把请求交给插件路由处理；插件没有路由时返回 False（由调用方回 404）。
+
+        ``scope`` 应已按 Starlette ``Mount`` 的约定设置好 ``root_path``，插件
+        路由据此只看到 ``/api/v1/plugins/<id>`` 之后的路径。插件处理函数抛出的
+        异常计入熔断后继续向上抛，由宿主的异常中间件返回 500。
+        """
+        found = self._apps.get(plugin_id)
+        if found is None:
+            return False
+        app: ASGIApp = found[1]
+        try:
+            await app(scope, receive, send)
+        except HTTPException:
+            # 未匹配的路径（404/405）或插件主动抛出的 HTTP 错误，不算插件故障
+            raise
+        except Exception as e:
+            self._breaker.record_failure(
+                plugin_id, f"{points.API_ROUTER}: {type(e).__name__}: {e}"
+            )
+            raise
+        self._breaker.record_success(plugin_id)
+        return True

backend/src/module/mcp/plugins.py

@@ -0,0 +1,182 @@
+"""插件提供的 MCP 工具与资源（points.MCP_TOOL / points.MCP_RESOURCE）。
+
+每次 list / call 都从扩展注册表现取，插件加载、重载、停用后立即生效。对外
+名称加插件 id 前缀以避免与内置及其它插件冲突：
+
+- 工具名 ``<plugin-id>__<id>``：不少 MCP 客户端会把工具名原样交给 LLM API，
+  而 Anthropic / OpenAI 的工具名只接受 ``^[a-zA-Z0-9_-]{1,64}$``，所以不用 ``.``。
+  插件 id 只含小写字母、数字与连字符，第一个 ``__`` 即为分隔符，不会歧义。
+- 资源 URI ``autobangumi://plugins/<plugin-id>/<id>``。
+
+插件处理函数失败或超时会计入该插件的熔断计数。
+"""
+
+import asyncio
+import logging
+import re
+from collections.abc import Callable
+from typing import Any
+
+from mcp import types
+from pydantic import AnyUrl
+
+from ab_sdk import points
+from ab_sdk.mcp import DEFAULT_TOOL_TIMEOUT, McpResource, McpTool
+from module.plugin.host import get_registry
+from module.plugin.registry import ProviderEntry
+
+from .runtime import get_context
+
+logger = logging.getLogger(__name__)
+
+TOOL_SEPARATOR = "__"
+RESOURCE_PREFIX = "autobangumi://plugins/"
+_TOOL_NAME = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
+
+
+def _record(plugin_id: str, reason: str | None) -> None:
+    """把调用结果计入插件熔断（MCP 运行时未绑定 AppContext 时跳过）。"""
+    ctx = get_context()
+    if ctx is None:
+        return
+    breaker = ctx.plugins.breaker
+    if reason is None:
+        breaker.record_success(plugin_id)
+    else:
+        breaker.record_failure(plugin_id, reason)
+
+
+def _resolve(
+    point: str, expected: type, spec: Callable[[ProviderEntry, Any], Any]
+) -> list[tuple[ProviderEntry, Any, Any]]:
+    """调用该扩展点的全部插件工厂并用 ``spec`` 生成对外的 MCP 描述，
+    跳过失败、类型不符或字段不合法（MCP 模型校验失败）的条目。"""
+    resolved = []
+    for entry in get_registry().providers(point).values():
+        try:
+            impl = entry.factory()
+            if not isinstance(impl, expected):
+                raise TypeError(f"应返回 {expected.__name__}，收到 {type(impl)}")
+            described = spec(entry, impl)
+        except Exception as e:
+            reason = f"{point} {entry.id}: {type(e).__name__}: {e}"
+            logger.warning("[Plugin:%s] %s", entry.plugin_id, reason)
+            _record(entry.plugin_id, reason)
+            continue
+        resolved.append((entry, impl, described))
+    return resolved
+
+
+# ------------------------------------------------------------------ tools
+
+
+def tool_name(plugin_id: str, tool_id: str) -> str:
+    return f"{plugin_id}{TOOL_SEPARATOR}{tool_id}"
+
+
+def _tool_spec(entry: ProviderEntry, tool: McpTool) -> types.Tool:
+    return types.Tool(
+        name=tool_name(entry.plugin_id, entry.id),
+        description=tool.description,
+        inputSchema=tool.input_schema,
+    )
+
+
+def _plugin_tools() -> dict[str, tuple[ProviderEntry, McpTool, types.Tool]]:
+    tools = {}
+    for entry, tool, spec in _resolve(points.MCP_TOOL, McpTool, _tool_spec):
+        name = spec.name
+        if not _TOOL_NAME.match(name):
+            logger.warning(
+                "[Plugin:%s] MCP 工具名 %r 不合法（仅限字母、数字、_、-，"
+                "不超过 64 字符），已跳过",
+                entry.plugin_id,
+                name,
+            )
+            continue
+        tools[name] = (entry, tool, spec)
+    return tools
+
+
+def list_plugin_tools() -> list[types.Tool]:
+    return [spec for _, (_, _, spec) in sorted(_plugin_tools().items())]
+
+
+async def call_plugin_tool(name: str, arguments: dict) -> tuple[bool, Any]:
+    """调用插件工具；``name`` 不是插件工具时返回 ``(False, None)``。
+
+    插件处理函数的异常会继续抛出（由 ``handle_tool`` 统一转成错误 JSON），
+    在此之前计入熔断。
+    """
+    if TOOL_SEPARATOR not in name:
+        return False, None
+    found = _plugin_tools().get(name)
+    if found is None:
+        return False, None
+    entry, tool, _ = found
+    try:
+        result = await asyncio.wait_for(tool.handler(arguments), tool.timeout)
+    except Exception as e:
+        reason = (
+            f"MCP 工具 {entry.id} 超时（{tool.timeout}s）"
+            if isinstance(e, TimeoutError)
+            else f"MCP 工具 {entry.id}: {type(e).__name__}: {e}"
+        )
+        _record(entry.plugin_id, reason)
+        raise
+    _record(entry.plugin_id, None)
+    return True, result
+
+
+# ------------------------------------------------------------------ resources
+
+
+def resource_uri(plugin_id: str, resource_id: str) -> str:
+    return f"{RESOURCE_PREFIX}{plugin_id}/{resource_id}"
+
+
+def _resource_spec(entry: ProviderEntry, resource: McpResource) -> types.Resource:
+    return types.Resource(
+        uri=AnyUrl(resource_uri(entry.plugin_id, entry.id)),
+        name=resource.name,
+        description=resource.description or None,
+        mimeType=resource.mime_type,
+    )
+
+
+def _plugin_resources() -> dict[str, tuple[ProviderEntry, McpResource, types.Resource]]:
+    # 以 AnyUrl 规范化后的 URI 为键：含空格或非 ASCII 字符的 id 会被百分号编码，
+    # 客户端读取时传回的正是列表里的编码形式
+    return {
+        str(spec.uri): (entry, resource, spec)
+        for entry, resource, spec in _resolve(
+            points.MCP_RESOURCE, McpResource, _resource_spec
+        )
+    }
+
+
+def list_plugin_resources() -> list[types.Resource]:
+    return [spec for _, (_, _, spec) in sorted(_plugin_resources().items())]
+
+
+async def read_plugin_resource(uri: str) -> tuple[bool, Any]:
+    """读取插件资源；``uri`` 不是插件资源时返回 ``(False, None)``。"""
+    if not uri.startswith(RESOURCE_PREFIX):
+        return False, None
+    found = _plugin_resources().get(uri)
+    if found is None:
+        return False, None
+    entry, resource, _ = found
+    # McpResource 没有超时字段，沿用工具的默认超时
+    try:
+        result = await asyncio.wait_for(resource.handler(), DEFAULT_TOOL_TIMEOUT)
+    except Exception as e:
+        reason = (
+            f"MCP 资源 {entry.id} 超时（{DEFAULT_TOOL_TIMEOUT}s）"
+            if isinstance(e, TimeoutError)
+            else f"MCP 资源 {entry.id}: {type(e).__name__}: {e}"
+        )
+        _record(entry.plugin_id, reason)
+        raise
+    _record(entry.plugin_id, None)
+    return True, result

backend/src/module/mcp/resources.py

@@ -17,6 +17,7 @@ from module.manager import TorrentManager
 from module.models import Bangumi
 from module.rss import RSSEngine
 
+from .plugins import list_plugin_resources, read_plugin_resource
 from .runtime import get_context
 from .tools import _bangumi_to_dict
 
@@ -53,6 +54,11 @@ RESOURCE_TEMPLATES = [
 ]
 
 
+def all_resources() -> list[types.Resource]:
+    """内置资源 + 当前已启用插件提供的资源（``autobangumi://plugins/<id>/...``）。"""
+    return RESOURCES + list_plugin_resources()
+
+
 async def handle_resource(uri: str) -> str:
     """Return a JSON string for the given MCP resource URI.
 
@@ -61,6 +67,7 @@ async def handle_resource(uri: str) -> str:
     - ``autobangumi://status`` - program version and running state
     - ``autobangumi://rss/feeds`` - configured RSS feeds
     - ``autobangumi://anime/{id}`` - single anime by integer ID
+    - ``autobangumi://plugins/<plugin-id>/<id>`` - 插件提供的资源
     """
     if uri == "autobangumi://anime/list":
         async with Database() as db:
@@ -108,4 +115,14 @@ async def handle_resource(uri: str) -> str:
             return json.dumps(_bangumi_to_dict(result), ensure_ascii=False)
         return json.dumps({"error": result.msg_en})
 
+    try:
+        handled, content = await read_plugin_resource(uri)
+    except Exception as e:
+        logger.warning("Plugin resource %s failed: %s", uri, e)
+        return json.dumps({"error": str(e)}, ensure_ascii=False)
+    if handled:
+        if isinstance(content, str):
+            return content
+        return json.dumps(content, ensure_ascii=False, default=str)
+
     return json.dumps({"error": f"Unknown resource: {uri}"})

backend/src/module/mcp/server.py

@@ -13,15 +13,15 @@ import logging
 from mcp import types
 from mcp.server import Server
 from mcp.server.sse import SseServerTransport
+from pydantic import AnyUrl
 from starlette.applications import Starlette
-from starlette.requests import Request
-from starlette.responses import Response
 from starlette.routing import Mount, Route
+from starlette.types import Receive, Scope, Send
 
-from .resources import RESOURCE_TEMPLATES, RESOURCES, handle_resource
+from .resources import RESOURCE_TEMPLATES, all_resources, handle_resource
 from .runtime import set_context
 from .security import McpAccessMiddleware
-from .tools import TOOLS, handle_tool
+from .tools import all_tools, handle_tool
 
 logger = logging.getLogger(__name__)
 
@@ -31,7 +31,7 @@ sse = SseServerTransport("/messages/")
 
 @server.list_tools()
 async def list_tools() -> list[types.Tool]:
-    return TOOLS
+    return all_tools()
 
 
 @server.call_tool()
@@ -42,7 +42,7 @@ async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
 
 @server.list_resources()
 async def list_resources() -> list[types.Resource]:
-    return RESOURCES
+    return all_resources()
 
 
 @server.list_resource_templates()
@@ -51,22 +51,30 @@ async def list_resource_templates() -> list[types.ResourceTemplate]:
 
 
 @server.read_resource()
-async def read_resource(uri: str) -> str:
+async def read_resource(uri: AnyUrl) -> str:
     logger.debug("Resource read: %s", uri)
-    return await handle_resource(uri)
+    # 底层 read_resource 传入的是 pydantic AnyUrl，handle_resource 按字符串匹配
+    return await handle_resource(str(uri))
 
 
-async def handle_sse(request: Request):
-    """Accept an SSE connection, run the MCP session until the client disconnects."""
-    async with sse.connect_sse(
-        request.scope, request.receive, request._send
-    ) as streams:
-        await server.run(
-            streams[0],
-            streams[1],
-            server.create_initialization_options(),
-        )
-    return Response()
+class _SseEndpoint:
+    """``GET /sse`` 的裸 ASGI 端点：建立 SSE 连接并运行 MCP 会话直到客户端断开。
+
+    ``connect_sse`` 自己经 ``send`` 写出整个 HTTP 响应，所以这里不能是返回
+    ``Response`` 的普通 endpoint 函数——Starlette 会在 SSE 结束后再发一次
+    ``http.response.start``，每次断开都触发 ``AssertionError``。
+    """
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        async with sse.connect_sse(scope, receive, send) as streams:
+            await server.run(
+                streams[0],
+                streams[1],
+                server.create_initialization_options(),
+            )
+
+
+handle_sse = _SseEndpoint()
 
 
 def create_mcp_starlette_app(ctx=None) -> Starlette:

backend/src/module/mcp/tools.py

@@ -1,5 +1,6 @@
 import json
 import logging
+from typing import Any
 
 from mcp import types
 
@@ -11,6 +12,7 @@ from module.models import Bangumi, BangumiUpdate, RSSItem
 from module.rss import RSSAnalyser, RSSEngine
 from module.searcher import SearchTorrent
 
+from .plugins import call_plugin_tool, list_plugin_tools
 from .runtime import get_context
 
 logger = logging.getLogger(__name__)
@@ -175,6 +177,11 @@ TOOLS = [
 ]
 
 
+def all_tools() -> list[types.Tool]:
+    """内置工具 + 当前已启用插件提供的工具（``<plugin-id>__<id>``）。"""
+    return TOOLS + list_plugin_tools()
+
+
 def _bangumi_to_dict(b: Bangumi) -> dict:
     return {
         "id": b.id,
@@ -202,7 +209,9 @@ async def handle_tool(name: str, arguments: dict) -> list[types.TextContent]:
     try:
         result = await _dispatch(name, arguments)
         return [
-            types.TextContent(type="text", text=json.dumps(result, ensure_ascii=False))
+            types.TextContent(
+                type="text", text=json.dumps(result, ensure_ascii=False, default=str)
+            )
         ]
     except Exception as e:
         logger.exception("Tool %s failed", name)
@@ -213,7 +222,7 @@ async def handle_tool(name: str, arguments: dict) -> list[types.TextContent]:
         ]
 
 
-async def _dispatch(name: str, args: dict) -> dict | list:
+async def _dispatch(name: str, args: dict) -> Any:
     if name == "list_anime":
         return await _list_anime(args.get("active_only", False))
     elif name == "get_anime":
@@ -234,8 +243,10 @@ async def _dispatch(name: str, args: dict) -> dict | list:
         return await _refresh_feeds()
     elif name == "update_anime":
         return await _update_anime(args)
-    else:
-        return {"error": f"Unknown tool: {name}"}
+    handled, result = await call_plugin_tool(name, args)
+    if handled:
+        return result
+    return {"error": f"Unknown tool: {name}"}
 
 
 async def _list_anime(active_only: bool) -> list[dict]:

backend/src/module/notification/base.py

@@ -7,6 +7,7 @@ from module.conf import settings
 from module.models.bangumi import Notification
 from module.network import RequestContent
 from module.notification.events import SystemEvent
+from module.notification.template import render_event
 
 if TYPE_CHECKING:
     from module.models.config import NotificationProvider as ProviderConfig
@@ -26,6 +27,8 @@ class NotificationProvider(ABC):
         # 单集通知模板（{{title}}/{{season}}/{{episode}}/{{poster_url}}）；
         # 未设置时 ``_format_message`` 回退到默认中文文案。
         self.template = config.template
+        # 渠道类型（如 telegram），传给 message_template 钩子按渠道定制文案
+        self.channel = str(config.type).lower()
 
     async def __aenter__(self) -> "NotificationProvider":
         await self._http.__aenter__()
@@ -62,7 +65,8 @@ class NotificationProvider(ABC):
         System events use each provider's default title/body delivery
         (:meth:`_deliver_text`) -- the per-episode template only covers
         {{title}}/{{season}}/{{episode}}/{{poster_url}}, which don't apply
-        here.
+        here. 标题与正文默认取 ``event.describe()``，插件可通过
+        ``message_template`` 钩子按事件类型与渠道改写。
 
         Args:
             event: The system event to send.
@@ -70,7 +74,7 @@ class NotificationProvider(ABC):
         Returns:
             True if the event was delivered successfully, False otherwise.
         """
-        title, body = event.describe()
+        title, body = await render_event(event, self.channel)
         return await self._deliver_text(title, body)
 
     @abstractmethod

backend/src/module/notification/events.py

@@ -1,294 +1,46 @@
 """非「新集数」通知的类型化事件负载。
 
 刻意与 ``Notification``（module/models/bangumi.py，仅描述「新集数下载完成」）
-分开：RSS 订阅失败、下载添加失败、偏移量待确认三类事件携带的字段与集数通知完全
+分开：RSS 订阅失败、下载添加失败、偏移量待确认等事件携带的字段与集数通知完全
 无关，硬塞进 ``Notification`` 只会制造出没有意义的 season/episode 占位值。
 
-每个事件同时服务两个消费方：
+4.0 起事件类定义在 ``ab_sdk.events``（插件订阅时需要 import 它们），这里只做
+再导出，宿主代码的 import 路径保持不变。每个事件同时服务三个消费方：
 
-- 外部推送（Telegram/Bark 等）：``describe() -> (标题, 正文)`` 渲染中文文案；
+- 外部推送（Telegram/Bark 等）：``describe() -> (标题, 正文)`` 渲染中文文案，
+  插件可通过 ``message_template`` 钩子改写；
 - 站内通知中心：``kind``/``severity``/``once``/``dedup_key()``/``payload()``
-  提供结构化字段，前端按 kind + payload 做多语言渲染，describe() 文案作为
-  未知 kind 的兜底展示。``once=True`` 表示同 dedup_key 终生只入库一次
-  （如"新版本可用"，已读后同一版本不再提醒）。
+  提供结构化字段，前端按 kind + payload 做多语言渲染（``i18n()`` 给出 key），
+  describe() 文案作为未知 kind 的兜底展示。``once=True`` 表示同 dedup_key
+  终生只入库一次（如"新版本可用"，已读后同一版本不再提醒）；
+- 事件总线：``NotificationManager.send_event`` 把事件发布到插件事件总线。
+
+``SystemEvent`` 从封闭的 Union 改为基类：插件定义的可通知事件同样可以送进
+``send_event``。
 """
 
-from dataclasses import dataclass
-from typing import ClassVar, Optional
-
-
-@dataclass(slots=True, frozen=True)
-class RssFailureEvent:
-    """RSS 订阅连接状态从正常变为异常（仅状态翻转时触发一次，而非每个 tick）。"""
-
-    kind: ClassVar[str] = "rss_failure"
-    severity: ClassVar[str] = "error"
-    once: ClassVar[bool] = False
-
-    rss_name: str
-    rss_url: str
-    error: str
-
-    def dedup_key(self) -> Optional[str]:
-        return f"rss_failure:{self.rss_url}"
-
-    def payload(self) -> dict:
-        return {"rss_name": self.rss_name, "rss_url": self.rss_url, "error": self.error}
-
-    def describe(self) -> tuple[str, str]:
-        """返回该事件的默认 (标题, 正文)。"""
-        return (
-            "RSS 订阅连接异常",
-            f"订阅：{self.rss_name}\n地址：{self.rss_url}\n错误：{self.error}",
-        )
-
-
-@dataclass(slots=True, frozen=True)
-class DownloadFailureEvent:
-    """匹配到的种子在（内置 HTTP 重试后）仍添加下载器失败。"""
-
-    kind: ClassVar[str] = "download_failure"
-    severity: ClassVar[str] = "error"
-    once: ClassVar[bool] = False
-
-    official_title: str
-    torrent_name: str
-
-    def dedup_key(self) -> Optional[str]:
-        return f"download_failure:{self.official_title}"
-
-    def payload(self) -> dict:
-        return {
-            "official_title": self.official_title,
-            "torrent_name": self.torrent_name,
-        }
-
-    def describe(self) -> tuple[str, str]:
-        return (
-            "种子添加失败",
-            f"番剧：{self.official_title}\n种子：{self.torrent_name}\n"
-            "重试后仍添加失败，请检查下载器连接。",
-        )
-
-
-@dataclass(slots=True, frozen=True)
-class OffsetReviewEvent:
-    """番剧被标记为需要人工确认季度/集数偏移。"""
-
-    kind: ClassVar[str] = "offset_review"
-    severity: ClassVar[str] = "warning"
-    once: ClassVar[bool] = False
-
-    official_title: str
-    reason: str
-
-    def dedup_key(self) -> Optional[str]:
-        return f"offset_review:{self.official_title}"
-
-    def payload(self) -> dict:
-        return {"official_title": self.official_title, "reason": self.reason}
-
-    def describe(self) -> tuple[str, str]:
-        return (
-            "集数偏移待确认",
-            f"番剧：{self.official_title}\n原因：{self.reason}\n请前往设置页确认偏移量。",
-        )
-
-
-@dataclass(slots=True, frozen=True)
-class DownloaderUnavailableEvent:
-    """下载器不可用：连不上、用户名/密码错误或 IP 被封禁。
-
-    ``reason`` 只进 payload 不进 dedup_key——同一台下载器从 unreachable 翻转
-    到 credentials 时合并为一条（内容以最新为准），不产生两行。
-    """
-
-    kind: ClassVar[str] = "downloader_unavailable"
-    severity: ClassVar[str] = "error"
-    once: ClassVar[bool] = False
-
-    host: str
-    reason: str  # unreachable | credentials | banned
-
-    def dedup_key(self) -> Optional[str]:
-        return f"downloader:{self.host}"
-
-    def payload(self) -> dict:
-        return {"host": self.host, "reason": self.reason}
-
-    def describe(self) -> tuple[str, str]:
-        details = {
-            "credentials": "用户名或密码错误，请在设置中检查下载器凭据。",
-            "banned": "IP 已被下载器封禁，请在下载器 WebUI 中解封或重启下载器。",
-            "unreachable": "无法连接下载器，请检查地址、端口和网络。",
-        }
-        detail = details.get(self.reason, details["unreachable"])
-        return ("下载器连接异常", f"下载器：{self.host}\n{detail}")
-
-
-@dataclass(slots=True, frozen=True)
-class UpdateAvailableEvent:
-    """检查到可用的新版本。"""
-
-    kind: ClassVar[str] = "update_available"
-    severity: ClassVar[str] = "info"
-    once: ClassVar[bool] = True
-
-    current: str
-    latest: str
-    channel: str
-    notes: str = ""
-
-    def dedup_key(self) -> Optional[str]:
-        return f"update_available:{self.latest}"
-
-    def payload(self) -> dict:
-        return {
-            "current": self.current,
-            "latest": self.latest,
-            "channel": self.channel,
-            "notes": self.notes,
-        }
-
-    def describe(self) -> tuple[str, str]:
-        return (
-            "发现新版本",
-            f"当前版本：{self.current}\n最新版本：{self.latest}（{self.channel} 频道）\n"
-            "可前往 设置 → 软件更新 升级。",
-        )
-
-
-@dataclass(slots=True, frozen=True)
-class UpdateAppliedEvent:
-    """更新应用结果（成功或失败）。每次结果都单独入库，不做去重。"""
-
-    once: ClassVar[bool] = False
-
-    version: str
-    success: bool
-    message: str = ""
-
-    @property
-    def kind(self) -> str:
-        return "update_applied" if self.success else "update_failed"
-
-    @property
-    def severity(self) -> str:
-        return "info" if self.success else "error"
-
-    def dedup_key(self) -> Optional[str]:
-        return None
-
-    def payload(self) -> dict:
-        return {"version": self.version, "message": self.message}
-
-    def describe(self) -> tuple[str, str]:
-        if self.success:
-            return ("程序更新完成", f"已更新到 {self.version}，重启后生效。")
-        return ("程序更新失败", f"版本：{self.version}\n原因：{self.message}")
-
-
-@dataclass(slots=True, frozen=True)
-class LLMAuthFailureEvent:
-    """订阅类 LLM 提供商凭据失效（刷新失败），需要用户重新连接。"""
-
-    kind: ClassVar[str] = "llm_auth_failure"
-    severity: ClassVar[str] = "error"
-    once: ClassVar[bool] = False
-
-    provider_id: str
-    account_label: str = ""
-    message: str = ""
-
-    def dedup_key(self) -> Optional[str]:
-        return f"llm_auth:{self.provider_id}"
-
-    def payload(self) -> dict:
-        return {
-            "provider_id": self.provider_id,
-            "account_label": self.account_label,
-            "message": self.message,
-        }
-
-    def describe(self) -> tuple[str, str]:
-        account = f"（{self.account_label}）" if self.account_label else ""
-        return (
-            "LLM 提供商凭据失效",
-            f"提供商：{self.provider_id}{account}\n{self.message}\n"
-            "请前往 设置 → LLM 解析器 重新连接。",
-        )
-
-
-@dataclass(slots=True, frozen=True)
-class LLMPluginInstallFailedEvent:
-    """LLM 提供商插件安装失败（下载/签名校验/兼容性）。"""
-
-    kind: ClassVar[str] = "llm_plugin_install_failed"
-    severity: ClassVar[str] = "error"
-    once: ClassVar[bool] = False
-
-    plugin_id: str
-    version: str = ""
-    message: str = ""
-
-    def dedup_key(self) -> Optional[str]:
-        return f"llm_plugin_install:{self.plugin_id}"
-
-    def payload(self) -> dict:
-        return {
-            "plugin_id": self.plugin_id,
-            "version": self.version,
-            "message": self.message,
-        }
-
-    def describe(self) -> tuple[str, str]:
-        return (
-            "LLM 插件安装失败",
-            f"插件：{self.plugin_id} {self.version}\n原因：{self.message}",
-        )
-
-
-@dataclass(slots=True, frozen=True)
-class RenameConflictEvent:
-    """A media rename reached a durable target-path conflict."""
-
-    kind: ClassVar[str] = "rename_conflict"
-    severity: ClassVar[str] = "warning"
-    once: ClassVar[bool] = False
-
-    task_id: str
-    torrent_name: str
-    target_path: str
-    reason: str
-
-    def dedup_key(self) -> Optional[str]:
-        return f"rename_conflict:{self.task_id}:{self.target_path}"
-
-    def payload(self) -> dict:
-        return {
-            "task_id": self.task_id,
-            "torrent_name": self.torrent_name,
-            "target_path": self.target_path,
-            "reason": self.reason,
-        }
-
-    def describe(self) -> tuple[str, str]:
-        return (
-            "媒体文件重命名冲突",
-            f"种子：{self.torrent_name}\n目标：{self.target_path}\n"
-            f"原因：{self.reason}",
-        )
-
-
-# 除「新集数」以外的通知事件的联合类型。
-SystemEvent = (
-    RssFailureEvent
-    | DownloadFailureEvent
-    | OffsetReviewEvent
-    | DownloaderUnavailableEvent
-    | UpdateAvailableEvent
-    | UpdateAppliedEvent
-    | LLMAuthFailureEvent
-    | LLMPluginInstallFailedEvent
-    | RenameConflictEvent
+from ab_sdk.events import (
+    DownloaderUnavailableEvent,
+    DownloadFailureEvent,
+    LLMAuthFailureEvent,
+    LLMPluginInstallFailedEvent,
+    OffsetReviewEvent,
+    RenameConflictEvent,
+    RssFailureEvent,
+    SystemEvent,
+    UpdateAppliedEvent,
+    UpdateAvailableEvent,
 )
+
+__all__ = [
+    "DownloadFailureEvent",
+    "DownloaderUnavailableEvent",
+    "LLMAuthFailureEvent",
+    "LLMPluginInstallFailedEvent",
+    "OffsetReviewEvent",
+    "RenameConflictEvent",
+    "RssFailureEvent",
+    "SystemEvent",
+    "UpdateAppliedEvent",
+    "UpdateAvailableEvent",
+]

backend/src/module/notification/inbox.py

@@ -2,15 +2,20 @@
 
 ``record_event`` 把 SystemEvent 落库（``async with Database()``，
 session-per-operation）；``inbox_revision`` 是进程内单调递增的修订号，
-任何入库/已读/删除操作都会 bump，SSE 循环据此决定是否需要查库推送，
-避免每个 tick 都打一次数据库。AB 为单进程部署，进程内计数即可。
+任何入库/已读/删除操作都会 bump，并在事件总线上发布 :class:`InboxChanged`。
+SSE 连接订阅该事件，收到后才查库推送未读数（取代原先每 3s 比较修订号的轮询）。
+AB 为单进程部署，进程内计数即可。
 """
 
 import json
 import logging
+from dataclasses import dataclass
+from typing import ClassVar
 
+from ab_sdk import Event
 from module.database import Database
 from module.notification.events import SystemEvent
+from module.plugin.host import publish
 
 logger = logging.getLogger(__name__)
 
@@ -19,6 +24,14 @@ INBOX_KEEP = 500
 _revision: int = 1
 
 
+@dataclass(frozen=True, slots=True)
+class InboxChanged(Event):
+    """站内通知中心有变化（新消息、已读、删除），``revision`` 为新的修订号。"""
+
+    kind: ClassVar[str] = "inbox.changed"
+    revision: int
+
+
 def inbox_revision() -> int:
     return _revision
 
@@ -26,6 +39,7 @@ def inbox_revision() -> int:
 def bump_inbox_revision() -> None:
     global _revision
     _revision += 1
+    publish(InboxChanged(revision=_revision))
 
 
 async def record_event(event: SystemEvent) -> int:

backend/src/module/notification/manager.py

@@ -9,6 +9,7 @@ from module.database import Database
 from module.models.bangumi import Notification
 from module.notification.events import SystemEvent
 from module.notification.inbox import record_event
+from module.plugin.host import publish
 
 if TYPE_CHECKING:
     from module.models.config import NotificationProvider as ProviderConfig
@@ -118,8 +119,9 @@ class NotificationManager:
     async def send_event(self, event: SystemEvent):
         """Persist a system event to the in-app inbox, then broadcast it.
 
-        持久化不受 ``settings.notification.enable`` 影响（该开关只管外部
-        推送），失败也不阻塞外部广播。
+        顺序：写入站内通知中心 → 发布到插件事件总线 → 推送到外部渠道。
+        持久化与总线发布不受 ``settings.notification.enable`` 影响（该开关只管
+        外部推送），持久化失败也不阻塞后两步。
 
         Args:
             event: The system event to send.
@@ -132,6 +134,8 @@ class NotificationManager:
                 type(event).__name__,
                 exc_info=True,
             )
+        # 插件 @subscribe(event.kind) 的订阅者在各自队列中异步处理，不阻塞这里
+        publish(event)
         if not settings.notification.enable:
             return
 

backend/src/module/notification/template.py

@@ -0,0 +1,37 @@
+"""系统事件推送文案：默认 ``describe()``，经 ``message_template`` 钩子改写。"""
+
+import logging
+
+from ab_sdk import points
+from ab_sdk.notify import RenderedMessage
+from module.notification.events import SystemEvent
+from module.plugin.host import get_runner
+
+logger = logging.getLogger(__name__)
+
+
+async def render_event(event: SystemEvent, channel: str) -> tuple[str, str]:
+    """返回推送到 ``channel`` 渠道的 (标题, 正文)。
+
+    没有插件挂 ``message_template`` 钩子时与 ``event.describe()`` 完全一致；
+    钩子失败、超时或返回值不是 ``RenderedMessage`` 时沿用上一步的结果（并计入
+    该插件的熔断）。
+    """
+    title, body = event.describe()
+    runner = get_runner()
+    if runner is None:
+        return title, body
+    try:
+        message = await runner.transform(
+            points.MESSAGE_TEMPLATE,
+            RenderedMessage(title, body),
+            event,
+            channel,
+            expect=RenderedMessage,
+        )
+    except Exception:
+        # 单个钩子的失败已在 runner 内隔离；这里兜底的是运行时本身的异常
+        # （如扩展点未声明），任何情况下都不能让通知因模板而发不出去
+        logger.warning("[Notification] 渲染通知模板失败，使用默认文案", exc_info=True)
+        return title, body
+    return message.title, message.body

backend/src/module/plugin/host.py

@@ -12,9 +12,10 @@ import logging
 from collections.abc import Callable
 from typing import Any
 
-from ab_sdk import points
+from ab_sdk import Event, points
 from ab_sdk.downloader import DownloaderConnection
 
+from .bus import EventBus
 from .registry import ExtensionPoint, ExtensionRegistry, ProviderEntry
 from .runner import HookRunner
 
@@ -40,10 +41,23 @@ POINTS = (
     ExtensionPoint(points.TITLE_PARSED, "transform", "修正标题解析结果"),
     ExtensionPoint(points.TORRENT_ADDING, "transform", "修改发给下载器的添加请求"),
     ExtensionPoint(points.HTTP_REQUEST, "transform", "修改宿主 GET 请求的请求头"),
+    # --- P5 events/api ---
+    ExtensionPoint(
+        points.API_ROUTER,
+        "provider",
+        "插件 REST 路由（/api/v1/plugins/<id>/）",
+        scoped=True,
+    ),
+    ExtensionPoint(points.MCP_TOOL, "provider", "MCP 工具", scoped=True),
+    ExtensionPoint(points.MCP_RESOURCE, "provider", "MCP 资源", scoped=True),
+    ExtensionPoint(points.MESSAGE_TEMPLATE, "transform", "系统事件通知文案"),
 )
 
 _registry: ExtensionRegistry | None = None
-# 由 AppContext 在构建时设置；未设置时（测试、CLI）钩子一律跳过，只执行宿主逻辑
+# 进程级事件总线与钩子执行器，由 AppContext 在构造时设置（即 PluginManager
+# 持有的那一份）；未设置时（单元测试、脚本、CLI）发布事件为空操作、钩子一律
+# 跳过，只执行宿主逻辑
+_bus: EventBus | None = None
 _runner: HookRunner | None = None
 
 
@@ -159,3 +173,30 @@ def _register_core(registry: ExtensionRegistry) -> None:
             points.METADATA_PROVIDER,
             ProviderEntry(CORE, provider_id, metadata(provider_id)),
         )
+
+
+# --- P5 events/api ---------------------------------------------------------
+
+
+def get_bus() -> EventBus | None:
+    return _bus
+
+
+def set_bus(bus: EventBus | None) -> None:
+    global _bus
+    _bus = bus
+
+
+def publish(event: Event) -> None:
+    """把宿主事件发布到进程级总线；总线未设置时什么也不做。
+
+    发布只负责入队，订阅者的失败不会影响调用方。没有运行中的事件循环时
+    （同步上下文）无法投递，记录日志后忽略。
+    """
+    bus = _bus
+    if bus is None:
+        return
+    try:
+        bus.publish(event)
+    except RuntimeError:
+        logger.debug("[EventBus] 无事件循环，丢弃事件 %s", event.kind)

backend/src/module/plugin/registry.py

@@ -18,6 +18,9 @@ class ExtensionPoint:
     description: str = ""
     # 仅 filter：钩子出错/超时时是否放行（True）还是视为拒绝（False）
     fail_open: bool = True
+    # 仅 provider：True 时 Provider id 只需在插件内唯一（宿主对外暴露时会加
+    # 插件 id 前缀，如插件路由、MCP 工具），注册表按 ``<plugin_id>/<id>`` 存储
+    scoped: bool = False
 
 
 @dataclass(frozen=True, slots=True)
@@ -68,13 +71,14 @@ class ExtensionRegistry:
         if point.kind != "provider":
             raise RegistryError(f"{point_name} 是 {point.kind} 扩展点，请使用 @hook")
         slot = self._providers.setdefault(point_name, {})
-        owner = slot.get(entry.id)
+        key = f"{entry.plugin_id}/{entry.id}" if point.scoped else entry.id
+        owner = slot.get(key)
         if owner is not None:
             raise RegistryError(
                 f"{point_name} 的 Provider id {entry.id!r} 已被插件 "
                 f"{owner.plugin_id} 占用"
             )
-        slot[entry.id] = entry
+        slot[key] = entry
 
     def remove_plugin(self, plugin_id: str) -> None:
         for name, entries in self._hooks.items():
@@ -100,5 +104,6 @@ class ExtensionRegistry:
         return bool(self._hooks.get(point_name))
 
     def providers(self, point_name: str) -> dict[str, ProviderEntry]:
+        """按 Provider id 索引；``scoped`` 扩展点的键为 ``<plugin_id>/<id>``。"""
         self.point(point_name)
         return dict(self._providers.get(point_name, {}))

backend/src/test/test_api_events.py

@@ -214,7 +214,18 @@ class TestLogPayload:
 
 
 class TestNotificationEvent:
-    def _generator(self, tmp_path):
+    """notification 帧由事件总线上的 inbox.changed 驱动（P5），线格式不变。"""
+
+    @pytest.fixture
+    def bus(self, monkeypatch):
+        from module.plugin import host
+        from module.plugin.bus import EventBus
+
+        bus = EventBus()
+        monkeypatch.setattr(host, "_bus", bus)
+        return bus
+
+    def _generator(self, tmp_path, *, tick=0.01, payload=None):
         from module.api.events import _event_generator
 
         request = AsyncMock()
@@ -222,17 +233,17 @@ class TestNotificationEvent:
         ctx = MagicMock()
         ctx.is_running = True
         ctx.first_run = False
+        notification = payload or AsyncMock(
+            return_value={"unread_count": 2, "latest_id": 9}
+        )
         patches = (
             patch(
                 "module.api.events._downloader_payload",
                 new=AsyncMock(return_value=None),
             ),
             patch("module.api.events.LOG_PATH", tmp_path / "missing.log"),
-            patch(
-                "module.api.events._notification_payload",
-                new=AsyncMock(return_value={"unread_count": 2, "latest_id": 9}),
-            ),
-            patch("module.api.events.asyncio.sleep", new=AsyncMock()),
+            patch("module.api.events._notification_payload", new=notification),
+            patch("module.api.events._TICK_SECONDS", tick),
         )
         return _event_generator(request, ctx), patches
 
@@ -242,7 +253,7 @@ class TestNotificationEvent:
             events.append(await asyncio.wait_for(gen.__anext__(), timeout=1.0))
         return events
 
-    async def test_first_tick_emits_notification_frame(self, tmp_path):
+    async def test_first_tick_emits_notification_frame(self, tmp_path, bus):
         from module.notification.inbox import inbox_revision
 
         gen, patches = self._generator(tmp_path)
@@ -252,14 +263,12 @@ class TestNotificationEvent:
             finally:
                 await gen.aclose()
 
-        by_name = {e["event"]: e for e in events}
-        assert set(by_name) == {"status", "downloader", "notification"}
-        data = json.loads(by_name["notification"]["data"])
-        assert data["unread_count"] == 2
-        assert data["latest_id"] == 9
-        assert data["revision"] == inbox_revision()
+        assert [e["event"] for e in events] == ["status", "downloader", "notification"]
+        # 线格式与 3.x 一致：unread_count / latest_id / revision 三个字段
+        data = json.loads(events[2]["data"])
+        assert data == {"unread_count": 2, "latest_id": 9, "revision": inbox_revision()}
 
-    async def test_no_frame_when_revision_unchanged(self, tmp_path):
+    async def test_no_frame_without_inbox_change(self, tmp_path, bus):
         gen, patches = self._generator(tmp_path)
         with patches[0], patches[1], patches[2], patches[3]:
             try:
@@ -271,18 +280,55 @@ class TestNotificationEvent:
 
         assert [e["event"] for e in more] == ["status", "downloader"]
 
-    async def test_frame_emitted_after_revision_bump(self, tmp_path):
+    async def test_inbox_change_pushes_frame_immediately(self, tmp_path, bus):
         from module.notification.inbox import bump_inbox_revision, inbox_revision
 
-        gen, patches = self._generator(tmp_path)
+        # tick 设为 30s：若不是事件推送，1s 内不可能收到下一帧
+        gen, patches = self._generator(tmp_path, tick=30)
         with patches[0], patches[1], patches[2], patches[3]:
             try:
                 await self._collect(gen, 3)  # 第一个 tick
                 bump_inbox_revision()
-                # tick3：status + notification（修订号变了）
-                more = await self._collect(gen, 2)
+                [frame] = await self._collect(gen, 1)
             finally:
                 await gen.aclose()
 
-        assert [e["event"] for e in more] == ["status", "notification"]
-        assert json.loads(more[1]["data"])["revision"] == inbox_revision()
+        assert frame["event"] == "notification"
+        assert json.loads(frame["data"])["revision"] == inbox_revision()
+
+    async def test_unsubscribes_when_stream_closes(self, tmp_path, bus):
+        gen, patches = self._generator(tmp_path)
+        with patches[0], patches[1], patches[2], patches[3]:
+            await self._collect(gen, 3)
+            assert len(bus._subscribers) == 1
+            await gen.aclose()
+        assert bus._subscribers == []
+
+    async def test_failed_query_is_retried_next_tick(self, tmp_path, bus):
+        payload = AsyncMock(
+            side_effect=[RuntimeError("db busy"), {"unread_count": 1, "latest_id": 3}]
+        )
+        gen, patches = self._generator(tmp_path, payload=payload)
+        with patches[0], patches[1], patches[2], patches[3]:
+            try:
+                events = await self._collect(gen, 3)
+            finally:
+                await gen.aclose()
+
+        # tick0：status + downloader（notification 查库失败）；tick1 重试成功
+        assert [e["event"] for e in events] == ["status", "downloader", "notification"]
+        assert json.loads(events[2]["data"])["latest_id"] == 3
+        assert payload.await_count == 2
+
+    async def test_without_bus_only_initial_frame(self, tmp_path, monkeypatch):
+        from module.plugin import host
+
+        monkeypatch.setattr(host, "_bus", None)
+        gen, patches = self._generator(tmp_path)
+        with patches[0], patches[1], patches[2], patches[3]:
+            try:
+                events = await self._collect(gen, 5)
+            finally:
+                await gen.aclose()
+
+        assert [e["event"] for e in events].count("notification") == 1

backend/src/test/test_plugin_events_api.py

@@ -0,0 +1,695 @@
+"""P5：系统事件上总线、message_template、插件 REST 路由与 MCP 工具/资源。"""
+
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import APIRouter, FastAPI, HTTPException
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+
+from ab_sdk import Event, Plugin, hook, points, provider, subscribe
+from ab_sdk.events import (
+    RssFailureEvent,
+    SystemEvent,
+    UpdateAppliedEvent,
+)
+from ab_sdk.mcp import McpResource, McpTool
+from ab_sdk.notify import RenderedMessage
+from module.api import v1
+from module.api.deps import get_context
+from module.core.plugin_routes import PluginRoutes
+from module.models.config import NotificationProvider as ProviderConfig
+from module.models.config import Plugins
+from module.notification import NotificationManager
+from module.notification.inbox import InboxChanged, bump_inbox_revision
+from module.notification.template import render_event
+from module.plugin import host
+from module.plugin.bus import EventBus
+from module.plugin.loader import PluginCandidate
+from module.plugin.manager import PluginManager
+from module.plugin.manifest import parse_manifest
+from module.plugin.registry import ExtensionRegistry, HookEntry, ProviderEntry
+from module.plugin.runner import CircuitBreaker, HookRunner
+from module.security.api import get_current_user
+
+
+@pytest.fixture
+def registry(monkeypatch):
+    monkeypatch.setattr(host, "_registry", None)
+    return host.get_registry()
+
+
+@pytest.fixture
+def bus(monkeypatch):
+    bus = EventBus()
+    monkeypatch.setattr(host, "_bus", bus)
+    return bus
+
+
+@pytest.fixture
+def breaker():
+    return CircuitBreaker()
+
+
+@pytest.fixture
+def runner(monkeypatch, registry, breaker):
+    runner = HookRunner(registry, breaker)
+    monkeypatch.setattr(host, "_runner", runner)
+    return runner
+
+
+def add(registry, point, provider_id, impl, plugin_id="ext"):
+    registry.add_provider(point, ProviderEntry(plugin_id, provider_id, lambda: impl))
+
+
+# ---------------------------------------------------------------- system events
+
+
+class TestSystemEvents:
+    def test_system_events_are_sdk_events(self):
+        event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom")
+        assert isinstance(event, Event)
+        assert isinstance(event, SystemEvent)
+        assert event.kind == "rss_failure"
+        # 宿主旧的 import 路径仍然可用，且是同一个类
+        from module.notification import RssFailureEvent as HostRssFailureEvent
+
+        assert HostRssFailureEvent is RssFailureEvent
+
+    def test_i18n_key_matches_frontend_and_payload_is_params(self):
+        event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom")
+        key, params = event.i18n()
+        assert key == "notifications.kind.rss_failure"
+        assert params == event.payload()
+        failed = UpdateAppliedEvent(version="4.0.0", success=False, message="x")
+        assert failed.i18n()[0] == "notifications.kind.update_failed"
+
+    def test_plugin_defined_system_event_defaults(self):
+        from dataclasses import dataclass
+        from typing import ClassVar
+
+        @dataclass(frozen=True, slots=True)
+        class Synced(SystemEvent):
+            kind: ClassVar[str] = "demo.synced"
+            count: int
+
+        event = Synced(count=3)
+        assert event.payload() == {"count": 3}
+        assert event.dedup_key() is None
+        assert event.severity == "info"
+        assert event.describe() == ("demo.synced", "")
+
+    async def test_send_event_publishes_to_bus_even_when_push_disabled(self, bus):
+        received: list[Event] = []
+        bus.subscribe("rss_failure", received.append)
+        manager = NotificationManager(load_providers=False)
+        event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom")
+        with (
+            patch(
+                "module.notification.manager.record_event", new_callable=AsyncMock
+            ) as record,
+            patch("module.notification.manager.settings") as mock_settings,
+        ):
+            mock_settings.notification.enable = False
+            await manager.send_event(event)
+        await bus.drain()
+        record.assert_awaited_once_with(event)
+        assert received == [event]
+
+    async def test_send_event_publishes_even_if_inbox_write_fails(self, bus):
+        received: list[Event] = []
+        bus.subscribe("*", received.append)
+        manager = NotificationManager(load_providers=False)
+        event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom")
+        with (
+            patch(
+                "module.notification.manager.record_event",
+                new=AsyncMock(side_effect=RuntimeError("db")),
+            ),
+            patch("module.notification.manager.settings") as mock_settings,
+        ):
+            mock_settings.notification.enable = False
+            await manager.send_event(event)
+        await bus.drain()
+        assert received == [event]
+
+    async def test_bump_publishes_inbox_changed(self, bus):
+        received: list[Event] = []
+        bus.subscribe(InboxChanged.kind, received.append)
+        bump_inbox_revision()
+        await bus.drain()
+        [event] = received
+        assert isinstance(event, InboxChanged)
+
+    def test_publish_without_bus_is_noop(self, monkeypatch):
+        monkeypatch.setattr(host, "_bus", None)
+        host.publish(InboxChanged(revision=1))  # 不抛错
+
+
+# ---------------------------------------------------------------- message_template
+
+
+class TestMessageTemplate:
+    EVENT = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom")
+
+    async def test_default_is_describe_without_runner(self, monkeypatch):
+        monkeypatch.setattr(host, "_runner", None)
+        assert await render_event(self.EVENT, "telegram") == self.EVENT.describe()
+
+    async def test_default_is_describe_without_hooks(self, runner):
+        assert await render_event(self.EVENT, "telegram") == self.EVENT.describe()
+
+    async def test_hook_rewrites_per_kind_and_channel(self, runner, registry):
+        def render(message, event, channel):
+            if event.kind != "rss_failure":
+                return None
+            return RenderedMessage(f"[{channel}] {message.title}", event.rss_name)
+
+        registry.add_hook(points.MESSAGE_TEMPLATE, HookEntry("ext", render, 100, None))
+        title, body = await render_event(self.EVENT, "bark")
+        assert title == f"[bark] {self.EVENT.describe()[0]}"
+        assert body == "Feed"
+        other = UpdateAppliedEvent(version="1", success=True)
+        assert await render_event(other, "bark") == other.describe()
+
+    async def test_bad_return_falls_back(self, runner, registry):
+        registry.add_hook(
+            points.MESSAGE_TEMPLATE,
+            HookEntry("ext", lambda m, e, c: "not a message", 100, None),
+        )
+        assert await render_event(self.EVENT, "bark") == self.EVENT.describe()
+
+    async def test_failing_hook_falls_back_and_counts(self, runner, registry, breaker):
+        def boom(message, event, channel):
+            raise RuntimeError("template bug")
+
+        registry.add_hook(points.MESSAGE_TEMPLATE, HookEntry("ext", boom, 100, None))
+        assert await render_event(self.EVENT, "bark") == self.EVENT.describe()
+        assert breaker._failures["ext"] == 1
+
+    async def test_render_event_bad_return_counts_against_its_own_plugin(
+        self, runner, registry, breaker
+    ):
+        def rewrite(message, event, channel):
+            return RenderedMessage(message.title, "rewritten")
+
+        registry.add_hook(
+            points.MESSAGE_TEMPLATE,
+            HookEntry("bad", lambda m, e, c: "not a message", 10, None),
+        )
+        registry.add_hook(
+            points.MESSAGE_TEMPLATE, HookEntry("good", rewrite, 100, None)
+        )
+        title, body = await render_event(self.EVENT, "bark")
+        assert (title, body) == (self.EVENT.describe()[0], "rewritten")
+        assert breaker._failures.get("bad") == 1
+        assert not breaker._failures.get("good")
+
+    async def test_runtime_error_never_blocks_delivery(self, monkeypatch):
+        bad = MagicMock()
+        bad.transform = AsyncMock(side_effect=RuntimeError("undeclared point"))
+        monkeypatch.setattr(host, "_runner", bad)
+        assert await render_event(self.EVENT, "bark") == self.EVENT.describe()
+
+    async def test_provider_send_event_uses_template(self, runner, registry):
+        from module.notification.providers.bark import BarkProvider
+
+        registry.add_hook(
+            points.MESSAGE_TEMPLATE,
+            HookEntry("ext", lambda m, e, c: RenderedMessage("T", c), 100, None),
+        )
+        provider_obj = BarkProvider(
+            ProviderConfig(type="bark", enabled=True, device_key="k")
+        )
+        with patch.object(
+            provider_obj, "_deliver_text", new=AsyncMock(return_value=True)
+        ) as deliver:
+            assert await provider_obj.send_event(self.EVENT)
+        deliver.assert_awaited_once_with("T", "bark")
+
+
+# ---------------------------------------------------------------- registry
+
+
+def test_scoped_points_allow_same_id_in_different_plugins(registry):
+    add(registry, points.MCP_TOOL, "search", object(), plugin_id="a")
+    add(registry, points.MCP_TOOL, "search", object(), plugin_id="b")
+    assert set(registry.providers(points.MCP_TOOL)) == {"a/search", "b/search"}
+    with pytest.raises(Exception, match="已被插件"):
+        add(registry, points.MCP_TOOL, "search", object(), plugin_id="a")
+
+
+# ---------------------------------------------------------------- api_router
+
+
+def demo_router() -> APIRouter:
+    router = APIRouter()
+
+    @router.get("/")
+    async def root():
+        return {"root": True}
+
+    @router.get("/hello")
+    async def hello(name: str = "world"):
+        return {"hello": name}
+
+    class Body(BaseModel):
+        value: int
+
+    @router.post("/echo")
+    async def echo(body: Body):
+        return {"value": body.value * 2}
+
+    @router.get("/teapot")
+    async def teapot():
+        raise HTTPException(status_code=418, detail="teapot")
+
+    @router.get("/boom")
+    async def boom():
+        raise RuntimeError("plugin bug")
+
+    return router
+
+
+@pytest.fixture
+def routes_app(registry, breaker):
+    routes = PluginRoutes(registry, breaker)
+    plugins_mgr = MagicMock()
+    plugins_mgr.statuses.return_value = []
+    ctx = SimpleNamespace(plugin_routes=routes, plugins=plugins_mgr)
+    app = FastAPI()
+    app.include_router(v1, prefix="/api")
+    app.dependency_overrides[get_context] = lambda: ctx
+    return app, routes
+
+
+@pytest.fixture
+def routes_client(routes_app):
+    app, routes = routes_app
+
+    async def user():
+        return "testuser"
+
+    app.dependency_overrides[get_current_user] = user
+    return TestClient(app, raise_server_exceptions=False), routes
+
+
+class TestApiRouter:
+    def test_routes_follow_plugin_lifecycle(self, routes_client, registry):
+        client, routes = routes_client
+        assert client.get("/api/v1/plugins/demo/hello").status_code == 404
+
+        add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="demo")
+        routes.sync()
+        resp = client.get("/api/v1/plugins/demo/hello", params={"name": "ab"})
+        assert resp.status_code == 200
+        assert resp.json() == {"hello": "ab"}
+        assert client.get("/api/v1/plugins/demo/").json() == {"root": True}
+
+        registry.remove_plugin("demo")
+        routes.sync()
+        assert client.get("/api/v1/plugins/demo/hello").status_code == 404
+
+    def test_body_and_errors_forwarded(self, routes_client, registry, breaker):
+        client, routes = routes_client
+        add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="demo")
+        routes.sync()
+        assert client.post("/api/v1/plugins/demo/echo", json={"value": 21}).json() == {
+            "value": 42
+        }
+        assert (
+            client.post("/api/v1/plugins/demo/echo", json={"value": "x"}).status_code
+            == 422
+        )
+        assert client.get("/api/v1/plugins/demo/teapot").status_code == 418
+        assert client.get("/api/v1/plugins/demo/missing").status_code == 404
+        assert breaker._failures == {}
+        assert client.get("/api/v1/plugins/demo/boom").status_code == 500
+        assert breaker._failures["demo"] == 1
+
+    def test_multiple_routers_are_merged(self, routes_client, registry):
+        client, routes = routes_client
+        extra = APIRouter()
+
+        @extra.get("/extra")
+        async def extra_route():
+            return {"extra": True}
+
+        add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="demo")
+        add(registry, points.API_ROUTER, "more", extra, plugin_id="demo")
+        routes.sync()
+        assert client.get("/api/v1/plugins/demo/extra").json() == {"extra": True}
+        assert client.get("/api/v1/plugins/demo/hello").status_code == 200
+
+    def test_bad_factory_is_skipped(self, routes_client, registry, breaker):
+        client, routes = routes_client
+        add(registry, points.API_ROUTER, "api", "not a router", plugin_id="demo")
+        routes.sync()
+        assert routes.plugin_ids() == []
+        assert breaker._failures["demo"] == 1
+
+    def test_management_routes_not_shadowed(self, routes_client, registry):
+        client, routes = routes_client
+        add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="settings")
+        add(registry, points.API_ROUTER, "api2", demo_router(), plugin_id="providers")
+        routes.sync()
+        assert client.get("/api/v1/plugins").status_code == 200
+        providers = client.get("/api/v1/plugins/providers")
+        assert providers.status_code == 200
+        assert "downloader" in providers.json()
+        # PUT /plugins/{id} 仍走插件管理路由（未知插件 → 404 且 detail 来自管理路由）
+        resp = client.put("/api/v1/plugins/demo", json={})
+        assert resp.status_code == 404
+        assert resp.json()["detail"] == "Unknown plugin: demo"
+        # 插件 id 恰好是 settings/providers 时，其子路径仍可访问
+        assert client.get("/api/v1/plugins/settings/hello").status_code == 200
+
+    @patch("module.security.api.DEV_AUTH_BYPASS", False)
+    def test_auth_is_enforced(self, routes_app, registry):
+        app, routes = routes_app
+        add(registry, points.API_ROUTER, "api", demo_router(), plugin_id="demo")
+        routes.sync()
+        client = TestClient(app)
+        assert client.get("/api/v1/plugins/demo/hello").status_code == 401
+        assert client.get("/api/v1/plugins/unknown/x").status_code == 401
+
+
+# ---------------------------------------------------------------- MCP
+
+
+async def _search(args):
+    return {"q": args.get("q"), "hits": [1, 2]}
+
+
+class TestMcp:
+    def test_plugin_tools_listed_with_namespace(self, registry):
+        from module.mcp.tools import TOOLS, all_tools
+
+        add(
+            registry,
+            points.MCP_TOOL,
+            "search",
+            McpTool("Search", _search, {"type": "object"}),
+            plugin_id="demo",
+        )
+        names = [t.name for t in all_tools()]
+        assert names[: len(TOOLS)] == [t.name for t in TOOLS]
+        assert "demo__search" in names
+
+    async def test_call_dispatches_to_plugin(self, registry):
+        from module.mcp.tools import handle_tool
+
+        add(registry, points.MCP_TOOL, "search", McpTool("S", _search), "demo")
+        [content] = await handle_tool("demo__search", {"q": "x"})
+        assert json.loads(content.text) == {"q": "x", "hits": [1, 2]}
+
+    async def test_unknown_names_keep_error(self, registry):
+        from module.mcp.tools import handle_tool
+
+        add(registry, points.MCP_TOOL, "search", McpTool("S", _search), "demo")
+        for name in ("nope", "demo__nope", "other__search"):
+            [content] = await handle_tool(name, {})
+            assert json.loads(content.text) == {"error": f"Unknown tool: {name}"}
+
+    async def test_failure_returns_error_and_counts(self, registry, breaker):
+        from module.mcp.tools import handle_tool
+
+        async def boom(args):
+            raise RuntimeError("plugin bug")
+
+        add(registry, points.MCP_TOOL, "boom", McpTool("B", boom), "demo")
+        ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker))
+        with patch("module.mcp.plugins.get_context", return_value=ctx):
+            [content] = await handle_tool("demo__boom", {})
+        assert json.loads(content.text) == {"error": "plugin bug"}
+        assert breaker._failures["demo"] == 1
+
+    def test_invalid_names_and_bad_factories_skipped(self, registry):
+        from module.mcp.plugins import list_plugin_tools
+
+        add(registry, points.MCP_TOOL, "has.dot", McpTool("x", _search), "demo")
+        add(registry, points.MCP_TOOL, "wrong", "not a tool", "demo")
+        add(registry, points.MCP_TOOL, "ok", McpTool("x", _search), "demo")
+        assert [t.name for t in list_plugin_tools()] == ["demo__ok"]
+
+    async def test_plugin_resources(self, registry):
+        from module.mcp.resources import RESOURCES, all_resources, handle_resource
+
+        async def stats():
+            return {"count": 3}
+
+        async def text():
+            return "plain"
+
+        add(registry, points.MCP_RESOURCE, "stats", McpResource("Stats", stats), "demo")
+        add(registry, points.MCP_RESOURCE, "text", McpResource("Text", text), "demo")
+        uris = [str(r.uri) for r in all_resources()]
+        assert uris[: len(RESOURCES)] == [str(r.uri) for r in RESOURCES]
+        assert "autobangumi://plugins/demo/stats" in uris
+        stats_json = await handle_resource("autobangumi://plugins/demo/stats")
+        assert json.loads(stats_json) == {"count": 3}
+        assert await handle_resource("autobangumi://plugins/demo/text") == "plain"
+        missing = await handle_resource("autobangumi://plugins/demo/none")
+        assert json.loads(missing) == {
+            "error": "Unknown resource: autobangumi://plugins/demo/none"
+        }
+
+    @pytest.mark.parametrize("resource_id", ["stats", "Weekly Stats", "统计"])
+    async def test_read_resource_via_protocol_returns_plugin_content(
+        self, registry, resource_id
+    ):
+        from mcp import types
+
+        from module.mcp import server
+
+        async def stats():
+            return {"count": 3}
+
+        add(
+            registry,
+            points.MCP_RESOURCE,
+            resource_id,
+            McpResource("Stats", stats),
+            "demo",
+        )
+        [listed] = [
+            r
+            for r in await server.list_resources()
+            if str(r.uri).startswith("autobangumi://plugins/")
+        ]
+        handler = server.server.request_handlers[types.ReadResourceRequest]
+        result = await handler(
+            types.ReadResourceRequest(
+                method="resources/read",
+                params=types.ReadResourceRequestParams(uri=listed.uri),
+            )
+        )
+        assert isinstance(result.root, types.ReadResourceResult)
+        [content] = result.root.contents
+        assert isinstance(content, types.TextResourceContents)
+        assert json.loads(content.text) == {"count": 3}
+
+    @pytest.mark.parametrize(
+        "point, impl",
+        [
+            (points.MCP_TOOL, McpTool("bad", _search, None)),  # type: ignore[arg-type]
+            (points.MCP_RESOURCE, McpResource(None, _search)),  # type: ignore[arg-type]
+        ],
+    )
+    def test_list_malformed_plugin_spec_skipped_and_counted(
+        self, registry, breaker, point, impl
+    ):
+        from module.mcp.resources import RESOURCES, all_resources
+        from module.mcp.tools import TOOLS, all_tools
+
+        add(registry, point, "bad", impl, "demo")
+        ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker))
+        with patch("module.mcp.plugins.get_context", return_value=ctx):
+            assert len(all_tools()) == len(TOOLS)
+            assert len(all_resources()) == len(RESOURCES)
+        assert breaker._failures["demo"] == 1
+
+    async def test_read_plugin_resource_hung_handler_times_out_and_counts(
+        self, registry, breaker, monkeypatch
+    ):
+        import asyncio
+
+        from module.mcp import plugins
+        from module.mcp.resources import handle_resource
+
+        async def hang():
+            await asyncio.Event().wait()
+
+        monkeypatch.setattr(plugins, "DEFAULT_TOOL_TIMEOUT", 0.01)
+        add(registry, points.MCP_RESOURCE, "hang", McpResource("H", hang), "demo")
+        ctx = SimpleNamespace(plugins=SimpleNamespace(breaker=breaker))
+        with patch("module.mcp.plugins.get_context", return_value=ctx):
+            raw = await handle_resource("autobangumi://plugins/demo/hang")
+        assert "error" in json.loads(raw)
+        assert breaker._failures["demo"] == 1
+
+    async def test_mcp_server_handlers_include_plugins(self, registry):
+        from module.mcp import server
+
+        add(registry, points.MCP_TOOL, "search", McpTool("S", _search), "demo")
+        assert "demo__search" in [t.name for t in await server.list_tools()]
+
+
+# ---------------------------------------------------------------- end to end
+
+
+class Options(BaseModel):
+    greeting: str = "hi"
+
+
+class FullPlugin(Plugin[Options]):
+    """一个同时使用全部 P5 扩展点的插件。"""
+
+    config_model = Options
+    seen: list[Event] = []
+
+    @provider(points.API_ROUTER, id="api")
+    def api(self):
+        router = APIRouter()
+
+        @router.get("/greet")
+        async def greet():
+            return {"greeting": self.config.greeting}
+
+        return router
+
+    @provider(points.MCP_TOOL, id="greet")
+    def greet_tool(self):
+        async def handler(args):
+            return {"greeting": self.config.greeting}
+
+        return McpTool("Greet", handler)
+
+    @subscribe("rss_failure")
+    async def on_rss_failure(self, event):
+        FullPlugin.seen.append(event)
+
+    @hook(points.MESSAGE_TEMPLATE)
+    def template(self, message, event, channel):
+        return RenderedMessage(f"{self.config.greeting}: {message.title}", message.body)
+
+
+MANIFEST = """
+[plugin]
+id = "full"
+name = "Full"
+version = "1.0.0"
+sdk = ">=0.3,<1"
+entry = "x:Y"
+"""
+
+
+async def test_plugin_end_to_end(registry, monkeypatch, tmp_path):
+    from module.core import AppContext
+    from module.core.scheduler import Scheduler
+    from module.mcp.tools import all_tools, handle_tool
+
+    # AppContext 会设置进程级 bus/runner；用 monkeypatch 让测试结束后复原
+    monkeypatch.setattr(host, "_bus", None)
+    monkeypatch.setattr(host, "_runner", None)
+    candidate = PluginCandidate(
+        manifest=parse_manifest(MANIFEST),
+        source="builtin",
+        load=lambda: FullPlugin,  # type: ignore[arg-type]
+    )
+    settings_obj = SimpleNamespace(
+        plugins=Plugins(options={"full": {"greeting": "yo"}})
+    )
+    manager = PluginManager(
+        settings_obj,
+        registry=registry,
+        discover_fn=lambda: ([candidate], []),
+        data_root=tmp_path,
+    )
+    notifier = NotificationManager(load_providers=False)
+    ctx = AppContext(
+        settings_obj, notifier, Scheduler([]), MagicMock(), plugins=manager
+    )
+    assert host.get_bus() is manager.bus
+    assert host.get_runner() is manager.runner
+    FullPlugin.seen = []
+    try:
+        with patch("module.core.context.reset_llm_parser"):
+            await manager.start()
+        assert manager.statuses()[0].state == "active"
+
+        # 事件：send_event → 总线 → 插件订阅者
+        event = RssFailureEvent(rss_name="Feed", rss_url="http://x", error="boom")
+        with patch("module.notification.manager.record_event", new_callable=AsyncMock):
+            await notifier.send_event(event)
+        await manager.bus.drain()
+        assert FullPlugin.seen == [event]
+
+        # 模板
+        title, _ = await render_event(event, "bark")
+        assert title == f"yo: {event.describe()[0]}"
+
+        # MCP
+        assert "full__greet" in [t.name for t in all_tools()]
+        [content] = await handle_tool("full__greet", {})
+        assert json.loads(content.text) == {"greeting": "yo"}
+
+        # 路由
+        app = FastAPI()
+        app.include_router(v1, prefix="/api")
+        app.dependency_overrides[get_context] = lambda: ctx
+
+        async def user():
+            return "u"
+
+        app.dependency_overrides[get_current_user] = user
+        client = TestClient(app)
+        assert client.get("/api/v1/plugins/full/greet").json() == {"greeting": "yo"}
+
+        # 停用后全部消失
+        settings_obj.plugins.enabled["full"] = False
+        await manager.apply_settings()
+        assert client.get("/api/v1/plugins/full/greet").status_code == 404
+        assert "full__greet" not in [t.name for t in all_tools()]
+        assert await render_event(event, "bark") == event.describe()
+    finally:
+        await manager.stop()
+
+
+# ---------------------------------------------------------------- MCP SSE
+
+
+async def test_mcp_sse_endpoint_sends_response_only_once(monkeypatch):
+    """/sse 结束后不能再发第二次 http.response.start（3.x 起每次断开都报
+    AssertionError）。"""
+    from contextlib import asynccontextmanager
+
+    from module.mcp import server as mcp_server
+
+    @asynccontextmanager
+    async def fake_connect_sse(scope, receive, send):
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        await send({"type": "http.response.body", "body": b"", "more_body": False})
+        yield (object(), object())
+
+    async def fake_run(*args, **kwargs):
+        return None
+
+    monkeypatch.setattr(mcp_server.sse, "connect_sse", fake_connect_sse)
+    monkeypatch.setattr(mcp_server.server, "run", fake_run)
+    sent: list[dict] = []
+
+    async def receive():
+        return {"type": "http.disconnect"}
+
+    async def send(message):
+        sent.append(message)
+
+    await mcp_server.handle_sse(
+        {"type": "http", "method": "GET", "path": "/sse", "headers": []},
+        receive,
+        send,
+    )
+    assert [m["type"] for m in sent].count("http.response.start") == 1

docs/dev/plugins.md

@@ -14,6 +14,9 @@
 | 搜索站点 | `points.SEARCH_SITE` | `SearchSite` | 搜索框的站点列表 |
 | 定时任务 | `points.SCHEDULED_TASK` | `ScheduledTask` | 插件启用即生效 |
 | 元数据源 | `points.METADATA_PROVIDER` | `MetadataProvider` | RSS 订阅的「解析器」 |
+| REST 路由 | `points.API_ROUTER` | `fastapi.APIRouter` | 插件启用即生效 |
+| MCP 工具 | `points.MCP_TOOL` | `McpTool` | 插件启用即生效 |
+| MCP 资源 | `points.MCP_RESOURCE` | `McpResource` | 插件启用即生效 |
 
 以及以下钩子（`@hook`，插件启用即生效）：
 
@@ -23,8 +26,9 @@
 | `title.parsed` | `points.TITLE_PARSED` | transform | 修正标题解析结果 |
 | `torrent.adding` | `points.TORRENT_ADDING` | transform | 修改发给下载器的保存路径、分类、标签 |
 | `http.request` | `points.HTTP_REQUEST` | transform | 修改 AB 发出的 GET 请求头（私有站 Cookie 等） |
+| `message_template` | `points.MESSAGE_TEMPLATE` | transform | 改写系统事件推送到外部渠道的文案 |
 
-此外，插件可以订阅事件（`@subscribe`），并使用私有的键值存储和数据目录。重命名等整理流水线的扩展点会在后续版本开放。
+此外，插件可以订阅系统事件（`@subscribe`），并使用私有的键值存储和数据目录。重命名等整理流水线的扩展点会在后续版本开放。
 
 ## 目录结构
 
@@ -296,7 +300,112 @@ class MyPlugin(Plugin):
         return BangumiTv()
 ```
 
-`request.kind` 区分番剧与电影，`request.torrent.homepage` 是种子的详情页。返回值是完整的新元数据，通常基于 `request.current` 修改。失败或超时时保留原值，并计入熔断。目前 WebUI 的订阅表单只提供内置取值，插件元数据源需要通过 API 设置订阅的 `parser` 字段。
+`request.kind` 区分番剧与电影，`request.torrent.homepage` 是种子的详情页。返回值是完整的新元数据，通常基于 `request.current` 修改。失败或超时时保留原值，并计入熔断。添加订阅时，「解析器」下拉框会列出插件提供的元数据源。
+
+### REST 路由
+
+```python
+from fastapi import APIRouter
+
+@provider(points.API_ROUTER, id="api")
+def api(self):
+    router = APIRouter()
+
+    @router.get("/stats")
+    async def stats():
+        return {"count": await self.ctx.kv.get("count", 0)}
+
+    return router
+```
+
+- 路由挂载在 `/api/v1/plugins/<插件 id>/` 下，上例即 `GET /api/v1/plugins/my-plugin/stats`。
+- **所有插件路由都强制登录鉴权**，与 WebUI 其它接口使用同一套凭据（会话 Cookie 或 `scope=api` 的 API 令牌）。插件无法注册匿名端点；未登录返回 401。
+- 路由随插件启用出现、随停用消失，无需重启。同一插件可以提供多个 `API_ROUTER`，它们会合并到同一前缀下。
+- 处理函数抛出 `HTTPException` 正常返回对应状态码；抛出其它异常返回 500，并计入熔断。
+- 插件路由不出现在 `/docs` 的 OpenAPI 文档中。
+
+### MCP 工具与资源
+
+```python
+from ab_sdk.mcp import McpResource, McpTool
+
+@provider(points.MCP_TOOL, id="search")
+def search_tool(self):
+    async def handler(args: dict):
+        return {"results": [...], "keyword": args["keyword"]}
+
+    return McpTool(
+        description="在私有站点搜索",
+        handler=handler,
+        input_schema={
+            "type": "object",
+            "properties": {"keyword": {"type": "string"}},
+            "required": ["keyword"],
+        },
+    )
+
+@provider(points.MCP_RESOURCE, id="stats")
+def stats_resource(self):
+    async def handler():
+        return {"count": 3}
+
+    return McpResource(name="统计", handler=handler)
+```
+
+- 对外名称自动加插件 id 前缀，避免冲突：工具名为 `<插件 id>__<id>`（双下划线），资源 URI 为 `autobangumi://plugins/<插件 id>/<id>`。不用 `.` 分隔，是因为不少 MCP 客户端会把工具名原样交给 LLM API，而后者通常只接受 `^[a-zA-Z0-9_-]{1,64}$`；不合规的工具名会被跳过并记录日志。
+- 工具处理函数返回可 JSON 序列化的对象；资源处理函数返回字符串（原样返回）或可 JSON 序列化的对象。抛出的异常以 `{"error": "..."}` 返回给客户端，并计入熔断。工具默认超时 60 秒（`McpTool(timeout=...)`）。
+- MCP 端点沿用 AB 的访问控制（IP 白名单或 `scope=mcp` 令牌），插件无需自行鉴权。
+
+### 订阅系统事件
+
+```python
+from ab_sdk import subscribe
+from ab_sdk.events import RssFailureEvent
+
+@subscribe("rss_failure")
+async def on_rss_failure(self, event: RssFailureEvent):
+    self.ctx.log.warning("订阅 %s 失败：%s", event.rss_name, event.error)
+```
+
+AB 发出的、会进入通知中心的事件都是 `ab_sdk.events.SystemEvent` 的子类，带有 `severity`、`payload()`、`describe()`（默认中文标题与正文）和 `i18n()`（前端 i18n key 与参数）。它们先写入通知中心，再发布到事件总线，最后推送到外部通知渠道；关闭「通知」开关只影响外部推送，不影响插件收到事件。
+
+| kind | 事件类 | 何时发布 |
+| --- | --- | --- |
+| `rss_failure` | `RssFailureEvent` | RSS 订阅从正常变为连接异常 |
+| `download_failure` | `DownloadFailureEvent` | 种子重试后仍添加失败 |
+| `offset_review` | `OffsetReviewEvent` | 番剧的季度 / 集数偏移需要人工确认 |
+| `downloader_unavailable` | `DownloaderUnavailableEvent` | 下载器连不上、凭据错误或 IP 被封 |
+| `update_available` | `UpdateAvailableEvent` | 检查到新版本 |
+| `update_applied` / `update_failed` | `UpdateAppliedEvent` | 在线更新成功 / 失败（`kind` 随 `success` 变化） |
+| `llm_auth_failure` | `LLMAuthFailureEvent` | 订阅类 LLM 提供商凭据失效 |
+| `llm_plugin_install_failed` | `LLMPluginInstallFailedEvent` | LLM 插件安装失败 |
+| `rename_conflict` | `RenameConflictEvent` | 媒体文件重命名遇到目标路径冲突 |
+| `plugin.loaded` | `PluginLoaded` | 插件加载成功 |
+| `plugin.disabled` | `PluginDisabled` | 插件加载失败或被熔断 |
+| `inbox.changed` | — | 通知中心有新消息、已读或删除（宿主内部使用，供 SSE 推送） |
+
+系统事件的 `kind` 沿用 3.x 的取值（通知中心按它存储和翻译），因此不带 `.` 前缀；插件自己的事件仍必须以 `<插件 id>.` 开头。后续版本会陆续增加流水线事件（如 `torrent.added`、`file.renamed`）。
+
+### 通知文案模板
+
+系统事件推送到外部渠道前，会依次经过 `message_template` 钩子。钩子拿到默认文案、事件和渠道类型，返回新的 `RenderedMessage` 即可改写，返回 `None` 表示不修改：
+
+```python
+from ab_sdk import hook, points
+from ab_sdk.notify import RenderedMessage
+
+@hook(points.MESSAGE_TEMPLATE)
+def template(self, message: RenderedMessage, event, channel: str):
+    if event.kind != "rss_failure":
+        return None
+    if channel == "telegram":
+        return RenderedMessage(f"[告警] {message.title}", f"<b>{event.rss_name}</b>\n{event.error}")
+    return None
+```
+
+- 只影响外部推送；通知中心里的文案由前端按 `i18n()` 渲染，不受影响。
+- 钩子对每个启用的渠道各调用一次。钩子出错或超时时沿用上一步的文案，并计入熔断；返回值类型不对时整条消息回退到默认文案。
+- 「新集数」通知暂不经过该钩子，仍使用通知渠道里配置的单集模板。
 
 ## 测试
 

docs/plans/2026-10-08-plugin-architecture-4.0-design.md

@@ -34,6 +34,17 @@
 - **推迟**：`feed_source`（与 `FeedConfig.source` 替代 `RSSItem.parser` 一起设计）、`title_parser` 链、`admission_policy`、`bangumi_matcher`、`release_ranker`、`rule.created`、`save_path` Provider，以及第 11 节的批量 `accept_many`。
 - `module/plugin/context.py` 改为延迟 import `module.database`，否则网络层引用 `module.plugin.host` 会形成循环依赖。
 
+### 实施中的调整（P5）
+
+- **系统事件类移入 `ab_sdk.events`**，`SystemEvent` 由封闭 Union 改为基类（插件也可定义可通知事件）；`module.notification.events` 只做再导出。`kind` 沿用 3.x 取值（`rss_failure` 等，通知中心按它存储与翻译），不改成 `rss.failure` 式的点分名。
+- **通知管理器仍是事件入口**：第 4.2 节设想的「通知管理器、inbox 都作为总线订阅者」未实施。`NotificationManager.send_event` 依次写通知中心、发布到总线、推送外部渠道，调用方与外部行为不变；改成纯总线驱动要等 P3/P4 的流水线事件稳定后再统一。
+- **SSE 只有 `notification` 改为订阅总线**（`inbox.changed`，通知中心写入 / 已读 / 删除时发布，立即推送）。status / downloader / log / update 是状态快照而非事件，仍按节拍采样；`inbox_revision` 计数保留，作为帧里的 `revision` 字段。
+- **插件路由用分发路由实现**：FastAPI 不能卸载路由，宿主只注册 `/api/v1/plugins/{plugin_id}/{path:path}`（鉴权依赖在此强制），按插件 id 转发给由插件 `APIRouter` 合并成的 ASGI 应用，插件变更时重建。插件路由不进 OpenAPI。
+- **按插件作用域的 Provider id**：`ExtensionPoint` 新增 `scoped`，`api_router` / `mcp_tool` / `mcp_resource` 的 id 只需在插件内唯一（注册表键为 `<plugin_id>/<id>`）。MCP 工具名用 `<plugin-id>__<id>` 而非 `.`，因为 Anthropic / OpenAI 等 LLM API 的工具名只接受 `^[a-zA-Z0-9_-]{1,64}$`。
+- **`message_template` 只作用于系统事件的外部推送**，钩子签名 `(RenderedMessage, event, channel)`，按渠道各调用一次；「新集数」通知仍用渠道里的单集模板（`_format_message` 是同步接口，改造留到通知渠道迁到 `config_model` 时）。
+- **i18n（第 11 节风险 3）先给出 key**：`SystemEvent.i18n()` 返回 `notifications.kind.<kind>` 与 `payload()`，与前端现有文案键一致；外部推送与通知中心的中文兜底文案不变。插件事件的前端翻译依赖 P6 的 `host.i18n`。
+- 进程级访问器 `module.plugin.host.get_bus/set_bus`、`get_runner/set_runner`、`publish()`，由 `AppContext` 构造时设置，未设置时为空操作。MCP 的 `tools/list_changed` 通知未实现（客户端重新 list 即可看到插件工具变化）。
+
 ## 1. 背景与目标
 
 AB 目前只有 **LLM 提供商** 是真正的运行时插件系统：签名下载、目录加载、懒导入、热重载。
@@ -581,7 +592,7 @@ organize: downloader.completed → media_files.classify → file_parser
 | **P2.5 多下载器** | 下载器多实例；`downloader_id` 列与迁移；按实例路由 add / rename / delete；organize 逐实例扫描 | qb + aria2 并存的 e2e 用例；单实例行为不变 |
 | **P3 流水线插件化：ingest** | `torrent.filter`、`title.parsed`、`torrent.adding`、`http.request` 钩子；`metadata_provider`（mikan / tmdb 以 `core` 登记）；内置插件 `ingest-filters`（包含过滤） | 已完成；无插件时行为不变（全量测试）。`feed_source`、`title_parser` 链、`admission_policy`、`matcher`、`ranker`、`save_path`、size 过滤、按订阅覆盖推迟，见第 0 节 |
 | **P4 流水线插件化：organize** | `media_files`、`file_parser`、`rename_strategy`（含 `template`）、`conflict_policy`、`file.renamed` 等事件 | 模板重命名；硬链接示例插件 |
-| **P5 事件与外部接口** | SSE 改订阅 bus；`api_router`、`mcp_tool` 扩展点；`message_template` | 删掉 SSE 轮询；插件 MCP 工具 |
+| **P5 事件与外部接口** | SSE 改订阅 bus；`api_router`、`mcp_tool` 扩展点；`message_template` | 已完成：系统事件上总线、通知中心 SSE 改为事件推送、插件路由 / MCP 工具与资源 / 通知模板；status 等快照类 SSE 仍按节拍采样。调整见第 0 节 |
 | **P6 前端插件** | Web Component 挂载点、`AbHost` 桥接、错误边界、`/plugins/<id>/web` 静态资源、`@autobangumi/plugin-ui` 包 | 示例插件「手动选种」以详情页标签形式可用 |
 | **P7 生态** | 插件管理页（安装、启停、日志、错误）、签名目录发布流程、模板仓库（含前端模板）、`ab-plugin` CLI、文档（中 / 英 / 日） | 6 个以上示例插件上架 |
 | **P8 发布** | beta 测试、性能对比（RSS 刷新耗时、内存）、升级指南、`docs/changelog/4.0.md` | `4.0.0-beta.1` → `4.0.0` |