Mirth Connect Channel Health Sentinel for Kestra

id: mirth-connect-channel-health-sentinel
namespace: company.team
description: |
  On a schedule, read-only check every deployed channel of a Mirth Connect
  (NextGen Connect) integration engine through its REST API: channels that
  are stopped or paused, messages piling up in destination queues, and new
  errors since the previous run. Classify the engine as HEALTHY, WARNING,
  CRITICAL or UNREACHABLE, page Slack when HL7 messages are not flowing, and
  keep the last error counts in the KV store so old errors never re-alert.

inputs:
  - id: mirth_url
    type: STRING
    defaults: "https://mirth.example.org:8443"
    displayName: "Mirth Connect URL"
    description: "Base URL of the Mirth Connect server (the HTTPS port that serves
      /api), without a trailing slash."
  - id: warning_queued
    type: INT
    defaults: 50
    displayName: "Warning Queued Messages"
    description: "At or above this many messages waiting in one channel's
      destination queues, the check is classified WARNING."
  - id: critical_queued
    type: INT
    defaults: 500
    displayName: "Critical Queued Messages"
    description: "At or above this many queued messages in one channel, the check is
      classified CRITICAL."
  - id: critical_new_errors
    type: INT
    defaults: 10
    displayName: "Critical New Errors"
    description: "At or above this many new errors across all channels since the
      previous run, the check is classified CRITICAL. Any new error is at least
      WARNING."

triggers:
  - id: engine_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check the integration engine every 10 minutes. Shipped disabled;
      enable it once mirth_url and the MIRTH_USERNAME and MIRTH_PASSWORD secrets
      point at a real server and a manual run has been verified.
    cron: "*/10 * * * *"
    disabled: true
    inputs:
      mirth_url: "https://mirth.example.org:8443"

