leap-feishu-webhook-notify

Async Python notifications for Lark/Feishu custom bot webhooks.

The PyPI distribution is leap-feishu-webhook-notify. The import package remains feishu_notify.

Features

• Async httpx transport with reusable connections.
• Lark/Feishu custom bot HMAC-SHA256 signing.
• CardKit v2 interactive cards.
• Built-in templates for lifecycle events, alerts, heartbeats, pings, and period reports.
• Payload splitting for large cards.
• Single-target and multi-target webhook fan-out.
• Runtime reconfiguration and reset support.
• Failure-safe sends: network and API failures are logged and returned as False, not raised.

Installation

pip install leap-feishu-webhook-notify

With uv:

uv add leap-feishu-webhook-notify

Quick Start

import asyncio
import os

from feishu_notify import FeishuBot, ServiceContext


async def main() -> None:
    bot = FeishuBot(
        webhook_url=os.environ["FEISHU_WEBHOOK_URL"],
        secret=os.environ.get("FEISHU_WEBHOOK_SECRET", ""),
        context=ServiceContext(service="orders-api", env="prod"),
    )
    try:
        ok = await bot.send_heartbeat(message="notification channel is ready")
        print(ok)
    finally:
        await bot.aclose()


asyncio.run(main())

Alert Cards

await bot.send_warning(
    title="Payment provider timeout",
    message="The provider did not respond after 3 retries.",
    error_message="TimeoutError: request timed out",
    trace_id="trace-123",
    details={"provider": "example-pay", "retry": 3},
)

Set optional URLs on ServiceContext to render action buttons:

context = ServiceContext(
    service="orders-api",
    env="prod",
    host="worker-1",
    log_url_template="https://logs.example.com/trace/{trace_id}",
    commit_url_template="https://git.example.com/repo/commit/{commit}",
    dashboard_url="https://metrics.example.com/d/orders",
)

Multi-target Webhooks

Use webhook_urls when the same message should be sent to several groups.

bot = FeishuBot(
    webhook_urls=[
        os.environ["FEISHU_PRIMARY_WEBHOOK"],
        os.environ["FEISHU_ONCALL_WEBHOOK"],
    ],
    secrets=[
        os.environ.get("FEISHU_PRIMARY_SECRET", ""),
        os.environ.get("FEISHU_ONCALL_SECRET", ""),
    ],
    context=ServiceContext(service="orders-api", env="prod"),
)

result = await bot.send_error(
    title="Database unavailable",
    message="All replicas failed the health check.",
    error_message="ConnectionError: no healthy upstream",
)

oks = result if isinstance(result, list) else [result]
failed_count = oks.count(False)

Return values:

Runtime Reconfiguration

reconfigure() updates the active webhook(s) in memory. It does not persist anything and does not read environment variables.

bot.reconfigure(webhook_url=new_url, secret=new_secret)

bot.reconfigure(
    webhook_urls=[primary_url, oncall_url],
    secrets=[primary_secret, oncall_secret],
)

reset() restores the webhook(s) and secret(s) passed to __init__.

if override_url:
    bot.reconfigure(webhook_url=override_url, secret=override_secret)
else:
    bot.reset()

BotManager

Use BotManager when different messages should go to different bots.

from feishu_notify import BotManager

manager = BotManager()
manager.register(ops_bot)
manager.register(alerts_bot)

await manager.broadcast_via("send_startup")

BotManager.broadcast*() folds a multi-target bot result with all(...), so each bot name maps to one success boolean.

Period Reports

Any object implementing the PeriodSummary protocol can be sent as a report.

from dataclasses import dataclass, field

from feishu_notify import FailureItem, StatGroup


@dataclass
class DailySummary:
    period_label: str = "2026-06-09"
    period_type: str = "daily"
    total_counts: dict[str, int] = field(default_factory=lambda: {"ok": 120, "failed": 2})
    groups: list[StatGroup] = field(default_factory=list)
    failures: list[FailureItem] = field(default_factory=list)
    prev_total_counts: dict[str, int] | None = None


await bot.send_period_report(summary=DailySummary())

Low-level Card API

Use feishu_notify.primitives to build custom CardKit v2 payloads.

from feishu_notify import primitives

payload = primitives.card(
    "Custom report",
    "blue",
    [
        primitives.md("**Status**: OK"),
        primitives.hr(),
        primitives.action_button("Open dashboard", "https://metrics.example.com"),
    ],
)

await bot.send(payload)

Failure Behavior

All send methods are designed to be safe in application code:

Avoid putting webhook URLs, secrets, tokens, passwords, or other sensitive data inside card message, details, or logs.

Development

uv sync --extra dev
uv run ruff check src tests
uv run pyright src
uv run pytest tests/unit/ --cov=feishu_notify --cov-report=term --cov-fail-under=85
uv run python -m compileall -q src/

Real webhook smoke tests are skipped unless these environment variables are set:

• FEISHU_TEST_WEBHOOK
• FEISHU_TEST_SECRET
• FEISHU_TEST_WEBHOOK_LIST

Run them explicitly:

uv run pytest tests/integration/ -v

License

MIT