Stage 7 · Master

ChatOps & On-Call Copilots

Slack & Teams Bots

Event APIs, slash commands, and streaming responses into a channel.

7 min readAI for SRE WorkflowsMaster

Bot Architecture

A ChatOps bot listens for messages in Slack or Teams, sends them to an LLM, and posts the response back to the channel. The bot needs to handle event webhooks, manage conversation context, and stream long responses so users see progress.

Basic bot structure


from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

app = App(token="xoxb-your-bot-token")

@app.message("ask-ops")
def handle_mention(message, say):
    """Handle @bot mentions in channels."""
    user_query = extract_query(message["text"])
    response = rag_answer(user_query)
    say(response)

@app.command("/ops")
def handle_slash_command(ack, command, say):
    """Handle /ops slash commands."""
    ack()  # Must acknowledge within 3 seconds
    user_query = command["text"]
    response = rag_answer(user_query)
    say(response)

if __name__ == "__main__":
    handler = SocketModeHandler(app, "xapp-your-app-token")
    handler.start()

Slack Bolt handles the boilerplate — authentication, event routing, and message sending. You focus on the AI logic.

Slack Event API

The Slack Event API sends webhooks when events occur in your workspace — new messages, reactions, channel changes. Your bot subscribes to specific events and processes them.

Listening for mentions in incident channels


import re

@app.event("app_mention")
def handle_app_mention(event, say, client):
    """Respond when someone mentions the bot in a channel."""
    channel = event["channel"]
    user = event["user"]
    text = event["text"]

    # Clean the mention from the text
    clean_text = re.sub(r"<@[A-Z0-9]+>", "", text).strip()

    # Check if this is an incident channel
    is_incident = channel.startswith("C_INCIDENT")

    if is_incident:
        # Use incident-aware context
        response = incident_copilot_answer(clean_text, channel)
    else:
        response = rag_answer(clean_text)

    say(
        text=response,
        thread_ts=event.get("ts"),  # Reply in thread
        reply_broadcast=False
    )

Thread replies keep incident channels clean. The bot responds in a thread rather than posting a new message.

Slash Commands

Slash commands give users a clear, intentional way to invoke the bot. They are explicit — users must type /ops — which prevents accidental triggers and makes the bot's capabilities discoverable.

Multiple slash commands


@app.command("/ops-triage")
def triage_command(ack, command, say):
    """Classify an alert."""
    ack()
    alert_text = command["text"]
    classification = classify_alert(alert_text)
    say(f"Severity: {classification['severity']}\nReason: {classification['reason']}")

@app.command("/ops-runbook")
def runbook_command(ack, command, say):
    """Look up a runbook procedure."""
    ack()
    query = command["text"]
    result = search_runbook(query)
    say(result)

@app.command("/ops-status")
def status_command(ack, command, say):
    """Get current system status."""
    ack()
    status = get_system_status()
    say(format_status(status))

Multiple commands let users choose the exact type of assistance they need. Each command can have its own validation and response format.

Streaming Responses

Long LLM responses can take 10 to 30 seconds. Streaming shows the response as it generates, improving the user experience. Slack supports message updates, so you can start with a placeholder and update it as the response streams in.

Streaming with message updates


import openai

def stream_response(say, initial_message, user_query):
    """Stream LLM response into a Slack channel."""
    # Post initial placeholder
    response = say(text="Thinking...")

    # Stream the LLM response
    full_response = ""
    for chunk in openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_query}],
        stream=True
    ):
        delta = chunk["choices"][0].get("delta", {})
        if "content" in delta:
            full_response += delta["content"]
            # Update message every 100 chars to avoid rate limits
            if len(full_response) % 100 < 10:
                app.client.chat_update(
                    ts=response["ts"],
                    channel=response["channel"],
                    text=full_response
                )

    # Final update with complete response
    app.client.chat_update(
        ts=response["ts"],
        channel=response["channel"],
        text=full_response
    )

Update the message periodically rather than on every token to avoid Slack API rate limits.

Use ephemeral messages for debugging

Send verbose debug info as ephemeral messages visible only to the invoking user. This keeps the channel clean while giving the requester detailed output.

Error Handling

Always acknowledge slash commands within 3 seconds or Slack retries.
Handle LLM timeouts gracefully with a fallback message.
Catch tool execution errors and report them without exposing internals.
Rate-limit per user to prevent abuse.

Never expose API keys in error messages

If the LLM API fails, show a generic error to the user. Log the full error details server-side for debugging.

Mark this lesson complete to store local progress and unlock a cleaner resume path the next time you visit.