Écrire un bot IRC en Python : du socket brut à asyncio

J'ai écrit mon premier bot IRC en Perl en 1999. Un script de 200 lignes qui répondait à !weather en scrapant une page météo. Ça plantait toutes les six heures. Depuis, les outils ont évolué, et Python est devenu le langage de choix pour ce type de projet. Dans cet article, vous allez construire un bot IRC fonctionnel en trois étapes : d'abord avec un socket TCP brut pour comprendre le protocole, puis avec la bibliothèque irc pour gagner en fiabilité, et enfin avec asyncio pour gérer la concurrence proprement.

Pourquoi Python pour un bot IRC

Python n'est pas le langage le plus rapide. Ce n'est pas le plus concis non plus. Mais pour un bot IRC, il coche toutes les cases qui comptent : bibliothèque standard riche (sockets, SSL, asyncio), écosystème de packages énorme, et lisibilité du code qui facilite la maintenance six mois plus tard quand vous avez oublié pourquoi vous aviez mis un try/except autour de cette ligne.

Le protocole IRC est basé sur du texte. Chaque message est une ligne terminée par \r\n, avec un format prévisible défini dans la RFC 2812. Python excelle dans le traitement de texte. Les f-strings, le slicing, les regex — tout est là nativement. Pas besoin de dépendance externe pour parser un message IRC.

Un bot IRC typique fait trois choses : se connecter à un serveur, écouter les messages, et répondre selon des règles. C'est un programme réseau simple, et Python a les outils natifs pour chacune de ces étapes.

Méthode 1 : le socket brut

Commençons par la méthode la plus directe. Un socket TCP, pas de bibliothèque tierce. Vous allez comprendre exactement ce qui transite sur le fil entre votre bot et le serveur IRC.

import socket
import ssl

def connect(server, port, nick, channel, use_tls=True):
    raw_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    if use_tls:
        ctx = ssl.create_default_context()
        sock = ctx.wrap_socket(raw_sock, server_hostname=server)
    else:
        sock = raw_sock

    sock.connect((server, port))
    send(sock, f"NICK {nick}")
    send(sock, f"USER {nick} 0 * :Bot IRC Python")

    buffer = ""
    while True:
        data = sock.recv(4096).decode("utf-8", errors="replace")
        if not data:
            break
        buffer += data
        while "\r\n" in buffer:
            line, buffer = buffer.split("\r\n", 1)
            print(f"<< {line}")
            handle_line(sock, line, channel)

def send(sock, msg):
    print(f">> {msg}")
    sock.sendall(f"{msg}\r\n".encode("utf-8"))

def handle_line(sock, line, channel):
    # Répondre au PING pour rester connecté
    if line.startswith("PING"):
        send(sock, f"PONG {line.split()[1]}")
        return

    # Joindre le channel après l'enregistrement (RPL_WELCOME = 001)
    if " 001 " in line:
        send(sock, f"JOIN {channel}")
        return

    # Parser les PRIVMSG
    if "PRIVMSG" in line:
        parts = line.split(" ", 3)
        sender = parts[0].split("!")[0][1:]
        target = parts[2]
        message = parts[3][1:]
        if message.startswith("!ping"):
            reply_to = target if target.startswith("#") else sender
            send(sock, f"PRIVMSG {reply_to} :pong, {sender}")

if __name__ == "__main__":
    connect("irc.libera.chat", 6697, "monbot", "#test-bot")

Ce code fait environ 50 lignes et gère l'essentiel : connexion TLS, enregistrement, réponse au PING, join automatique, et une commande !ping. C'est volontairement minimal. Vous voyez chaque octet qui passe.

Le problème du socket brut, c'est la gestion des erreurs. Que se passe-t-il si la connexion tombe ? Si le serveur envoie un message mal formé ? Si votre bot se fait kick ? Vous devez tout gérer à la main. C'est formateur, mais pas viable en production.

Méthode 2 : la bibliothèque irc

La bibliothèque irc (anciennement python-irclib, maintenue par Jason R. Coombs) abstrait la couche réseau et fournit un système d'événements propre. Installez-la avec pip :

