GTFS-Realtime service alert watcher with Kestra

id: gtfs-realtime-service-alert-watcher
namespace: company.team

concurrency:
  limit: 1

inputs:
  - id: feed_url
    type: STRING
    displayName: GTFS-Realtime service alerts URL
    description: Public URL of a GTFS-Realtime ServiceAlerts feed. The default is
      the MBTA feed, which needs no API key.
    defaults: https://cdn.mbta.com/realtime/Alerts.pb

  - id: route_ids
    type: STRING
    displayName: Routes to watch
    description: Comma-separated route IDs. Only alerts that affect one of these
      routes are reported. Leave empty to watch every route.
    defaults: Red,Orange,Blue,Mattapan,Green-B,Green-C,Green-D,Green-E

  - id: min_severity
    type: SELECT
    displayName: Lowest severity to report
    description: Alerts below this GTFS-Realtime severity level are skipped. Alerts
      without a severity count as INFO.
    values:
      - INFO
      - WARNING
      - SEVERE
    defaults: WARNING

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Post new and updated alerts to Slack. Needs the SLACK_WEBHOOK_URL secret.
    defaults: false

triggers:
  - id: feed_changed
    type: io.kestra.plugin.core.http.Trigger
    description: Check the feed's ETag with a HEAD request every five minutes and
      start a run only when the agency has published a new version of the file.
    uri: https://cdn.mbta.com/realtime/Alerts.pb
    method: HEAD
    interval: PT5M
    responseCondition: "{{ response.statusCode == 200 and (response.headers['etag']
      | first) != kv(flow.id ~ '_etag', errorOnMissing=false) }}"

tasks:
  - id: download
    type: io.kestra.plugin.core.http.Download
    description: Fetch the current alerts snapshot.
    uri: "{{ inputs.feed_url }}"

  - id: previous
    type: io.kestra.plugin.core.kv.Get
    description: Read the fingerprints of the alerts reported so far.
    key: "{{ flow.id }}_seen"
    errorOnMissing: false

  - id: diff
    type: io.kestra.plugin.scripts.python.Script
    description: Decode the feed, keep the alerts for the watched routes and
      severities, and find the ones that are new or changed since the last run.
    dependencies:
      - kestra
      - gtfs-realtime-bindings
    inputFiles:
      feed.pb: "{{ outputs.download.uri }}"
      seen.json: "{{ (outputs.previous.value ?? {}) | toJson }}"
    script: |
      import hashlib
      import json
      from google.transit import gtfs_realtime_pb2
      from kestra import Kestra

      Alert = gtfs_realtime_pb2.Alert
      routes = {r.strip() for r in "{{ inputs.route_ids }}".split(",") if r.strip()}
      levels = ["INFO", "WARNING", "SEVERE"]
      min_level = levels.index("{{ inputs.min_severity }}")

      with open("seen.json") as f:
          seen = json.load(f) or {}
      feed = gtfs_realtime_pb2.FeedMessage()
      with open("feed.pb", "rb") as f:
          feed.ParseFromString(f.read())

      def text(translated):
          for t in translated.translation:
              if t.language in ("", "en"):
                  return t.text.strip()
          return translated.translation[0].text.strip() if translated.translation else ""

      current, reported = {}, []
      for entity in feed.entity:
          if not entity.HasField("alert"):
              continue
          alert = entity.alert
          alert_routes = sorted({e.route_id for e in alert.informed_entity if e.route_id})
          if routes and not routes.intersection(alert_routes):
              continue
          severity = Alert.SeverityLevel.Name(alert.severity_level) if alert.HasField("severity_level") else "INFO"
          if severity not in levels or levels.index(severity) < min_level:
              continue
          header = text(alert.header_text)
          effect = Alert.Effect.Name(alert.effect)
          fingerprint = hashlib.sha256(json.dumps(
              [header, text(alert.description_text), effect, severity]).encode()).hexdigest()[:16]
          current[entity.id] = fingerprint
          if seen.get(entity.id) != fingerprint:
              status = "updated" if entity.id in seen else "new"
              where = ", ".join(r for r in alert_routes if not routes or r in routes) or "all routes"
              link = text(alert.url)
              reported.append(f"[{severity}] {where}: {header} ({effect}, {status})" + (f" {link}" if link else ""))

      Kestra.outputs({
          "count": len(reported),
          "watched": len(current),
          "cleared": len(set(seen) - set(current)),
          "seen": current,
          "summary": "\n".join(["New or updated service alerts:"] + [f"• {line}" for line in reported]),
      })

  - id: report
    type: io.kestra.plugin.core.flow.If
    description: Log the new or updated alerts and post them to Slack when
      notify_slack is true. When nothing changed, log a one-line status instead.
    condition: "{{ outputs.diff.vars.count > 0 }}"
    then:
      - id: log_alerts
        type: io.kestra.plugin.core.log.Log
        level: WARN
        message: "{{ outputs.diff.vars.summary }}"

      - id: notify
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": {{ outputs.diff.vars.summary | toJson }}
              }
    else:
      - id: nothing_new
        type: io.kestra.plugin.core.log.Log
        message: "No new or updated alerts. {{ outputs.diff.vars.watched }} active, {{
          outputs.diff.vars.cleared }} cleared since the last run."

  - id: remember_alerts
    type: io.kestra.plugin.core.kv.Set
    description: Save the fingerprints of the active alerts, so the next run reports
      only what changed.
    key: "{{ flow.id }}_seen"
    kvType: JSON
    value: "{{ outputs.diff.vars.seen | toJson }}"

  - id: remember_etag
    type: io.kestra.plugin.core.kv.Set
    description: Save the feed's ETag. The trigger starts a run only when the feed's
      ETag differs from this one.
    key: "{{ flow.id }}_etag"
    value: "{{ outputs.download.headers['etag'] | first }}"

