6. Webhooks and Incoming Messages
6. Webhooks and Incoming Messages
# webhooks.py
# ABOUTME: Incoming webhook integration for external services
# ABOUTME: CI/CD notifications, alerts, and external triggers
import requests
import json
from typing import Optional, List, Dict
import hmac
import hashlib
import time
class SlackWebhook:
"""Incoming webhook client for Slack"""
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
def send(
self,
text: str,
blocks: Optional[List[Dict]] = None,
attachments: Optional[List[Dict]] = None,
thread_ts: Optional[str] = None,
unfurl_links: bool = True,
unfurl_media: bool = True
) -> dict:
"""Send a message via webhook"""
payload = {
"text": text,
"unfurl_links": unfurl_links,
"unfurl_media": unfurl_media
}
if blocks:
payload["blocks"] = blocks
if attachments:
payload["attachments"] = attachments
if thread_ts:
payload["thread_ts"] = thread_ts
response = requests.post(
self.webhook_url,
json=payload,
headers={"Content-Type": "application/json"}
)
response.raise_for_status()
return {"ok": True, "status": response.status_code}
def send_deployment_notification(
self,
app_name: str,
environment: str,
version: str,
status: str,
commit_sha: str,
author: str,
url: Optional[str] = None
):
"""Send a deployment notification"""
color_map = {
"success": "#36a64f",
"failure": "#ff0000",
"started": "#ffcc00",
"pending": "#808080"
}
status_emoji = {
"success": ":white_check_mark:",
"failure": ":x:",
"started": ":rocket:",
"pending": ":hourglass:"
}
blocks = [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{status_emoji.get(status, ':grey_question:')} Deployment {status.title()}: {app_name}",
"emoji": True
}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Environment:*\n{environment}"},
{"type": "mrkdwn", "text": f"*Version:*\n{version}"},
{"type": "mrkdwn", "text": f"*Commit:*\n`{commit_sha[:8]}`"},
{"type": "mrkdwn", "text": f"*Author:*\n{author}"}
]
}
]
if url:
blocks.append({
"type": "actions",
"elements": [
{
"type": "button",
"text": {"type": "plain_text", "text": "View Deployment"},
"url": url,
"style": "primary" if status == "success" else None
}
]
})
attachments = [
{
"color": color_map.get(status, "#808080"),
"blocks": blocks
}
]
return self.send(
text=f"Deployment {status}: {app_name} to {environment}",
attachments=attachments
)
def send_alert(
self,
title: str,
message: str,
severity: str = "warning",
source: str = "System",
details: Optional[Dict] = None
):
"""Send an alert notification"""
severity_config = {
"critical": {"emoji": ":rotating_light:", "color": "#ff0000"},
"error": {"emoji": ":x:", "color": "#ff4444"},
"warning": {"emoji": ":warning:", "color": "#ffcc00"},
"info": {"emoji": ":information_source:", "color": "#0088ff"}
}
config = severity_config.get(severity, severity_config["info"])
blocks = [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"{config['emoji']} {title}",
"emoji": True
}
},
{
"type": "section",
"text": {"type": "mrkdwn", "text": message}
},
{
"type": "context",
"elements": [
{"type": "mrkdwn", "text": f"*Source:* {source} | *Severity:* {severity.upper()}"}
]
}
]
if details:
detail_text = "\n".join(f"*{k}:* {v}" for k, v in details.items())
blocks.insert(2, {
"type": "section",
"text": {"type": "mrkdwn", "text": detail_text}
})
return self.send(
text=f"[{severity.upper()}] {title}",
attachments=[{"color": config["color"], "blocks": blocks}]
)
# Webhook signature verification
def verify_slack_signature(
signing_secret: str,
request_body: str,
timestamp: str,
signature: str
) -> bool:
"""Verify Slack request signature"""
# Check timestamp to prevent replay attacks
*Content truncated — see parent skill for full reference.*
vamseeachanta