PricingLoginStart free
AI Agents · Lekcja

Tworzenie bota do powiadomień zespołowych

Wiadomości planowane, podsumowania w wiadomościach prywatnych i agenci alertów kanałowych.

Lekcja 4 z 413 kroki

Tworzenie bota do powiadomień zespołowych to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.

Architektura bota powiadomień zespołu

Bot powiadomień zespołu monitoruje systemy zewnętrzne — wdrożenia, potoki CI/CD, alerty monitoringu i śledzenie błędów — oraz publikuje sformatowane aktualizacje na odpowiednich kanałach Slacka. Główny schemat to: zdarzenie zewnętrzne → webhook → agent → wiadomość Slack. Agent obsługuje kierowanie, formatowanie i dostarczanie wiadomości.

# 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')

Odbieranie zewnętrznych webhooków

Usługi zewnętrzne wysyłają zdarzenia do Państwa bota za pośrednictwem webhooków HTTP. Należy skonfigurować endpoint Flask, który odbiera żądania POST, weryfikuje je (sprawdzając podpis, jeśli dana usługa to obsługuje), a następnie przekazuje payload do modułu obsługi powiadomień.

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

Kierowanie zdarzeń do właściwego kanału

Różne typy zdarzeń powinny trafiać do różnych kanałów. Należy zdefiniować mapę routingu: zdarzenia CI/CD trafiają do #deployments, błędy do #alerts, a przeglądy PR do #engineering. Identyfikatory kanałów należy przechowywać w zmiennych środowiskowych, aby można było je konfigurować bez zmian w kodzie.

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))

Formatowanie powiadomień o wdrożeniach

Powiadomienia o wdrożeniach powinny zawierać informacje: co wdrożono, kto wykonał wdrożenie, do jakiego środowiska oraz czy zakończyło się ono powodzeniem, czy niepowodzeniem. Należy użyć bloków section i context Block Kit, aby uzyskać przejrzysty format, który można szybko przeskanować wzrokiem.

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)

Wysyłanie podsumowań w wiadomościach bezpośrednich

Niektóre powiadomienia lepiej wysłać jako wiadomość bezpośrednią do odpowiedniej osoby, zamiast publikować je na kanale. Należy użyć client.conversations_open(users=[user_id]), aby otworzyć kanał wiadomości bezpośrednich, a następnie opublikować wiadomość, korzystając ze zwróconego identyfikatora kanału.

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')

Planowane dostarczanie wiadomości

APScheduler służy do wysyłania zaplanowanych raportów — podsumowań dziennych, zestawień tygodniowych lub poniedziałkowych porannych odpraw. Zadania należy planować na konkretne godziny za pomocą wyrażeń w stylu cron. Harmonogram działa w wątku działającym w tle obok modułu obsługi zdarzeń.

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')

Agregowanie alertów — unikanie zmęczenia powiadomieniami

Wysyłanie wiadomości Slack dla każdego pojedynczego błędu szybko prowadzi do zmęczenia powiadomieniami. Należy agregować alerty: zbierać błędy w określonym przedziale czasu (np. 5 minut), a następnie publikować jedno zbiorcze podsumowanie. Należy użyć licznika i opróżniać go zgodnie z harmonogramem.

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

Formatowanie rozbudowanych bloków alertów

Alerty powinny umożliwiać szybkie rozpoznanie poziomu ważności. Należy używać oznaczeń kontekstowych z kodowaniem kolorami, emoji oraz uporządkowanych pól. Warto dodać przycisk działania prowadzący do runbooka lub panelu alertów, aby inżynierowie dyżurni mogli natychmiast zareagować z poziomu Slacka.

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)

Wzmianka o osobie pełniącej dyżur

Alerty krytyczne powinny zawierać wzmiankę o inżynierze pełniącym dyżur. Należy używać wzmianek o grupach użytkowników Slacka (<!subteam^SUBTEAM_ID>) w przypadku grup dyżurnych albo pobrać osobę pełniącą dyżur z PagerDuty/OpsGenie i wspomnieć o niej bezpośrednio za pomocą <@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}'
    )

Wątki wiadomości dla powiązanych alertów

Gdy wiele alertów dotyczy tego samego incydentu, należy publikować je jako odpowiedzi w wątku do pierwotnej wiadomości z alertem. Dzięki temu główny kanał pozostaje uporządkowany, a pełna historia alertów jest zachowana w wątku. Należy zapisać znacznik czasu pierwotnej wiadomości, aby dodawać kolejne odpowiedzi w jej wątku.

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')

Testowanie bota powiadomień

Przed wdrożeniem należy przetestować bota powiadomień, wysyłając testowe webhooki za pomocą requests.post() i sprawdzając, czy wiadomości pojawiają się na kanale #bot-testing. Należy napisać skrypt testowy, który symuluje każdy typ zdarzenia i sprawdza format danych wyjściowych.

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')

Szybki test: wiadomości efemeryczne a wiadomości kanałowe

Proszę sprawdzić swoją wiedzę na temat routingu powiadomień.

Odpowiedz w interaktywnej lekcji

Podsumowanie: bot powiadomień zespołowych

Można już zbudować kompletnego bota do powiadomień zespołowych:

  • Punkt końcowy webhooka: odbieranie i weryfikowanie zdarzeń z systemów zewnętrznych (weryfikacja podpisów)
  • Routing kanałów: przypisywanie typów zdarzeń do właściwych kanałów Slacka za pomocą słownika konfiguracji
  • Alerty Block Kit: uporządkowany nagłówek, pola i przyciski akcji z odnośnikami do procedur operacyjnych
  • Dostarczanie wiadomości prywatnych: najpierw conversations_open(users=[id]), a następnie chat_postMessage do kanału wiadomości prywatnych
  • Raporty cykliczne: APScheduler z wyrażeniami cron do tworzenia dziennych i tygodniowych podsumowań
  • Agregowanie alertów: buforowanie błędów w określonym przedziale czasu i wysyłanie ich jako jednego podsumowania
  • Zarządzanie wątkami: umieszczanie kolejnych alertów w wątkach, aby zachować porządek na głównym kanale
Bezpłatny start

Ucz się AI Agents dzięki korepetycjom AI — za darmo

Pisz i uruchamiaj kod w przeglądarce, otrzymuj natychmiastową pomoc od korepetytora AI dostępnego 24/7 i kontynuuj naukę w sieci lub w aplikacji.

Kursy
60
Lekcje
239
Zacznij naukę bezpłatnieWyświetl pełny kurs

Często zadawane pytania

Czy lekcja „Tworzenie bota do powiadomień zespołowych” jest bezpłatna?

Tak — pełny tekst „Tworzenie bota do powiadomień zespołowych” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.

Co nauczysz się w „Tworzenie bota do powiadomień zespołowych”?

Wiadomości planowane, podsumowania w wiadomościach prywatnych i agenci alertów kanałowych. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Agents?

Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.

Ile czasu zajmuje lekcja „Tworzenie bota do powiadomień zespołowych”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?

Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

  1. Podstawy Slack Bolt SDK
  2. Nasłuchiwanie zdarzeń i poleceń slash
  3. Wysyłanie wiadomości i rozbudowanych bloków
  4. Tworzenie bota do powiadomień zespołowych
← Powrót do AI Agents