Creazione di un bot per le notifiche del team
Messaggi pianificati, riepiloghi tramite DM e agenti per gli avvisi nei canali
Creazione di un bot per le notifiche del team è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.
Architettura di un bot per le notifiche del team
Un bot per le notifiche del team monitora sistemi esterni, come deploy, pipeline CI/CD, avvisi di monitoraggio e monitoraggio degli errori, e pubblica aggiornamenti formattati nei canali Slack pertinenti. Il pattern di base è: evento esterno → webhook → agente → messaggio Slack. L'agente gestisce l'instradamento, la formattazione e la consegna.
# 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')Ricezione di webhook esterni
I servizi esterni inviano eventi al Suo bot tramite webhook HTTP. Configuri un endpoint Flask che riceva richieste POST, le convalidi, verificando la firma se il servizio lo supporta, e passi il payload al gestore delle notifiche.
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'}), 200Instradamento degli eventi al canale corretto
I diversi tipi di evento devono essere inviati a canali diversi. Definisca una mappa di instradamento: gli eventi CI/CD vanno in #deployments, gli errori in #alerts e le revisioni delle PR in #engineering. Memorizzi gli ID dei canali nelle variabili d'ambiente per poterli configurare senza modificare il codice.
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))
Formattazione delle notifiche dei deploy
Le notifiche dei deploy devono indicare che cosa è stato distribuito, chi ha eseguito il deploy, in quale ambiente e se l'operazione è riuscita o non è riuscita. Utilizzi sezioni e blocchi context di Block Kit per ottenere un formato chiaro e facilmente consultabile.
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)
Invio di riepiloghi DM ai singoli utenti
Alcune notifiche è preferibile inviarle come messaggio diretto alla persona interessata anziché pubblicarle in un canale. Utilizzi client.conversations_open(users=[user_id]) per aprire un canale DM, quindi pubblichi il messaggio nell'ID del canale restituito.
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')
Consegna programmata dei messaggi
Utilizzi APScheduler per inviare report programmati: riepiloghi giornalieri, digest settimanali o briefing del lunedì mattina. Pianifichi i job a orari specifici con espressioni in stile cron. Il pianificatore viene eseguito in un thread in background insieme al gestore degli eventi.
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')Aggregazione degli avvisi — Evitare l'affaticamento da notifiche
Inviare un messaggio Slack per ogni singolo errore causa rapidamente un affaticamento da notifiche. Aggrergi gli avvisi: raccolga gli errori in un intervallo di tempo, ad esempio 5 minuti, quindi pubblichi un unico messaggio riepilogativo. Utilizzi un contatore e lo azzeri secondo una pianificazione.
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
Formattazione di blocchi di avviso avanzati
Gli avvisi devono comunicare rapidamente la gravità. Utilizzi contesti con codifica a colori, emoji e campi strutturati. Aggiunga un pulsante di azione che rimandi al runbook o alla dashboard degli avvisi, così gli ingegneri on-call possono intervenire immediatamente da 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)
Menzione della rotazione on-call
Gli avvisi critici devono menzionare per nome l'ingegnere on-call. Utilizzi le menzioni dei gruppi di utenti Slack (<!subteam^SUBTEAM_ID>) per i gruppi on-call oppure recuperi l'utente on-call da PagerDuty/OpsGenie e lo menzioni direttamente con <@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}'
)Thread dei messaggi per avvisi correlati
Quando più avvisi riguardano lo stesso incidente, li pubblichi come risposte in un thread al messaggio di avviso originale. In questo modo il canale principale rimane ordinato, preservando al contempo la cronologia completa degli avvisi nel thread. Memorizzi il timestamp del messaggio originale per collegare i messaggi successivi al thread.
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')
Test del bot per le notifiche
Testi il bot per le notifiche prima del deploy inviando webhook di prova con requests.post() e verificando che i messaggi compaiano nel canale #bot-testing. Scriva uno script di test che simuli ogni tipo di evento e controlli il formato dell'output.
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')Verifica rapida: messaggi effimeri e messaggi nei canali
Verifichi la Sua comprensione dell'instradamento delle notifiche.
Riepilogo del bot di notifica del team
Ora è in grado di creare un bot completo per le notifiche del team:
- Endpoint webhook: ricevere e convalidare eventi da sistemi esterni (verificando le firme)
- Instradamento nei canali: associare i tipi di evento ai canali Slack corretti tramite un dizionario di configurazione
- Avvisi Block Kit: intestazione strutturata + campi + pulsanti di azione con collegamenti alle procedure operative
- Consegna tramite DM:
conversations_open(users=[id])quindichat_postMessagenel canale DM - Report pianificati: APScheduler con espressioni cron per riepiloghi giornalieri o settimanali
- Aggregazione degli avvisi: memorizzare gli errori in un intervallo di tempo e inviarli come un unico riepilogo
- Gestione dei thread: inserire gli avvisi di follow-up in un thread per mantenere pulito il canale principale
Domande Frequenti
La lezione «Creazione di un bot per le notifiche del team» è gratuita?
Sì — il testo completo di «Creazione di un bot per le notifiche del team» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.
Cosa imparerò in «Creazione di un bot per le notifiche del team»?
Messaggi pianificati, riepiloghi tramite DM e agenti per gli avvisi nei canali Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.
Ho bisogno di esperienza per iniziare AI Agents?
Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.
Quanto tempo richiede la lezione «Creazione di un bot per le notifiche del team»?
La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.
Posso scrivere ed eseguire codice in questa lezione AI Agents?
Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.
Tutte le lezioni di questo corso
- Nozioni di base di Slack Bolt SDK
- Ascolto degli eventi e dei comandi slash
- Invio di messaggi e blocchi avanzati
- Creazione di un bot per le notifiche del team