こんにちは！虎の穴ラボの鷺山です。

前回の記事で、アプリケーションのエラーログをSlackに転送する方法をご紹介しました。

日々流れてくるログには「原因が判明しているもの」や「静観してよいもの」が少なからず混ざっています。本来はログレベルを見直したり問題そのものを修正すべきですが、すぐに対応できないケースも現実的には多いと思います。しかし、大量のログを一件ずつ判読するのは骨が折れます。

そこで今回は、Slackに投稿されるエラーログが既知のものだった場合に自動返信するボットを作ってみたのでご紹介します。事前に「既知のパターン」だと把握できるだけで、確認時の心理的・時間的コストはぐっと下がります。

前提環境

項目	バージョン・値
Python	3.14
AWSリージョン	`ap-northeast-1`

Slack Appのセットアップ

「Slackのメッセージを読み取る」「読み取ったスレッドに返信する」ためのSlack Appをセットアップします。

Slackアプリの管理ページ https://api.slack.com/apps に移動します。
Create New AppまたはCreate an Appを押し、From scratchを選択します。
App Nameにアプリ名を入力し、Pick a workspace to develop your app in: にアプリの作成先のワークスペースを選択します。
- 今回は既知のアラートですというアプリ名とします。
- Create Appを押すとアプリが作成されます。
アプリが作成されたらOAuth & Permissionsメニューを開き、ボットトークンのスコープに以下を追加します。
- chat:write … チャンネルへのメッセージ送信を許可
- channels:history … パブリックチャンネルの読み取りを許可（パブリックチャンネルの場合に追加）
- groups:history … プライベートチャンネルの読み取りを許可（プライベートチャンネルの場合に追加）
次にApp Homeメニューを開き、Your App's Presence in SlackのApp Display Nameを設定します。
- Display Name (Bot Name): ボットの表示名を入力します。
- Default username: ボットのユーザー名を入力します。
Install Appメニューを開き、Install to Workspaceを押してこのアプリのSlackへのアクセスを許可します。ボット用のOAuthトークンxoxb-...が生成されます。このトークンは次のセクションで使用します。
最後に、以下のcURLコマンドでボットのIDを取得します。Bearerには上記のトークンxoxb-...を指定します。

curl -H "Authorization: Bearer xoxb-..." https://slack.com/api/auth.test

リクエストに成功するとJSONが返ります。bot_idがボットのIDです。このIDも次のセクションで使用します。

{
    "ok": true,
    ...
    "bot_id": "BXXXXXXXXXX",
    ...
}

Lambda関数の作成

Slackからイベントを受け取り、特定のキーワードを含む場合に返信するAWS Lambda関数を作成します。

今回はreplyToKnownIssueOnSlackという関数名にしました。ランタイムはPythonを使用します。

今回はSlackからのイベントを受け取れるようにするため、その他の設定から関数URLを有効化します。
認証タイプはLambda側で認証を実装するためNONEを選択します。

Lambda関数が作成されたら、「設定」タブから関数URLを控えておいてください（次のセクションで使用します）。

コードは以下のようになります。

import json
import os
from urllib.request import Request, urlopen

# `when` にマッチするワードを検出したら `then` のテキストを返信します
rules = [
    {"when": "Ping", "then": "Pong"},
    {"when": "The cat sat on the keyboard.", "then": "一時的なエラーです"},
]

API_KEY = os.environ["API_KEY"]
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_BOT_ID = os.environ["SLACK_BOT_ID"]


def lambda_handler(payload, context):
    # クエリによる簡易認証
    api_key = payload.get("queryStringParameters", {}).get("apiKey", None)
    if api_key != API_KEY:
        return {"statusCode": 401, "body": "Unauthorized"}

    body = json.loads(payload.get("body", "{}"))

    # Slackイベントへのサブスクリプション用
    if "challenge" in body:
        return {"statusCode": 200, "body": body["challenge"]}

    # パラメータを抽出
    event = body.get("event", {})
    bot_id = event.get("bot_id", "")
    text = event.get("text", "")
    channel = event.get("channel", "")
    ts = event.get("ts", "")

    print(f"bot_id={bot_id}, text={text}, channel={channel}, ts={ts}")

    # 自分自身のメッセージはスキップ (無限ループ防止)
    if bot_id and bot_id == SLACK_BOT_ID:
        return {"statusCode": 200, "body": "SKIPPED"}

    # ルールにマッチするワードを検出したらスレッドに返信
    for rule in rules:
        if rule["when"] in text:
            when = rule["when"].replace("`", "'")  # バッククォートをサニタイズ
            message = f"{rule['then']}＞ `{when}`"
            send_reply_to_slack(channel=channel, thread_ts=ts, text=message)

    return {"statusCode": 200, "body": "OK"}


