A 0-token event-driven notification framework for OpenClaw agents.
Status: Proof-of-concept / Userland workaround. This exists because OpenClaw currently lacks native event-driven hooks. When native support lands (see Related Issues), this framework can be retired or adapted as a plugin.
OpenClaw agents have no native way to proactively notify users when events occur. All existing mechanisms are poll-based:
| Mechanism | Model | Latency | Token cost |
|---|---|---|---|
cron (systemEvent) |
Poll every N min | Up to N min | Burns tokens when idle |
heartbeat |
Poll on schedule | Minutes | Burns tokens when idle |
sessions_spawn announce |
Push (completion only) | Instant | Only fires once |
When you have multiple notification sources (cost monitoring, task completion, error alerts, health checks), you need:
- Rate limiting — don't spam the user
- Deduplication — don't repeat unchanged status
- Quiet hours — batch non-urgent notifications during sleep
- Immediate alerts — bypass quiet hours for critical events
- Digest batching — combine multiple events into one message
┌─────────────────┐ ┌──────────────┐ ┌─────────────────┐
│ Producer │ │ Event Inbox │ │ Event Router │
│ (any script) │────▶│ (JSONL) │────▶│ (Python) │
│ │ │ append-only │ │ │
│ cost-monitor │ └──────────────┘ │ ┌───────────┐ │
│ health-check │ │ │ Dedupe │ │
│ task-complete │ │ │ Rate-limit│ │
│ custom script │ │ │ Quiet hrs │ │
│ │ │ │ Batching │ │
└─────────────────┘ │ └───────────┘ │
│ │ │
│ ▼ │
│ ┌───────────┐ │
│ │ Telegram │ │
│ │ (direct) │ │
│ └───────────┘ │
└─────────────────┘
- 0-token: Router is a pure Python script — no LLM calls, no token consumption
- Append-only inbox: Producers just write JSON lines; no coordination needed
- Dual-cursor architecture: Separate cursors for immediate alerts vs. digest batching
- Fingerprint-based dedup: Same event with same data won't re-notify
- Three output levels:
- L1: Short notification for humans (Telegram message)
- L2: Detailed debug info (CLI
detailcommand) - L3: Model-assisted analysis (only when user explicitly requests)
# Copy files to your OpenClaw host
cp events-router.py ~/.openclaw/tools/
cp events.example.json ~/.openclaw/config/events.jsonEdit events.json:
{
"version": 1,
"defaults": {
"quietHours": { "start": "23:00", "end": "08:00" },
"digestEverySeconds": 600,
"minIntervalSeconds": 300,
"alertMinIntervalSeconds": 300
},
"sources": [{
"id": "default",
"type": "jsonl",
"path": "~/.openclaw/events/inbox.jsonl",
"enabled": true
}]
}python3 events-router.py emit \
--code "COST/ALERT" \
--title "Balance Low" \
--level alert \
--summary "SiliconFlow balance dropped below ¥10"Or write directly to inbox:
import json, datetime
event = {
"ts": datetime.datetime.now().isoformat(),
"level": "info",
"code": "TASK/DONE",
"title": "Sub-agent completed",
"summary": "architect finished system design (3 files, 2min)"
}
with open("~/.openclaw/events/inbox.jsonl", "a") as f:
f.write(json.dumps(event) + "\n")# Dry run (print only, don't send)
python3 events-router.py tick --dry-run --print-only
# Actually send
python3 events-router.py tick
# Schedule with cron/launchd for periodic ticking{
"ts": "2026-02-07T16:30:00+08:00",
"level": "info|progress|warn|alert",
"code": "COST/HOURLY",
"title": "Hourly Cost Report",
"summary": "Total spend today: $2.34",
"details": ["DeepSeek: $1.20", "SiliconFlow: $0.89"],
"action": "No action needed",
"dedupeKey": "COST/HOURLY",
"fingerprint": "spend=2.34",
"actionRequired": false
}Required: ts, level, code, title
Recommended: summary, details, dedupeKey, fingerprint
# Validate configuration
python3 events-router.py validate-config
# Self-test
python3 events-router.py selftest
# Run one routing tick
python3 events-router.py tick [--print-only] [--dry-run]
# Emit a test event
python3 events-router.py emit --code "TEST/HELLO" --title "Test" --level info --summary "..."
# View router status
python3 events-router.py status [--json]
# View event details (L2)
python3 events-router.py detail [--code CODE] [--topic TOPIC]
# Acknowledge/snooze an event stream
python3 events-router.py ack --code "OPS/STALL" --for-seconds 28800level=alertoractionRequired=true→ sent immediately- Subject to rate limiting (configurable
alertMinIntervalSeconds) - Not affected by quiet hours
level=info|progress|warn→ batched into periodic digest- Sent every
digestEverySeconds(default 600s) - Suppressed during quiet hours (accumulated, sent when quiet hours end)
- Events with same
dedupeKey+ unchangedfingerprint→ skipped - Prevents "nothing changed" spam
- Still poll-based: Router must be ticked externally (cron/launchd); not truly event-driven
- Outside OpenClaw: Cannot access session state, agent context, or native message routing
- Telegram only: Notification channel is hardcoded to Telegram API
Native OpenClaw event hooks would make this framework unnecessary:
message:received/message:transcribedhooks (PR #9859, #7545, #9387)session:spawn/session:completed/session:errorevents (#8995)- Built-in notification routing with rate-limiting and quiet hours
When these land, this framework's policy engine (dedupe, rate-limit, quiet hours, batching) could be ported as a native OpenClaw plugin.
| # | Title | Status |
|---|---|---|
| #9859 | message:received and message:transcribed hooks |
PR Open |
| #7545 | message:received hook for pre-turn automation |
PR Open |
| #9387 | Bridge plugin message hooks to internal hooks | PR Open |
| #8995 | Spawn Verification & Worker Health Monitoring | Open |
| #5541 | Agent activity visibility and lifecycle tracking | Closed |
- Python 3.10+
TELEGRAM_BOT_TOKENenvironment variable (for sending notifications)- Telegram chat ID configured in events.json or environment
MIT