使用asyncio开发Discord Bot遇事件循环错误及功能咨询

启动错误原因 #

你遇到的asyncio.run() cannot be called from a running event loop错误，是因为discord.py的bot.run()内部已经调用了asyncio.run()启动事件循环。如果你的代码在已有运行中事件循环的环境（比如Jupyter Notebook、IPython交互式终端）里执行，就会触发循环冲突。

启动错误修复方案 #

方案1：用普通命令行运行脚本 #

直接在Windows命令提示符、Linux终端或Mac的Terminal里执行脚本，避开有内置事件循环的交互式环境，这是最简单的解决方式。

方案2：手动管理事件循环（适配已有循环的环境） #

如果必须在已有事件循环的环境里运行，把代码最后一段替换为：

if __name__ == '__main__':
    import asyncio
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    loop.run_until_complete(bot.start('你的Bot令牌'))

代码功能逻辑的关键修复 #

除了启动问题，你的代码还有几个会导致功能失效的问题，必须修正：

1. invite命令参数处理错误 #

discord.py的命令装饰器会自动剥离命令前缀和命令名，用户输入!invite 服务器链接时，invite_url参数直接就是链接内容，不需要判断startswith('!invite ')，这段多余判断会导致命令报错，直接删除即可。

2. 同步requests阻塞Bot #

requests是同步HTTP库，调用它会阻塞整个Bot的事件循环，导致Bot完全无响应。换成异步HTTP库aiohttp，所有HTTP请求用异步方式执行。

3. OAuth2授权逻辑错误 #

你用的BackendApplicationClient是客户端凭证流，用于获取Bot自身的API权限，而非用户授权。要实现用户授权Bot代其加入服务器，需用授权码流，并在redirect_uri对应的服务器端处理用户授权后的回调，获取用户的访问令牌。

4. 批量添加用户的异步处理错误 #

原代码把同步requests请求放进列表再await是错误的，同步函数不能被await，需用aiohttp的异步请求配合asyncio.gather()批量处理。

修正后的完整代码 #

import os
import discord
from discord.ext import commands
import aiohttp
import asyncio
from urllib.parse import urlparse, parse_qs
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import WebApplicationClient

# 配置参数
intents = discord.Intents.default()
intents.members = True  # 开启成员权限意图
client_id = '你的客户端ID'
client_secret = '你的客户端密钥'
redirect_uri = 'https://example.com/xyz'  # 你的回调地址
# 用WebApplicationClient实现授权码流（用户授权）
oauth_client = WebApplicationClient(client_id)
# 生成用户授权链接（需包含guilds.join权限）
authorization_url, state = oauth_client.generate_authorize_url(
    'https://discord.com/api/oauth2/authorize',
    scope=['guilds.join'],
    redirect_uri=redirect_uri
)
# 存储已授权用户的令牌（实际建议用数据库存储，这里用字典示例）
authorized_users = {}

bot = commands.Bot(command_prefix='!', intents=intents)

async def authorize_user(user):
    if user.id not in authorized_users:
        embed = discord.Embed(
            title="授权请求",
            description=f"点击链接授权我代你加入服务器：\n{authorization_url}"
        )
        await user.send(embed=embed)
    else:
        await user.send("你已经完成授权了！")

@bot.event
async def on_ready():
    print(f'Bot已登录：{bot.user}')

@bot.command()
async def authorize(ctx):
    await authorize_user(ctx.author)

@bot.command()
@commands.has_permissions(administrator=True)
async def invite(ctx, invite_url):
    # 解析邀请链接获取服务器ID
    parsed_url = urlparse(invite_url)
    query_params = parse_qs(parsed_url.query)
    guild_id = query_params.get('guild_id', [None])[0]
    if not guild_id:
        await ctx.send("无效的邀请链接，无法获取服务器ID")
        return
    
    guild = bot.get_guild(int(guild_id))
    if not guild:
        await ctx.send("无法找到该服务器，请确保Bot已经加入该服务器")
        return
    
    # 异步批量处理用户加入请求
    async with aiohttp.ClientSession() as session:
        tasks = []
        for user_id, token in authorized_users.items():
            url = f'https://discord.com/api/v9/guilds/{guild_id}/members/{user_id}'
            headers = {'Authorization': f'Bot {bot.http.token}'}
            data = {'access_token': token['access_token']}
            tasks.append(session.put(url, headers=headers, json=data))
        
        # 批量执行请求
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        # 处理结果
        error_count = 0
        for idx, response in enumerate(responses):
            user_id = list(authorized_users.keys())[idx]
            if isinstance(response, Exception):
                error_count +=1
                await ctx.send(f"用户ID {user_id} 加入失败：{str(response)}")
            elif response.status_code != 201:
                error_count +=1
                await ctx.send(f"用户ID {user_id} 加入失败：状态码 {response.status_code}")
        
        await ctx.send(f"处理完成！成功加入{len(authorized_users)-error_count}人，失败{error_count}人")

# 注意：需自行实现回调接口处理授权码（示例用Flask）
# from flask import Flask, request
# app = Flask(__name__)
# @app.route('/xyz')
# def callback():
#     code = request.args.get('code')
#     token = oauth_client.fetch_token(
#         'https://discord.com/api/oauth2/token',
#         client_secret=client_secret,
#         code=code,
#         redirect_uri=redirect_uri
#     )
#     # 获取用户ID并存入authorized_users
#     import requests
#     user_info = requests.get('https://discord.com/api/v9/users/@me', headers={'Authorization': f'Bearer {token["access_token"]}'}).json()
#     authorized_users[user_info['id']] = token
#     return "授权成功！"

if __name__ == '__main__':
    bot.run('你的Bot令牌')

提示：用户授权的回调逻辑需要用Web框架（如Flask/FastAPI）实现，因为Bot是长运行进程，无法直接处理HTTP回调请求。

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