from cordslite.core import *
from fastcore.utils import *
import asyncio,httpx,json,os,random
import websockets, websockets.asyncio.clientGateway API
Real-time Discord events over the WebSocket gateway
dc = DiscordClient()
gid = '1493461895615873044'
ch = await dc.channel('1493461896139903028')The Gateway is Discord’s WebSocket connection for real-time events. Unlike the REST API (request/response), the Gateway pushes events to you as they happen—new messages, reactions, user joins, etc.
The connection lifecycle is: 1. Fetch WSS URL from /gateway/bot 2. Connect to WebSocket 3. Receive Hello event with heartbeat interval 4. Start heartbeat loop to keep connection alive 5. Send Identify with your token and intents 6. Receive Ready event—you’re connected!
Intents tell Discord which events you want. They’re bitwise flags you OR together: - 1 << 0 = GUILDS (guild/channel events) - 1 << 9 = GUILD_MESSAGES (message events) - 1 << 15 = MESSAGE_CONTENT (privileged—see message content)
class GatewayClient:
def __init__(self, intents, client, token=None):
self.intents,self.dc = intents,client
self.token = token or os.environ['DISCORD_BOT_TOKEN']
self.ws = self.hb_int = self.session_id = self.seq = None
self.running = False
self._tries = 0
gw_info = httpx.get(f'{client.base_url}/gateway/bot', headers={'Authorization': f'Bot {self.token}'}).json()
if 'url' not in gw_info: raise ConnectionError(f"Gateway auth failed: {gw_info.get('message', gw_info)}")
self.url = f"{gw_info['url']}?v=10&encoding=json"
self.handlers = {'READY': self.on_rdy}
async def on_rdy(self, evt):
self.session_id, self.user_id, = evt['session_id'], evt['user']['id']
self.resume_url = evt['resume_gateway_url'] + '?v=10&encoding=json' if 'resume_gateway_url' in evt else self.url
def __repr__(self): return f"GatewayClient({self.intents=}, {self.url=})"# For now, let's use basic intents for guilds and messages
intents = (1 << 0) | (1 << 7) | (1 << 9) | (1 << 15) # GUILDS | VOICE_EVENTS | GUILD_MESSAGES | MESSAGE_CONTENT
gc = GatewayClient(intents, dc); gcGatewayClient(self.intents=33409, self.url='wss://gateway.discord.gg?v=10&encoding=json')Op is a thin helper for constructing Gateway opcodes. Discord’s Gateway expects JSON payloads with an op field (operation code) and d field (payload data). Rather than hand-building dicts each time, Op.identify() and Op.heartbeat() return properly structured payloads ready to send over the WebSocket.
class Op(AttrDict):
def __repr__(self): return f"Op(op={self.op}, d={self.d})"
@classmethod
def identify(cls, token, intents):
d = AttrDict(token=token, intents=intents, properties=dict(os='linux',browser='discord_wrapper', device='discord_wrapper'))
return cls(op=2, d=d)
@classmethod
def heartbeat(cls, seq): return cls(op=1, d=seq)
@classmethod
def resume(cls, token, session_id, seq): return cls(op=6, d=AttrDict(token=token, session_id=session_id, seq=seq))The Gateway sends events as JSON with four fields: - op — operation code: 0 = dispatch (real events), 1 = heartbeat request, 10 = hello, 11 = heartbeat ACK - t — event type name (only for dispatches): MESSAGE_CREATE, GUILD_CREATE, etc. - s — sequence number (only for dispatches): increments per event, needed for heartbeats and resuming - d — the payload data
We track the latest s value so heartbeats can tell Discord “I’ve received everything up to here.” The Event class wraps raw JSON and auto-converts known dispatch types (like MESSAGE_CREATE) into their corresponding wrapper classes (Message, Channel, etc.) via evt_typs.
evt_typs = {'MESSAGE_CREATE': Message,
'MESSAGE_UPDATE': Message,
'MESSAGE_DELETE': Message,
'GUILD_CREATE': Guild,
'CHANNEL_CREATE': Channel}
class Event(DiscordObject):
def __init__(self, data, client):
super().__init__(data, client)
typ = evt_typs.get(self.t)
self.d = typ(self.d, client) if typ else self.d
@property
def type(self): return self.t
@property
def seq(self): return self.s
def __repr__(self): return f"Event({self.op=}, {self.type=}, {self.seq=})"
@patch
async def send(self:websockets.asyncio.client.ClientConnection, msg, **kw):
if isinstance(msg, dict): msg = json.dumps(msg)
return await self._orig_send(msg, **kw)recv_evt is the low-level building block for receiving events. It reads one raw WebSocket message, wraps it as an Event (auto-converting known dispatch types), and updates the sequence counter. You can use it directly to pull events one at a time — useful for debugging or understanding what Discord is sending.
@patch
async def recv_evt(self:GatewayClient):
evt = Event(json.loads(await self.ws.recv()), self.dc)
if evt.op == 0 and evt.seq: self.seq,self._tries = evt.seq,0
return evtRather than special-casing the READY event inside _listen, we register it as a regular dispatch handler via gc.on(). When Discord confirms a successful Identify, on_rdy caches session_id, user_id, and resume_gateway_url — the three values needed to Resume after a disconnect. Note the query params appended to resume_gateway_url — Discord returns it bare, but the docs require the same version and encoding as the initial connection.
@patch
def on(self:GatewayClient, event_type, handler): self.handlers[event_type] = handlerThe gateway runs on two cooperating loops. _hb sends heartbeats at the interval Discord specifies and watches for ACK responses — if one is missed, the connection is zombied and gets closed with code 4000 (which preserves the session for Resuming). _listen reads events and dispatches them by opcode. On disconnect (exception from recv_evt), it calls _reconnect to swap in a fresh WebSocket. The new connection triggers Hello (op 10), which restarts the heartbeat and sends either Identify (first connect) or Resume (reconnect) depending on whether a session already exists. Op 7 (Reconnect) follows the same path — close and reopen with a resume.
All reconnection funnels through _reconnect. A resume closes with code 4000 (keeping the session valid) and opens a new socket to resume_gateway_url; a re-identify deliberately closes with 1000 (invalidating the session) and goes back to the original gateway URL. The listener’s op 10 handler then takes over automatically.
@patch
async def _hb(self:GatewayClient):
await asyncio.sleep(self.hb_int / 1_000 * random.random()) # jitter
while self.running:
self._got_ack = False
try: await self.ws.send(Op.heartbeat(self.seq))
except Exception: return
await asyncio.sleep(self.hb_int / 1_000)
if not self._got_ack: return await self.ws.close(code=4000)
@patch
async def _reconnect(self:GatewayClient, resume=False):
resume = resume and bool(self.session_id)
code, url = (4000, self.resume_url) if resume else (1000, self.url)
if not resume: self.session_id = None
try: await self.ws.close(code=code)
except Exception: pass
while self.running:
await asyncio.sleep(min(2 ** self._tries, 60))
self._tries += 1
try:
self.ws = await websockets.connect(url, max_size=None)
return
except OSError as e: print('gateway reconnect failed:', repr(e))
@patch
async def _listen(self:GatewayClient):
while self.running:
try:
evt = await self.recv_evt()
if self.debug: print('DEBUG: Received Opcode:', evt.op)
if evt.op == 10:
self.hb_int = evt.d['heartbeat_interval']
if hasattr(self, '_hb_task') and self._hb_task: self._hb_task.cancel()
self._hb_task = asyncio.create_task(self._hb())
if self.session_id: await self.ws.send(Op.resume(self.token, self.session_id, self.seq))
else: await self.ws.send(Op.identify(self.token, self.intents))
elif evt.op == 0 and evt.type in self.handlers: asyncio.create_task(self.handlers[evt.type](evt.d))
elif evt.op == 11: self._got_ack = True
elif evt.op == 1: await self.ws.send(Op.heartbeat(self.seq))
elif evt.op == 7: await self._reconnect(resume=True)
elif evt.op == 9: await self._reconnect(resume=evt.d)
except websockets.exceptions.ConnectionClosed as e:
if not self.running: break
print(f'gateway closed: {e.code} {e.reason}')
if e.code in (4004, 4010, 4011, 4012, 4013, 4014): # fatal: bad token/shard/intents
self.running = False
break
await self._reconnect(resume=e.code not in (4007, 4009))
continue
except Exception as e:
print('gateway listen error:', repr(e))
await self._reconnect(resume=True)
continuestart opens the initial WebSocket and spawns the listener — nothing else. The listener handles Hello, kicks off the heartbeat, and sends Identify, so the full connection lifecycle is driven by events rather than imperative setup.
@patch
async def start(self:GatewayClient, debug=False):
self.debug = debug
self.running = True
self.ws = await websockets.connect(self.url, max_size=None)
self._listen_task = asyncio.create_task(self._listen())await gc.start(debug=True)DEBUG: Received Opcode: 10
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DBuddy: Test our event listener! Otters are awesome 🦦
gateway closed: 4000
DEBUG: Received Opcode: 10
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 10
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DBuddy: Houston, do we have a problem?
gateway closed: 1000
DEBUG: Received Opcode: 10
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 10
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DBuddy: Did we re-identify?
gateway closed: 4000
DEBUG: Received Opcode: 10
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 0
DBuddy: Back from the dead? (listener auto-resume)
DEBUG: Received Opcode: 0
DEBUG: Received Opcode: 11Let’s now test that our gateway is getting events. Here is a simple handlers for new messages.
async def on_msg(msg):
if msg.get('guild_id') != gid: return
if not msg.author.get('bot'): return
print(f"{msg.author['username']}: {msg.content}")
gc.on('MESSAGE_CREATE', on_msg)Now, we can send a message and it should be printed out.
await ch.send('Test our event listener! Otters are awesome 🦦')Message(id=1524528166285082704, author='DBuddy', content='Test our event listener! Otter…')Let’s trigger and reconnection that resumes and one that doesn’t and see if everything works correctly and we still end up with our printed message.
await gc._reconnect(resume=True)
await asyncio.sleep(2)
await ch.send('Houston, do we have a problem?')Message(id=1524528193283948816, author='DBuddy', content='Houston, do we have a problem?')await gc._reconnect(resume=False)
await asyncio.sleep(2)
await ch.send('Did we re-identify?')Message(id=1524528240155430924, author='DBuddy', content='Did we re-identify?')Those were polite reconnects we initiated. Real disconnects arrive as a ConnectionClosed in the listener, which classifies the close code per the docs: transient codes resume, 4007/4009 re-identify, and 4004/4010–4014 (bad token, bad shard, bad intents) stop rather than loop forever. We can trigger the transient path for real by killing our own live socket with a non-1000 code — exactly what a zombied connection’s heartbeat watchdog does — and confirm the listener resumes on its own.
await gc.ws.close(code=4000) # simulate a dead connection; the listener should resume unaided
await asyncio.sleep(3)
await ch.send('Back from the dead? (listener auto-resume)')Message(id=1524528333700862142, author='DBuddy', content='Back from the dead? (listener …')And the fatal path for real: identify with a corrupted token and Discord closes us with 4004 — the client must give up instead of hammering the gateway.
gc_bad = GatewayClient(intents, dc)
gc_bad.token = gc_bad.token[:-4] + 'zzzz'
await gc_bad.start()
for _ in range(30):
if not gc_bad.running: break
await asyncio.sleep(0.5)
assert not gc_bad.running # 4004 authentication failed: no reconnect loopgateway closed: 4004 Authentication failed.stop is a full shutdown — cancels both background tasks and closes the WebSocket with the default code (1000), which tells Discord to invalidate the session. After this, a fresh start() would need to Identify from scratch.
@patch
async def stop(self:GatewayClient):
self.running = False
for t in (self._hb_task, self._listen_task):
if t: t.cancel()
if self.ws: await self.ws.close()await gc.stop()