PricingLoginStart free
AI Agents · レッスン

チーム通知ボットの構築

スケジュール済みメッセージ、DMの要約、チャンネル通知エージェントを構築します。

レッスン 4/413 ステップ

「チーム通知ボットの構築」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

チーム通知ボットのアーキテクチャ

チーム通知ボットは、デプロイ、CI/CD パイプライン、監視アラート、エラートラッキングなどの外部システムを監視し、整形した更新情報を関連する Slack チャンネルに投稿します。基本パターンは、外部イベント → Webhook → エージェント → Slack メッセージ です。エージェントがルーティング、書式設定、配信を担当します。

# Team Notification Bot Flow:
#
# External System (GitHub, PagerDuty, Sentry, etc.)
#    |
#    | HTTP POST (webhook)
#    v
# Flask/FastAPI webhook endpoint
#    |
#    | Parse event
#    v
# Agent: classify, format, route
#    |
#    | Slack API
#    v
# Team channel / DM / thread

print('Webhook -> Agent -> Slack is the core notification pattern')

外部 Webhook の受信

外部サービスは HTTP Webhook を介してボットにイベントを送信します。Flask のエンドポイントを設定して POST リクエストを受信し、検証（サービスが対応している場合は署名チェック）を行ってから、通知ハンドラーにペイロードを渡します。

from flask import Flask, request, jsonify
import hmac
import hashlib
import os

flask_app = Flask(__name__)

