0Pricing
AI Agents · Lektion

Einen Benachrichtigungs-Bot für Teams erstellen

Geplante Nachrichten, DM-Zusammenfassungen und Agenten für Kanalbenachrichtigungen.

Einen Benachrichtigungs-Bot für Teams erstellen ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Architektur eines Team-Benachrichtigungs-Bots

Ein Team-Benachrichtigungs-Bot überwacht externe Systeme – Deployments, CI/CD-Pipelines, Monitoring-Warnungen und Fehlerverfolgung – und postet formatierte Aktualisierungen in die jeweils relevanten Slack-Kanäle. Das grundlegende Muster lautet: externes Ereignis → Webhook → Agent → Slack-Nachricht. Der Agent übernimmt Routing, Formatierung und Zustellung.

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

Externe Webhooks empfangen

Externe Dienste senden Ereignisse über HTTP-Webhooks an Ihren Bot. Richten Sie einen Flask-Endpunkt ein, der POST-Anfragen empfängt, sie validiert (einschließlich einer Signaturprüfung, falls der Dienst dies unterstützt) und die Payload an Ihren Benachrichtigungs-Handler weitergibt.

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

Ereignisse an den richtigen Kanal weiterleiten

Unterschiedliche Ereignistypen sollten in unterschiedliche Kanäle geleitet werden. Definieren Sie eine Routing-Tabelle: CI/CD-Ereignisse gehen an #deployments, Fehler an #alerts und PR-Reviews an #engineering. Speichern Sie Kanal-IDs in Umgebungsvariablen, damit sie ohne Codeänderungen konfiguriert werden können.

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

Deployment-Benachrichtigungen formatieren

Deployment-Benachrichtigungen müssen zeigen, was bereitgestellt wurde, wer es bereitgestellt hat, in welcher Umgebung dies geschah und ob der Vorgang erfolgreich war oder fehlgeschlagen ist. Verwenden Sie Section- und Context-Blöcke von Block Kit für ein übersichtliches, schnell erfassbares Format.

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)

Zusammenfassungen als Direktnachrichten senden

Manche Benachrichtigungen sollten besser als Direktnachricht an die zuständige Person gesendet werden, statt sie in einem Kanal zu veröffentlichen. Verwenden Sie client.conversations_open(users=[user_id]), um einen DM-Kanal zu öffnen, und posten Sie anschließend in die zurückgegebene Kanal-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')

Geplante Nachrichtenübermittlung

Verwenden Sie APScheduler, um geplante Berichte zu senden – tägliche Zusammenfassungen, wöchentliche Zusammenfassungen oder Briefings am Montagmorgen. Planen Sie Jobs mit Cron-ähnlichen Ausdrücken für bestimmte Zeitpunkte. Der Scheduler läuft neben Ihrem Event-Handler in einem Hintergrund-Thread.

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

Warnungen zusammenfassen – Benachrichtigungsmüdigkeit vermeiden

Wenn für jeden einzelnen Fehler eine Slack-Nachricht gesendet wird, führt das schnell zu Benachrichtigungsmüdigkeit. Fassen Sie Warnungen zusammen: Sammeln Sie Fehler über einen bestimmten Zeitraum (z. B. 5 Minuten) und posten Sie anschließend eine einzelne Zusammenfassungsnachricht. Verwenden Sie einen Zähler und leeren Sie ihn nach einem festen Zeitplan.

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

Umfangreiche Warnungsblöcke formatieren

Warnungen müssen den Schweregrad schnell vermitteln. Verwenden Sie farbcodierte Context-Blöcke, Emojis und strukturierte Felder. Fügen Sie eine Aktionsschaltfläche hinzu, die auf das Runbook oder das Warnungs-Dashboard verweist, damit Bereitschaftstechniker direkt aus Slack handeln können.

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)

Erwähnung des Bereitschaftstechnikers

Kritische Warnungen sollten den Bereitschaftstechniker namentlich erwähnen. Verwenden Sie Slacks Erwähnungen von Benutzergruppen (<!subteam^SUBTEAM_ID>) für Bereitschaftsgruppen oder ermitteln Sie den zuständigen Benutzer über PagerDuty/OpsGenie und erwähnen Sie ihn direkt mit <@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}'
    )

Nachrichten-Threads für zusammengehörige Warnungen

Wenn sich mehrere Warnungen auf denselben Vorfall beziehen, posten Sie sie als Antworten in einem Thread zur ursprünglichen Warnungsnachricht. So bleibt der Hauptkanal übersichtlich, während der vollständige Verlauf der Warnungen im Thread erhalten bleibt. Speichern Sie den Zeitstempel der ursprünglichen Nachricht, um Folgebeiträge im Thread zu veröffentlichen.

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

Den Benachrichtigungs-Bot testen

Testen Sie Ihren Benachrichtigungs-Bot vor der Bereitstellung, indem Sie Test-Webhooks mit requests.post() senden und überprüfen, ob Nachrichten in einem #bot-testing-Kanal erscheinen. Schreiben Sie ein Testskript, das jeden Ereignistyp simuliert und das Ausgabeformat überprüft.

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

Kurztest: Ephemere Nachrichten vs. Channel-Nachrichten

Testen Sie Ihr Verständnis der Benachrichtigungsweiterleitung.

Zusammenfassung: Team-Benachrichtigungs-Bot

Sie können jetzt einen vollständigen Team-Benachrichtigungs-Bot erstellen:

  • Webhook-Endpunkt: Ereignisse von externen Systemen empfangen und validieren (Signaturen überprüfen)
  • Channel-Routing: Ereignistypen mithilfe eines Konfigurations-Dictionaries den richtigen Slack-Channels zuordnen
  • Block-Kit-Alarme: strukturierter Header mit Feldern und Aktionsschaltflächen mit Runbook-Links
  • DM-Zustellung: zuerst conversations_open(users=[id]), anschließend chat_postMessage an den DM-Channel senden
  • Geplante Berichte: APScheduler mit Cron-Ausdrücken für tägliche oder wöchentliche Zusammenfassungen
  • Alarmaggregation: Fehler über einen Zeitraum sammeln und als eine einzige Zusammenfassung ausgeben
  • Thread-Verwaltung: Folgealarme in Threads veröffentlichen, damit der Haupt-Channel übersichtlich bleibt

Häufig gestellte Fragen

Ist die Lektion „Einen Benachrichtigungs-Bot für Teams erstellen“ kostenlos?

Ja — der vollständige Text von „Einen Benachrichtigungs-Bot für Teams erstellen“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Einen Benachrichtigungs-Bot für Teams erstellen“?

Geplante Nachrichten, DM-Zusammenfassungen und Agenten für Kanalbenachrichtigungen. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Agents zu starten?

Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Einen Benachrichtigungs-Bot für Teams erstellen“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?

Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Grundlagen des Slack Bolt SDK
  2. Auf Ereignisse und Slash Commands reagieren
  3. Nachrichten und Rich Blocks senden
  4. Einen Benachrichtigungs-Bot für Teams erstellen
← Zurück zu AI Agents