tasks:
  - id: fetch_channel_statuses
    type: io.kestra.plugin.core.http.Request
    description: Read the dashboard status of every deployed channel (name and
      state) from /api/channels/statuses. This is a GET; nothing in Mirth is
      changed. Mirth's API requires the X-Requested-With header on every call.
    uri: "{{ inputs.mirth_url }}/api/channels/statuses"
    method: GET
    headers:
      Accept: application/json
      X-Requested-With: OpenAPI
    options:
      auth:
        type: BASIC
        username: "{{ secret('MIRTH_USERNAME') }}"
        password: "{{ secret('MIRTH_PASSWORD') }}"
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT30S
    retry:
      type: constant
      interval: PT15S
      maxAttempts: 3
    allowFailure: true

  - id: fetch_channel_statistics
    type: io.kestra.plugin.core.http.Request
    description: Read the message counters of every channel (received, sent, error,
      filtered, queued) from /api/channels/statistics. Skipped when the statuses
      could not be read.
    runIf: "{{ outputs.fetch_channel_statuses.code is defined and
      outputs.fetch_channel_statuses.code == 200 and
      (outputs.fetch_channel_statuses.body | trim | startsWith('{')) }}"
    uri: "{{ inputs.mirth_url }}/api/channels/statistics"
    method: GET
    headers:
      Accept: application/json
      X-Requested-With: OpenAPI
    options:
      auth:
        type: BASIC
        username: "{{ secret('MIRTH_USERNAME') }}"
        password: "{{ secret('MIRTH_PASSWORD') }}"
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT30S
    retry:
      type: constant
      interval: PT15S
      maxAttempts: 3
    allowFailure: true

  - id: load_previous_error_counts
    type: io.kestra.plugin.core.kv.Get
    description: Read the per-channel error counters saved by the previous run, so
      only errors that happened since then count. Missing on the first run.
    key: "mirth_error_counts_{{ inputs.mirth_url | slugify }}"
    errorOnMissing: false

  - id: classify_engine
    type: io.kestra.plugin.core.debug.Return
    description: Join statuses, statistics and the previous error counters in one jq
      query. Mirth's JSON returns a single object instead of a list when only
      one channel is deployed and null when none is, so both shapes are
      normalised. A missing response, a non-200 code or a body that is not JSON
      becomes UNREACHABLE instead of looking healthy.
    format: >-
      {{ (outputs.fetch_channel_statuses.code is not defined or
      outputs.fetch_channel_statuses.code != 200 or not
      (outputs.fetch_channel_statuses.body | trim | startsWith('{')) or
      outputs.fetch_channel_statistics is not defined or
      outputs.fetch_channel_statistics.code is not defined or
      outputs.fetch_channel_statistics.code != 200 or not
      (outputs.fetch_channel_statistics.body | trim | startsWith('{'))) ?
      '{"status":"UNREACHABLE","channels":0,"not_running":0,"new_errors":0,"max_queued":0,"not_running_channels":"none","error_channels":"none","queued_channels":"none","error_counts":{}}'
      : ({'st': fromJson(outputs.fetch_channel_statuses.body), 'sx':
      fromJson(outputs.fetch_channel_statistics.body), 'prev':
      (outputs.load_previous_error_counts.value ?? {})} | jq('
      (.st.list.dashboardStatus // [] | if type == "array" then . else [.] end)
      as $status | (.sx.list.channelStatistics // [] | if type == "array" then .
      else [.] end) as $stats | .prev as $prev | ([$stats[] | {key: .channelId,
      value: .}] | from_entries) as $byid | [$status[] | . as $s |
      ($byid[$s.channelId] // {}) as $c | ($c.error // 0) as $err |
      ($prev[$s.channelId] // null) as $p | {id: $s.channelId, name: $s.name,
      state: $s.state, queued: ($c.queued // 0), error: $err, new_errors: (if $p
      == null then 0 elif $err >= $p then $err - $p else $err end)}] as $ch |
      [$ch[] | select(.state != "STARTED")] as $down | [$ch[] |
      select(.new_errors > 0)] as $erring | ([$ch[].new_errors] | add // 0) as
      $newerr | ([$ch[].queued] | max // 0) as $maxq | {channels: ($ch |
      length), not_running: ($down | length), new_errors: $newerr, max_queued:
      $maxq, not_running_channels: (if ($down | length) == 0 then "none" else
      ([$down[] | .name + " (" + .state + ")"] | join(", ")) end),
      error_channels: (if ($erring | length) == 0 then "none" else ([$erring[] |
      .name + " (+" + (.new_errors | tostring) + ")"] | join(", ")) end),
      queued_channels: ([$ch[] | select(.queued >= ' ~ inputs.warning_queued ~
      ') | .name + " (" + (.queued | tostring) + ")"] | if length == 0 then
      "none" else join(", ") end), largest_queue: (if $maxq == 0 then "none"
      else ($ch | max_by(.queued) | .name + " (" + (.queued | tostring) + ")")
      end), error_counts: ([$ch[] | {key: .id, value: .error}] | from_entries),
      first_run: ($prev | length == 0), status: (if ($ch | length) == 0 then
      "UNREACHABLE" elif ($down | length) > 0 or $maxq >= ' ~
      inputs.critical_queued ~ ' or $newerr >= ' ~ inputs.critical_new_errors ~
      ' then "CRITICAL" elif $newerr > 0 or $maxq >= ' ~ inputs.warning_queued ~
      ' then "WARNING" else "HEALTHY" end)} ') | first | toJson) }}

  - id: save_error_counts
    type: io.kestra.plugin.core.kv.Set
    description: Save this run's per-channel error counters for the next run.
      Skipped when the engine could not be read, so an outage never resets the
      baseline.
    runIf: "{{ fromJson(outputs.classify_engine.value).status != 'UNREACHABLE' }}"
    key: "mirth_error_counts_{{ inputs.mirth_url | slugify }}"
    value: "{{ fromJson(outputs.classify_engine.value).error_counts | toJson }}"
    kvType: JSON

  - id: route_by_engine_status
    type: io.kestra.plugin.core.flow.Switch
    description: Route to a quiet log when every channel is started with no new
      errors and short queues, an informational Slack message for new errors or
      a growing queue, or a Slack page when a channel is stopped, queues or
      errors cross the critical line, or the engine could not be read.
    value: "{{ fromJson(outputs.classify_engine.value).status }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          description: Log a quiet confirmation when every channel is started, no new
            errors appeared and no queue reached the warning line.
          message: "Mirth Connect healthy: {{
            fromJson(outputs.classify_engine.value).channels }} channel(s)
            started, no new errors{{
            fromJson(outputs.classify_engine.value).first_run ? ' (first run,
            error baseline saved)' : '' }}, largest queue {{
            fromJson(outputs.classify_engine.value).max_queued }}."
      "WARNING":
        - id: alert_warning_engine
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Post an informational Slack message naming the channels with new
            errors and the channels with a growing queue.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Mirth Connect WARNING ({{ fromJson(outputs.classify_engine.value).channels }} channel(s)).{% if fromJson(outputs.classify_engine.value).new_errors > 0 %} New errors since the last check: {{ fromJson(outputs.classify_engine.value).error_channels }}.{% endif %}{% if fromJson(outputs.classify_engine.value).queued_channels != 'none' %} Queues at {{ inputs.warning_queued }}+: {{ fromJson(outputs.classify_engine.value).queued_channels }}.{% endif %} Execution {{ execution.id }}."
            }
      "CRITICAL":
        - id: alert_critical_engine
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when a channel is not running, a queue crosses
            critical_queued, or new errors cross critical_new_errors; HL7
            messages are not reaching their destination.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Mirth Connect CRITICAL ({{ fromJson(outputs.classify_engine.value).channels }} channel(s)).{% if fromJson(outputs.classify_engine.value).not_running > 0 %} Not running: {{ fromJson(outputs.classify_engine.value).not_running_channels }}.{% endif %}{% if fromJson(outputs.classify_engine.value).new_errors > 0 %} New errors since the last check: {{ fromJson(outputs.classify_engine.value).error_channels }}.{% endif %}{% if fromJson(outputs.classify_engine.value).max_queued > 0 %} Largest queue: {{ fromJson(outputs.classify_engine.value).largest_queue }}.{% endif %} HL7 messages may not be reaching their destination. Execution {{ execution.id }}."
            }
      "UNREACHABLE":
        - id: alert_unreachable_mirth
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the Mirth Connect API could not be read, rejected
            the credentials, or returned no deployed channels.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Mirth Connect UNREACHABLE: {{ inputs.mirth_url }} did not return channel statuses and statistics (server down, wrong URL or port, rejected credentials, or no channel deployed). Integration status is unknown. Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        description: Log any classification outside the four known values, so a change
          in the jq query never fails silently.
        message: "Mirth Connect channel health sentinel completed with an unexpected
          classification: {{ outputs.classify_engine.value }}."

  - id: log_audit_result
    type: io.kestra.plugin.core.log.Log
    description: Record one audit line for every run, whichever branch fired, with
      both HTTP codes and the classification, so the check's own history stays
      queryable even if Slack delivery fails.
    message: "Mirth Connect audit: server={{ inputs.mirth_url }} statuses_http={{
      outputs.fetch_channel_statuses.code ?? 'n/a' }} statistics_http={{
      outputs.fetch_channel_statistics.code ?? 'n/a' }} result={{
      outputs.classify_engine.value }} execution={{ execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the sentinel flow itself fails outright, so a
      broken check is never mistaken for a healthy integration engine.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "Mirth Connect channel health sentinel flow {{ flow.id }} FAILED (execution {{ execution.id }}). Integration status at {{ inputs.mirth_url }} is unknown; check the task logs and the Mirth dashboard manually."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.classify_engine is defined ?
      fromJson(outputs.classify_engine.value).status : 'UNREACHABLE' }}"
  - id: engine
    type: JSON
    value: "{{ outputs.classify_engine is defined ? outputs.classify_engine.value :
      '{\"status\":\"UNREACHABLE\"}' }}"

