webhook-push

Webhook 推送技能

概述

此技能提供统一接口，可向三大中国企业通信平台发送 webhook 通知：

• WeCom（企业微信） - 腾讯旗下的企业微信
• DingTalk（钉钉） - 阿里巴巴旗下的企业通信平台
• Feishu（飞书） - 字节跳动旗下的协作平台

何时使用此技能

在以下情况下使用此技能：

• 向企业聊天群发送通知
• 将监控告警集成到企业平台
• 自动发送 CI/CD 流水线通知
• 构建事件管理工作流
• 向团队频道发送报告和更新
• 创建多平台通知系统

核心功能

1. 统一消息模型

使用单一的平台无关接口发送消息：

from webhook_push import UnifiedMessage, MessageSender

message = UnifiedMessage(
    content={
        "type": "text",
        "body": {"text": "Deployment completed successfully!"}
    }
)

sender = MessageSender()
result = await sender.send(message, "dingtalk", webhook_url="https://oapi.dingtalk.com/robot/send?access_token=xxx")

2. 丰富的消息类型

支持多种消息类型：

• 文本 - 可包含 @提及的简单文本消息
• Markdown - 格式丰富的消息
• 图片 - 发送图片（WeCom、Feishu）
• 链接 - 链接卡片（DingTalk）
• 卡片 - 带按钮的交互式卡片
• 文件 - 文件附件（WeCom、Feishu）
• Feed - 多个链接卡片（DingTalk）

3. 自动平台转换

消息会自动转换为平台专属格式：

# Same message works across platforms
message = UnifiedMessage(
    content={
        "type": "markdown",
        "title": "Daily Report",
        "body": {
            "content": "# Daily Stats\n- Users: 128\n- Revenue: $5,000"
        }
    }
)

# Send to all platforms automatically
result = await sender.send_auto(message)

4. 平滑降级

不支持的功能会自动降级：

• 卡片 → 带链接的 Markdown
• Markdown → 纯文本
• 丰富功能 → 简单替代方案

5. 重试机制

内置指数退避重试逻辑，可处理暂时性故障。

消息类型

文本消息

message = UnifiedMessage(
    content={
        "type": "text",
        "body": {"text": "Hello, team!"},
        "mentions": [
            {"type": "mobile", "value": "13800000000", "display_name": "Zhang San"}
        ]
    }
)

Markdown 消息

message = UnifiedMessage(
    content={
        "type": "markdown",
        "title": "Deployment Report",
        "body": {
            "content": """# Deployment Complete

## Status: ✅ Success
- Version: v1.2.3
- Environment: Production
- Deployed by: @zhangsan

[View Logs](https://example.com/logs)"""
        }
    }
)

卡片消息

卡片消息提供带按钮和丰富内容的交互元素。每个平台都有自己的卡片格式：

WeCom 模板卡片（text_notice）

from webhook_push import UnifiedMessage, MessageContent

message = UnifiedMessage(
    content=MessageContent(
        type="card",
        body={
            "card_type": "text_notice",
            "title": "系统告警",
            "description": "CPU使用率超过阈值",
            "emphasis": {
                "title": "95%",
                "desc": "当前CPU使用率"
            },
            "horizontal_content_list": [
                {"keyname": "服务器", "value": "web-01"},
                {"keyname": "阈值", "value": "80%"}
            ],
            "jump_list": [
                {"type": 1, "url": "https://example.com/alerts", "title": "查看详情"}
            ],
            "action": {"type": 1, "url": "https://example.com/alerts"}
        }
    )
)

result = await sender.send(message, "wecom")

WeCom 图文消息（文章卡片）

message = UnifiedMessage(
    content={
        "type": "news",
        "body": {
            "links": [
                {
                    "title": "技术分享: Webhook 最佳实践",
                    "description": "了解如何设计可靠的 webhook 系统",
                    "url": "https://example.com/article",
                    "image_url": "https://example.com/cover.jpg"
                }
            ]
        }
    }
)

result = await sender.send(message, "wecom")

