webhook-push

Webhook Push Skill

概要

このスキルは、中国の主要な企業向けコミュニケーションプラットフォーム3つにWebhook通知を送信するための統一インターフェースを提供します。

• WeCom (企业微信) - Tencentの企業向けWeChat
• DingTalk (钉钉) - Alibabaの企業向けコミュニケーションプラットフォーム
• Feishu (飞书) - ByteDanceのコラボレーションプラットフォーム

このスキルを使用する場面

次のような場合にこのスキルを使用します。

• 企業チャットグループに通知を送信する
• 監視アラートを企業向けプラットフォームに統合する
• 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）
• フィード - 複数のリンクカード（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アクションカード（ボタン1つ）

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

プラットフォーム設定

Enterprise WeChat

# 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/分
DingTalk	20/分
Feishu	約100/分

送信側でレート制限エラーを自動的に処理します。

高度な機能

カスタム再試行ポリシー

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 のドキュメント