如何用Telethon Bot API主动发消息？运行报错求助

我正在测试Telethon Bot，尝试向测试频道发送内容为“bot”的消息，使用的代码如下：

bot = TelegramClient(
    Path("tgapi.bot.session"),
    config.api_id,
    config.api_hash
)
bot.start(bot_token=config.bot_token)


async def bot_main():
    await bot.send_message(config.demo, "bot")


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

运行后出现如下错误：

错误信息 #

Traceback (most recent call last):
  File "/home/trdthg/repo/tgapi/tgapi-py/demo.py", line 20, in <module>
    asyncio.run(bot_main())
  File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/home/trdthg/repo/tgapi/tgapi-py/demo.py", line 16, in bot_main
    await bot.send_message(config.demo, "bot")
  File "/home/trdthg/repo/tgapi/tgapi-py/.venv/lib/python3.10/site-packages/telethon/client/messages.py", line 905, in send_message
    result = await self(request)
  File "/home/trdthg/repo/tgapi/tgapi-py/.venv/lib/python3.10/site-packages/telethon/client/users.py", line 30, in __call__
    return await self._call(self._sender, request, ordered=ordered)
  File "/home/trdthg/repo/tgapi/tgapi-py/.venv/lib/python3.10/site-packages/telethon/client/users.py", line 34, in _call
    raise RuntimeError('The asyncio event loop must not change after connection (see the FAQ for details)')
RuntimeError: The asyncio event loop must not change after connection (see the FAQ for details)
Task was destroyed but it is pending!

...

请问Telethon Bot无法主动发送消息吗？

Telethon Bot完全可以主动发送消息，你遇到的错误是事件循环使用不当导致的：bot.start()会绑定到当前线程的默认事件循环，但后续asyncio.run()会创建新的事件循环，两者冲突触发报错。

修正后的代码示例 #

场景1：仅发送单次消息后断开

import asyncio
from pathlib import Path
import config
from telethon import TelegramClient

async def bot_main():
    # 在异步函数内初始化并启动客户端
    bot = TelegramClient(
        Path("tgapi.bot.session"),
        config.api_id,
        config.api_hash
    )
    await bot.start(bot_token=config.bot_token)
    await bot.send_message(config.demo, "bot")
    await bot.disconnect()

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

场景2：保持客户端长期运行（可同时处理消息更新）

import asyncio
from pathlib import Path
import config
from telethon import TelegramClient

async def send_test_message(bot):
    await bot.send_message(config.demo, "bot")

async def main():
    bot = TelegramClient(
        Path("tgapi.bot.session"),
        config.api_id,
        config.api_hash
    )
    await bot.start(bot_token=config.bot_token)
    await send_test_message(bot)
    # 保持客户端运行以接收后续更新
    await bot.run_until_disconnected()

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

关键注意点 #

• 不要在异步函数外调用bot.start()，它会绑定到初始事件循环，与asyncio.run()创建的新循环冲突。
• 所有与Telethon客户端的异步交互操作，必须在同一个事件循环内完成，确保客户端启动和后续调用使用同一循环。

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