Model Card Publishing for Registry Promotions

id: model-card-on-promote
namespace: company.team
description: |
  Publish a documentation-ready model card the moment a model registry marks a version as
  promoted, so the card stops depending on whoever remembers to write it. The registry POSTs
  the promotion payload to a keyed webhook, a python task renders the markdown card from that
  payload or from the bundled promotion fixture, and a promotion missing required fields opens
  a GitHub issue plus a Slack alert instead of publishing. A complete payload is POSTed to the
  registry docs endpoint with a bearer token, polled until the artifact is readable, and the
  execution ends with a verdict, the model, the card uri, and the metric count as outputs.

triggers:
  - id: on_promotion
    type: io.kestra.plugin.core.trigger.Webhook
    description: The model registry POSTs the promotion event - model name, version,
      metrics, intended use, limitations, and owner - the instant a version is
      approved, so the card is written while the decision is fresh.
    key: "{{ secret('PROMOTION_WEBHOOK_KEY') }}"
  - id: nightly_backfill
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily 06:00 UTC net that re-renders the bundled sample promotion as
      a smoke test of the publishing path - stays disabled until the webhook
      path is proven in your environment.
    cron: "0 6 * * *"
    disabled: true

inputs:
  - id: registry_docs_endpoint
    type: STRING
    defaults: https://registry.acme.example/v1/models/{model}/{version}/docs/model-card.md
    description: Template for the card uri - {model} and {version} are filled from
      the promotion payload, the same uri receives the POST and answers every
      sync probe.
  - id: github_repository
    type: STRING
    defaults: acme/model-registry
    description: Repository that receives the tracking issue when a promotion
      payload arrives without the fields a card needs.