Transit agencies publish disruptions, detours and shuttle replacements in a GTFS-Realtime ServiceAlerts feed. The feed is a full snapshot of every active alert, so the people running a rider app, a station screen or an operations channel have to work out which alerts are new for the routes they own. This flow checks the feed with an HTTP trigger, starts a run only when the agency has published a new version, and reports the alerts that are new or updated since the last run, filtered by route and severity.

It runs with no setup. The default feed is the MBTA's public alerts feed, watching the subway lines, and Slack is off.

How it works

feed_changed (io.kestra.plugin.core.http.Trigger) sends a HEAD request every five minutes and fires only when the feed's ETag differs from the one saved in the KV store. It uses HEAD because the trigger reads the response body as text, and a protobuf body is binary. The MBTA republishes its feed every minute, so the default feed starts a run on every check. A feed that is only republished when an alert changes starts no runs in between.
download (io.kestra.plugin.core.http.Download) fetches the protobuf snapshot from feed_url.
previous (io.kestra.plugin.core.kv.Get) reads the fingerprints of the alerts reported by earlier runs.
diff (io.kestra.plugin.scripts.python.Script) decodes the feed with gtfs-realtime-bindings, keeps the alerts that affect route_ids at min_severity or above, and fingerprints each one from its text, effect and severity. An alert whose fingerprint is not in the KV store is new or updated. Active periods are left out because some agencies, the MBTA included, move the end time of an ongoing alert forward every minute, which would report the same alert again on every run. It outputs count, watched, cleared, seen and summary.
report (io.kestra.plugin.core.flow.If) logs the new or updated alerts as one warning and posts them to Slack when notify_slack is true. When nothing changed, it logs how many alerts are active and how many cleared.
remember_alerts and remember_etag (io.kestra.plugin.core.kv.Set) save the fingerprints and the feed's ETag for the next run.

The flow runs one execution at a time (concurrency.limit: 1), so a slow run never reports the same alerts twice.

Inputs

feed_url (STRING, default the MBTA alerts feed): the GTFS-Realtime ServiceAlerts URL.
route_ids (STRING, default the MBTA subway lines): comma-separated route IDs to watch. Empty watches every route.
min_severity (SELECT, default WARNING): the lowest severity to report, one of INFO, WARNING or SEVERE.
notify_slack (BOOL, default false): post new and updated alerts to Slack.

Prerequisites

A Kestra worker that can run Docker containers, for the Python script.
Network access to the feed URL. Some agencies require an API key; add it to both download and feed_changed, in the URL or as a header.
A feed URL that answers a HEAD request with an ETag header. Most CDNs do. If yours does not, replace feed_changed with a Schedule trigger and delete remember_etag. The flow still reports only new or updated alerts.

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook. Only needed when notify_slack is true.

Quick start

Save the flow. feed_changed checks the feed right away and starts the first run, which logs a warning listing each current subway alert at WARNING or above.
Run the flow yourself. Unless an alert changed in between, it logs only how many alerts are active and how many cleared.
To watch other routes, change the route_ids default. Set the min_severity default to INFO to see every alert. Triggered runs always use the defaults.
Leave the feed_changed trigger on. It checks the feed every five minutes and starts a run when the agency has published a new version. Lower its interval for faster alerts.
When pointing at another agency, set feed_url and the trigger's uri to the same URL.

Expected outputs

outputs.diff.vars.count: how many alerts are new or updated.
outputs.diff.vars.summary: a heading followed by one line per new or updated alert, with severity, routes, header, effect and link.
outputs.diff.vars.watched: how many active alerts match the filters.
outputs.diff.vars.cleared: how many previously reported alerts are no longer in the feed.

Report new and updated GTFS-Realtime service alerts for the routes you watch

How it works

Inputs

Prerequisites

Secrets

Quick start

Expected outputs

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Report new and updated GTFS-Realtime service alerts for the routes you watch

How it works

Inputs

Prerequisites

Secrets

Quick start

Expected outputs

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Gate a Dashboard Publish on a Sifflet Rule

Run a Sifflet Rule After a Warehouse Load

Schedule a Daily Sifflet Observability Sweep

Recheck a Sifflet Rule After an Incident Fix

Validate Releases with a Sifflet Rule Run

Fan Out a Sifflet Rule Pack with Loop

New to Kestra?