Slack Webhook Integration

Send alerts and notifications to Slack using the configured webhook.

Quick Start

The webhook URL is automatically loaded from the SLACK_WEBHOOK_URL environment variable.

import os import json from urllib.request import Request, urlopen

def send_slack_alert(message: dict) -> bool: """Send a message to Slack via webhook.

Args:
    message: Slack Block Kit message payload

Returns:
    True if sent successfully, False otherwise
"""
webhook_url = os.environ.get('SLACK_WEBHOOK_URL')
if not webhook_url:
    raise ValueError("SLACK_WEBHOOK_URL environment variable not set")

request = Request(
    webhook_url,
    data=json.dumps(message).encode('utf-8'),
    headers={'Content-Type': 'application/json'}
)
with urlopen(request, timeout=10) as response:
    return response.status == 200

Message Formats

Simple Text Message

message = { "text": "Hello from the agent!" } send_slack_alert(message)

Alert with Block Kit (Recommended)

message = { "blocks": [ { "type": "header", "text": {"type": "plain_text", "text": "🔴 Anomaly Detected"} }, { "type": "section", "fields": [ {"type": "mrkdwn", "text": "Metric:\nConversion Rate"}, {"type": "mrkdwn", "text": "Severity:\nCritical"} ] }, { "type": "section", "fields": [ {"type": "mrkdwn", "text": "Expected:\n4.2%"}, {"type": "mrkdwn", "text": "Actual:\n1.8%"} ] }, {"type": "divider"}, { "type": "section", "text": { "type": "mrkdwn", "text": "Potential Causes:\n• Checkout flow issue\n• Payment gateway error\n• Landing page test impact" } }, { "type": "context", "elements": [ {"type": "mrkdwn", "text": "🤖 Alert generated by AI Agent"} ] } ] }

Anomaly Alert Template

def build_anomaly_alert( metric_name: str, date: str, expected: float, actual: float, severity: str = "Warning", potential_causes: list = None ) -> dict: """Build a formatted anomaly alert message.

Args:
    metric_name: Name of the metric (e.g., "Conversion Rate")
    date: Date of the anomaly
    expected: Expected value
    actual: Actual value
    severity: "Critical" or "Warning"
    potential_causes: List of potential cause strings

Returns:
    Slack Block Kit message payload
"""
emoji = "🔴" if severity == "Critical" else "🟡"
pct_change = ((actual - expected) / expected) * 100
direction = "drop" if pct_change < 0 else "spike"

blocks = [
    {
        "type": "header",
        "text": {"type": "plain_text", "text": f"{emoji} Anomaly Detected: {metric_name}"}
    },
    {
        "type": "section",
        "fields": [
            {"type": "mrkdwn", "text": f"*Date:*\n{date}"},
            {"type": "mrkdwn", "text": f"*Severity:*\n{severity}"}
        ]
    },
    {
        "type": "section",
        "fields": [
            {"type": "mrkdwn", "text": f"*Expected:*\n{expected:,.2f}"},
            {"type": "mrkdwn", "text": f"*Actual:*\n{actual:,.2f}"}
        ]
    },
    {
        "type": "section",
        "text": {
            "type": "mrkdwn",
            "text": f"*Change:* {pct_change:+.1f}% {direction}"
        }
    }
]

if potential_causes:
    causes_text = "\n".join([f"• {cause}" for cause in potential_causes])
    blocks.extend([
        {"type": "divider"},
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"*Potential Causes:*\n{causes_text}"
            }
        }
    ])

blocks.append({
    "type": "context",
    "elements": [
        {"type": "mrkdwn", "text": "🤖 Alert generated by AI Agent"}
    ]
})

return {"blocks": blocks}

Usage Example

Build and send an anomaly alert

alert = build_anomaly_alert( metric_name="Trial Conversion Rate", date="2025-11-28", expected=37.6, actual=10.5, severity="Critical", potential_causes=[ "Onboarding flow broken", "Integration setup failing", "New user segment with different behavior" ] )

success = send_slack_alert(alert) if success: print("✅ Alert sent to Slack") else: print("❌ Failed to send alert")

Testing

To test without sending (dry run):

import json print(json.dumps(alert, indent=2))

Error Handling

from urllib.error import URLError, HTTPError

try: send_slack_alert(message) except ValueError as e: print(f"Configuration error: {e}") except HTTPError as e: print(f"Slack API error: {e.code} - {e.reason}") except URLError as e: print(f"Network error: {e.reason}")