如何同时运行Flask Webhook与aiogram3机器人并实现数据互通？

核心问题分析 #

你之前尝试的异步/线程方案失败，本质是因为Flask（默认同步）和aiogram3（基于asyncio）运行在不同的事件循环/线程中，直接跨上下文调用机器人方法会导致循环不兼容、数据无法同步的问题。最可靠的解决方案是用消息队列实现跨服务的异步通信，这里以轻量的Redis Pub/Sub为例。

方案一：Redis Pub/Sub实现数据传递 #

1. 依赖安装 #

pip install redis flask aiogram python-dotenv

2. Flask Webhook服务（发送数据到队列） #

创建flask_webhook.py：

from flask import Flask, request, jsonify
import redis
import json

app = Flask(__name__)
# 连接Redis，本地默认配置，生产环境需调整参数
redis_client = redis.Redis(host='localhost', port=6379, db=0)
CHANNEL_NAME = "bot_updates"

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.get_json()
    if not data:
        return jsonify({"status": "error", "msg": "No data provided"}), 400
    
    # 将数据序列化后发布到Redis频道（生产环境用json更安全）
    redis_client.publish(CHANNEL_NAME, json.dumps(data))
    return jsonify({"status": "success"}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

3. aiogram3机器人服务（从队列订阅数据） #

创建aiogram_bot.py：

import asyncio
import redis.asyncio as redis
import json
from aiogram import Bot, Dispatcher, types
from aiogram.filters.command import Command
from dotenv import load_dotenv
import os

load_dotenv()
BOT_TOKEN = os.getenv("BOT_TOKEN")
CHANNEL_NAME = "bot_updates"

bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()

# 存储需要通知的用户ID（可根据业务调整，比如从数据库读取）
TARGET_USER_ID = 123456789  # 替换成你的目标用户ID

async def redis_subscriber():
    # 异步Redis客户端，适配aiogram的asyncio环境
    redis_client = redis.Redis(host='localhost', port=6379, db=0)
    pubsub = redis_client.pubsub()
    await pubsub.subscribe(CHANNEL_NAME)
    
    while True:
        message = await pubsub.get_message(ignore_subscribe_messages=True)
        if message:
            # 解析收到的数据
            data = json.loads(message['data'].decode('utf-8'))
            print(f"Received from webhook: {data}")
            # 调用机器人方法处理数据，比如给指定用户发消息
            try:
                await bot.send_message(chat_id=TARGET_USER_ID, text=f"Webhook推送数据：{str(data)}")
            except Exception as e:
                print(f"消息发送失败：{e}")
        await asyncio.sleep(0.1)

@dp.message(Command("start"))
async def start_command(message: types.Message):
    await message.answer("机器人已启动，等待Webhook数据推送...")

async def main():
    # 在机器人启动时启动Redis订阅任务
    asyncio.create_task(redis_subscriber())
    await dp.start_polling(bot)

if __name__ == '__main__':
    asyncio.run(main())

4. 运行说明 #

• 先启动Redis服务（本地安装后直接运行redis-server）
• 启动Flask服务：python flask_webhook.py
• 启动aiogram机器人：python aiogram_bot.py
• 向http://localhost:5000/webhook发送POST请求（比如用curl或Postman），机器人会收到数据并推送给指定用户

方案二：同进程异步服务（适用于Flask与机器人同部署） #

如果你的Flask和机器人可以运行在同一个进程中，可改用异步Web框架（如FastAPI）共享asyncio loop：

import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters.command import Command
from fastapi import FastAPI
from pydantic import BaseModel

BOT_TOKEN = "你的机器人Token"
TARGET_USER_ID = 123456789

bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()
app = FastAPI()

class WebhookData(BaseModel):
    content: str

@app.post("/webhook")
async def webhook(data: WebhookData):
    # 直接在异步上下文调用机器人方法
    try:
        await bot.send_message(chat_id=TARGET_USER_ID, text=f"Webhook数据：{data.content}")
    except Exception as e:
        print(f"消息发送失败：{e}")
    return {"status": "success"}

@dp.message(Command("start"))
async def start(message: types.Message):
    await message.answer("机器人已启动")

async def run_bot():
    await dp.start_polling(bot)

async def main():
    # 同时启动FastAPI和机器人
    import uvicorn
    config = uvicorn.Config(app, host="0.0.0.0", port=5000)
    server = uvicorn.Server(config)
    await asyncio.gather(server.serve(), run_bot())

if __name__ == "__main__":
    asyncio.run(main())

关键注意事项 #

• 生产环境中必须用json序列化数据，禁止使用eval避免安全风险
• 如果需要处理高并发或消息持久化，可替换为RabbitMQ等更专业的消息队列
• 确保目标用户ID是已与机器人发起过对话的用户，否则会因权限问题发送失败

内容的提问来源于stack exchange，提问作者Marz1k