DingTalk 操作卡片（单个按钮）

from webhook_push import UnifiedMessage, MessageContent

message = UnifiedMessage(
    content=MessageContent(
        type="card",
        body={
            "card_type": "single",
            "config": {"hide_avatar": "0"},
            "title": "审批请求",
            "elements": [{"type": "div", "text": "您有一个新的审批请求待处理"}],
            "actions": [
                {"text": "立即审批", "url": "https://example.com/approve", "style": "positive"}
            ]
        }
    )
)

result = await sender.send(message, "dingtalk")

DingTalk 操作卡片（多个按钮）

message = UnifiedMessage(
    content=MessageContent(
        type="card",
        body={
            "card_type": "multi",
            "config": {"hide_avatar": "0", "btn_orientation": "1"},
            "title": "满意度调查",
            "elements": [{"type": "div", "text": "请对本次服务进行评价"}],
            "actions": [
                {"text": "非常满意", "url": "https://example.com/survey/1", "style": "positive"},
                {"text": "满意", "url": "https://example.com/survey/2", "style": "default"},
                {"text": "不满意", "url": "https://example.com/survey/3", "style": "default"}
            ]
        }
    )
)

result = await sender.send(message, "dingtalk")

Feishu 交互式卡片

message = UnifiedMessage(
    content={
        "type": "card",
        "body": {
            "config": {
                "wide_screen_mode": True,
                "enable_forward": True
            },
            "elements": [
                {
                    "tag": "div",
                    "text": {"tag": "lark_md", "content": "**审批提醒**\n您有一个新的待审批事项"}
                }
            ],
            "actions": [
                {
                    "tag": "button",
                    "text": {"tag": "plain_text", "content": "立即审批"},
                    "url": "https://example.com/approve",
                    "type": "primary"
                }
            ]
        }
    }
)

result = await sender.send(message, "feishu")

图片消息

message = UnifiedMessage(
    content={
        "type": "image",
        "body": {
            "url": "https://example.com/screenshot.png",
            "alt": "System dashboard screenshot"
        }
    }
)

使用模式

基本模式

from webhook_push import MessageSender, UnifiedMessage

# Create message
message = UnifiedMessage(
    content={
        "type": "text",
        "body": {"text": "Your notification message"}
    }
)

# Send to specific platform
sender = MessageSender()
result = await sender.send(
    message,
    "dingtalk",
    webhook_url="https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"
)

if result.success:
    print("Sent!")
else:
    print(f"Failed: {result.error}")

多平台

# Send to multiple platforms
result = await sender.send_multi(
    message,
    platforms=["wecom", "dingtalk", "feishu"],
    webhook_urls={
        "wecom": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=XXX",
        "dingtalk": "https://oapi.dingtalk.com/robot/send?access_token=YYY",
        "feishu": "https://open.feishu.cn/open-apis/bot/v2/hook/ZZZ"
    }
)

print(f"Success: {result.success_count}, Failed: {result.failed_count}")

自动选择

# Send to all configured platforms
result = await sender.send_auto(message)

print(f"Sent to: {result.sent_platforms}")
print(f"Skipped: {result.skipped_platforms}")

平台配置

企业微信

# Via webhook URL
webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"

message = UnifiedMessage(content={"type": "text", "body": {"text": "Hello WeCom!"}})
await sender.send(message, "wecom", webhook_url)

消息类型：text、markdown、markdown_v2、image、news、file、voice、template_card

DingTalk

# Via webhook URL
webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"

message = UnifiedMessage(content={"type": "markdown", "body": {"content": "Hello DingTalk!"}})
await sender.send(message, "dingtalk", webhook_url)

消息类型：text、markdown、link、actionCard、feedCard

注意：支持签名验证以增强安全性

Feishu

# Via webhook URL
webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/YOUR_ID"

# With signature verification (recommended)
from webhook_push import FeishuAdapter

adapter = FeishuAdapter(
    webhook_id="YOUR_ID",
    secret="YOUR_SECRET"  # Enable signature verification
)

