如何通过Discord API监听新消息(不依赖现有库)
1. 在Discord.py中通过on_message创建并发送响应
先修正你代码里的小问题:@client.event不需要加括号。另外要注意跳过机器人自身的消息,避免触发无限循环。以下是完整的响应示例:
import discord from discord.ext import commands # 必须开启message_content意图才能读取消息内容 intents = discord.Intents.default() intents.message_content = True client = commands.Bot(command_prefix='!', intents=intents) @client.event async def on_message(message): # 跳过机器人自己的消息,防止循环回复 if message.author == client.user: return # 打印收到的消息内容 print(message.content) # 示例1:直接回复消息发送者 await message.reply(f"我收到你的消息了:{message.content}") # 示例2:发送消息到当前频道 await message.channel.send(f"频道新增消息:{message.content}") # 示例3:@提及消息发送者 await message.channel.send(f"{message.author.mention} 你好!") # 替换为你的机器人TOKEN client.run('YOUR_BOT_TOKEN')
⚠️ 注意:要在Discord开发者平台的机器人设置中开启message_content权限,否则无法读取消息内容。
2. Discord.py高效获取新消息(规避速率限制)
Discord.py的on_message本身是事件驱动的——Discord服务器会主动把新消息推送给机器人,不需要你主动拉取全频道历史,完全不会触发速率限制,这是最优的实时监听方案。
如果机器人离线后需要补取期间的消息,可以通过以下方式高效拉取:
- 使用
channel.history()方法,指定after参数(上次记录的最后一条消息ID),只拉取该ID之后的新消息:
async def fetch_recent_messages(channel, last_msg_id): # 仅拉取last_msg_id之后的100条最新消息 messages = await channel.history(after=last_msg_id, limit=100).flatten() for msg in messages: print(f"补取消息:{msg.content}")
- 单次监听特定消息可以用
client.wait_for('message'):
# 等待来自指定用户的消息 target_msg = await client.wait_for('message', check=lambda m: m.author.id == 123456789) print(f"收到目标用户消息:{target_msg.content}")
3. 用Python的requests模块实现Discord消息监听
Discord的实时消息推送依赖WebSocket网关,而非普通HTTP请求。requests是HTTP库,无法直接处理WebSocket连接,但有两种替代方案:
方案1:HTTP轮询(不推荐)
定期调用GET /channels/{channel_id}/messages接口,只拉取最新消息,需控制轮询间隔避免触发速率限制:
import requests import time DISCORD_TOKEN = "YOUR_BOT_TOKEN" CHANNEL_ID = "TARGET_CHANNEL_ID" last_msg_id = None # 记录上次获取的最后一条消息ID headers = {"Authorization": f"Bot {DISCORD_TOKEN}"} while True: params = {"limit": 1} if last_msg_id: params["after"] = last_msg_id resp = requests.get( f"https://discord.com/api/v10/channels/{CHANNEL_ID}/messages", headers=headers, params=params ) resp.raise_for_status() messages = resp.json() if messages: new_msg = messages[0] if new_msg["id"] != last_msg_id: print(f"新消息:{new_msg['content']}") last_msg_id = new_msg["id"] time.sleep(5) # 5秒轮询一次,符合Discord的速率限制
⚠️ 注意:GET /channels/{channel_id}/messages的速率限制是10次/5秒,轮询间隔不能过短。
方案2:WebSocket网关(推荐,需websocket库)
先用requests获取网关地址,再用websocket库连接实现实时监听:
import requests import websocket import json DISCORD_TOKEN = "YOUR_BOT_TOKEN" # 第一步:获取WebSocket网关地址 resp = requests.get( "https://discord.com/api/v10/gateway/bot", headers={"Authorization": f"Bot {DISCORD_TOKEN}"} ) resp.raise_for_status() gateway_url = resp.json()["url"] + "?v=10&encoding=json" # 第二步:处理WebSocket消息 def on_ws_message(ws, message): data = json.loads(message) # 处理心跳包 if data["op"] == 10: heartbeat_interval = data["d"]["heartbeat_interval"] ws.send(json.dumps({"op": 1, "d": None})) # 发送身份验证请求 elif data["op"] == 1: ws.send(json.dumps({ "op": 2, "d": { "token": DISCORD_TOKEN, "intents": 513, # 开启guilds和message_content意图 "properties": { "$os": "linux", "$browser": "custom_bot", "$device": "custom_bot" } } })) # 监听新消息事件 elif data["op"] == 0 and data["t"] == "MESSAGE_CREATE": msg = data["d"] print(f"收到消息:{msg['content']} 来自:{msg['author']['username']}") ws = websocket.WebSocketApp(gateway_url, on_message=on_ws_message) ws.run_forever()
⚠️ 注意:需要提前安装websocket-client库(pip install websocket-client),并在Discord开发者平台开启对应意图。
内容的提问来源于stack exchange,提问作者Walker
相关产品推荐
相关产品推荐

