feat: 为插件添加统一webhook

Author: moemoli

astrbot/core/star/base.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from collections.abc import Awaitable, Callable
 from typing import Any, Protocol
 
 from astrbot.core import html_renderer
@@ -21,6 +22,14 @@ class Star(CommandParserMixin, PluginKVStoreMixin):
     class _ContextLike(Protocol):
         def get_config(self, umo: str | None = None) -> Any: ...
 
+        def register_unified_webhook(
+            self,
+            webhook_uuid: str,
+            view_handler: Callable[..., Awaitable[Any]],
+            methods: list[str] | None = None,
+            desc: str = "",
+        ) -> None: ...
+
     def __init__(self, context: _ContextLike, config: dict | None = None) -> None:
         self.context = context
 
@@ -77,6 +86,28 @@ async def html_render(
             options=options,
         )
 
+    def register_unified_webhook(
+        self,
+        webhook_uuid: str,
+        view_handler: Callable[..., Awaitable[Any]],
+        methods: list[str] | None = None,
+        desc: str = "",
+    ) -> None:
+        """注册统一 Webhook 回调。
+
+        插件可以通过该方法注册回调，Dashboard 会通过
+        /api/plug/webhook/<webhook_uuid> 转发到对应处理函数。
+        """
+        register = getattr(self.context, "register_unified_webhook", None)
+        if not callable(register):
+            raise RuntimeError("Context does not support unified webhook registration")
+        register(
+            webhook_uuid=webhook_uuid,
+            view_handler=view_handler,
+            methods=methods,
+            desc=desc,
+        )
+
     async def initialize(self) -> None:
         """当插件被激活时会调用这个方法"""
 

astrbot/core/star/context.py

@@ -57,6 +57,9 @@ class Context:
     """暴露给插件的接口上下文。"""
 
     registered_web_apis: list = []
+    registered_unified_webhooks: dict[
+        str, tuple[Callable[..., Awaitable[Any]], list[str], str]
+    ] = {}
 
     # 向后兼容的变量
     _register_tasks: list[Awaitable] = []
@@ -519,6 +522,31 @@ def register_web_api(
                 return
         self.registered_web_apis.append((route, view_handler, methods, desc))
 
+    def register_unified_webhook(
+        self,
+        webhook_uuid: str,
+        view_handler: Callable[..., Awaitable[Any]],
+        methods: list[str] | None = None,
+        desc: str = "",
+    ) -> None:
+        """注册统一 Webhook 回调。
+
+        Args:
+            webhook_uuid: Webhook 唯一标识。
+            view_handler: 异步视图处理函数。
+            methods: HTTP 方法列表，默认 ["GET", "POST"]。
+            desc: 回调描述。
+
+        Note:
+            如果相同 webhook_uuid 已注册，会覆盖原有回调。
+        """
+        normalized_methods = [method.upper() for method in (methods or ["GET", "POST"])]
+        self.registered_unified_webhooks[webhook_uuid] = (
+            view_handler,
+            normalized_methods,
+            desc,
+        )
+
     """
     以下的方法已经不推荐使用。请从 AstrBot 文档查看更好的注册方式。
     """

astrbot/dashboard/server.py

@@ -145,6 +145,11 @@ def __init__(
             view_func=self.srv_plug_route,
             methods=["GET", "POST"],
         )
+        self.app.add_url_rule(
+            "/api/plug/webhook/<webhook_uuid>",
+            view_func=self.srv_plug_webhook_route,
+            methods=["GET", "POST"],
+        )
 
         self.shutdown_event = shutdown_event
 
@@ -159,6 +164,21 @@ async def srv_plug_route(self, subpath, *args, **kwargs):
                 return await view_handler(*args, **kwargs)
         return jsonify(Response().error("未找到该路由").__dict__)
 
+    async def srv_plug_webhook_route(self, webhook_uuid, *args, **kwargs):
+        """插件统一 Webhook 路由"""
+        registered_unified_webhooks = (
+            self.core_lifecycle.star_context.registered_unified_webhooks
+        )
+        callback = registered_unified_webhooks.get(webhook_uuid)
+        if not callback:
+            return jsonify(Response().error("未找到对应 webhook").__dict__), 404
+
+        view_handler, methods, _ = callback
+        if request.method not in methods:
+            return jsonify(Response().error("请求方法不被允许").__dict__), 405
+
+        return await view_handler(*args, **kwargs)
+
     async def auth_middleware(self):
         if not request.path.startswith("/api"):
             return None
@@ -199,6 +219,7 @@ async def auth_middleware(self):
         allowed_endpoints = [
             "/api/auth/login",
             "/api/file",
+            "/api/plug/webhook",
             "/api/platform/webhook",
             "/api/stat/start-time",
             "/api/backup/download",  # 备份下载使用 URL 参数传递 token