tasks:
  # --- 1. render the model card from the promotion payload ---
  - id: generate_model_card
    type: io.kestra.plugin.scripts.python.Script
    description: Read the promotion payload, fall back to the bundled promotion.json
      fixture when the webhook body is absent, verify the six required card
      fields, render the markdown card, and write model-card.md.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      PROMOTION: "{{ (trigger.body ?? {}) | toJson }}"
      ENDPOINT: "{{ inputs.registry_docs_endpoint }}"
    script: |
      import json
      import os

      REQUIRED = ("model", "version", "metrics", "intended_use", "limitations", "owner")
      TEXT_FIELDS = ("model", "version", "intended_use", "limitations", "owner")
      FIXTURE = {"model": "churn-propensity-lgbm", "version": "4.2.0",
                 "metrics": {"auc": 0.913, "f1": 0.782, "log_loss": 0.214},
                 "intended_use": "Rank subscription accounts by 30-day churn risk for the retention team.",
                 "limitations": "Fitted on 2024-2025 billing history, accuracy degrades after pricing changes.",
                 "owner": "risk-modeling@acme.example"}


      def load_fixture():
          if not os.path.exists("promotion.json"):
              with open("promotion.json", "w") as handle:
                  json.dump(FIXTURE, handle, indent=2, sort_keys=True)
          try:
              data = json.load(open("promotion.json"))
          except (OSError, ValueError):
              data = {}
          return data if isinstance(data, dict) else {}


      raw = os.environ.get("PROMOTION", "").strip()
      payload, source = {}, "unparsed"
      if not raw or raw in ("{}", "null"):
          payload, source = load_fixture(), "fixture"
      else:
          try:
              parsed = json.loads(raw)
          except ValueError:
              parsed = None
          if isinstance(parsed, dict) and parsed:
              payload, source = parsed, "payload"

      values = {}
      for field in TEXT_FIELDS:
          value = payload.get(field)
          values[field] = value.strip() if isinstance(value, str) else ""
      metrics = {}
      if isinstance(payload.get("metrics"), dict):
          for key, value in payload["metrics"].items():
              if str(key).strip():
                  metrics[str(key).strip()] = value
      present = dict(values, metrics=bool(metrics))
      missing = [field for field in REQUIRED if not present[field]]

      def cell(value):
          if value is None or value == "":
              return "TBD"
          return str(value).replace("|", "/").replace("\n", " ")

      name = values["model"] or "unknown-model"
      version = values["version"] or "unversioned"
      lines = ["# Model card - " + name, "",
               "Version " + version + " was promoted to the model registry.", "",
               "## Details", "", "| Field | Value |", "| --- | --- |",
               "| Name | " + cell(values["model"]) + " |",
               "| Version | " + cell(values["version"]) + " |",
               "| Owner | " + cell(values["owner"]) + " |",
               "| Metrics reported | " + str(len(metrics)) + " |", "",
               "## Metrics", ""]
      if metrics:
          lines.append("| Metric | Value |")
          lines.append("| --- | --- |")
          for key in sorted(metrics):
              lines.append("| " + cell(key) + " | " + cell(metrics[key]) + " |")
      else:
          lines.append("No metrics arrived with this promotion.")
      lines += ["", "## Intended use", "",
                values["intended_use"] or "TBD - intended_use was not supplied with this promotion.",
                "", "## Limitations", "",
                values["limitations"] or "TBD - limitations was not supplied with this promotion.", ""]

      card = "\n".join(lines)
      with open("model-card.md", "w") as handle:
          handle.write(card)
      endpoint = os.environ.get("ENDPOINT", "").strip()
      try:
          card_uri = endpoint.format(model=name, version=version)
      except (IndexError, KeyError, ValueError):
          card_uri = endpoint

      result = {"model": name, "version": version, "metrics": len(metrics),
                "missing_fields": ",".join(missing), "missing_count": len(missing),
                "has_missing": bool(missing), "source": source, "card_uri": card_uri, "card_text": card}
      print("model card %s version=%s metrics=%d missing=%s source=%s" % (name, version, len(metrics), ",".join(missing) or "none", source))
      print("::" + json.dumps({"outputs": result}) + "::")
  # --- 2. required-field gate ---
  - id: check_required_fields
    type: io.kestra.plugin.core.flow.If
    description: Branch on whether the promotion carried every required card field -
      an incomplete promotion escalates instead of publishing a half-written
      card.
    condition: "{{ outputs.generate_model_card.vars.has_missing == true }}"
    then:
      - id: open_card_gap_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open a tracking issue in the registry repository listing exactly
          which fields the promotion payload did not carry, so the card is
          completed before any publish attempt.
        jwtToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "{{ inputs.github_repository }}"
        title: "Model card held - {{ outputs.generate_model_card.vars.model }} v{{
          outputs.generate_model_card.vars.version }} is missing {{
          outputs.generate_model_card.vars.missing_count }} field(s)"
        labels:
          - model-card
          - documentation
        body: |
          ## Promotion payload is incomplete

          Execution {{ execution.id }} refused to publish a model card.

          | Field | Value |
          | --- | --- |
          | Model | {{ outputs.generate_model_card.vars.model }} |
          | Missing fields | {{ outputs.generate_model_card.vars.missing_fields }} |

          Supply the missing fields and replay the webhook.
      - id: alert_missing_fields
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the model governance channel which promotion was held back and
          which card fields the payload did not carry, naming the uri that was
          not published.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"text": "Model card held - {{ outputs.generate_model_card.vars.model }} v{{ outputs.generate_model_card.vars.version }} is missing {{ outputs.generate_model_card.vars.missing_fields }}. Nothing was published to {{ outputs.generate_model_card.vars.card_uri }}. Execution {{ execution.id }}."}
    else:
      # --- 3. publish the rendered card ---
      - id: post_card_to_registry
        type: io.kestra.plugin.core.http.Request
        description: POST the rendered markdown to the registry docs endpoint with a
          bearer token, keeping a rejection as a status for the verdict instead
          of failing the execution.
        uri: "{{ outputs.generate_model_card.vars.card_uri }}"
        method: POST
        timeout: PT30S
        headers:
          Authorization: "Bearer {{ secret('REGISTRY_API_TOKEN') }}"
          Content-Type: text/markdown
          X-Model-Version: "{{ outputs.generate_model_card.vars.version }}"
        body: "{{ outputs.generate_model_card.vars.card_text }}"
        options:
          allowFailed: true
      - id: log_card_post
        type: io.kestra.plugin.core.log.Log
        description: Record the card uri, the size of the markdown, and the registry
          response status in the execution history before polling begins.
        message: "Posted {{ outputs.generate_model_card.vars.card_text | length }}
          characters to {{ outputs.generate_model_card.vars.card_uri }} with
          status {{ outputs.post_card_to_registry.code }}."
      - id: settle_before_sync
        type: io.kestra.plugin.core.flow.Sleep
        description: Give the registry a moment to index the new document before the
          first artifact probe, so the poll does not race the write it just
          made.
        duration: PT10S
      # --- 4. poll until the registry serves the artifact ---
      - id: poll_registry_sync
        type: io.kestra.plugin.core.flow.LoopUntil
        description: Probe the published card uri until the registry serves it or the
          polling budget runs out, so the verdict reflects a readable artifact
          rather than an accepted POST.
        condition: "{{ outputs.check_sync.vars.visible == true }}"
        checkFrequency:
          interval: PT5S
          maxDuration: PT2M
          maxIterations: 6
        failOnMaxReached: false
        tasks:
          - id: probe_registry_artifact
            type: io.kestra.plugin.core.http.Request
            description: Fetch the published card with the same bearer token, letting a
              replication lag 404 surface as a status the python probe can read
              instead of failing the loop.
            uri: "{{ outputs.generate_model_card.vars.card_uri }}"
            method: GET
            timeout: PT15S
            headers:
              Authorization: "Bearer {{ secret('REGISTRY_API_TOKEN') }}"
            options:
              allowFailed: true
          - id: check_sync
            type: io.kestra.plugin.scripts.python.Script
            description: Turn the probe status into the boolean that LoopUntil evaluates,
              true only when the registry answers with the card itself.
            taskRunner:
              type: io.kestra.plugin.core.runner.Process
            env:
              PROBE_CODE: "{{ outputs.probe_registry_artifact.code }}"
            script: |
              import json
              import os

              raw = os.environ.get("PROBE_CODE", "").strip()
              try:
                  code = int(raw)
              except ValueError:
                  code = 0
              visible = 200 <= code < 300
              print("registry artifact probe status=%d visible=%s" % (code, visible))
              print("::" + json.dumps({"outputs": {"visible": visible, "probe_code": code}}) + "::")
  # --- 5. final verdict ---
  - id: publish_verdict
    type: io.kestra.plugin.scripts.python.Script
    description: Combine the field check, the registry response, and the sync probe
      into the four published outputs of this execution, one verdict string for
      both branches.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      MODEL: "{{ outputs.generate_model_card.vars.model }}"
      CARD_URI: "{{ outputs.generate_model_card.vars.card_uri }}"
      METRICS: "{{ outputs.generate_model_card.vars.metrics }}"
      MISSING_COUNT: "{{ outputs.generate_model_card.vars.missing_count }}"
      POST_CODE: "{{ outputs.post_card_to_registry.code ?? '' }}"
      SYNCED: "{{ outputs.check_sync.vars.visible ?? '' }}"
    script: |
      import json
      import os


      def as_int(name):
          raw = os.environ.get(name, "").strip()
          return int(raw) if raw.isdigit() else 0


      def as_flag(name):
          return os.environ.get(name, "").strip().lower() in ("1", "true", "yes")


      model = os.environ.get("MODEL", "").strip() or "unknown-model"
      card_uri = os.environ.get("CARD_URI", "").strip()
      metrics = as_int("METRICS")
      missing_count = as_int("MISSING_COUNT")
      post_code = as_int("POST_CODE")
      synced = as_flag("SYNCED")

      if missing_count:
          verdict = "escalated_missing_fields"
      elif post_code and 200 <= post_code < 300:
          verdict = "published" if synced else "published_sync_pending"
      elif post_code:
          verdict = "publish_rejected"
      else:
          verdict = "not_published"

      payload = {"verdict": verdict, "model": model, "card_uri": card_uri,
                 "metrics": metrics, "post_code": post_code, "synced": synced}
      print("verdict=%s model=%s metrics=%d post_code=%d synced=%s" % (verdict, model, metrics, post_code, synced))
      print("::" + json.dumps({"outputs": payload}) + "::")

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: published, published_sync_pending, publish_rejected, not_published,
      or escalated_missing_fields.
    value: "{{ outputs.publish_verdict.vars.verdict }}"
  - id: model
    type: STRING
    description: Model name taken from the promotion payload and used in the card title.
    value: "{{ outputs.publish_verdict.vars.model }}"
  - id: card_uri
    type: STRING
    description: Registry uri the card was rendered for, templated from the endpoint
      input with model and version.
    value: "{{ outputs.publish_verdict.vars.card_uri }}"
  - id: metrics
    type: INT
    description: Number of metrics that arrived with the promotion and were rendered
      into the card table.
    value: "{{ outputs.publish_verdict.vars.metrics }}"

