Build an AI ChatOps Incident Bot for Slack with Claude API

pip install slack-bolt anthropic requests

import os
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
import anthropic
 
app = App(token=os.environ["SLACK_BOT_TOKEN"])
client = anthropic.Anthropic()
 
SERVICE_OWNERS = {
    "api": {"team": "platform", "pagerduty": "PXXXXX", "runbook": "https://wiki.internal/runbooks/api"},
    "checkout": {"team": "payments", "pagerduty": "PYYYYY", "runbook": "https://wiki.internal/runbooks/checkout"},
    "auth": {"team": "identity", "pagerduty": "PZZZZZ", "runbook": "https://wiki.internal/runbooks/auth"},
}
 
 
@app.command("/incident")
def handle_incident(ack, respond, command, client):
    ack()
    description = command["text"]
 
    service = identify_service(description)
    owner_info = SERVICE_OWNERS.get(service, {})
 
    # Create dedicated incident channel
    channel_name = f"incident-{service}-{int(__import__('time').time())}"
    channel = client.conversations_create(name=channel_name)
    channel_id = channel["channel"]["id"]
 
    # Pull recent metrics/logs context if available
    context = gather_context(service)
 
    assessment = get_ai_assessment(description, service, context)
 
    client.chat_postMessage(
        channel=channel_id,
        text=f"*Incident Assessment*\n\n{assessment}\n\n"
             f"*Owning team:* {owner_info.get('team', 'unknown')}\n"
             f"*Runbook:* {owner_info.get('runbook', 'not found')}\n"
    )
 
    # Invite the right people
    if owner_info.get("team"):
        invite_team(client, channel_id, owner_info["team"])
        page_oncall(owner_info.get("pagerduty"), description)
 
    respond(f"Incident channel created: <#{channel_id}>")
 
 
def identify_service(description: str) -> str:
    """Simple keyword match — swap for embeddings-based routing at scale."""
    description_lower = description.lower()
    for service in SERVICE_OWNERS:
        if service in description_lower:
            return service
    return "unknown"

ASSESSMENT_PROMPT = """A DevOps engineer just reported this incident:
"{description}"
 
Recent context (metrics/logs, may be empty):
{context}
 
Provide a brief initial assessment for the incident channel:
1. **Likely severity** (SEV1/SEV2/SEV3) with one-line reasoning
2. **First 3 things to check** — specific, not generic ("check CPU" is bad,
   "check pod restart count with kubectl get pods -n {{service}}" is good)
3. **Similar past incidents this resembles** (based on the description pattern, if any)
 
Keep it under 150 words. This is going into a live incident channel — be direct and actionable, not cautious."""
 
 
def get_ai_assessment(description: str, service: str, context: dict) -> str:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": ASSESSMENT_PROMPT.format(description=description, context=context)
        }]
    )
    return response.content[0].text
 
 
def gather_context(service: str) -> dict:
    """Pull last 15 min of error rate / restart count if metrics available."""
    try:
        import requests
        error_rate = requests.get(
            f"http://prometheus:9090/api/v1/query",
            params={"query": f'rate(http_requests_total{{service="{service}",status=~"5.."}}[15m])'}
        ).json()
        return {"error_rate_query": error_rate}
    except Exception:
        return {"note": "metrics unavailable, assess from description alone"}

def invite_team(client, channel_id: str, team: str):
    members = TEAM_MEMBERS.get(team, [])
    for user_id in members:
        client.conversations_invite(channel=channel_id, users=user_id)
 
 
def page_oncall(pagerduty_service_id: str, description: str):
    if not pagerduty_service_id:
        return
    import requests
    requests.post(
        "https://api.pagerduty.com/incidents",
        headers={"Authorization": f"Token token={os.environ['PAGERDUTY_TOKEN']}"},
        json={
            "incident": {
                "type": "incident",
                "title": description[:100],
                "service": {"id": pagerduty_service_id, "type": "service_reference"},
            }
        }
    )

@app.event("app_mention")
def handle_update_request(event, client):
    """Reply to '@bot summarize' with a status recap pulled from channel history."""
    if "summarize" not in event["text"].lower():
        return
 
    history = client.conversations_history(channel=event["channel"], limit=50)
    messages = "\n".join(m.get("text", "") for m in reversed(history["messages"]))
 
    summary = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": f"Summarize this incident channel's current status in 3 bullet points, "
                       f"suitable for posting to a #status-updates channel:\n\n{messages}"
        }]
    ).content[0].text
 
    client.chat_postMessage(channel=event["channel"], thread_ts=event["ts"], text=summary)