- Authors
- Name
はじめに
Slackは開発チームの中核コミュニケーションプラットフォームです。よくできたSlack Botは、障害アラート、デプロイ管理、コードレビューリマインダー、AI問い合わせなど、さまざまな業務を自動化します。
Bolt for PythonはSlack公式フレームワークで、簡潔なAPIでさまざまなSlack機能を活用できます。
Slackアプリの設定
1. アプリの作成
1. https://api.slack.com/apps にアクセス
2. "Create New App" → "From scratch"
3. App Name: "DevOps Bot"
4. Workspaceを選択
2. Bot Token Scopesの設定
OAuth & Permissions → Bot Token Scopes:
- chat:write (メッセージ送信)
- commands (スラッシュコマンド)
- im:history (DM履歴読み取り)
- im:read (DMチャンネル情報)
- channels:history (チャンネルメッセージ読み取り)
- channels:read (チャンネル情報)
- users:read (ユーザー情報)
- reactions:write (リアクション追加)
- files:write (ファイルアップロード)
3. Event Subscriptions
Event Subscriptions → Enable Events
Subscribe to bot events:
- message.im (DM受信)
- message.channels (チャンネルメッセージ)
- app_mention (メンション)
- app_home_opened (アプリホームタブを開く)
プロジェクト設定
# 依存関係のインストール
pip install slack-bolt slack-sdk python-dotenv openai
# プロジェクト構造
slack-bot/
├── app.py # メインアプリ
├── handlers/
│ ├── __init__.py
│ ├── commands.py # スラッシュコマンド
│ ├── events.py # イベントハンドラー
│ ├── actions.py # ボタン/アクションハンドラー
│ └── modals.py # モーダルビュー
├── services/
│ ├── __init__.py
│ ├── ai.py # LLM統合
│ ├── deploy.py # デプロイサービス
│ └── github.py # GitHub連携
├── .env
└── requirements.txt
# .env
SLACK_BOT_TOKEN=xoxb-your-bot-token
SLACK_APP_TOKEN=xapp-your-app-token
SLACK_SIGNING_SECRET=your-signing-secret
OPENAI_API_KEY=sk-your-key
基本アプリ構成
# app.py
import os
from dotenv import load_dotenv
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
load_dotenv()
# Socket Modeでアプリ作成（Webhookサーバー不要）
app = App(token=os.environ["SLACK_BOT_TOKEN"])
# ===== イベントハンドラー =====
# メンションに応答
@app.event("app_mention")
def handle_mention(event, say, client):
user = event["user"]
text = event["text"]
# ユーザー情報を取得
user_info = client.users_info(user=user)
name = user_info["user"]["real_name"]
say(f"こんにちは{name}さん！何かお手伝いしましょうか？ 🤖")
# DMメッセージの処理
@app.event("message")
def handle_dm(event, say, client, logger):
# ボット自身のメッセージは無視
if event.get("bot_id"):
return
channel_type = event.get("channel_type")
if channel_type == "im":
text = event.get("text", "")
say(f"お伝えいただいた内容を処理いたします: '{text}'")
# ===== スラッシュコマンド =====
@app.command("/deploy")
def handle_deploy_command(ack, body, client):
ack() # 3秒以内の応答が必須！
# モーダルを開く
client.views_open(
trigger_id=body["trigger_id"],
view={
"type": "modal",
"callback_id": "deploy_modal",
"title": {"type": "plain_text", "text": "デプロイ管理"},
"submit": {"type": "plain_text", "text": "デプロイ開始"},
"blocks": [
{
"type": "input",
"block_id": "service_block",
"element": {
"type": "static_select",
"placeholder": {"type": "plain_text", "text": "サービスを選択"},
"options": [
{"text": {"type": "plain_text", "text": "API Server"}, "value": "api"},
{"text": {"type": "plain_text", "text": "Web Frontend"}, "value": "web"},
{"text": {"type": "plain_text", "text": "Worker"}, "value": "worker"},
],
"action_id": "service_select",
},
"label": {"type": "plain_text", "text": "サービス"},
},
{
"type": "input",
"block_id": "env_block",
"element": {
"type": "static_select",
"options": [
{"text": {"type": "plain_text", "text": "Staging"}, "value": "staging"},
{"text": {"type": "plain_text", "text": "Production"}, "value": "production"},
],
"action_id": "env_select",
},
"label": {"type": "plain_text", "text": "環境"},
},
{
"type": "input",
"block_id": "version_block",
"element": {
"type": "plain_text_input",
"placeholder": {"type": "plain_text", "text": "v1.2.3 または latest"},
"action_id": "version_input",
},
"label": {"type": "plain_text", "text": "バージョン"},
},
],
},
)
# モーダル送信の処理
@app.view("deploy_modal")
def handle_deploy_submission(ack, body, client, view):
ack()
values = view["state"]["values"]
service = values["service_block"]["service_select"]["selected_option"]["value"]
env = values["env_block"]["env_select"]["selected_option"]["value"]
version = values["version_block"]["version_input"]["value"]
user_id = body["user"]["id"]
# デプロイ通知を送信
client.chat_postMessage(
channel="#deployments",
blocks=[
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"🚀 *デプロイ開始*\n"
f"• サービス: `{service}`\n"
f"• 環境: `{env}`\n"
f"• バージョン: `{version}`\n"
f"• 依頼者: <@{user_id}>"
}
},
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {"type": "plain_text", "text": "✅ 承認"},
"style": "primary",
"value": f"{service}|{env}|{version}",
"action_id": "deploy_approve",
},
{
"type": "button",
"text": {"type": "plain_text", "text": "❌ 拒否"},
"style": "danger",
"value": f"{service}|{env}|{version}",
"action_id": "deploy_reject",
},
],
},
],
)
# ボタンアクションの処理
@app.action("deploy_approve")
def handle_approve(ack, body, client, action):
ack()
service, env, version = action["value"].split("|")
user_id = body["user"]["id"]
# 元のメッセージを更新
client.chat_update(
channel=body["channel"]["id"],
ts=body["message"]["ts"],
blocks=[
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"✅ *デプロイ承認済み*\n"
f"• サービス: `{service}` → `{env}` ({version})\n"
f"• 承認者: <@{user_id}>\n"
f"• ステータス: デプロイ進行中... ⏳"
}
}
],
)
# 実際のデプロイロジックを実行
# deploy_service(service, env, version)
# ===== ステータス照会コマンド =====
@app.command("/status")
def handle_status(ack, say, command):
ack()
say(
blocks=[
{
"type": "header",
"text": {"type": "plain_text", "text": "📊 サービスステータス"}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": "*API Server*\n🟢 Healthy (v1.2.3)"},
{"type": "mrkdwn", "text": "*Web Frontend*\n🟢 Healthy (v2.0.1)"},
{"type": "mrkdwn", "text": "*Worker*\n🟡 Degraded (v1.1.0)"},
{"type": "mrkdwn", "text": "*Database*\n🟢 Healthy (CPU: 45%)"},
]
},
{"type": "divider"},
{
"type": "context",
"elements": [
{"type": "mrkdwn", "text": f"最終更新: <!date^{int(__import__('time').time())}^{{date_short_pretty}} {{time}}|just now>"}
]
}
]
)
# ===== AIチャットボット機能 =====
@app.command("/ask")
def handle_ask(ack, say, command):
ack()
question = command["text"]
if not question:
say("質問を入力してください。例: `/ask Kubernetes PodがCrashLoopBackOffの時の対処法`")
return
# ローディング表示
say(f"🤔 '{question}'について考えています...")
# LLM呼び出し
from openai import OpenAI
client_ai = OpenAI()
response = client_ai.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "You are a helpful DevOps assistant. Answer in Korean. Be concise."},
{"role": "user", "content": question}
],
max_tokens=1000,
)
answer = response.choices[0].message.content
say(
blocks=[
{
"type": "section",
"text": {"type": "mrkdwn", "text": f"*Q: {question}*"}
},
{"type": "divider"},
{
"type": "section",
"text": {"type": "mrkdwn", "text": answer}
},
{
"type": "context",
"elements": [
{"type": "mrkdwn", "text": "💡 AI生成回答 — 正確性をご確認ください"}
]
}
]
)
# ===== アプリホームタブ =====
@app.event("app_home_opened")
def update_home_tab(client, event):
client.views_publish(
user_id=event["user"],
view={
"type": "home",
"blocks": [
{
"type": "header",
"text": {"type": "plain_text", "text": "🤖 DevOps Bot Dashboard"}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "*利用可能なコマンド:*\n"
"• `/deploy` — サービスデプロイ\n"
"• `/status` — サービスステータス確認\n"
"• `/ask [質問]` — AIに質問\n"
"• `/incident` — 障害報告"
}
},
{"type": "divider"},
{
"type": "section",
"text": {"type": "mrkdwn", "text": "*最近のデプロイ:*\n✅ API Server v1.2.3 → Production（2時間前）\n✅ Web v2.0.1 → Production（昨日）"}
},
],
},
)
# ===== スケジュールメッセージ =====
def send_daily_standup_reminder():
"""毎朝のスタンドアップリマインダー"""
import time
app.client.chat_postMessage(
channel="#engineering",
text="🌅 *スタンドアップリマインダー*\n今日の計画をスレッドで共有してください！\n1. 昨日やったこと\n2. 今日やること\n3. ブロッカー",
)
# ===== エラーハンドリング =====
@app.error
def handle_errors(error, body, logger):
logger.exception(f"Error: {error}")
logger.info(f"Request body: {body}")
# ===== 実行 =====
if __name__ == "__main__":
# Socket Mode（Webhookサーバー不要、ファイアウォール内でも動作）
handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
handler.start()
障害管理ボット
@app.command("/incident")
def handle_incident(ack, body, client):
ack()
client.views_open(
trigger_id=body["trigger_id"],
view={
"type": "modal",
"callback_id": "incident_modal",
"title": {"type": "plain_text", "text": "🚨 障害報告"},
"submit": {"type": "plain_text", "text": "報告"},
"blocks": [
{
"type": "input",
"block_id": "severity",
"element": {
"type": "static_select",
"options": [
{"text": {"type": "plain_text", "text": "🔴 P1 - Critical"}, "value": "P1"},
{"text": {"type": "plain_text", "text": "🟠 P2 - Major"}, "value": "P2"},
{"text": {"type": "plain_text", "text": "🟡 P3 - Minor"}, "value": "P3"},
],
"action_id": "severity_select",
},
"label": {"type": "plain_text", "text": "深刻度"},
},
{
"type": "input",
"block_id": "description",
"element": {
"type": "plain_text_input",
"multiline": True,
"placeholder": {"type": "plain_text", "text": "障害状況を説明してください"},
"action_id": "desc_input",
},
"label": {"type": "plain_text", "text": "説明"},
},
],
},
)
@app.view("incident_modal")
def handle_incident_submit(ack, body, client, view):
ack()
values = view["state"]["values"]
severity = values["severity"]["severity_select"]["selected_option"]["value"]
description = values["description"]["desc_input"]["value"]
reporter = body["user"]["id"]
# 障害チャンネルを作成
import time
channel_name = f"incident-{int(time.time())}"
channel = client.conversations_create(
name=channel_name,
is_private=False,
)
channel_id = channel["channel"]["id"]
# 障害の初期メッセージ
client.chat_postMessage(
channel=channel_id,
blocks=[
{
"type": "header",
"text": {"type": "plain_text", "text": f"🚨 障害報告 ({severity})"}
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*報告者:* <@{reporter}>\n"
f"*深刻度:* {severity}\n"
f"*説明:*\n{description}"
}
},
{"type": "divider"},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "⏱ *タイムラインをこのスレッドに記録してください*"
}
},
],
)
# 通知チャンネルにアナウンス
client.chat_postMessage(
channel="#incidents",
text=f"🚨 新規障害発生: {severity}\n📝 {description[:100]}...\n🔗 <#{channel_id}>",
)
Dockerデプロイ
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "app.py"]
# docker-compose.yml
version: '3.8'
services:
slack-bot:
build: .
env_file: .env
restart: unless-stopped
healthcheck:
test: ['CMD', 'python', '-c', "import requests; requests.get('http://localhost:3000/health')"]
interval: 30s
timeout: 10s
retries: 3
クイズ
Q1. Slack Boltでスラッシュコマンドハンドラーが最初にやるべきことは？
**ack()**の呼び出しです。Slackは3秒以内に応答（acknowledge）を受け取れないとタイムアウトエラーを表示します。
Q2. Socket Modeの利点は？
WebSocket接続を使用するため、公開URLやWebhookサーバーが不要です。ファイアウォール内でも、ローカル開発でもすぐに動作します。
Q3. Block Kitの主要なブロックタイプ3つは？
section（テキスト/フィールド）、actions（ボタン/セレクトなどのインタラクティブ要素）、input（モーダルでのユーザー入力）です。
Q4. @app.actionと@app.viewの違いは？
@app.actionはボタンクリックなどのインタラクションを、@app.viewはモーダルの送信（submit）を処理します。それぞれaction_idとcallback_idでマッチングされます。
Q5. Bot Token（xoxb-）とApp Token（xapp-）の違いは？
Bot Tokenはチャンネルメッセージ送信などのSlack API呼び出しに使用し、App TokenはSocket ModeのWebSocket接続に使用します。
Q6. 障害発生時に自動でSlackチャンネルを作成する理由は？
障害関連のコミュニケーションを一か所に集中させ、タイムラインを記録し、事後分析（ポストモーテム）のための履歴を残すためです。
Q7. Slack BotにLLMを統合する際の注意点は？
LLMの応答時間が3秒を超える可能性があるため、まずack()で応答した後、非同期でLLMを呼び出し、結果を別メッセージで送信する必要があります。
まとめ
Slack Botはチームの生産性を大きく向上させるツールです。Bolt SDKを使えば、スラッシュコマンド、モーダル、ボタンインタラクションなどを簡潔なPythonコードで実装でき、Socket Modeで別途サーバーなしですぐにデプロイできます。