Model cards record what a promoted version is for, what it scores, and what it must not be used for, yet they are almost always written late. This blueprint makes the promotion event the trigger - the registry POSTs model, version, and metrics to a keyed webhook, the card is rendered from that payload, and a promotion that arrives without its required fields is escalated instead of published.

How it works

on_promotion (keyed io.kestra.plugin.core.trigger.Webhook) receives a JSON body of model, version, metrics, intended_use, limitations, and owner whenever a version is approved for production.
nightly_backfill (io.kestra.plugin.core.trigger.Schedule, cron 0 6 * * *) re-renders the bundled sample promotion as a smoke test of the publishing path; it ships disabled until the webhook path is trusted.
generate_model_card (python on the Process runner) uses the webhook body when it arrives or the promotion.json fixture it writes when it does not, checks the six required fields, renders a card with a details table, a metrics table, intended use, limitations, and the owner, writes model-card.md, and emits the card text, the missing fields, and the card uri through the repo outputs protocol.
check_required_fields (io.kestra.plugin.core.flow.If) branches on has_missing == true. Missing fields open a labeled issue through open_card_gap_issue (io.kestra.plugin.github.issues.Create) and post alert_missing_fields (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) naming the uri that was not published.
A complete payload is POSTed by post_card_to_registry (io.kestra.plugin.core.http.Request) with a Bearer token, recorded by log_card_post, and held for settle_before_sync (io.kestra.plugin.core.flow.Sleep) before the first probe.
poll_registry_sync (io.kestra.plugin.core.flow.LoopUntil) GETs the same uri through probe_registry_artifact and lets check_sync turn the response code into a boolean, repeating every 5 seconds for up to 2 minutes until the registry serves the artifact.
publish_verdict folds the field check, the POST status, and the sync flag into verdict, model, card_uri, and metrics, while the errors block alerts Slack when the pipeline itself breaks.