pip install irc

Voici le même bot, réécrit avec cette bibliothèque :

import irc.bot
import irc.strings

class MonBot(irc.bot.SingleServerIRCBot):
    def __init__(self, server, port, nick, channel):
        factory = irc.connection.Factory(wrapper=irc.connection.Factory().wrapper)
        irc.bot.SingleServerIRCBot.__init__(
            self,
            [(server, port)],
            nick,
            nick,
            connect_factory=factory
        )
        self.target_channel = channel

    def on_welcome(self, connection, event):
        connection.join(self.target_channel)
        print(f"Connecté et joint à {self.target_channel}")

    def on_pubmsg(self, connection, event):
        message = event.arguments[0]
        sender = irc.strings.lower(event.source.nick)
        if message.startswith("!ping"):
            connection.privmsg(event.target, f"pong, {sender}")
        elif message.startswith("!quit") and sender == "marco":
            self.die("Arrêt demandé par l'admin.")

    def on_disconnect(self, connection, event):
        print("Déconnecté. Reconnexion dans 60 secondes...")
        # La reconnexion est gérée automatiquement par SingleServerIRCBot

if __name__ == "__main__":
    bot = MonBot("irc.libera.chat", 6697, "monbot", "#test-bot")
    bot.start()

La différence est visible : plus de parsing manuel, plus de gestion du PING/PONG (la bibliothèque s'en charge), reconnexion automatique. Les événements IRC sont mappés sur des méthodes Python (on_welcome, on_pubmsg, on_privmsg, on_kick, etc.). Vous vous concentrez sur la logique métier.

La bibliothèque gère aussi les modes, les topics, les listes de nicks, les CTCP, et une bonne partie des numéros de réponse IRC. Pour un bot de modération ou un bot de service, c'est suffisant.

Méthode 3 : asyncio pour la concurrence

Le modèle synchrone de la bibliothèque irc a une limite : votre bot ne peut faire qu'une chose à la fois. Si vous appelez une API HTTP dans un handler, tout le bot se fige pendant l'attente de la réponse. C'est là qu'asyncio entre en jeu.

import asyncio
import ssl

class AsyncIRCBot:
    def __init__(self, server, port, nick, channel):
        self.server = server
        self.port = port
        self.nick = nick
        self.channel = channel
        self.writer = None

    async def connect(self):
        ctx = ssl.create_default_context()
        reader, self.writer = await asyncio.open_connection(
            self.server, self.port, ssl=ctx
        )
        await self.send(f"NICK {self.nick}")
        await self.send(f"USER {self.nick} 0 * :Async Bot")

        async for raw in reader:
            line = raw.decode("utf-8", errors="replace").strip()
            if line:
                await self.handle(line)

    async def send(self, msg):
        self.writer.write(f"{msg}\r\n".encode("utf-8"))
        await self.writer.drain()

    async def handle(self, line):
        print(f"<< {line}")
        if line.startswith("PING"):
            await self.send(f"PONG {line.split()[1]}")
        elif " 001 " in line:
            await self.send(f"JOIN {self.channel}")
        elif "PRIVMSG" in line:
            parts = line.split(" ", 3)
            sender = parts[0].split("!")[0][1:]
            target = parts[2]
            text = parts[3][1:]
            await self.on_message(sender, target, text)

    async def on_message(self, sender, target, text):
        if text.startswith("!weather"):
            # Appel HTTP non bloquant
            city = text.split(" ", 1)[1] if " " in text else "Paris"
            result = await self.fetch_weather(city)
            reply_to = target if target.startswith("#") else sender
            await self.send(f"PRIVMSG {reply_to} :{result}")

    async def fetch_weather(self, city):
        # Exemple avec aiohttp (non inclus pour garder le code minimal)
        # async with aiohttp.ClientSession() as session:
        #     async with session.get(f"https://api.example.com/{city}") as resp:
        #         data = await resp.json()
        #         return f"{city}: {data['temp']}°C"
        await asyncio.sleep(0.1)  # Simulation
        return f"{city}: 18°C, couvert"

async def main():
    bot = AsyncIRCBot("irc.libera.chat", 6697, "asyncbot", "#test-bot")
    await bot.connect()

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

Avec asyncio, l'appel à fetch_weather ne bloque pas la boucle d'événements. Votre bot continue de répondre aux PING et aux autres commandes pendant que la requête HTTP est en cours. Pour un bot qui interroge des API externes, c'est un gain de fiabilité considérable.

Gérer les commandes utilisateur

Un bot IRC sans système de commandes, c'est un script qui reste connecté sans rien faire. Voici un pattern que j'utilise depuis des années pour organiser les commandes de façon maintenable :

class CommandHandler:
    def __init__(self, prefix="!"):
        self.prefix = prefix
        self.commands = {}

    def command(self, name, admin_only=False):
        """Décorateur pour enregistrer une commande."""
        def decorator(func):
            self.commands[name] = {
                "func": func,
                "admin_only": admin_only
            }
            return func
        return decorator

    async def dispatch(self, bot, sender, target, text):
        if not text.startswith(self.prefix):
            return
        parts = text[len(self.prefix):].split(" ", 1)
        cmd_name = parts[0].lower()
        args = parts[1] if len(parts) > 1 else ""

        cmd = self.commands.get(cmd_name)
        if not cmd:
            return
        if cmd["admin_only"] and sender not in ADMINS:
            return
        await cmd["func"](bot, sender, target, args)

# Utilisation
handler = CommandHandler(prefix="!")
ADMINS = ["marco", "admin"]

@handler.command("ping")
async def cmd_ping(bot, sender, target, args):
    await bot.send(f"PRIVMSG {target} :pong, {sender}")

@handler.command("quit", admin_only=True)
async def cmd_quit(bot, sender, target, args):
    await bot.send(f"QUIT :Arrêt par {sender}")

Ce pattern par décorateur permet d'ajouter des commandes sans toucher au code de dispatch. Chaque commande est une fonction isolée, testable indépendamment. Vous pouvez ajouter des contrôles de rate-limiting, de permissions par channel, ou de journalisation sans modifier les commandes existantes.

Pour le rate-limiting — et vous en aurez besoin tôt ou tard — un dictionnaire avec des timestamps suffit :

import time

rate_limits = {}  # {nick: [timestamp, ...]}

def check_rate(nick, max_per_minute=5):
    now = time.time()
    if nick not in rate_limits:
        rate_limits[nick] = []
    # Nettoyer les entrées de plus d'une minute
    rate_limits[nick] = [t for t in rate_limits[nick] if now - t < 60]
    if len(rate_limits[nick]) >= max_per_minute:
        return False
    rate_limits[nick].append(now)
    return True

Déploiement et supervision

Un bot IRC doit tourner en permanence. Sur un serveur Linux, la méthode la plus fiable est un service systemd :

# /etc/systemd/system/ircbot.service
[Unit]
Description=Bot IRC Python
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=ircbot
WorkingDirectory=/opt/ircbot
ExecStart=/opt/ircbot/venv/bin/python bot.py
Restart=always
RestartSec=30
Environment=PYTHONUNBUFFERED=1

[Install]
WantedBy=multi-user.target

# Activation
sudo systemctl daemon-reload
sudo systemctl enable ircbot
sudo systemctl start ircbot

# Vérification
sudo systemctl status ircbot
journalctl -u ircbot -f

Le Restart=always avec RestartSec=30 assure que votre bot revient en ligne après un crash ou une déconnexion non gérée. C'est plus fiable qu'un screen ou un tmux oublié, et les logs sont centralisés dans journald.

Pour la configuration, évitez de hardcoder les paramètres dans le script. Un fichier .env ou un config.toml fait l'affaire :

# config.toml
[irc]
server = "irc.libera.chat"
port = 6697
nick = "monbot"
channel = "#moncanal"
tls = true

[admin]
nicks = ["marco", "admin"]