api-monitoring-bots

# API Monitoring Bots Build lightweight monitoring bots that poll REST APIs and send notifications when state changes (new items, price alerts, status updates). ## When to Use - User wants notifications for new listings/posts/items - Need to track price changes or threshold alerts - Monitor API for specific conditions - Watchdog for service status changes ## Architecture Pattern **Stateful polling bot:** 1. Fetch current state from API 2. Compare with last known state (stored in file) 3. Detect changes (new IDs, price deltas, status transitions) 4. Format and send notifications 5. Update state file **Key principle:** Silent when no changes (watchdog pattern, no spam). ## Implementation ### 1. Core Script Structure ```python #!/usr/bin/env python3 import requests import json from pathlib import Path from datetime import datetime API_URL = "https://api.example.com/items" STATE_FILE = Path.home() / ".hermes" / "monitor_state.json" def load_state(): """Load last seen state""" if STATE_FILE.exists(): with open(STATE_FILE) as f: return json.load(f) return {"seen_ids": [], "last_check": None} def save_state(state): """Save current state""" STATE_FILE.parent.mkdir(parents=True, exist_ok=True) with open(STATE_FILE, "w") as f: json.dump(state, f, indent=2) def fetch_items(): """Fetch current items from API""" try: r = requests.get(API_URL, timeout=10) r.raise_for_status() return r.json() except Exception as e: return None def check_new_items(): """Check for new items and return notifications""" state = load_state() items = fetch_items() if not items: return [] seen_ids = set(state["seen_ids"]) new_items = [] for item in items: if item["id"] not in seen_ids: new_items.append(item) seen_ids.add(item["id"]) # Update state (keep last 1000 IDs to prevent bloat) state["seen_ids"] = list(seen_ids)[-1000:] state["last_check"] = datetime.now().isoformat() save_state(state) return new_items # Initialize state on first run (don't notify) state = load_state() if not state["seen_ids"]: items = fetch_items() if items: state["seen_ids"] = [item["id"] for item in items] state["last_check"] = datetime.now().isoformat() save_state(state) print(f"Initialized with {len(state['seen_ids'])} existing items") # Check for new items new_items = check_new_items() if new_items: for item in new_items: print(format_notification(item)) # Silent when no new items (watchdog pattern) ``` ### 2. Hermes Cron Integration **Create cron job:** ```bash hermes cron create \ --name "api-monitor" \ --schedule "every 1m" \ --script "monitor.py" \ --no-agent \ --deliver "telegram:Username" ``` **Key flags:** - `--no-agent`: Pure script execution, no LLM cost - `--deliver "telegram:Username"`: Send to specific Telegram user/group (use `send_message list` to see targets) - `--schedule "every 1m"`: Run every minute (adjust as needed) **⚠️ CRITICAL:** `--deliver origin` does NOT work for cron jobs — it doesn't know the chat ID. Always use explicit target like `telegram:Username` or `telegram:GroupName`. **Delivery target patterns:** - `telegram:Username` — DM to specific user (use `send_message list` to see available names) - `telegram:GroupName` — Group chat (must be in available targets list) - `telegram:-1001234567890` — Chat ID (numeric, for channels/groups not in target list) - ❌ `telegram:https://t.me/channelname` — **DOES NOT WORK** (causes `invalid literal for int()` error) **Getting chat ID for channels:** 1. Forward a message from the channel to @userinfobot 2. Bot replies with chat ID (format: `-100xxxxxxxxxx`) 3. Use that numeric ID: `--deliver "telegram:-1003963927119"` **Delivery target patterns:** - `telegram:Username` — DM to specific user (use `send_message list` to see available names) - `telegram:GroupName` — Group chat (must be in available targets list) - `telegram:-1001234567890` — Chat ID (numeric, for channels/groups not in target list) - ❌ `telegram:https://t.me/channelname` — **DOES NOT WORK** (causes `invalid literal for int()` error) **Getting chat ID for channels:** 1. Forward a message from the channel to @userinfobot 2. Bot replies with chat ID (format: `-100xxxxxxxxxx`) 3. Use that numeric ID: `--deliver "telegram:-1003963927119"` **Management:** ```bash # List jobs hermes cron list # Pause monitoring hermes cron pause <job_id> # Resume monitoring hermes cron resume <job_id> # Check logs cat ~/.hermes/cron/logs/<job_id>.log ``` ### 3. Notification Formatting **Markdown for Telegram:** ```python def format_notification(item): # Detect anomalies alert = "" if item["price"] < threshold: alert = "🚨 ALERT! " return f""" {alert}**New Item Detected** 💰 **Price:** {item['price']} 📊 **Quantity:** {item['quantity']} 🆔 **ID:** `{item['id'][:8]}...` ⏰ **Created:** {item['created_at']} 🔗 https://example.com/item/{item['id']} """.strip() ``` **Alert levels:** - 🚨 Critical (price anomaly, urgent action) - ⚠️ Warning (notable but not urgent) - ℹ️ Info (normal notification) - (no emoji) Standard update ### 4. State Management **State file structure:** ```json { "seen_ids": ["id1", "id2", "..."], "last_check": "2026-05-10T13:00:00", "last_price": 0.5, "alert_count": 3 } ``` **Best practices:** - Keep last 1000 IDs max (prevent unbounded growth) - Store timestamp for debugging - Include metadata for threshold tracking - Use `.hermes/` directory for state files **State file location:** ```python STATE_FILE = Path.home() / ".hermes" / f"{bot_name}_state.json" ``` ## Patterns ### Pattern 1: New Item Detection (rpow2swap.com case study) **Use case:** Notify on every new token listing. **API:** `https://rpow2swap.com/api/listings` **Response:** ```json [ { "id": "uuid", "type": "sell", "amount": 10000, "sol_price": 10, "status": "active", "created_at": "2026-05-10 02:43:08", "solana_wallet_address": "..." } ] ``` **Detection logic:** ```python def check_new_listings(): state = load_state() listings = fetch_listings() seen_ids = set(state["seen_ids"]) new_listings = [] for listing in listings: if listing["status"] == "active" and listing["id"] not in seen_ids: new_listings.append(listing) seen_ids.add(listing["id"]) state["seen_ids"] = list(seen_ids)[-1000:] save_state(state) return new_listings ``` **Notification:** ```python def format_listing(listing): price_per_token = listing["sol_price"] / listing["amount"] alert = "" if price_per_token < 0.0001: alert = "🚨 SANGAT MURAH! " elif price_per_token < 0.0005: alert = "⚠️ Harga Menarik! " return f""" {alert}**Token Listing Baru** 💰 Amount: {listing['amount']:,} RPOW2 📊 Total: {listing['sol_price']:.4f} SOL 💵 Price/Token: {price_per_token:.6f} SOL 🆔 ID: `{listing['id'][:8]}...` 👤 Wallet: `{listing['solana_wallet_address'][:8]}...` ⏰ {listing['created_at']} 🔗 https://rpow2swap.com/ """ ``` **Cron setup:** ```bash hermes cron create \ --name "rpow2swap-monitor" \ --schedule "every 1m" \ --script "rpow2_monitor.py" \ --no-agent \ --deliver "telegram:Aboy" # Use explicit target, NOT "origin" ``` **Files:** - Script: `~/.hermes/scripts/rpow2_monitor.py` - State: `~/.hermes/rpow2_monitor_state.json` See `references/rpow2swap-monitor.md` for full implementation. ### Pattern 2: Price Threshold Alert **Use case:** Alert when price drops below threshold. ```python def check_price_alert(): state = load_state() data = fetch_price() if not data: return None current_price = data["price"] last_price = state.get("last_price", current_price) # Alert on 10% drop if current_price < last_price * 0.9: alert = f"🚨 Price dropped {((last_price - current_price) / last_price * 100):.1f}%" state["last_price"] = current_price save_state(state) return alert # Update price silently state["last_price"] = current_price save_state(state) return None ``` ### Pattern 3: Status Change Detection **Use case:** Monitor service status, alert on downtime. ```python def check_status(): state = load_state() try: r = requests.get(API_URL, timeout=5) current_status = "up" if r.status_code == 200 else "down" except: current_status = "down" last_status = state.get("last_status", "up") if current_status != last_status: state["last_status"] = current_status state["status_changed_at"] = datetime.now().isoformat() save_state(state) if current_status == "down": return "🚨 Service is DOWN!" else: return "✅ Service is back UP" return None # Silent when no change ``` ## Pitfalls - **First run spam:** Initialize state without notifying on first run - **State file bloat:** Limit stored IDs to last 1000 - **API rate limits:** Don't poll too frequently (1-5 min intervals recommended) - **Timeout handling:** Always use timeouts on API calls (10s recommended) - **Silent failures:** Log errors but don't spam user with connection failures - **Duplicate notifications:** Ensure ID-based deduplication, not timestamp-based - **State file corruption:** Use `try/except` when loading state, fallback to empty state - **Telegram delivery target format:** NEVER use `telegram:https://t.me/...` — causes `invalid literal for int()` error. Use `telegram:Username`, `telegram:GroupName`, or numeric chat ID `telegram:-1001234567890`. Get chat ID by forwarding message to @userinfobot ## Verification **Test script manually:** ```bash python3 ~/.hermes/scripts/monitor.py ``` **Simulate new items (reset state):** ```bash rm ~/.hermes/monitor_state.json python3 ~/.hermes/scripts/monitor.py ``` **Check cron job status:** ```bash hermes cron list cat ~/.hermes/cron/logs/<job_id>.log ``` **Verify state file:** ```bash cat ~/.hermes/monitor_state.json | jq '.' ``` ## Real-Time Monitoring (<1 Minute Intervals) **Problem:** Hermes cron minimum interval is 1 minute. For near-instant notifications (1-10 second checks), you need a different approach.