@flask_app.route('/webhook/github', methods=['POST'])
def github_webhook():
    # Verify GitHub signature
    signature = request.headers.get('X-Hub-Signature-256', '')
    secret = os.environ['GITHUB_WEBHOOK_SECRET'].encode()
    body = request.get_data()
    expected = 'sha256=' + hmac.new(secret, body, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return jsonify({'error': 'Invalid signature'}), 403

    event_type = request.headers.get('X-GitHub-Event', '')
    payload = request.json

    handle_github_event(event_type, payload)
    return jsonify({'status': 'ok'}), 200

イベントの適切なチャンネルへのルーティング

イベントの種類ごとに送信先のチャンネルを分けます。ルーティングマップを定義し、CI/CD イベントは #deployments、エラーは #alerts、PR レビューは #engineering に送ります。コードを変更せずに設定を切り替えられるよう、チャンネル ID は環境変数に保存します。

import os

# Channel routing configuration
CHANNEL_MAP = {
    'deploy': os.environ.get('DEPLOY_CHANNEL', 'C0DEPLOY123'),
    'error': os.environ.get('ERROR_CHANNEL', 'C0ERROR456'),
    'pr_review': os.environ.get('PR_CHANNEL', 'C0PR789'),
    'general': os.environ.get('GENERAL_CHANNEL', 'C0GENERAL'),
}

def route_github_event(event_type, payload):
    if event_type == 'push':
        branch = payload.get('ref', '').replace('refs/heads/', '')
        if branch in ('main', 'master'):
            return CHANNEL_MAP['deploy']
        return CHANNEL_MAP['general']
    elif event_type == 'pull_request':
        return CHANNEL_MAP['pr_review']
    elif event_type == 'workflow_run':
        if payload.get('workflow_run', {}).get('conclusion') == 'failure':
            return CHANNEL_MAP['error']
        return CHANNEL_MAP['deploy']
    return CHANNEL_MAP['general']

# --- demo ---
push_payload = {'ref': 'refs/heads/main'}
pr_payload = {}
wf_payload = {'workflow_run': {'conclusion': 'failure'}}

print('push to main       ->', route_github_event('push', push_payload))
print('pull_request        ->', route_github_event('pull_request', pr_payload))
print('failed workflow_run ->', route_github_event('workflow_run', wf_payload))

デプロイ通知の書式設定

デプロイ通知には、何をデプロイしたか、誰がデプロイしたか、どの環境にデプロイしたか、成功したか失敗したかを表示する必要があります。Block Kit の section ブロックと context ブロックを使うと、すっきりして確認しやすい形式になります。

def build_deploy_blocks(repo, branch, commit_sha, deployer, status, env):
    status_emoji = ':white_check_mark:' if status == 'success' else ':x:'
    status_text = 'Success' if status == 'success' else 'Failed'

    blocks = [
        {
            'type': 'header',
            'text': {
                'type': 'plain_text',
                'text': f'{status_emoji} Deploy {status_text}: {repo}'
            }
        },
        {
            'type': 'section',
            'fields': [
                {'type': 'mrkdwn', 'text': f'*Repo:*\n{repo}'},
                {'type': 'mrkdwn', 'text': f'*Environment:*\n{env}'},
                {'type': 'mrkdwn', 'text': f'*Branch:*\n{branch}'},
                {'type': 'mrkdwn', 'text': f'*Deployed by:*\n{deployer}'},
                {'type': 'mrkdwn', 'text': f'*Commit:*\n`{commit_sha[:8]}`'}
            ]
        }
    ]
    return blocks

# --- demo ---
blocks = build_deploy_blocks('coddy-agents', 'main', 'a1b2c3d4e5f6', 'alice', 'success', 'production')
for b in blocks:
    print(b)

個人への DM サマリーの送信

チャンネルに一斉送信するよりも、関係する個人に ダイレクトメッセージ として送るほうが適した通知もあります。client.conversations_open(users=[user_id]) を使用して DM チャンネルを開き、返されたチャンネル ID に投稿します。

def send_dm(client, user_id, text, blocks=None):
    # Open DM channel with the user
    dm_result = client.conversations_open(users=[user_id])
    dm_channel = dm_result['channel']['id']

    # Post message to the DM channel
    msg = {'channel': dm_channel, 'text': text}
    if blocks:
        msg['blocks'] = blocks

    return client.chat_postMessage(**msg)

# Example: DM a developer when their PR build fails
def notify_pr_author_of_failure(client, pr_author_slack_id, pr_title, build_url):
    blocks = [
        {
            'type': 'section',
            'text': {
                'type': 'mrkdwn',
                'text': f'Your PR build failed: *{pr_title}*\n<{build_url}|View build logs>'
            }
        }
    ]
    send_dm(client, pr_author_slack_id, f'Build failed: {pr_title}', blocks)

# --- demo: minimal stand-in for the Slack client ---
class _FakeClient:
    def conversations_open(self, users):
        return {'channel': {'id': f'D_{users[0]}'}}
    def chat_postMessage(self, **kwargs):
        print(f"[slack DM] to {kwargs['channel']}: {kwargs['text']}")
        return {'ts': '1700000000.000200'}

notify_pr_author_of_failure(_FakeClient(), 'U_ALICE', 'Add retry logic to fetcher', 'https://ci.example.com/build/42')

スケジュールされたメッセージの配信

APScheduler を使うと、日次サマリー、週次ダイジェスト、月曜朝のブリーフィングなどの定期レポートを送信できます。cron 形式の式で特定の時刻にジョブをスケジュールします。スケジューラーはイベントハンドラーと並行してバックグラウンドスレッドで動作します。

from apscheduler.schedulers.background import BackgroundScheduler
from slack_sdk import WebClient
import os

client = WebClient(token=os.environ['SLACK_BOT_TOKEN'])

def send_daily_summary():
    stats = fetch_daily_stats()  # query your data source
    blocks = [
        {
            'type': 'header',
            'text': {'type': 'plain_text', 'text': 'Daily Team Summary'}
        },
        {
            'type': 'section',
            'text': {'type': 'mrkdwn',
                     'text': f'*PRs merged:* {stats["prs"]}\n'
                             f'*Deploys:* {stats["deploys"]}\n'
                             f'*Incidents:* {stats["incidents"]}'}
        }
    ]
    client.chat_postMessage(
        channel=os.environ['DAILY_CHANNEL'],
        text='Daily Team Summary',
        blocks=blocks
    )

scheduler = BackgroundScheduler()
scheduler.add_job(send_daily_summary, 'cron', hour=9, minute=0)
scheduler.start()
print('Scheduler started: daily summary at 9:00 AM')

アラートの集約 — 通知疲れを防ぐ

エラーが発生するたびに Slack メッセージを送信すると、すぐに通知疲れを引き起こします。アラートを集約し、一定の時間枠（例: 5 分間）に発生したエラーを収集してから、1 件のサマリーメッセージとして投稿します。カウンターを使用し、スケジュールに従って内容を送信して空にします。

import threading
import time
from collections import defaultdict

class AlertAggregator:
    def __init__(self, client, channel, flush_interval=300):
        self.client = client
        self.channel = channel
        self.flush_interval = flush_interval
        self.buffer = defaultdict(int)  # error_type -> count
        self.lock = threading.Lock()
        self._start_flusher()

    def add_alert(self, error_type):
        with self.lock:
            self.buffer[error_type] += 1

    def _flush(self):
        with self.lock:
            if not self.buffer:
                return
            lines = [f'• {err}: {count}x' for err, count in self.buffer.items()]
            self.buffer.clear()

        self.client.chat_postMessage(
            channel=self.channel,
            text=f'Alert summary ({len(lines)} error types):\n' + '\n'.join(lines)
        )

    def _start_flusher(self):
        def loop():
            while True:
                time.sleep(self.flush_interval)
                self._flush()
        threading.Thread(target=loop, daemon=True).start()

# --- demo (flush immediately instead of waiting flush_interval seconds) ---
class _FakeClient:
    def chat_postMessage(self, **kwargs):
        print(f"[slack] postMessage to {kwargs['channel']}: {kwargs['text']}")

agg = AlertAggregator(_FakeClient(), '#alerts', flush_interval=9999)
agg.add_alert('TimeoutError')
agg.add_alert('TimeoutError')
agg.add_alert('ConnectionError')
agg._flush()  # normally the background thread does this every flush_interval seconds

リッチなアラートブロックの書式設定

アラートでは、重要度をすばやく伝える必要があります。色分けしたコンテキスト、絵文字、構造化されたフィールドを使用します。ランブックやアラートダッシュボードへのアクションボタンを追加すると、オンコールのエンジニアが Slack からすぐに対応できます。

def build_incident_alert_blocks(service, error_rate, threshold,
                                 runbook_url, pagerduty_url):
    blocks = [
        {
            'type': 'header',
            'text': {'type': 'plain_text', 'text': ':rotating_light: Incident Alert'}
        },
        {
            'type': 'section',
            'text': {
                'type': 'mrkdwn',
                'text': (
                    f'*Service:* `{service}`\n'
                    f'*Error Rate:* {error_rate:.1f}% (threshold: {threshold}%)\n'
                    f'*Status:* Investigating'
                )
            }
        },
        {
            'type': 'actions',
            'elements': [
                {
                    'type': 'button',
                    'text': {'type': 'plain_text', 'text': 'View Runbook'},
                    'url': runbook_url,
                    'action_id': 'view_runbook'
                },
                {
                    'type': 'button',
                    'text': {'type': 'plain_text', 'text': 'PagerDuty'},
                    'url': pagerduty_url,
                    'style': 'danger',
                    'action_id': 'view_pagerduty'
                }
            ]
        }
    ]
    return blocks

# --- demo ---
blocks = build_incident_alert_blocks('checkout-api', 12.4, 5.0,
                                      'https://runbooks.example.com/checkout-api',
                                      'https://pagerduty.example.com/incidents/1')
for b in blocks:
    print(b)

オンコール担当者へのメンション

重大なアラートでは、オンコールのエンジニアを名前でメンションしてください。オンコールグループには Slack のユーザーグループメンション（<!subteam^SUBTEAM_ID>）を使用するか、PagerDuty／OpsGenie からオンコールユーザーを検索し、<@USER_ID> で直接メンションします。

import requests
import os

def get_oncall_slack_user():
    # Query PagerDuty for current on-call
    headers = {'Authorization': f'Token token={os.environ["PAGERDUTY_TOKEN"]}'}
    r = requests.get(
        'https://api.pagerduty.com/oncalls?include[]=users&limit=1',
        headers=headers
    )
    oncalls = r.json().get('oncalls', [])
    if not oncalls:
        return None
    email = oncalls[0]['user']['email']
    return email

def send_oncall_alert(client, channel, alert_text):
    oncall_email = get_oncall_slack_user()

    if oncall_email:
        # Look up Slack user by email
        user_result = client.users_lookupByEmail(email=oncall_email)
        user_id = user_result['user']['id']
        mention = f'<@{user_id}>'
    else:
        mention = '<!channel>'

    client.chat_postMessage(
        channel=channel,
        text=f'{mention} - CRITICAL ALERT: {alert_text}'
    )

関連するアラートのメッセージスレッド化

複数のアラートが同じインシデントに関連する場合は、元のアラートメッセージへのスレッド返信として投稿します。これによりメインチャンネルを整理された状態に保ちながら、スレッド内にアラートの履歴全体を残せます。後続のメッセージをスレッド化できるよう、元のメッセージのタイムスタンプを保存します。

class IncidentThread:
    def __init__(self, client, channel):
        self.client = client
        self.channel = channel
        self.active_incidents = {}  # service_name -> thread_ts

    def open_incident(self, service, initial_text, blocks=None):
        msg = self.client.chat_postMessage(
            channel=self.channel,
            text=initial_text,
            blocks=blocks
        )
        self.active_incidents[service] = msg['ts']
        return msg['ts']

    def update_incident(self, service, update_text):
        thread_ts = self.active_incidents.get(service)
        if thread_ts:
            self.client.chat_postMessage(
                channel=self.channel,
                thread_ts=thread_ts,
                text=update_text
            )
        else:
            self.open_incident(service, f'[New] {update_text}')

    def close_incident(self, service, resolution_text):
        thread_ts = self.active_incidents.pop(service, None)
        if thread_ts:
            self.client.chat_postMessage(
                channel=self.channel,
                thread_ts=thread_ts,
                text=f':white_check_mark: RESOLVED: {resolution_text}'
            )

# --- demo: minimal stand-in for the Slack client ---
class _FakeClient:
    def __init__(self):
        self._counter = 0
    def chat_postMessage(self, **kwargs):
        self._counter += 1
        ts = f'ts_{self._counter}'
        print(f"[slack] {kwargs.get('text')} (thread_ts={kwargs.get('thread_ts')})")
        return {'ts': ts}

thread = IncidentThread(_FakeClient(), '#incidents')
thread.open_incident('checkout-api', 'Checkout API error rate spiking')
thread.update_incident('checkout-api', 'Rolled back the last deploy')
thread.close_incident('checkout-api', 'Error rate back to normal')

通知ボットのテスト

デプロイ前に、requests.post() でテスト用 Webhook を送信し、#bot-testing チャンネルにメッセージが表示されることを確認して、通知ボットをテストします。各イベントタイプをシミュレートし、出力形式を確認するテストスクリプトを作成してください。

import requests
import json

def test_webhook(webhook_url, event_type, payload):
    response = requests.post(
        webhook_url,
        json=payload,
        headers={'X-GitHub-Event': event_type, 'Content-Type': 'application/json'}
    )
    print(f'Webhook test {event_type}: {response.status_code}')
    return response

# Test a deploy notification
test_webhook(
    webhook_url='http://localhost:3000/webhook/github',
    event_type='push',
    payload={
        'ref': 'refs/heads/main',
        'pusher': {'name': 'alice'},
        'repository': {'full_name': 'myorg/myapp'},
        'head_commit': {'id': 'abc12345', 'message': 'Fix: auth bug'}
    }
)

print('Check #bot-testing channel for the notification')

クイックチェック: エフェメラルメッセージとチャンネルメッセージ

通知のルーティングについての理解度を確認しましょう。

対話的なレッスンで答える

チーム通知ボットの振り返り

これで、チーム通知ボットを一通り構築できるようになりました:

  • Webhookエンドポイント: 外部システムからイベントを受信して検証します（署名を確認します）
  • チャンネルルーティング: 設定用dictを使って、イベントタイプを適切なSlackチャンネルに割り当てます
  • Block Kitアラート: runbookリンク付きの構造化されたヘッダー、フィールド、アクションボタンを使用します
  • DM配信: conversations_open(users=[id])を実行してから、DMチャンネルに対してchat_postMessageを実行します
  • スケジュールレポート: cron式を使った日次・週次のダイジェストにAPSchedulerを使用します
  • アラートの集約: 一定期間エラーをバッファーに蓄積し、1つのサマリーとしてまとめて送信します
  • スレッド管理: 追加のアラートをスレッドに投稿して、メインチャンネルを整理された状態に保ちます
無料で開始

AI チューターと学ぶ AI Agents — 無料

ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。

コース
60
レッスン
239
無料で学習を始めるコース全体を表示

よくある質問

「チーム通知ボットの構築」レッスンは無料ですか？

はい。「チーム通知ボットの構築」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。

「チーム通知ボットの構築」で何を学びますか？

スケジュール済みメッセージ、DMの要約、チャンネル通知エージェントを構築します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Agentsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。

「チーム通知ボットの構築」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Agentsレッスンでコードを書いて実行できますか？

はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. Slack Bolt SDKの基礎
  2. イベントとスラッシュコマンドの受信
  3. メッセージとリッチブロックの送信
  4. チーム通知ボットの構築
← AI Agentsに戻る