pollers

Pollers — Event-Driven Monitoring

Pollers are lightweight scripts that check external services on a schedule and report back when something needs attention. They live inside skills and are discovered automatically by the scheduler.

How It Works

1. A skill includes a pollers.json file alongside its SKILL.md
2. The scheduler discovers all pollers.json files at startup and when reload_pollers is called
3. On each cron tick, the scheduler runs the poller command as a subprocess
4. Each line of stdout is parsed as JSON and delivered to the agent as an event
5. If there's nothing to report, the poller outputs nothing — silence is the filter

Creating a Poller

1. Write the poller script

The script runs in the skill directory. It receives these environment variables automatically:

Variable	Description
STATE_DIR	The skill directory (writable, for cursors/state)
POLLER_NAME	The poller's name from pollers.json

Plus any custom env vars from the env field, plus the agent's existing environment.

Output contract:

• stdout: JSONL (one JSON object per line). Each line must have poller (string) and prompt (string) fields.
• stderr: Free-form logging. Not forwarded to the agent.
• Exit 0: Success. Non-zero: Error, this cycle is skipped.

Example poller script:

#!/usr/bin/env python3
"""Check for new items since last poll."""
import json, os, sys
from pathlib import Path

STATE_DIR = Path(os.environ.get("STATE_DIR", "."))
CURSOR_FILE = STATE_DIR / "cursor.json"

def load_cursor():
    if CURSOR_FILE.exists():
        return json.loads(CURSOR_FILE.read_text())
    return {}

def save_cursor(cursor):
    CURSOR_FILE.write_text(json.dumps(cursor, indent=2))

def main():
    cursor = load_cursor()
    # ... check your service, compare against cursor ...

    new_items = []  # your logic here

    for item in new_items:
        event = {
            "poller": os.environ.get("POLLER_NAME", "my-poller"),
            "prompt": f"New item: {item['title']}"
        }
        print(json.dumps(event))

    # Update cursor so next run skips these items
    save_cursor(cursor)

if __name__ == "__main__":
    main()