内置一个简易的 HTTP API 接口
为了让外部系统（如其他的独立脚本、青龙面板、宝塔监控、或者你的 Web 服务器）能够把报错直接传给 AstrBot 并推送到你的个人微信，最好的方案是在插件里内置一个简易的 HTTP API 接口（基于 FastAPI / Sanic，AstrBot 内部通常自带）。
这样，任何外部系统只需要向 AstrBot 发送一个标准的 POST 请求，微信就能立即收到报错通知。
以下是升级后的完整代码实现。
🐍 修改后的 __init__.py 核心代码
请用以下代码完全替换你之前的 data/plugins/wx_pusher/__init__.py。这段代码利用了 AstrBot 的组件，通过启动一个独立的轻量级异步服务（或直接挂载到 AstrBot 现有的 Web 服务上）来接收外部通知：
python
import logging
from astrbot.api.all import *
from astrbot.api.message_components import Plain
from aiohttp import web # 异步 HTTP 库，无需额外安装复杂依赖
logger = logging.getLogger("astrbot")
@register("wx_pusher", "你的名字", "个人微信外部报错接收推送插件", "1.1.0")
class WxPusherPlugin(Plugin):
def __init__(self, context: Context):
super().__init__(context)
# 1. 从配置文件读取默认接收者的微信 ID
self.config = self.context.get_config("wx_pusher") or {}
self.target_wxid = self.config.get("target_wxid", "")
# 2. 启动异步 HTTP 服务器监听外部报错
# 默认监听 19090 端口，可根据服务器安全组自行修改
self.api_port = 19090
self.runner = None
# 使用 AstrBot 的异步事件循环启动 Web 服务
import asyncio
asyncio.create_task(self.start_http_server())
async def start_http_server(self):
"""启动轻量级 HTTP 服务以接收外部系统请求"""
app = web.Application()
# 注册路由：POST http://IP:19090/webhook
app.router.add_post('/webhook', self.handle_webhook)
self.runner = web.AppRunner(app)
await self.runner.setup()
site = web.TCPSite(self.runner, '0.0.0.0', self.api_port)
await site.start()
logger.info(f"【报错推送】WebHook 服务已成功运行在 http://0.0.0:{self.api_port}/webhook")
async def handle_webhook(self, request):
"""处理外部系统传过来的报错数据"""
try:
data = await request.json()
# 解析外部系统传来的参数，例如: {"event": "server_error", "message": "内存爆满！"}
event_type = data.get("event", "系统通知")
message = data.get("message", "未提供详细报错内容")
# 拼装微信推送格式
push_content = f"🚨 **【服务器报警】**\n- **类型**: {event_type}\n- **详情**: {message}"
# 异步调用微信发送
success = await self.send_wx_message(self.target_wxid, push_content)
if success:
return web.json_response({"status": "success", "msg": "已推送到微信"})
else:
return web.json_response({"status": "error", "msg": "微信发送失败，请检查机器人状态"}, status=500)
except Exception as e:
logger.error(f"【报错推送】解析 Webhook 请求失败: {e}")
return web.json_response({"status": "error", "msg": str(e)}, status=400)
async def send_wx_message(self, target_id: str, text: str) -> bool:
"""核心推送函数"""
if not target_id or "需替换" in target_id:
logger.error("【微信推送】未配置有效的 target_wxid，请前往 WebUI 插件设置中填写。")
return False
platform = self.context.get_platform("weixin_oc")
if not platform:
logger.error("【微信推送】未找到 weixin_oc 机器人实例！请确认已扫码登录。")
return False
try:
message_chain = MessageChain().message(Plain(text))
await platform.send_message(target_id, message_chain)
logger.info(f"【微信推送】成功向 {target_id} 推送报错通知。")
return True
except Exception as e:
logger.error(f"【微信推送】消息发送失败: {e}")
return False
async def terminate(self):
"""插件卸载或重启时关掉端口占用"""
if self.runner:
await self.runner.cleanup()
logger.info("【报错推送】HTTP 服务已关闭。")
请谨慎使用此类代码。
📡 外部系统如何调用（发送报错）
配置好插件并重启 AstrBot 后，你的服务器就在 19090 端口开放了一个接收通道。
1. Python 外部脚本调用示例
如果你有其他 Python 脚本在运行，当抓取到 try...except 异常时，可以这样把报错甩给微信：
python
import requests
def report_to_wechat(event_name, error_msg):
# 替换为你部署 AstrBot 的服务器 IP
url = "http://127.0.0"
payload = {
"event": event_name,
"message": error_msg
}
try:
response = requests.post(url, json=payload, timeout=5)
print("通知结果:", response.json())
except Exception as e:
print("通知 AstrBot 失败:", e)
# 模拟业务出错
try:
1 / 0
except ZeroDivisionError as e:
report_to_wechat("MySQL 数据库断开", f"错误详情: {str(e)}。无法连接到 3306 端口，请立刻检查集群状态！")
请谨慎使用此类代码。
2. Shell 脚本 / 宝塔面板监控调用示例
如果是 Linux 服务器的 Shell 监控脚本（比如监控 Nginx 是否挂掉），可以直接用 curl 发送：
bash
curl -X POST http://127.0.0 \
-H "Content-Type: application/json" \
-d '{"event": "Nginx 服务挂死", "message": "检测到 80 端口无响应，尝试重启失败！"}'
请谨慎使用此类代码。
⚠️ 安全注意事项
防火墙安全组：如果你的外部系统和 AstrBot 不在同一台服务器上，你需要去云服务器后台（如腾讯云、阿里云）放行
19090端口，并且将代码中的127.0.0.1改为 AstrBot 所在服务器的公网 IP。防恶意刷消息：目前此接口为公网裸奔状态。如果你的接口暴露在公网上，建议在代码或 Payload 中加一个
token校验字段（例如验证请求头里是否包含密钥），防止不法分子扫描到端口后往你的微信狂发垃圾广告。