Build an AI Slack Incident Classifier with Claude API

pip install slack-bolt anthropic python-dotenv

# .env
SLACK_BOT_TOKEN=xoxb-your-bot-token
SLACK_APP_TOKEN=xapp-your-app-token   # only for Socket Mode
ANTHROPIC_API_KEY=sk-ant-your-key
INCIDENTS_CHANNEL_ID=C0123456789
SUMMARY_CHANNEL_ID=C9876543210

CLASSIFICATION_PROMPT = """You are an incident classifier for a DevOps team.
 
Analyze this Slack message from an incidents channel and return a JSON object with exactly these fields:
- severity: "P1", "P2", or "P3"
  P1 = customer-facing outage or data loss, needs immediate response
  P2 = degraded performance or partial outage, affects customers but workaround exists
  P3 = internal issue, no customer impact, can wait until business hours
- service: the affected service/system name (string, "unknown" if not mentioned)
- incident_type: one of "infra", "app", "security", "network"
  infra = servers, clusters, nodes, databases down
  app = application error, deployment failure, bug in production
  security = auth failure, unusual access, credential leak, vulnerability
  network = DNS, routing, load balancer, connectivity issues
- summary: one sentence describing the incident (max 120 chars)
- confidence: "high", "medium", or "low" (how confident you are in this classification)
 
Return ONLY valid JSON. No explanation. No markdown.
 
Message: {message}"""

import json
import os
import logging
from dotenv import load_dotenv
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
import anthropic
 
load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
app = App(token=os.environ["SLACK_BOT_TOKEN"])
claude = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
 
INCIDENTS_CHANNEL = os.environ["INCIDENTS_CHANNEL_ID"]
SUMMARY_CHANNEL = os.environ["SUMMARY_CHANNEL_ID"]
 
SEVERITY_EMOJI = {"P1": ":red_circle:", "P2": ":yellow_circle:", "P3": ":white_circle:"}
TYPE_EMOJI = {"infra": ":computer:", "app": ":bug:", "security": ":lock:", "network": ":globe_with_meridians:"}
 
 
def classify_incident(message_text: str) -> dict:
    prompt = CLASSIFICATION_PROMPT.format(message=message_text)
 
    response = claude.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=256,
        messages=[{"role": "user", "content": prompt}],
    )
 
    raw = response.content[0].text.strip()
    return json.loads(raw)
 
 
def format_summary_block(classification: dict, original_text: str, user: str, channel: str) -> list:
    severity = classification.get("severity", "P3")
    service = classification.get("service", "unknown")
    inc_type = classification.get("incident_type", "app")
    summary = classification.get("summary", original_text[:120])
    confidence = classification.get("confidence", "medium")
 
    sev_emoji = SEVERITY_EMOJI.get(severity, ":white_circle:")
    type_emoji = TYPE_EMOJI.get(inc_type, ":bell:")
 
    return [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": f"{sev_emoji} {severity} Incident Detected"},
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*Service:*\n{service}"},
                {"type": "mrkdwn", "text": f"*Type:*\n{type_emoji} {inc_type}"},
                {"type": "mrkdwn", "text": f"*Reported by:*\n<@{user}>"},
                {"type": "mrkdwn", "text": f"*Channel:*\n<#{channel}>"},
            ],
        },
        {"type": "section", "text": {"type": "mrkdwn", "text": f"*Summary:*\n{summary}"}},
        {
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": f"Classification confidence: {confidence} | Model: claude-haiku-4-5"}],
        },
        {"type": "divider"},
    ]
 
 
@app.event("message")
def handle_message(event, say, client):
    channel_id = event.get("channel")
    text = event.get("text", "")
    user = event.get("user", "unknown")
    subtype = event.get("subtype")
 
    # Only process messages from the incidents channel, skip bot messages
    if channel_id != INCIDENTS_CHANNEL or subtype == "bot_message" or not text:
        return
 
    # Skip very short messages (reactions, acks)
    if len(text) < 20:
        return
 
    logger.info(f"Classifying message from {user}: {text[:80]}...")
 
    try:
        classification = classify_incident(text)
        blocks = format_summary_block(classification, text, user, channel_id)
 
        client.chat_postMessage(
            channel=SUMMARY_CHANNEL,
            blocks=blocks,
            text=f"{classification.get('severity', 'P3')} incident: {classification.get('summary', text[:80])}",
        )
 
        logger.info(f"Posted classification: {classification}")
 
    except json.JSONDecodeError as e:
        logger.error(f"Claude returned invalid JSON: {e}")
    except Exception as e:
        logger.error(f"Classification failed: {e}")
 
 
if __name__ == "__main__":
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    handler.start()

FROM python:3.12-slim
 
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY . .
 
CMD ["python", "app.py"]

pip install slack-bolt[aws-lambda]

from slack_bolt.adapter.aws_lambda import SlackRequestHandler
 
def lambda_handler(event, context):
    slack_handler = SlackRequestHandler(app=app)
    return slack_handler.handle(event, context)