discord.py の tasks.loop で無料で簡易的に死活監視してみる
はじめに
discord.py には tasks.loop という機能が組み込まれていて、これを使うとファイル1つ追加するだけで簡易的な死活監視ができます。今回はこの方法を試してみました。
完成イメージ
10分ごとに Discord チャンネルへこんな Embed が届くようになります:
Heartbeat — Healthy
Uptime: 0h 10m 2s | Latency: 22ms | Guilds: 1
Boot: 2026-01-27 21:44 JST
レイテンシに応じて色が変わります:
- 緑: 200ms 未満(正常)
- 黄: 200〜500ms(低下)
- 赤: 500ms 以上(異常)
ログにも出るので、デプロイ先のログビューアでも確認できます。
INFO:src.cogs.health:[Heartbeat] Healthy | uptime=0h 10m 2s latency=22ms guilds=1
環境
- Python 3.12+
- discord.py 2.6+
やること
今回やることはざっくり以下の3つだけです。
-
health.pyを書く - ボットに登録する
- 送信先チャンネルIDを
.envに書く
1. health.py を書く
src/cogs/health.py を新規作成します。全体のコードはこんな感じです。
"""Health monitoring cog for sending periodic heartbeat embeds."""
# Python 3.10+ のアノテーション記法を使うためのインポート
from __future__ import annotations
# 標準ライブラリ
import logging
import time
from datetime import datetime, timedelta, timezone
# discord.py
import discord
from discord.ext import commands, tasks
# 自分のプロジェクトの設定(後述)
from src.config import settings
# このファイル用のロガーを取得
logger = logging.getLogger(__name__)
# ハートビートの間隔(分)
_HEARTBEAT_MINUTES = 10
# 日本時間のタイムゾーン(UTC+9)
# 必要に応じて変更してください
_JST = timezone(timedelta(hours=9))
class HealthCog(commands.Cog):
"""定期的にハートビートを送信する Cog"""
def __init__(self, bot: commands.Bot) -> None:
# Bot インスタンスを保存
self.bot = bot
# 起動時刻を記録(Uptime 計算用)
# time.monotonic() はシステム起動からの経過秒数を返す
# datetime と違って時刻変更の影響を受けない
self._start_time = time.monotonic()
# 起動時刻を日本時間で保存(Embed のフッター表示用)
self._boot_jst = datetime.now(_JST)
async def cog_load(self) -> None:
"""Cog が読み込まれたときに呼ばれる"""
# ハートビートのループを開始
self._heartbeat.start()
async def cog_unload(self) -> None:
"""Cog がアンロードされたときに呼ばれる"""
# ハートビートのループを停止
self._heartbeat.cancel()
@tasks.loop(minutes=_HEARTBEAT_MINUTES)
async def _heartbeat(self) -> None:
"""定期的に実行されるハートビート処理"""
# === Uptime の計算 ===
# 現在時刻 - 起動時刻 = 経過秒数
uptime_sec = int(time.monotonic() - self._start_time)
# divmod で商と余りを同時に取得
# 例: 3661秒 → 1時間, 余り61秒 → 1分, 余り1秒
hours, remainder = divmod(uptime_sec, 3600)
minutes, seconds = divmod(remainder, 60)
# 表示用の文字列に整形
uptime_str = f"{hours}h {minutes}m {seconds}s"
# === Bot の状態を取得 ===
# Bot が参加しているサーバー(ギルド)の数
guild_count = len(self.bot.guilds)
# Discord との接続遅延を取得
# bot.latency は秒単位なのでミリ秒に変換
latency_ms = round(self.bot.latency * 1000)
# === レイテンシに応じてステータスを判定 ===
if latency_ms < 200:
status = "Healthy" # 正常
elif latency_ms < 500:
status = "Degraded" # やや遅い
else:
status = "Unhealthy" # 異常
# === ログに出力 ===
logger.info(
"[Heartbeat] %s | uptime=%s latency=%dms guilds=%d",
status, uptime_str, latency_ms, guild_count,
)
# === Discord チャンネルに送信 ===
# チャンネルIDが設定されている場合のみ送信
if settings.health_channel_id:
# チャンネルIDからチャンネルオブジェクトを取得
channel = self.bot.get_channel(settings.health_channel_id)
# チャンネルが存在し、テキストチャンネルであることを確認
if channel is not None and isinstance(channel, discord.TextChannel):
# Embed を作成
embed = self._build_embed(
status=status,
uptime_str=uptime_str,
latency_ms=latency_ms,
guild_count=guild_count,
)
# 送信
await channel.send(embed=embed)
@_heartbeat.before_loop
async def _before_heartbeat(self) -> None:
"""ループ開始前に1回だけ呼ばれる"""
# Bot が Discord に接続完了するまで待機
# これがないと、接続前にハートビートを送ろうとしてエラーになる
await self.bot.wait_until_ready()
def _build_embed(
self,
*,
status: str,
uptime_str: str,
latency_ms: int,
guild_count: int,
) -> discord.Embed:
"""ハートビート用の Embed を作成する"""
# === レイテンシに応じて色を決定 ===
if latency_ms < 200:
color = discord.Color.green() # 緑
elif latency_ms < 500:
color = discord.Color.gold() # 黄
else:
color = discord.Color.red() # 赤
# === Embed を作成 ===
embed = discord.Embed(
title=f"Heartbeat — {status}",
color=color,
# timestamp は UTC で指定する(Discord が自動でユーザーの
# タイムゾーンに変換して表示してくれる)
timestamp=datetime.now(timezone.utc),
)
# === フィールドを追加 ===
# inline=True で横並びに表示
embed.add_field(name="Uptime", value=uptime_str, inline=True)
embed.add_field(name="Latency", value=f"{latency_ms}ms", inline=True)
embed.add_field(name="Guilds", value=str(guild_count), inline=True)
# === フッターに起動時刻を表示 ===
embed.set_footer(text=f"Boot: {self._boot_jst:%Y-%m-%d %H:%M JST}")
return embed
async def setup(bot: commands.Bot) -> None:
"""
Bot にこの Cog を登録する
load_extension("src.cogs.health") で呼ばれる
"""
await bot.add_cog(HealthCog(bot))
こんな感じで動きます。順番に見ていきます。
tasks.loop でループを作る
@tasks.loop(minutes=_HEARTBEAT_MINUTES)
async def _heartbeat(self) -> None:
...
@tasks.loop(minutes=10) を付けるだけで、10分ごとに自動実行されるメソッドになります。seconds や hours も指定できます。
cog_load / cog_unload でライフサイクル管理
async def cog_load(self) -> None:
self._heartbeat.start()
async def cog_unload(self) -> None:
self._heartbeat.cancel()
__init__ でタスクを開始するのは避けた方がよさそうです。__init__ は同期メソッドなので、ボットがまだ Discord に接続していない状態でタスクが走ってしまいます。cog_load は非同期メソッドなので、こちらで開始するのが安全かと思います。
before_loop で接続完了を待つ
@_heartbeat.before_loop
async def _before_heartbeat(self) -> None:
await self.bot.wait_until_ready()
before_loop はループ開始前に1回だけ呼ばれる準備用のフックです。wait_until_ready() で Discord への接続完了を待ちます。
tasks.loop は before_loop 完了後に即座に初回実行される仕様のようなので、起動直後にハートビートが送られます。before_loop 内でタスク本体を手動で呼ぶと二重実行になるので、そこだけ注意が必要です。
bot.latency で応答性をチェック
latency_ms = round(self.bot.latency * 1000)
bot.latency は Discord Gateway との WebSocket Heartbeat 往復遅延を秒単位で返します。ミリ秒に変換して、200ms / 500ms を閾値にステータスを判定しています。
2. ボットに登録する
setup_hook に1行追加するだけです。
async def setup_hook(self) -> None:
await self.load_extension("src.cogs.health")
3. 送信先チャンネルIDを設定する
.env に追加します:
HEALTH_CHANNEL_ID=123456789
設定クラス側(pydantic-settings を使っている場合):
class Settings(BaseSettings):
discord_token: str
health_channel_id: int = 0 # 0 なら Discord への送信をスキップ
health_channel_id = 0 のままでもログには出力されるので、Discord に送らなくてもOKです。
これで完成
ボットを起動すると、すぐにハートビートが送られます。あとは10分ごとに自動で繰り返されます。
追加のパッケージも不要、外部サービスへの登録も不要。discord.py の標準機能だけで動くのは手軽でいいですね。
もう少しやるなら
エラーハンドリング
@_heartbeat.error
async def _heartbeat_error(self, error: Exception) -> None:
logger.exception("Heartbeat failed: %s", error)
tasks.loop はデフォルトでエラーが起きるとタスクを停止するようです。error ハンドラを追加すると、エラーをログに記録しつつタスクを継続できます。
実運用では、一時的なネットワークエラーで channel.send が失敗することがあります。エラーハンドラがないとループ自体が止まってしまうので、本番環境では追加しておいた方がよさそうです。
インターバルを動的に変更
self._heartbeat.change_interval(minutes=5)
「死んだ」をどう判断するか
死活監視で一番大事なのは、Bot が死んだことにどうやって気づくかです。
この方式は Bot 自身がハートビートを送る仕組み(Push 型)なので、Bot がフリーズ・クラッシュすると通知も一緒に止まります。つまり:
- 通知が来ている → 生きている
- 通知が来ない → 死んでいる かもしれない(ただし人間が気づく必要がある)
10分間隔で送っているので、「20分以上 Embed が来ていない」なら異常と判断できます。しかし、これは Discord チャンネルを目視で確認しないとわかりません。
一方、UptimeRobot のような外部サービス(Pull 型)は「生きてる?」と外から問い合わせる方式です。Bot が応答しなければ即座にアラートが飛ぶため、より確実に異常を検知できます。
自動で検知したい場合は、外部からの監視が必要です。次のセクションで紹介します。
もっと本格的にやるなら
tasks.loop はあくまで簡易的な方法です。本格的な監視サービスを使えば、グラフ付きの通知やフリーズ検知など、より高度なことができます。もっと良いやり方があるかもしれませんが、参考程度に比較表を載せておきます。
| サービス | 無料枠 | グラフ通知 | フリーズ検知 | 導入コスト |
|---|---|---|---|---|
| tasks.loop(本記事) | - | × | × | ファイル1つ |
| UptimeRobot | 50モニター | × | ○ | HTTP エンドポイント追加 |
| Datadog | 5ホスト / 1日保持 | △(有料) | △(有料) | エージェント導入 |
| Mackerel | 無料プランあり | Slack/Discordにグラフ画像付き通知 | ○ | エージェント導入 |
UptimeRobot で無料で自動検知する
Bot に aiohttp で簡易 HTTP サーバーを追加し、UptimeRobot から定期的に叩く方法です。
from aiohttp import web
async def health_handler(request: web.Request) -> web.Response:
return web.json_response({"status": "ok"})
# bot の setup_hook 等で起動
app = web.Application()
app.router.add_get("/health", health_handler)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "0.0.0.0", int(os.environ.get("PORT", 8080)))
await site.start()
UptimeRobot(無料プラン)で https://your-app.example.com/health を5分間隔で監視し、応答がなければメール通知を受け取れます。Bot がフリーズしても HTTP サーバーごと止まるので、外部から検知できます。
Datadog でやってみる
Datadog Agent をインストールすれば、コード変更なしにログやメトリクスを収集できます。多くの PaaS では Buildpack や Add-on として簡単に導入できます。
本記事の [Heartbeat] ログを Datadog に流せば、Logs Monitor で「一定時間ログが出なければアラート」という監視ができます。ただし、Logs Monitor やグラフ付き通知は有料プランが必要です。
...と思ったけど Mackerel の方が楽かも
Discord Bot の監視なら、Mackerel の方が相性が良いかもしれません。
Mackerel は Slack にグラフ画像のスナップショット付きで通知を飛ばせます。そして Discord は Slack 互換の Webhook に対応しているので、Webhook URL の末尾に /slack を付ければ Discord チャンネルにも同じ通知が届きます。
# Discord の Slack 互換 Webhook URL
https://discord.com/api/webhooks/xxxxx/yyyyyy/slack
Bot の監視通知を Discord に統一できるのは地味に便利です。グラフ画像付きで「CPU 使用率が跳ねた」「メモリが逼迫している」といった状況が一目でわかります。
まずは本記事の tasks.loop で始めて、運用が安定してきたら外部サービスを検討するのが良いのではないでしょうか。
まとめ
- ファイル1つ + 設定1行で死活監視が動く
-
tasks.loopは discord.py 標準機能。追加パッケージ不要 -
before_loop+wait_until_ready()で接続完了を待ってから即時実行 - まずはこれで始めて、必要に応じて外部監視を追加するのが良さそう
Discussion