基于aiogram(Python)的双Telegram Bot同项目Webhook整合问题

出现AttributeError: 'NoneType' object has no attribute 'current_state'的核心原因是当前任务上下文未绑定有效的Dispatcher实例：

• 自定义Webhook服务器分发更新时，若未将更新通过对应Dispatcher的process_update方法处理，而是直接在Web框架的路由上下文（如FastAPI/Flask的视图函数）中调用Dispatcher.get_current()，此时aiogram的任务上下文未注入Dispatcher，导致返回None。
• 多Dispatcher（店铺Bot和配送Bot）场景下，若上下文隔离处理不当，全局的get_current()无法定位到对应Bot的Dispatcher实例。
• 此前用asyncio.gather启动长轮询时，可能未确保每个Dispatcher的start_polling任务在独立的上下文环境中运行，导致上下文混乱。

1. 正确分发更新到对应Dispatcher #

自定义Webhook服务器收到Telegram的更新后，需先判断更新归属（通过update.effective_bot.token或业务逻辑区分店铺/配送Bot），然后调用对应Dispatcher的process_update方法处理，确保更新在Dispatcher的上下文环境中流转：

# 假设用FastAPI搭建Webhook服务器
@app.post("/webhook/{bot_token}")
async def webhook_handler(bot_token: str, update: dict):
    # 根据bot_token匹配对应的Dispatcher
    if bot_token == STORE_BOT_TOKEN:
        dp = store_dp
    elif bot_token == COURIER_BOT_TOKEN:
        dp = courier_dp
    else:
        return JSONResponse(status_code=400, content={"error": "Invalid bot token"})
    
    # 将更新交给Dispatcher处理，确保上下文绑定
    telegram_update = types.Update(**update)
    await dp.process_update(telegram_update)
    return JSONResponse(status_code=200)

在Dispatcher注册的handler中调用Dispatcher.get_current().current_state()，即可正常获取状态。

2. 多Bot长轮询的正确实现 #

若采用长轮询模式，需为每个Bot创建独立的Dispatcher和Bot实例，并通过asyncio.gather并行启动，确保每个任务的上下文隔离：

import asyncio
from aiogram import Bot, Dispatcher, types

STORE_BOT_TOKEN = "your_store_token"
COURIER_BOT_TOKEN = "your_courier_token"

async def setup_store_bot():
    bot = Bot(token=STORE_BOT_TOKEN)
    dp = Dispatcher(bot)
    
    # 注册店铺Bot的handler，示例：
    @dp.message_handler(commands=["start"])
    async def store_start(message: types.Message):
        state = Dispatcher.get_current().current_state()
        await state.set_state("store_menu")
        await message.answer("欢迎来到店铺Bot")
    
    await dp.start_polling()

async def setup_courier_bot():
    bot = Bot(token=COURIER_BOT_TOKEN)
    dp = Dispatcher(bot)
    
    # 注册配送Bot的handler，示例：
    @dp.message_handler(commands=["start"])
    async def courier_start(message: types.Message):
        state = Dispatcher.get_current().current_state()
        await state.set_state("courier_dashboard")
        await message.answer("欢迎来到配送Bot")
    
    await dp.start_polling()

async def main():
    await asyncio.gather(setup_store_bot(), setup_courier_bot())

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

3. 避免在非Handler上下文调用current_state() #

如果需要在工具函数中使用状态，不要依赖Dispatcher.get_current()，而是从handler中把state作为参数传递进去：

# 错误示例：在工具函数中直接调用
async def some_tool_func():
    state = Dispatcher.get_current().current_state()  # 可能返回None
    # ...

# 正确示例：从handler传递state
async def some_tool_func(state):
    await state.set_state("some_state")
    # ...

@dp.message_handler()
async def handler(message: types.Message):
    state = await Dispatcher.get_current().current_state()
    await some_tool_func(state)

4. 显式绑定Dispatcher上下文（进阶） #

若必须在自定义上下文（如Webhook路由）中操作状态，可使用dp.ctx.run方法显式绑定Dispatcher上下文：

@app.post("/webhook/{bot_token}")
async def webhook_handler(bot_token: str, update: dict):
    dp = get_dispatcher_by_token(bot_token)
    async with dp.ctx.run():
        # 在此上下文中可正常调用Dispatcher.get_current()
        state = Dispatcher.get_current().current_state()
        # ...

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