Webhookを搭載したdiscord botをつくる

次の要件でdiscordのbotを作りたかったのですが、良いサンプルがなかったので、備忘録も兼ねてここに残しておきます。

• unix上でsystemdを使用して常駐する
• botとして拡張が容易
• 自身がHTTPサーバーを内蔵し、webhook経由で特定のチャンネルに文章を投稿できる

discord自体にwebhookがありますが、botの発言として投稿させたかったので、bot自身にwebhook機能を内蔵させようと考えました。

技術選定

要はdiscordのbotと簡単なhttpサーバが動けば良いです。普段はjsが好きで良く使いますが、discordのbotに関するドキュメントが充実していること、VPSでサクッと動かしたいことを踏まえてpythonを使うことにしました。
discord.botとhttpサーバーをそれぞれ非同期で動かせればよいかな、と軽い気持ちで作り始めましたが、これが意外と大変でした。

完成品

先に完成品を掲載します。

from aiohttp import web
import discord
from discord.ext import commands, tasks
import asyncio
import datetime
import signal

# Discord BotのトークンとチャンネルIDを設定
TOKEN = 'Your Token'
NOTIFY_CHANNEL_ID = 9999 #Channel ID

# Discordクライアントの設定
# 必要なintentsを有効化
intents = discord.Intents.default()
intents.messages = True
intents.message_content = True  # メッセージ内容にアクセスするために必要
intents.guilds = True
intents.guild_scheduled_events = True

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

@bot.listen()
async def on_message(message: discord.Message):
    # ボット自身のメッセージを無視
    if message.author.bot:
        return
    if message.channel.id != NOTIFY_CHANNEL_ID:
        return

    # ボットがメンションされたかチェック
    if bot.user.mentioned_in(message) and len(message.mentions) == 1:
        #リアクションを追加
        await message.add_reaction("👀")

        # メッセージ内容をエコーして返信
        await message.reply(message)

# aiohttpサーバーの設定
async def handle_request(request):
    try:
        # リクエストからデータを取得
        data = await request.json()
        message = data.get("message", "デフォルトメッセージ")

        # Discordチャンネルにメッセージを送信
        channel = bot.get_channel(NOTIFY_CHANNEL_ID)
        if channel:
            await channel.send(message)
            return web.Response(text="Message sent to Discord!")
        else:
            return web.Response(text="Discord channel not found.", status=500)
    except Exception as e:
        return web.Response(text=f"Error: {e}", status=500)

# aiohttpアプリの初期化
app = web.Application()
app.router.add_post("/send", handle_request)

# Discordボットをバックグラウンドで実行
async def start_discord_bot():
    try:
        await bot.start(TOKEN)
    finally:
        loop.stop()
        await bot.close()

async def start_server():
    try:
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, "localhost", 8080)
        await site.start()
    
        print("Server is running...")
        await asyncio.Event().wait()  # 永続的に待機する
    finally:
        print("close server...")

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(start_discord_bot())
        tg.create_task(start_server())

def sigterm_handler(signum, frame):
    raise KeyboardInterrupt

# aiohttpサーバーとDiscordボットを同時に実行
if __name__ == "__main__":
    # SIGTERMのハンドラを設定
    signal.signal(signal.SIGTERM, sigterm_handler)
    
    discord.utils.setup_logging()
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("keyboard interrupt!")
    finally:
        print("exiting...")

ハマった点

aiohttpを使う

pythonでHTTPサーバーを実現できるライブラリはいくつもありますが、「非同期で」「他の動作を阻害しない」となると、試した限りaiohttpが一番相性がよかったです。chatGPT等に聞くとFlaskを使ったアプリを出力してくれるのですが、動作はしても期待通りに終了しないなど、あまりいい動作をしてくれませんでした。今回非同期で動かしているので、それが影響したかもしれません。
aiohttpはあまりリッチな機能はありませんが、今回の用途では十分でした。

SIGTERMを捕捉してKeyboardInterruptを送出する

systemctl stopはアプリケーションにSIGTERMを送出するのですが、これを適切にハンドルしてあげないとdiscord botが正しくcloseせず、終了後もしばらくdiscord上でオンラインになってしまいました。KeyboardInterruptで解決しました。（時間が経てばオフラインになります）

まとめ

これをお手持ちのVPSなどで稼働させ、apacheやnginxでリバプロを設定してあげればOKです。今回は特定のサーバーでしか動かないので認証などをサボっていますが、必要であれば適切に設定してください。チャンネルIDをパラメータで渡すなどしてもよいですね。よきDiscordライフを！