What you get

A markdown card with the six required sections rendered straight from the promotion event, including a metrics table built from whatever the registry reported.
A gate that refuses to publish an incomplete card and instead opens a GitHub issue and posts a Slack alert naming the missing fields.
A bearer-authenticated POST plus a bounded poll, so published means the artifact is readable rather than merely accepted.
verdict, model, card_uri, and metrics as typed outputs for dashboards, audits, and release checks.

Who it's for

ML platform teams that require a card for every promoted version and want the registry event to drive the documentation.
Model governance owners who need intended use and limitations captured before a version serves production traffic.
Platform engineers standardizing publication behind one endpoint template and one set of secrets.

Why orchestrate this with Kestra

Card writing usually happens in a wiki tab nobody updates after the first release, so the record drifts from the model it describes. Kestra puts the trigger, the render, the completeness gate, the escalation, the authenticated publish, and the sync poll in one versioned file - every promotion leaves an execution with its card, its verdict, and its missing fields, and the disabled schedule gives you a rehearsal path before the webhook goes live.

Prerequisites

A model registry that POSTs a promotion JSON body to a Kestra webhook and accepts markdown on a docs endpoint.
Python 3 on the Kestra worker for the Process runner, plus outbound access to the registry endpoint.
A GitHub token with issue write access and a Slack incoming webhook for the escalation path.

Secrets

PROMOTION_WEBHOOK_KEY: key segment guarding the webhook URL, anyone holding it can submit promotion events.
REGISTRY_API_TOKEN: bearer token the registry accepts on the docs endpoint for the publish POST and the sync probes.
GITHUB_TOKEN: token open_card_gap_issue uses to create the tracking issue.
SLACK_WEBHOOK_URL: Slack incoming webhook for held-card alerts and failure alerts.

Quick start

Add the four secrets and point registry_docs_endpoint at your docs route, keeping the {model} and {version} placeholders.
POST a full promotion body to the webhook URL and confirm verdict: published with the reported metric count.
Replay it with owner removed and confirm the GitHub issue, the Slack alert, and verdict: escalated_missing_fields.
Keep nightly_backfill disabled until both paths behave.

How to extend

Add required fields, for example training_data, by extending the REQUIRED tuple in generate_model_card.
Fail the execution instead of alerting with io.kestra.plugin.core.execution.Fail after the escalation branch.
Render the card from a template file kept in git by swapping the line builder for a template read.
Notify a second channel, for example email with io.kestra.plugin.email.MailSend, alongside Slack.
Mirror the card to object storage with io.kestra.plugin.aws.s3.Upload before the registry POST.

Publish a Model Card Automatically When a Promotion Lands

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Publish a Model Card Automatically When a Promotion Lands

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

ML Training Dataset Security Guard and Poisoning Prevention Gate

Back up a MongoDB collection and prove it restores, every night

MongoDB Expired Document Retention Guard

MongoDB Duplicate Document Dedup Guard

MongoDB Document Validation Drift Repair Guard

MongoDB GridFS Orphaned Chunk and Dangling File Reclamation Guard

New to Kestra?