Bot Discord + Claude API : architecture complète et production

claude-discord-bot/
├── bot/
│   ├── __init__.py
│   ├── main.py          # Point d'entrée
│   ├── claude_client.py # Wrapper API Claude
│   ├── commands.py      # Commandes slash
│   └── config.py        # Configuration centralisée
├── .env
├── .env.example
├── .gitignore
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── README.md

import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()


@dataclass(frozen=True)
class Config:
    discord_token: str
    anthropic_api_key: str
    claude_model: str = "claude-sonnet-4-20250514"
    max_tokens: int = 2048
    max_history: int = 30
    rate_limit_per_user: int = 10  # requêtes par minute
    system_prompt: str = (
        "Tu es un assistant technique dans un serveur Discord francophone. "
        "Réponds de manière précise et structurée. "
        "Utilise le markdown Discord : **gras**, `code inline`, "
        "```blocs de code``` avec le langage spécifié. "
        "Si tu ne sais pas, dis-le."
    )

    @classmethod
    def from_env(cls) -> "Config":
        token = os.getenv("DISCORD_TOKEN")
        api_key = os.getenv("ANTHROPIC_API_KEY")
        if not token:
            raise EnvironmentError("DISCORD_TOKEN manquant dans .env")
        if not api_key:
            raise EnvironmentError("ANTHROPIC_API_KEY manquant dans .env")
        return cls(
            discord_token=token,
            anthropic_api_key=api_key,
            claude_model=os.getenv("CLAUDE_MODEL", cls.claude_model),
            max_tokens=int(os.getenv("MAX_TOKENS", cls.max_tokens)),
            max_history=int(os.getenv("MAX_HISTORY", cls.max_history)),
            rate_limit_per_user=int(os.getenv("RATE_LIMIT", cls.rate_limit_per_user)),
        )

import time
import logging
from collections import defaultdict
from anthropic import Anthropic, APIError, RateLimitError

logger = logging.getLogger(__name__)


