AI Agents · Lección

Creación de un bot de notificaciones para equipos

Mensajes programados, resúmenes por mensaje directo y agentes de alertas para canales.

Lección 4 de 413 pasos

Creación de un bot de notificaciones para equipos es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.

Arquitectura de un bot de notificaciones de equipo

Un bot de notificaciones de equipo supervisa sistemas externos —implementaciones, canalizaciones de CI/CD, alertas de monitorización y seguimiento de errores— y publica actualizaciones con formato en los canales de Slack correspondientes. El patrón principal es: evento externo → webhook → agente → mensaje de Slack. El agente se encarga del enrutamiento, el formato y la entrega.

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

Recepción de webhooks externos

Los servicios externos envían eventos a su bot mediante webhooks HTTP. Configure un endpoint de Flask que reciba solicitudes POST, las valide (compruebe la firma si el servicio lo admite) y pase el payload a su controlador de notificaciones.

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

Enrutamiento de eventos al canal adecuado

Los distintos tipos de eventos deben dirigirse a canales diferentes. Defina un mapa de enrutamiento: los eventos de CI/CD van a #deployments, los errores a #alerts y las revisiones de PR a #engineering. Almacene los ID de los canales en variables de entorno para poder configurarlos sin cambiar el código.

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

Formato de las notificaciones de implementación

Las notificaciones de implementación deben indicar qué se implementó, quién lo hizo, en qué entorno y si la operación tuvo éxito o falló. Use secciones de Block Kit y bloques context para conseguir un formato limpio y fácil de consultar.

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)

Envío de resúmenes por mensaje directo

Algunas notificaciones se envían mejor como un mensaje directo a la persona correspondiente que como un anuncio en un canal. Use client.conversations_open(users=[user_id]) para abrir un canal de mensajes directos y, después, publique en el ID del canal devuelto.

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

Envío programado de mensajes

Use APScheduler para enviar informes programados: resúmenes diarios, recopilaciones semanales o informes de los lunes por la mañana. Programe los trabajos a horas específicas mediante expresiones con formato cron. El planificador se ejecuta en un hilo en segundo plano junto con el controlador de eventos.

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

Agregación de alertas: cómo evitar la fatiga por notificaciones

Enviar un mensaje de Slack por cada error provoca rápidamente fatiga por notificaciones. Agregue las alertas: recopile los errores durante un intervalo de tiempo (por ejemplo, 5 minutos) y, después, publique un único mensaje de resumen. Use un contador y vacíelo según un horario.

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

Formato de bloques de alerta enriquecidos

Las alertas deben transmitir rápidamente la gravedad. Use elementos context con colores, emojis y campos estructurados. Añada un botón de acción que enlace al runbook o al panel de alertas para que los ingenieros de guardia puedan actuar de inmediato desde 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)

Mención de la rotación de guardia

Las alertas críticas deben mencionar por su nombre al ingeniero de guardia. Use las menciones de grupos de usuarios de Slack (<!subteam^SUBTEAM_ID>) para los grupos de guardia, o busque al usuario de guardia en PagerDuty/OpsGenie y menciónelo directamente 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}'
    )

Hilos de mensajes para alertas relacionadas

Cuando varias alertas estén relacionadas con el mismo incidente, publíquelas como respuestas en un hilo al mensaje de alerta original. Así mantendrá limpio el canal principal y conservará todo el historial de alertas en el hilo. Almacene la marca de tiempo del mensaje original para añadir seguimientos al hilo.

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

Prueba del bot de notificaciones

Pruebe el bot de notificaciones antes de implementarlo: envíe webhooks de prueba con requests.post() y compruebe que los mensajes aparecen en un canal #bot-testing. Escriba un script de prueba que simule cada tipo de evento y compruebe el formato de salida.

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

Comprobación rápida: mensajes efímeros frente a mensajes de canal

Compruebe su comprensión del enrutamiento de notificaciones.

Responde en la lección interactiva

Resumen del bot de notificaciones del equipo

Ahora puede crear un bot completo de notificaciones para equipos:

Endpoint de webhook: recibe y valida eventos de sistemas externos (verifica las firmas)
Enrutamiento por canal: asigna tipos de eventos a los canales de Slack correspondientes mediante un diccionario de configuración
Alertas de Block Kit: encabezado estructurado, campos y botones de acción con enlaces a guías operativas
Entrega por mensaje directo: conversations_open(users=[id]) y luego chat_postMessage en el canal de mensajes directos
Informes programados: APScheduler con expresiones cron para resúmenes diarios o semanales
Agregación de alertas: almacena los errores en un búfer durante un intervalo de tiempo y los envía como un único resumen
Gestión de hilos: publica las alertas de seguimiento en hilos para mantener despejado el canal principal

Gratis para empezar

Aprende AI Agents con un tutor de IA — gratis

Escribe y ejecuta código real en tu navegador, obtén ayuda instantánea de un tutor de IA disponible 24/7 y continúa donde lo dejaste en la web o en la aplicación.

Cursos: 60
Lecciones: 239

Empezar a aprender gratis Ver el curso completo

Preguntas frecuentes

¿La lección «Creación de un bot de notificaciones para equipos» es gratis?

Sí — el texto completo de «Creación de un bot de notificaciones para equipos» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.

¿Qué aprenderé en «Creación de un bot de notificaciones para equipos»?

Mensajes programados, resúmenes por mensaje directo y agentes de alertas para canales. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Agents?

No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.

¿Cuánto tiempo toma la lección «Creación de un bot de notificaciones para equipos»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Agents?

Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

Fundamentos del SDK Slack Bolt
Escucha de eventos y comandos de barra
Envío de mensajes y bloques enriquecidos
Creación de un bot de notificaciones para equipos

← Volver a AI Agents