def send_reply_to_slack(channel, thread_ts, text):
    url = "https://slack.com/api/chat.postMessage"

    payload = {"channel": channel, "thread_ts": thread_ts, "text": text}

    headers = {
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {SLACK_BOT_TOKEN}",
    }

    req = Request(url, json.dumps(payload).encode("utf-8"), headers)
    urlopen(req).read()

💡ポイント

上記のコードの各ポイントを解説します。

# `when` にマッチするワードを検出したら `then` のテキストを返信します
rules = [
    {"when": "Ping", "then": "Pong"},
    {"when": "The cat sat on the keyboard.", "then": "一時的なエラーです"},
]

「マッチさせるキーワード」と「マッチした場合に返信する文章」がペアになった「ルール辞書」です。
自動返信させたい内容をこの辞書を追加していきます。

API_KEY = os.environ["API_KEY"]
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_BOT_ID = os.environ["SLACK_BOT_ID"]

このLambda関数で使用する環境変数です。

キー	値
API_KEY	この関数の認証キーです。十分な長さのランダムな文字列を設定してください。
SLACK_BOT_TOKEN	前のセクションで生成したSlackボットのトークンです。`xoxb-...`
SLACK_BOT_ID	前のセクションで生成したSlackボットのIDです。`B...`

環境変数は「設定」タブから事前に設定してください。

    # クエリによる簡易認証
    api_key = payload.get("queryStringParameters", {}).get("apiKey", None)
    if api_key != API_KEY:
        return {"statusCode": 401, "body": "Unauthorized"}

認証キー ?apiKey=xxxxxxxxxx が一致しない場合は認証エラーとしています。

    # Slackイベントへのサブスクリプション用
    if "challenge" in body:
        return {"statusCode": 200, "body": body["challenge"]}

Slackイベントへのサブスクリプションに必要なコードです。次のセクションで使用します。
リクエストにchallengeが含まれていたらそのままレスポンスで返します。

    # 自分自身のメッセージはスキップ (無限ループ防止)
    if bot_id and bot_id == SLACK_BOT_ID:
        return {"statusCode": 200, "body": "SKIPPED"}

ボットが返信したメッセージにボット自身が反応しないようにしています。
⚠️環境変数SLACK_BOT_IDを必ず設定してください。

    # ルールにマッチするワードを検出したらスレッドに返信
    for rule in rules:
        if rule["when"] in text:
            when = rule["when"].replace("`", "'")  # バッククォートをサニタイズ
            message = f"{rule['then']}＞ `{when}`"
            send_reply_to_slack(channel=channel, thread_ts=ts, text=message)

前述の「ルール辞書」とメッセージを照らし合わせて、キーワードがマッチした場合にそのスレッドに返信しています。

Slackボットのイベント受信設定

Slackアプリの設定画面のEvent Subscriptionsメニューから、ボットがメッセージを受け取れるように設定します。

Enable EventsをOnにします。
Request URLに、Lambdaの関数URLにAPIキーを付与したものを入力します。
- 例: https://xxxxxxxxxx.lambda-url.ap-northeast-1.on.aws/?apiKey=xxxxxxxxxx
- チャレンジ認証に成功するとVerifiedと表示されます。
Subscribe to bot eventsにサブスクライブするイベントを追加します。
- パブリックチャンネルの場合： message.channels
- プライベートチャンネルの場合： message.groups
最後にSave Changesで設定を保存します。

Slackチャンネルにインテグレーションを追加

アプリケーションログの転送先（今回の自動返信先）であるSlackチャンネルのインテグレーションに、今回セットアップしたアプリを追加します。

動作確認

チャンネルに「Ping」と投稿したら、ボットが「Pong」と返してくるはずです。

ルール辞書のwhenにマッチするメッセージが投稿された場合に、thenの内容が返信されるようになっています。

rules = [
    ...
    {"when": "The cat sat on the keyboard.", "then": "一時的なエラーです"},
]

まとめ

今回は、Slackに投稿されるエラーログが既知のものだった場合に自動返信するボットを作成しました。
ルール辞書にキーワードと返信内容を追加するだけで簡単に拡張できるので、ぜひ活用してみてください。
本来はログそのものの改善が望ましいですが、すぐに対応できない場面での運用の手助けになれば幸いです！

採用情報

虎の穴ラボでは一緒に働く仲間を募集中です！
この記事を読んで、興味を持っていただけた方はぜひ弊社の採用情報をご覧ください。
toranoana-lab.co.jp

虎の穴ラボ技術ブログ

虎の穴ラボ株式会社所属のエンジニアが書く技術ブログです

Slackのエラーログに自動返信するBotをAWS Lambdaで作ってみた