class RateLimiter:
    """Rate limiter par utilisateur avec fenêtre glissante."""

    def __init__(self, max_requests: int, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests: dict[str, list[float]] = defaultdict(list)

    def check(self, user_id: str) -> bool:
        now = time.time()
        # Nettoyer les requêtes expirées
        self.requests[user_id] = [
            t for t in self.requests[user_id] if now - t < self.window
        ]
        if len(self.requests[user_id]) >= self.max_requests:
            return False
        self.requests[user_id].append(now)
        return True

    def remaining(self, user_id: str) -> int:
        now = time.time()
        active = [t for t in self.requests.get(user_id, []) if now - t < self.window]
        return max(0, self.max_requests - len(active))


class ClaudeClient:
    """Wrapper autour de l'API Anthropic avec retry exponentiel."""

    def __init__(self, api_key: str, model: str, max_tokens: int, system_prompt: str):
        self.client = Anthropic(api_key=api_key)
        self.model = model
        self.max_tokens = max_tokens
        self.system_prompt = system_prompt

    def ask(self, messages: list[dict], retries: int = 3) -> str:
        for attempt in range(retries):
            try:
                response = self.client.messages.create(
                    model=self.model,
                    max_tokens=self.max_tokens,
                    system=self.system_prompt,
                    messages=messages,
                )
                return response.content[0].text

            except RateLimitError:
                wait = 2 ** attempt * 5
                logger.warning(f"Rate limit Anthropic, retry dans {wait}s")
                time.sleep(wait)

            except APIError as e:
                logger.error(f"Erreur API Anthropic : {e}")
                if attempt == retries - 1:
                    raise
                time.sleep(2 ** attempt)

        raise RuntimeError("Échec après plusieurs tentatives")

import discord
from discord import app_commands


def setup_commands(tree: app_commands.CommandTree, conversations: dict):
    """Enregistrer les commandes slash du bot."""

    @tree.command(name="ask", description="Poser une question à Claude")
    @app_commands.describe(question="Votre question pour Claude")
    async def ask(interaction: discord.Interaction, question: str):
        # Defer car la réponse peut prendre quelques secondes
        await interaction.response.defer(thinking=True)

        salon_id = str(interaction.channel_id)
        if salon_id not in conversations:
            conversations[salon_id] = []

        conversations[salon_id].append({"role": "user", "content": question})
        conversations[salon_id] = conversations[salon_id][-20:]

        try:
            from bot.main import claude_client
            reponse = claude_client.ask(conversations[salon_id])
            conversations[salon_id].append({"role": "assistant", "content": reponse})

            if len(reponse) <= 2000:
                await interaction.followup.send(reponse)
            else:
                morceaux = [reponse[i:i+2000] for i in range(0, len(reponse), 2000)]
                await interaction.followup.send(morceaux[0])
                for m in morceaux[1:]:
                    await interaction.channel.send(m)

        except Exception as e:
            await interaction.followup.send(f"Erreur : {e}")

    @tree.command(name="clear", description="Effacer l'historique de conversation")
    async def clear(interaction: discord.Interaction):
        salon_id = str(interaction.channel_id)
        conversations.pop(salon_id, None)
        await interaction.response.send_message(
            "Historique effacé pour ce salon.", ephemeral=True
        )

    @tree.command(name="model", description="Voir ou changer le modèle Claude utilisé")
    @app_commands.describe(nom="Nom du modèle (laisser vide pour voir le modèle actuel)")
    async def model_cmd(interaction: discord.Interaction, nom: str = None):
        from bot.main import claude_client
        if nom:
            claude_client.model = nom
            await interaction.response.send_message(
                f"Modèle changé : `{nom}`", ephemeral=True
            )
        else:
            await interaction.response.send_message(
                f"Modèle actuel : `{claude_client.model}`", ephemeral=True
            )

import logging
import discord
from discord import app_commands

from bot.config import Config
from bot.claude_client import ClaudeClient, RateLimiter
from bot.commands import setup_commands

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

config = Config.from_env()

claude_client = ClaudeClient(
    api_key=config.anthropic_api_key,
    model=config.claude_model,
    max_tokens=config.max_tokens,
    system_prompt=config.system_prompt,
)

rate_limiter = RateLimiter(max_requests=config.rate_limit_per_user)

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


class ClaudeBot(discord.Client):
    def __init__(self):
        super().__init__(intents=intents)
        self.tree = app_commands.CommandTree(self)
        self.conversations: dict[str, list[dict]] = {}

    async def setup_hook(self):
        setup_commands(self.tree, self.conversations)
        await self.tree.sync()
        logger.info("Commandes slash synchronisées")


bot = ClaudeBot()


@bot.event
async def on_ready():
    logger.info(f"Connecté en tant que {bot.user} (ID: {bot.user.id})")
    await bot.change_presence(
        activity=discord.Activity(
            type=discord.ActivityType.listening,
            name="vos questions | /ask"
        )
    )


@bot.event
async def on_message(message: discord.Message):
    if message.author.bot:
        return

    if bot.user not in message.mentions:
        return

    user_id = str(message.author.id)

    if not rate_limiter.check(user_id):
        remaining = rate_limiter.remaining(user_id)
        await message.reply(
            f"Doucement ! Tu as atteint la limite de {config.rate_limit_per_user} "
            f"messages par minute. Restant : {remaining}"
        )
        return

    texte = message.content.replace(f"<@{bot.user.id}>", "").strip()
    if not texte:
        return

    salon_id = str(message.channel.id)
    if salon_id not in bot.conversations:
        bot.conversations[salon_id] = []

    bot.conversations[salon_id].append({
        "role": "user",
        "content": f"{message.author.display_name}: {texte}",
    })
    bot.conversations[salon_id] = bot.conversations[salon_id][-config.max_history:]

    async with message.channel.typing():
        try:
            reponse = claude_client.ask(bot.conversations[salon_id])
            bot.conversations[salon_id].append(
                {"role": "assistant", "content": reponse}
            )

            if len(reponse) <= 2000:
                await message.reply(reponse)
            else:
                morceaux = [
                    reponse[i:i + 2000]
                    for i in range(0, len(reponse), 2000)
                ]
                for morceau in morceaux:
                    await message.channel.send(morceau)

        except Exception as e:
            logger.exception("Erreur lors de la réponse")
            await message.reply(f"Erreur interne : `{type(e).__name__}`")


def run():
    bot.run(config.discord_token, log_handler=None)


if __name__ == "__main__":
    run()

discord.py>=2.3.0
anthropic>=0.39.0
python-dotenv>=1.0.0

DISCORD_TOKEN=
ANTHROPIC_API_KEY=
CLAUDE_MODEL=claude-sonnet-4-20250514
MAX_TOKENS=2048
MAX_HISTORY=30
RATE_LIMIT=10

.env
venv/
__pycache__/
*.pyc
.DS_Store

FROM python:3.12-slim

WORKDIR /app

## Copier et installer les dépendances d'abord (cache Docker)
COPY requirements.txt .
RUN pip install —no-cache-dir -r requirements.txt

## Copier le code source
COPY bot/ bot/
COPY .env .env

## Utilisateur non-root pour la sécurité
RUN useradd -r -s /bin/false botuser
USER botuser

CMD ["python", "-m", "bot.main"]

version: "3.8"

services:
  bot:
    build: .
    container_name: claude-discord-bot
    restart: unless-stopped
    env_file:
      - .env
    logging:
      driver: json-file
      options:
        max-size: "10m"
        max-file: "3"
    deploy:
      resources:
        limits:
          memory: 256M
          cpus: "0.5"

## Build et lancement en arrière-plan
docker compose up -d —build

## Vérifier les logs
docker compose logs -f bot

## Redémarrer après modification
docker compose down && docker compose up -d —build

## Sur le VPS, cloner le projet
git clone https://votre-repo.git /opt/claude-discord-bot
cd /opt/claude-discord-bot

## Créer l'environnement Python
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

## Configurer le .env
cp .env.example .env
nano .env  # Remplir les clés

[Unit]
Description=Claude Discord Bot
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=www-data
Group=www-data
WorkingDirectory=/opt/claude-discord-bot
ExecStart=/opt/claude-discord-bot/venv/bin/python -m bot.main
Restart=on-failure
RestartSec=10
StartLimitBurst=5

## Sécurité
NoNewPrivileges=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/opt/claude-discord-bot
PrivateTmp=true

## Environnement
EnvironmentFile=/opt/claude-discord-bot/.env

[Install]
WantedBy=multi-user.target

## Activer et démarrer le service
sudo systemctl daemon-reload
sudo systemctl enable claude-bot
sudo systemctl start claude-bot

## Vérifier le statut
sudo systemctl status claude-bot

## Voir les logs en temps réel
sudo journalctl -u claude-bot -f

## Ajouter dans on_message, après la vérification des mentions
ALLOWED_CHANNELS = {123456789, 987654321}  # IDs des salons autorisés
if message.channel.id not in ALLOWED_CHANNELS:
    return

BLOCKED_USERS = set()  # Ajouter les IDs des utilisateurs à bloquer
if message.author.id in BLOCKED_USERS:
    return

MAX_INPUT_LENGTH = 4000
if len(texte) > MAX_INPUT_LENGTH:
    await message.reply("Message trop long. Limite : 4000 caractères.")
    return

import asyncio

async def cleanup_old_conversations():
    while True:
        await asyncio.sleep(3600)  # Toutes les heures
        bot.conversations.clear()
        logger.info("Historique des conversations vidé")

## Ajouter dans setup_hook :
asyncio.create_task(cleanup_old_conversations())

## Dans setup_hook, remplacer await self.tree.sync() par :
guild = discord.Object(id=VOTRE_GUILD_ID)
await self.tree.sync(guild=guild)

import json
from datetime import datetime

class Stats:
    def __init__(self):
        self.total_requests = 0
        self.errors = 0
        self.started_at = datetime.now().isoformat()

    def to_dict(self):
        return {
            "total_requests": self.total_requests,
            "errors": self.errors,
            "started_at": self.started_at,
            "uptime_hours": round(
                (datetime.now() - datetime.fromisoformat(self.started_at))
                .total_seconds() / 3600, 1
            ),
        }

stats = Stats()