slack-webhook（Slack webhook 前端）

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