Skip to main content
Cogs are a way to organize your bot’s commands, listeners, and functionality into modular, reusable classes. Think of them as plugins that can be loaded, unloaded, and reloaded without restarting your bot.

What are Cogs?

A Cog is a Python class that groups related commands and event listeners together. They help you:
  • Organize code into logical modules
  • Load and unload features dynamically
  • Share code between multiple bots
  • Maintain clean, manageable codebases
The Cog class is implemented in discord/cog.py:319 and uses metaclasses to automatically register commands and listeners.

Creating a Cog

Create a simple cog by inheriting from commands.Cog:
import discord
from discord.ext import commands

class Greetings(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
    
    @commands.slash_command()
    async def hello(self, ctx):
        await ctx.respond(f"Hello {ctx.author.mention}!")
    
    @commands.Cog.listener()
    async def on_member_join(self, member):
        channel = member.guild.system_channel
        if channel:
            await channel.send(f"Welcome {member.mention}!")

Loading Cogs

Load a cog defined in the same file:
import discord
from discord.ext import commands

bot = commands.Bot(command_prefix="!")

class MyCog(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
    
    @commands.slash_command()
    async def test(self, ctx):
        await ctx.respond("Test command!")

# Add the cog
bot.add_cog(MyCog(bot))

bot.run("TOKEN")

Cog Listeners

Add event listeners to cogs using the @commands.Cog.listener() decorator:
class Events(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
    
    @commands.Cog.listener()
    async def on_message(self, message):
        print(f"Message from {message.author}: {message.content}")
    
    @commands.Cog.listener()
    async def on_member_join(self, member):
        print(f"{member.name} joined {member.guild.name}")
    
    # Listen to the same event multiple times
    @commands.Cog.listener("on_message")
    async def check_spam(self, message):
        if message.content.isupper():
            await message.delete()

Cog Special Methods

Cogs have special methods for lifecycle management:
Called when the cog is loaded. Use this for async initialization:
class Database(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.db = None
    
    async def cog_load(self):
        # Async initialization
        self.db = await setup_database()
Called when the cog is unloaded. Clean up resources here:
class TaskManager(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.task = self.bot.loop.create_task(self.background_task())
    
    def cog_unload(self):
        # Cancel background tasks
        self.task.cancel()
    
    async def background_task(self):
        while True:
            # Do something
            await asyncio.sleep(60)
Add a check that applies to all commands in the cog:
class Admin(commands.Cog):
    async def cog_check(self, ctx):
        # Only allow admins
        return ctx.author.guild_permissions.administrator
    
    @commands.slash_command()
    async def admin_command(self, ctx):
        await ctx.respond("Admin only!")
Handle errors for all commands in the cog:
class MyCommands(commands.Cog):
    async def cog_command_error(self, ctx, error):
        if isinstance(error, commands.MissingPermissions):
            await ctx.respond("You don't have permission!", ephemeral=True)
        else:
            await ctx.respond(f"Error: {error}", ephemeral=True)
Run code before/after every command in the cog:
class Logging(commands.Cog):
    async def cog_before_invoke(self, ctx):
        print(f"Command {ctx.command.name} invoked by {ctx.author}")
    
    async def cog_after_invoke(self, ctx):
        print(f"Command {ctx.command.name} completed")

Managing Cogs at Runtime

@bot.command()
@commands.is_owner()
async def load(ctx, extension):
    try:
        bot.load_extension(f"cogs.{extension}")
        await ctx.send(f"Loaded {extension}")
    except Exception as e:
        await ctx.send(f"Error loading {extension}: {e}")

Cog Organization Patterns

By Feature

Organize cogs by functionality:
bot/
├── main.py
└── cogs/
    ├── moderation.py    # Kick, ban, mute
    ├── music.py         # Play, pause, skip
    ├── fun.py          # Memes, games
    └── utility.py      # Info, help

By Category

Organize by Discord categories:
bot/
├── main.py
└── cogs/
    ├── admin/
    │   ├── moderation.py
    │   └── server_management.py
    ├── members/
    │   ├── levels.py
    │   └── profiles.py
    └── general/
        ├── info.py
        └── help.py

Shared State

Share data between cogs:
# main.py
class Bot(commands.Bot):
    def __init__(self):
        super().__init__(command_prefix="!")
        self.db = None
    
    async def setup_hook(self):
        self.db = await setup_database()

# cogs/users.py
class Users(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
    
    @commands.slash_command()
    async def profile(self, ctx):
        # Access shared database
        user_data = await self.bot.db.get_user(ctx.author.id)
        await ctx.respond(f"Profile: {user_data}")

Complete Example

Here’s a complete bot with multiple cogs: main.py:
import discord
from discord.ext import commands
import os

intents = discord.Intents.default()
intents.message_content = True

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

@bot.event
async def on_ready():
    print(f"Logged in as {bot.user}")
    print(f"Loaded cogs: {', '.join(bot.cogs.keys())}")

# Load all cogs
for filename in os.listdir("./cogs"):
    if filename.endswith(".py") and not filename.startswith("_"):
        bot.load_extension(f"cogs.{filename[:-3]}")
        print(f"Loaded {filename}")

bot.run("TOKEN")
cogs/greetings.py:
import discord
from discord.ext import commands

class Greetings(commands.Cog):
    """Greeting commands and welcome messages"""
    
    def __init__(self, bot):
        self.bot = bot
        self._last_member = None
    
    @commands.Cog.listener()
    async def on_member_join(self, member):
        channel = member.guild.system_channel
        if channel is not None:
            await channel.send(f"Welcome {member.mention}!")
    
    @commands.slash_command()
    async def hello(self, ctx, member: discord.Member = None):
        """Say hello to someone"""
        member = member or ctx.author
        
        if self._last_member is None or self._last_member.id != member.id:
            await ctx.respond(f"Hello {member.mention}!")
        else:
            await ctx.respond(f"Hello again {member.mention}!")
        
        self._last_member = member

def setup(bot):
    bot.add_cog(Greetings(bot))
cogs/moderation.py:
import discord
from discord.ext import commands

class Moderation(commands.Cog, name="Mod Tools"):
    """Server moderation commands"""
    
    def __init__(self, bot):
        self.bot = bot
    
    async def cog_check(self, ctx):
        # Only allow members with manage_messages permission
        return ctx.author.guild_permissions.manage_messages
    
    @commands.slash_command()
    async def kick(self, ctx, member: discord.Member, reason: str = None):
        """Kick a member from the server"""
        await member.kick(reason=reason)
        await ctx.respond(f"Kicked {member.mention}")
    
    @commands.slash_command()
    async def ban(self, ctx, member: discord.Member, reason: str = None):
        """Ban a member from the server"""
        await member.ban(reason=reason)
        await ctx.respond(f"Banned {member.mention}")
    
    async def cog_command_error(self, ctx, error):
        if isinstance(error, commands.MissingPermissions):
            await ctx.respond(
                "You don't have permission to use moderation commands!",
                ephemeral=True
            )

def setup(bot):
    bot.add_cog(Moderation(bot))

Best Practices

  1. One cog per file - Keep cogs in separate files for better organization
  2. Use descriptive names - Name cogs after their functionality
  3. Group related commands - Put similar commands in the same cog
  4. Clean up resources - Use cog_unload to cancel tasks and close connections
  5. Use cog checks - Apply permission checks to entire cogs when appropriate
  6. Document your cogs - Add docstrings to cogs and commands
  7. Handle errors - Use cog_command_error for cog-specific error handling
  8. Avoid circular imports - Don’t import cogs from each other

Common Patterns

Configuration Cog

class Config(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.config = {}
    
    @commands.slash_command()
    async def set_prefix(self, ctx, prefix: str):
        self.config[ctx.guild.id] = {"prefix": prefix}
        await ctx.respond(f"Prefix set to {prefix}")

Background Task Cog

from discord.ext import tasks

class StatusUpdate(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.update_status.start()
    
    def cog_unload(self):
        self.update_status.cancel()
    
    @tasks.loop(minutes=5)
    async def update_status(self):
        await self.bot.change_presence(
            activity=discord.Game(f"with {len(self.bot.guilds)} servers")
        )
    
    @update_status.before_loop
    async def before_update_status(self):
        await self.bot.wait_until_ready()

Database Cog

class Database(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.pool = None
    
    async def cog_load(self):
        self.pool = await asyncpg.create_pool(database="mydb")
    
    def cog_unload(self):
        if self.pool:
            self.bot.loop.create_task(self.pool.close())
    
    async def get_user_data(self, user_id):
        async with self.pool.acquire() as conn:
            return await conn.fetchrow(
                "SELECT * FROM users WHERE id = $1",
                user_id
            )

Build docs developers (and LLMs) love