Hospitals move admissions, orders and lab results between systems through an integration engine, and Mirth Connect (NextGen Connect) is one of the most widely used. When a destination system goes down, a channel's queue silently grows; when a channel is stopped after maintenance, messages stop flowing; when a mapping breaks, messages end in ERROR. The Mirth dashboard shows all of it, but only to someone looking at it. This blueprint turns two read-only Mirth REST calls into a scheduled integration health gate: /api/channels/statuses returns every deployed channel and its state, /api/channels/statistics returns its counters, the KV store remembers the previous error counts, a single jq query works out what changed, and the result routes to a quiet log, an informational Slack note or a Slack page.

How it works

fetch_channel_statuses (io.kestra.plugin.core.http.Request) reads /api/channels/statuses with basic authentication and the X-Requested-With header Mirth requires. It is a GET with a timeout, three retries, allowFailed: true and allowFailure: true, so an outage never aborts the flow.
fetch_channel_statistics reads /api/channels/statistics: received, sent, error, filtered and queued for every channel. A runIf skips it when the statuses could not be read.
load_previous_error_counts (io.kestra.plugin.core.kv.Get) reads the error counters the previous run saved, with errorOnMissing: false so the first run starts cleanly.
classify_engine (io.kestra.plugin.core.debug.Return) normalises Mirth's JSON (a single channel comes back as an object, not a list), joins statuses, statistics and the previous counters, and computes for every channel its state, queue size and new errors since the last run (a counter that went down after a statistics reset counts from zero). Any channel not STARTED, a queue at critical_queued, or critical_new_errors new errors is CRITICAL; any new error or a queue at warning_queued is WARNING; no deployed channel, a non-200 code or a body that is not JSON is UNREACHABLE; otherwise HEALTHY.
save_error_counts (io.kestra.plugin.core.kv.Set) stores this run's counters for the next run, and is skipped when Mirth could not be read, so an outage never resets the baseline.
route_by_engine_status (io.kestra.plugin.core.flow.Switch) logs HEALTHY quietly, posts a Slack note for WARNING, pages Slack for CRITICAL and UNREACHABLE, and logs any unexpected value in defaults.
log_audit_result writes one audit line per run with both HTTP codes and the classification. The flow-level errors block alerts Slack if the flow fails outright, and the engine_check trigger runs every 10 minutes and ships disabled: true.

