The smallest look at the sensor layer. A poll sensor checks a health endpoint every few minutes and emits an Incident only when it is down. A second workflow reacts by opening a GitHub issue. Turning a raw signal into a typed event, and only when it matters, is the whole job of a sensor.
One poll sensor in sensors/sensors.py. It returns None on a healthy check, so nothing goes on the bus; only a non-200 becomes an Incident.
import urllib.request
from loopy import sensor
from loopy.events import Incident
HEALTH_URL = "https://example.com/health"
@sensor(poll="5m", emits="Incident")
def health_check(req) -> Incident | None:
try:
with urllib.request.urlopen(HEALTH_URL, timeout=10) as resp:
status = resp.status
except Exception:
status = 0 # unreachable, treat as down
if status == 200:
return None # healthy, emit nothing
return Incident(url=HEALTH_URL, status=status)1 step in workflows/respond/, triggered by the Incident the sensor emits. It has no idea a poll produced it; it just consumes the event.
---
on: Incident
agent: Responder
output: { issue_url: url }
emits: Acknowledged
---
The health check for {{ event.url }} returned {{ event.status }}
instead of 200. Open a GitHub issue for the outage (and skip it if
an open one already exists). Return the issue URL.Incident is emitted by the sensor and consumed by the workflow; Acknowledged is the terminal announcement. Both are declared because they cross the bus.
agents:
Responder: {} # opens a GitHub issue for the outage
events:
Incident:
url: url
status: int
Acknowledged:
issue_url: url