message = UnifiedMessage(content={"type": "text", "body": {"text": "Hello Feishu!"}})
await sender.send(message, "feishu", webhook_url)

消息类型：text、post、image、file、card、audio

注意：支持 HMAC-SHA256 签名验证，可增强安全性（推荐）

环境变量

# Optional: Configure default platforms
WECOM_WEBHOOK_KEY=your-key
DINGTALK_ACCESS_TOKEN=your-token
DINGTALK_SECRET=your-secret
FEISHU_WEBHOOK_ID=your-id
FEISHU_SECRET=your-secret  # Optional, for signature verification

CLI 用法

# Send a text message
webhook-push send dingtalk "https://oapi.dingtalk.com/robot/send?access_token=xxx" \
    --content "Hello from CLI!"

# Send a markdown message
webhook-push send wecom "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx" \
    --type markdown \
    --title "Report" \
    --content "# Daily Report\n- Metrics here"

# Send to multiple platforms
webhook-push send-multi wecom dingtalk --content "Notification"

# Send to all available platforms
webhook-push send-auto --content "All platforms"

# Show platform info
webhook-push info dingtalk

最佳实践

1. 使用 Markdown 展示丰富内容

Markdown 在各平台上具有最佳兼容性：

message = UnifiedMessage(
    content={
        "type": "markdown",
        "body": {
            "content": """# Report

## Summary
- Metric 1: **128**
- Metric 2: 3,421

> Last updated: 18:00"""
        }
    }
)

2. 妥善处理故障

result = await sender.send(message, "dingtalk", webhook_url)

if not result.success:
    if result.retry_suggested:
        # Retry later or queue for retry
        await retry_queue.add(message)
    else:
        # Log the error
        logger.error(f"Send failed: {result.error}")

3. 使用卡片消息执行操作

带按钮的卡片可提供更好的用户体验：

message = UnifiedMessage(
    content={
        "type": "card",
        "body": {
            "card_type": "interactive",
            "elements": [{"type": "div", "text": "Review required"}],
            "actions": [
                {"type": "button", "text": "Approve", "url": "...", "style": "primary"},
                {"type": "button", "text": "Reject", "url": "..."}
            ]
        }
    }
)

4. 监控速率限制

各平台的速率限制不同：

平台	速率限制
WeCom	20/min
DingTalk	20/min
Feishu	~100/min

发送器会自动处理速率限制错误。

高级功能

自定义重试策略

from webhook_push import MessageSender, RetryPolicy, SenderOptions

options = SenderOptions(
    retry_policy=RetryPolicy(
        max_retries=5,
        initial_delay=1000,
        max_delay=30000,
        backoff_multiplier=2.0
    )
)

sender = MessageSender(options)

消息去重

使用 message_id 进行去重：

message = UnifiedMessage(
    metadata={"message_id": "unique-id-123"},
    content={"type": "text", "body": {"text": "Same message"}}
)

关联追踪

使用 correlation_id 追踪相关消息：

message = UnifiedMessage(
    metadata={"correlation_id": "deployment-123"},
    content={"type": "markdown", "body": {"content": "Deployment update"}}
)

错误处理

常见错误代码：

代码	含义	操作
UNKNOWN_PLATFORM	平台名称无效	检查平台名称
UNSUPPORTED	平台不支持此消息类型	使用其他消息类型
RATE_LIMIT	请求过多	等待后重试
NETWORK_ERROR	网络故障	稍后重试
PLATFORM_ERROR	平台返回错误	查看错误消息

性能提示

1. 发送到多个平台时，使用 send_multi
2. 使用 send_auto 以利用所有可用平台
3. 为长时间运行的操作配置超时
4. 在高频场景中监控速率限制

依赖项

• Python 3.9+
• httpx（HTTP 客户端）
• pydantic（数据验证）
• tenacity（重试逻辑）

完整依赖项请参见 pyproject.toml。

参考资料

• 设计文档
• 统一消息设计
• API 参考
• 平台文档
• WeCom 文档
• DingTalk 文档
• Feishu 文档