0Pricing
AI Agents · 강의

팀 알림 봇 만들기

예약 메시지, DM 요약, 채널 알림 에이전트를 만듭니다.

팀 알림 봇 만들기은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

팀 알림 봇 아키텍처

팀 알림 봇은 배포, CI/CD 파이프라인, 모니터링 알림, 오류 추적과 같은 외부 시스템을 모니터링하고 관련 Slack 채널에 서식이 지정된 업데이트를 게시합니다. 핵심 패턴은 외부 이벤트 → 웹훅 → 에이전트 → 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')

외부 웹훅 수신하기

외부 서비스는 HTTP 웹훅을 통해 봇에 이벤트를 보냅니다. 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를 사용해 일일 요약, 주간 정리 또는 월요일 아침 브리핑과 같은 예약 보고서를 보내십시오. 크론 형식 표현식을 사용해 특정 시간에 작업을 예약합니다. 스케줄러는 이벤트 처리기와 함께 백그라운드 스레드에서 실행됩니다.

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분), 요약 메시지 하나를 게시합니다. 카운터를 사용하고 일정에 따라 내용을 비우십시오.

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

풍부한 알림 블록 서식 지정

알림은 심각도를 빠르게 전달해야 합니다. 색상으로 구분한 context, 이모지, 구조화된 필드를 사용하십시오. 런북 또는 알림 대시보드로 연결되는 작업 버튼을 추가하면 당번 엔지니어가 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()로 테스트 웹훅을 보내고 #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')

빠른 확인: 임시 메시지와 채널 메시지

알림 라우팅에 대한 이해도를 확인해 보십시오.

팀 알림 봇 복습

이제 완전한 팀 알림 봇을 구축할 수 있습니다.

  • 웹훅 엔드포인트: 외부 시스템의 이벤트를 수신하고 검증합니다(서명을 확인합니다).
  • 채널 라우팅: 설정 딕셔너리를 통해 이벤트 유형을 적절한 Slack 채널에 매핑합니다.
  • 블록 키트 알림: 런북 링크가 포함된 구조화된 헤더, 필드, 작업 버튼을 사용합니다.
  • DM 전달: conversations_open(users=[id])을 실행한 다음 DM 채널에 chat_postMessage를 실행합니다.
  • 예약 보고서: 일일 및 주간 요약을 위해 cron 표현식과 APScheduler를 사용합니다.
  • 알림 집계: 일정 시간 동안 오류를 버퍼링한 다음 하나의 요약으로 한꺼번에 전송합니다.
  • 스레드 관리: 후속 알림을 스레드에 추가하여 기본 채널을 깔끔하게 유지합니다.

자주 묻는 질문

“팀 알림 봇 만들기” 강의는 무료인가요?

네 — “팀 알림 봇 만들기” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

“팀 알림 봇 만들기”에서 뭘 배우나요?

예약 메시지, DM 요약, 채널 알림 에이전트를 만듭니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 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(으)로 돌아가기