What you get

A scheduled, read-only health check for every deployed Mirth channel, using only the REST API.
Alerts on new errors only: errors from yesterday, already handled, never page again, because the previous counters live in the KV store.
A status output (HEALTHY / WARNING / CRITICAL / UNREACHABLE) other flows can gate on, and an engine JSON output with channels, not_running, new_errors, max_queued, largest_queue and the channel names behind each.
Slack messages that name the channels, for example Not running: EHR feed (STOPPED). Largest queue: Lab results to EHR (4).

Who it's for

Hospital and health system integration teams running Mirth Connect or NextGen Connect.
Interface analysts who are on call for HL7 feeds between the EHR, laboratory, radiology and billing systems.
Data platform teams already using Kestra who want the same alerting for their integration engine.

Why orchestrate this with Kestra

Mirth can send alerts on errors from inside a channel, but it does not page you when a channel is stopped, when a queue keeps growing because a destination is down, or when the engine itself is unreachable, and its alerts have no memory of what was already reported. Kestra runs the same read-only check on a schedule, keeps the previous counters in its KV store, classifies the whole engine in one typed status, and keeps every run in the execution history, while the Switch branches are ready to grow into follow-up (a ticket for the interface team, a restart after an approval) without changing the check.

Prerequisites

A Mirth Connect or NextGen Connect server (3.x or 4.x) whose API port is reachable from Kestra.
A dedicated Mirth user for monitoring (with a role that can view channels and statistics on editions that support roles).
A Slack incoming webhook for the WARNING, CRITICAL, UNREACHABLE and flow-failure alerts.

Secrets

MIRTH_USERNAME and MIRTH_PASSWORD: credentials of the monitoring user (HTTP basic authentication).
SLACK_WEBHOOK: Slack incoming webhook used for every alert branch and the flow-failure alert.

In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_ and base64-encoded, then read in flows with {{ secret('NAME') }}, for example SECRET_MIRTH_PASSWORD=$(echo -n '<password>' | base64).

Quick start

To try it without a production engine, download Mirth Connect 4.5.2 (the last open-source release), start it with java -jar mirth-server-launcher.jar (change http.port in conf/mirth.properties if 8080 is taken), and create two channels: one that forwards to a second channel through a Channel Writer with queuing on, and that second channel.
Add MIRTH_USERNAME, MIRTH_PASSWORD and SLACK_WEBHOOK as secrets, set mirth_url (for example https://localhost:8443), and run the flow once: it is HEALTHY and saves the error baseline.
Stop the second channel and send a few messages to the first: they queue, and the next run is CRITICAL with the stopped channel named.
Enable engine_check once pointed at the real server.

How to extend

Ignore channels that are stopped on purpose by keeping an allow-list of channel ids in the KV store and filtering them out in the jq query.
For each channel with new errors, read the errored messages with /api/channels/{channelId}/messages?status=ERROR and include the first error in the Slack message.
Alert only on state changes by storing the last status in the KV store.
Send each team its own channel by mapping channel names to Slack webhooks.

Inputs

mirth_url: type STRING, default "https://mirth.example.org:8443".
warning_queued: type INT, default 50.
critical_queued: type INT, default 500.
critical_new_errors: type INT, default 10.

Outputs

status: one of HEALTHY, WARNING, CRITICAL, UNREACHABLE.
engine: the full classification as JSON, for example {"channels":3,"not_running":1,"new_errors":0,"max_queued":4,"not_running_channels":"EHR feed (STOPPED)","error_channels":"none","queued_channels":"none","largest_queue":"Lab results to EHR (4)","status":"CRITICAL",...}.

Pitfalls

Mirth's REST API rejects requests without the X-Requested-With header (cross-site request protection). Both requests set it; keep it if you add calls.
Mirth serves its API with a self-signed certificate out of the box, which Kestra rejects (the run reports UNREACHABLE). Install a certificate from your own CA. For a throwaway test server only, you can add ssl: { insecureTrustAllCertificates: true } under options of both requests.
Statistics are counters since the last reset. The flow only alerts on the difference to the previous run, so the first run after deploying the flow is a baseline and cannot report errors that happened before it.
Undeployed channels do not appear in the statuses at all. A channel that should exist but is missing is not reported; add an expected channel list if you need that.
The check only reads. It never starts, stops, redeploys or reprocesses channels or messages.