/api-monitoring-bots

name: api-monitoring-bots
description: Build monitoring bots that poll APIs and send notifications on state changes (new listings, price alerts, status updates)
tags: [monitoring, cron, notifications, polling, alerts]
related_skills: [web-scraping, hermes-agent]
triggers:
  - monitor API for changes
  - notify when new listing
  - alert on price change
  - track API state
  - polling bot
  - watchdog script
origin: unknown
source_license: see upstream
language: en

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

#!/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:**

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:**

# 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:**

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:**

{
  "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