New to Kestra?
Use blueprints to kickstart your first workflows.
API Evolution Watchtower
Monitor third-party OpenAPI specs with Kestra and oasdiff, detect consumer-impacting changes, persist baselines, and govern revisions as PASS/WARN/BLOCK.
Categories
BusinessInfrastructure
id: api-evolution-watchtower
namespace: company.team
description: |
Consumer-side API Evolution Watchtower.
Monitors third-party OpenAPI specifications, detects consumer-relevant
changes, applies governance policy, and maintains a persistent accepted
baseline using the Kestra KV store.
triggers:
- id: scheduled_check
type: io.kestra.plugin.core.trigger.Schedule
description: Run the API evolution check on a schedule.
cron: "0 9 * * *"
disabled: true
inputs:
- id: apis
type: ARRAY
itemType: STRING
required: true
displayName: APIs
description: |
APIs to monitor.
Each item contains:
name
baseline_url
revision_url
used_operations
defaults:
- |
{
"name": "loop-demo-1",
"baseline_url": "https://raw.githubusercontent.com/oasdiff/oasdiff/v1.32.1/data/simple1.yaml",
"revision_url": "https://raw.githubusercontent.com/oasdiff/oasdiff/v1.32.1/data/simple1.yaml",
"used_operations": ["GET /api/test"]
}
- |
{
"name": "loop-demo-warn",
"baseline_url": "https://raw.githubusercontent.com/oasdiff/oasdiff/v1.32.1/data/checker/request_property_one_of_wrapped_base.yaml",
"revision_url": "https://raw.githubusercontent.com/oasdiff/oasdiff/v1.32.1/data/checker/request_property_one_of_wrapped_original_preserved_revision.yaml",
"used_operations": ["POST /example"]
}
tasks:
# ============================================================
# PROCESS ALL APIS
# ============================================================
- id: validate_oasdiff_repository
type: io.kestra.plugin.core.flow.WorkingDirectory
description: |
Validate the public oasdiff repository fixtures used by the demo.
tasks:
- id: clone_oasdiff_repository
type: io.kestra.plugin.git.Clone
url: https://github.com/oasdiff/oasdiff
tag: v1.32.1
- id: verify_oasdiff_fixtures
type: io.kestra.plugin.scripts.shell.Commands
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.12-slim
commands:
- |
python3 - <<'PY'
from pathlib import Path
required = [
"data/simple1.yaml",
"data/checker/request_property_one_of_wrapped_base.yaml",
"data/checker/request_property_one_of_wrapped_original_preserved_revision.yaml",
]
missing = [p for p in required if not Path(p).is_file()]
if missing:
raise SystemExit(
"Missing fixtures: " + ", ".join(missing)
)
print("All oasdiff demo fixtures verified.")
PY
- id: monitor_apis
type: io.kestra.plugin.core.flow.Loop
description: |
Process every configured API independently.
A BLOCK verdict does not fail the loop immediately.
The final governance gate runs after all APIs are evaluated.
values: "{{ inputs.apis }}"
concurrencyLimit: 1
transmitFailed: true
fetchType: FETCH
# ----------------------------------------------------------
# Outputs exposed from every iteration
# ----------------------------------------------------------
outputs:
- id: api_name
type: STRING
value: "{{ fromJson(item.value).name }}"
- id: verdict
type: STRING
value: >-
{{
outputs.check_baseline.vars.unchanged
? 'PASS'
: outputs.evaluate_policy.vars.verdict
}}
- id: breaking_count
type: STRING
value: >-
{{
outputs.check_baseline.vars.unchanged
? '0'
: outputs.evaluate_policy.vars.breaking_count
}}
- id: warning_count
type: STRING
value: >-
{{
outputs.check_baseline.vars.unchanged
? '0'
: outputs.evaluate_policy.vars.warning_count
}}
- id: version_policy_count
type: STRING
value: >-
{{
outputs.check_baseline.vars.unchanged
? '0'
: outputs.evaluate_policy.vars.version_policy_count
}}
- id: baseline_source
type: STRING
value: "{{ outputs.check_baseline.vars.baseline_source }}"
- id: unchanged
type: STRING
value: "{{ outputs.check_baseline.vars.unchanged }}"
tasks:
# ========================================================
# 1. FETCH BOOTSTRAP BASELINE
# ========================================================
- id: fetch_base_spec
type: io.kestra.plugin.core.http.Download
description: Download the bootstrap baseline OpenAPI specification.
uri: "{{ fromJson(item.value).baseline_url }}"
# ========================================================
# 2. FETCH CANDIDATE
# ========================================================
- id: fetch_revision_spec
type: io.kestra.plugin.core.http.Download
description: Download the latest candidate OpenAPI specification.
uri: "{{ fromJson(item.value).revision_url }}"
# ========================================================
# 3. LOAD ACCEPTED KV BASELINE
# ========================================================
- id: get_baseline_spec
type: io.kestra.plugin.core.kv.Get
description: |
Load the previously accepted specification from KV.
key: "api-evolution-watchtower-{{ fromJson(item.value).name }}-spec"
errorOnMissing: false
# ========================================================
# 4. RESOLVE BASELINE FILE
# ========================================================
- id: resolve_baseline_file
type: io.kestra.plugin.core.storage.Write
description: |
Use the persistent KV baseline when available.
Otherwise use the bootstrap specification.
content: "{{ outputs.get_baseline_spec.value ??
read(outputs.fetch_base_spec.uri) }}"
extension: .yaml
# ========================================================
# 5. HASH CHECK
# ========================================================
- id: check_baseline
type: io.kestra.plugin.scripts.shell.Commands
description: |
Calculate SHA-256 hashes and detect whether the candidate
specification is unchanged.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.12-slim
inputFiles:
base.yaml: "{{ outputs.resolve_baseline_file.uri }}"
revision.yaml: "{{ outputs.fetch_revision_spec.uri }}"
outputFiles:
- baseline_state.json
commands:
- |
python3 - <<'PY'
import hashlib
import json
with open("base.yaml", "rb") as f:
baseline_bytes = f.read()
with open("revision.yaml", "rb") as f:
revision_bytes = f.read()
baseline_hash = hashlib.sha256(
baseline_bytes
).hexdigest()
revision_hash = hashlib.sha256(
revision_bytes
).hexdigest()
baseline_value = """{{ outputs.get_baseline_spec.value ?? '' }}"""
has_persistent_baseline = bool(
baseline_value.strip()
)
baseline_source = (
"persistent-kv"
if has_persistent_baseline
else "bootstrap-input"
)
unchanged = (
revision_hash == baseline_hash
)
result = {
"baseline_hash": baseline_hash,
"revision_hash": revision_hash,
"baseline_source": baseline_source,
"unchanged": unchanged
}
with open(
"baseline_state.json",
"w",
encoding="utf-8"
) as f:
json.dump(
result,
f,
indent=2
)
print(
"::"
+ json.dumps(
{"outputs": result}
)
+ "::"
)
PY
# ========================================================
# 6. INITIALIZE BASELINE
# ========================================================
- id: initialize_baseline
type: io.kestra.plugin.core.flow.If
description: |
Seed the persistent baseline when this API is seen
for the first time.
condition: "{{ outputs.check_baseline.vars.baseline_source == 'bootstrap-input'
}}"
then:
- id: store_initial_baseline_spec
type: io.kestra.plugin.core.kv.Set
description: Store the initial accepted OpenAPI specification.
key: "api-evolution-watchtower-{{ fromJson(item.value).name }}-spec"
kvType: STRING
overwrite: true
value: "{{ read(outputs.fetch_base_spec.uri) }}"
- id: store_initial_baseline_hash
type: io.kestra.plugin.core.kv.Set
description: Store the initial baseline hash.
key: "api-evolution-watchtower-{{ fromJson(item.value).name }}-hash"
kvType: STRING
overwrite: true
value: "{{ outputs.check_baseline.vars.baseline_hash }}"
# ========================================================
# 7. ONLY ANALYZE WHEN CHANGED
# ========================================================
- id: change_detected
type: io.kestra.plugin.core.flow.If
description: |
Run oasdiff only when the candidate differs from
the accepted baseline.
condition: "{{ outputs.check_baseline.vars.unchanged == false }}"
then:
# ==================================================
# 7A. OASDIFF
# ==================================================
- id: oasdiff_changelog
type: io.kestra.plugin.scripts.shell.Commands
description: |
Generate a machine-readable OpenAPI changelog.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: tufin/oasdiff:v1.32.1
inputFiles:
base.yaml: "{{ outputs.resolve_baseline_file.uri }}"
revision.yaml: "{{ outputs.fetch_revision_spec.uri }}"
commands:
- |
oasdiff changelog \
-f json \
base.yaml \
revision.yaml \
> changelog.json
outputFiles:
- changelog.json
# ==================================================
# 7B. CONSUMER IMPACT + POLICY
# ==================================================
- id: evaluate_policy
type: io.kestra.plugin.scripts.shell.Commands
description: |
Filter changes to consumer-used operations and
classify the candidate as PASS, WARN or BLOCK.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.12-slim
inputFiles:
changelog.json: "{{ outputs.oasdiff_changelog.outputFiles['changelog.json'] }}"
outputFiles:
- report.md
- verdict.json
commands:
- |
python3 - <<'PY'
import json
import fnmatch
# ------------------------------------------------
# Load oasdiff output
# ------------------------------------------------
with open(
"changelog.json",
"r",
encoding="utf-8"
) as f:
changes = json.load(f)
api_name = """{{ fromJson(item.value).name }}"""
# ------------------------------------------------
# Consumer configuration
# ------------------------------------------------
used_operations = {{
fromJson(item.value).used_operations | toJson
}}
# ------------------------------------------------
# Consumer impact filter
# ------------------------------------------------
def is_used(change):
if not used_operations:
return True
operation = change.get("operation")
path = change.get("path")
# Global/component findings may affect
# all consumers.
if not operation or not path:
return True
for rule in used_operations:
rule = rule.strip()
if not rule:
continue
parts = rule.split(" ", 1)
if len(parts) != 2:
continue
method, pattern = parts
if method.upper() != operation.upper():
continue
if fnmatch.fnmatch(path, pattern):
return True
return False
# ------------------------------------------------
# Split relevant and ignored
# ------------------------------------------------
relevant = []
ignored = []
for change in changes:
if is_used(change):
relevant.append(change)
else:
ignored.append(change)
# ------------------------------------------------
# Classification
# ------------------------------------------------
version_policy_ids = {
"api-version-not-bumped",
"api-version-decreased",
"api-major-version-not-bumped",
}
breaking = []
warnings = []
infos = []
version_policy = []
for change in relevant:
level = change.get(
"level",
1
)
change_id = change.get(
"id",
""
)
if change_id in version_policy_ids:
version_policy.append(change)
continue
if level >= 3:
breaking.append(change)
elif level == 2:
warnings.append(change)
else:
infos.append(change)
# ------------------------------------------------
# Version policy only matters when there is
# a breaking consumer-relevant change
# ------------------------------------------------
relevant_version_policy = (
version_policy
if breaking
else []
)
# ------------------------------------------------
# Verdict
# ------------------------------------------------
if (
breaking
or relevant_version_policy
):
verdict = "BLOCK"
elif warnings:
verdict = "WARN"
else:
verdict = "PASS"
# ------------------------------------------------
# Format change
# ------------------------------------------------
def format_change(change):
change_id = change.get(
"id",
"unknown"
)
text = change.get(
"text",
""
)
operation = change.get(
"operation",
""
)
path = change.get(
"path",
""
)
if operation and path:
location = (
f": "
f"{operation.upper()} "
f"{path}"
)
elif path:
location = (
f": {path}"
)
else:
location = ""
return (
f"- `{change_id}`"
f"{location}: {text}"
)
# ------------------------------------------------
# Markdown report
# ------------------------------------------------
lines = [
"# API Evolution Watchtower Report",
"",
f"**API:** "
f"`{fromJson(item.value).name}`",
"",
f"**Verdict:** `{verdict}`",
"",
"## Summary",
"",
f"- Relevant changes: "
f"{len(relevant)}",
f"- Ignored consumer-irrelevant changes: "
f"{len(ignored)}",
f"- Breaking changes: "
f"{len(breaking)}",
f"- Warnings: "
f"{len(warnings)}",
f"- Informational changes: "
f"{len(infos)}",
f"- Version-policy findings: "
f"{len(relevant_version_policy)}",
"",
]
if breaking:
lines += [
"## Breaking Changes",
"",
*[
format_change(c)
for c in breaking
],
"",
]
if relevant_version_policy:
lines += [
"## Version Policy",
"",
*[
format_change(c)
for c in relevant_version_policy
],
"",
]
if warnings:
lines += [
"## Warnings",
"",
*[
format_change(c)
for c in warnings
],
"",
]
if ignored:
lines += [
"## Ignored Consumer-Irrelevant Changes",
"",
*[
format_change(c)
for c in ignored
],
"",
]
if (
not breaking
and not warnings
and not relevant_version_policy
):
lines += [
"## Result",
"",
"No consumer-relevant breaking "
"or warning-level changes detected.",
"",
]
# ------------------------------------------------
# Write report
# ------------------------------------------------
with open(
"report.md",
"w",
encoding="utf-8"
) as f:
f.write(
"\n".join(lines)
)
# ------------------------------------------------
# Machine-readable verdict
# ------------------------------------------------
verdict_data = {
"api_name":
fromJson(item.value).name,
"verdict":
verdict,
"relevant_changes":
len(relevant),
"ignored_changes":
len(ignored),
"breaking_count":
len(breaking),
"warning_count":
len(warnings),
"info_count":
len(infos),
"version_policy_count":
len(relevant_version_policy),
}
with open(
"verdict.json",
"w",
encoding="utf-8"
) as f:
json.dump(
verdict_data,
f,
indent=2
)
print(
"::"
+ json.dumps(
{"outputs": verdict_data}
)
+ "::"
)
PY
# ==================================================
# 7C. LOG VERDICT: DO NOT FAIL HERE
# ==================================================
- id: governance_decision
type: io.kestra.plugin.core.log.Log
description: |
Record the governance decision for this API.
BLOCK does not fail the Loop iteration.
message: |
API governance verdict:
{{ outputs.evaluate_policy.vars.verdict }}
API:
{{ fromJson(item.value).name }}
Breaking changes:
{{ outputs.evaluate_policy.vars.breaking_count }}
Warnings:
{{ outputs.evaluate_policy.vars.warning_count }}
Version-policy findings:
{{ outputs.evaluate_policy.vars.version_policy_count }}
# ==================================================
# 7D. PROMOTE ONLY PASS OR WARN
# ==================================================
- id: promote_if_accepted
type: io.kestra.plugin.core.flow.If
description: |
Promote the candidate only when governance
returns PASS or WARN.
condition: |
{{
outputs.evaluate_policy.vars.verdict != 'BLOCK'
}}
then:
- id: promote_baseline_spec
type: io.kestra.plugin.core.kv.Set
description: |
Store the accepted candidate specification
as the new persistent baseline.
key: "api-evolution-watchtower-{{ fromJson(item.value).name }}-spec"
kvType: STRING
overwrite: true
value: "{{ read(outputs.fetch_revision_spec.uri) }}"
- id: promote_baseline_hash
type: io.kestra.plugin.core.kv.Set
description: Store the accepted revision hash.
key: "api-evolution-watchtower-{{ fromJson(item.value).name }}-hash"
kvType: STRING
overwrite: true
value: "{{ outputs.check_baseline.vars.revision_hash }}"
else:
- id: baseline_frozen
type: io.kestra.plugin.core.log.Log
message: |
Accepted baseline remains frozen.
API:
{{ fromJson(item.value).name }}
Candidate was BLOCKED.
# ======================================================
# UNCHANGED
# ======================================================
else:
- id: unchanged_log
type: io.kestra.plugin.core.log.Log
message: |
API specification unchanged.
API:
{{ fromJson(item.value).name }}
Baseline source:
{{ outputs.check_baseline.vars.baseline_source }}
Baseline hash:
{{ outputs.check_baseline.vars.baseline_hash }}
Candidate hash:
{{ outputs.check_baseline.vars.revision_hash }}
oasdiff analysis skipped.
# ============================================================
# FINAL SUMMARY
# ============================================================
- id: final_summary
type: io.kestra.plugin.scripts.shell.Commands
description: |
Aggregate governance results from all API iterations.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.12-slim
outputFiles:
- governance_summary.json
commands:
- |
python3 - <<'PY'
import json
api_names = {{
loopOutputs(
outputs.monitor_apis.outputs,
'api_name'
) | toJson
}}
verdicts = {{
loopOutputs(
outputs.monitor_apis.outputs,
'verdict'
) | toJson
}}
breaking_counts = {{
loopOutputs(
outputs.monitor_apis.outputs,
'breaking_count'
) | toJson
}}
warning_counts = {{
loopOutputs(
outputs.monitor_apis.outputs,
'warning_count'
) | toJson
}}
version_policy_counts = {{
loopOutputs(
outputs.monitor_apis.outputs,
'version_policy_count'
) | toJson
}}
baseline_sources = {{
loopOutputs(
outputs.monitor_apis.outputs,
'baseline_source'
) | toJson
}}
unchanged_values = {{
loopOutputs(
outputs.monitor_apis.outputs,
'unchanged'
) | toJson
}}
results = []
blocked_count = 0
warning_count = 0
passed_count = 0
unchanged_count = 0
for i, api_name in enumerate(api_names):
verdict = (
verdicts[i]
if i < len(verdicts)
else "UNKNOWN"
)
breaking = int(
breaking_counts[i]
) if i < len(breaking_counts) else 0
warnings = int(
warning_counts[i]
) if i < len(warning_counts) else 0
version_policy = int(
version_policy_counts[i]
) if i < len(version_policy_counts) else 0
baseline_source = (
baseline_sources[i]
if i < len(baseline_sources)
else "unknown"
)
unchanged = (
str(unchanged_values[i]).lower() == "true"
if i < len(unchanged_values)
else False
)
if verdict == "BLOCK":
blocked_count += 1
elif verdict == "WARN":
warning_count += 1
elif verdict == "PASS":
passed_count += 1
if unchanged:
unchanged_count += 1
results.append({
"api_name": api_name,
"verdict": verdict,
"breaking_count": breaking,
"warning_count": warnings,
"version_policy_count": version_policy,
"baseline_source":
baseline_source,
"unchanged":
unchanged,
})
overall_verdict = (
"BLOCK"
if blocked_count > 0
else (
"WARN"
if warning_count > 0
else "PASS"
)
)
summary = {
"overall_verdict":
overall_verdict,
"api_count":
len(api_names),
"blocked_count":
blocked_count,
"warning_count":
warning_count,
"passed_count":
passed_count,
"unchanged_count":
unchanged_count,
"apis":
results,
}
with open(
"governance_summary.json",
"w",
encoding="utf-8"
) as f:
json.dump(
summary,
f,
indent=2
)
print(
"::"
+ json.dumps(
{"outputs": summary}
)
+ "::"
)
print(
"Processed APIs:",
len(api_names)
)
print(
"Overall verdict:",
overall_verdict
)
print(
"BLOCK:",
blocked_count
)
print(
"WARN:",
warning_count
)
print(
"PASS:",
passed_count
)
print(
"UNCHANGED:",
unchanged_count
)
PY
# ============================================================
# FINAL GOVERNANCE GATE
# ============================================================
- id: final_governance_gate
type: io.kestra.plugin.core.flow.If
description: |
Fail the overall execution only after every API has
been evaluated.
condition: |
{{
outputs.final_summary.vars.blocked_count > 0
}}
then:
- id: final_block
type: io.kestra.plugin.core.execution.Fail
errorMessage: |
API governance BLOCKED one or more API revisions.
APIs processed:
{{ outputs.final_summary.vars.api_count }}
BLOCK:
{{ outputs.final_summary.vars.blocked_count }}
WARN:
{{ outputs.final_summary.vars.warning_count }}
PASS:
{{ outputs.final_summary.vars.passed_count }}
UNCHANGED:
{{ outputs.final_summary.vars.unchanged_count }}
Review:
outputs.final_summary.outputFiles['governance_summary.json']
else:
- id: final_success
type: io.kestra.plugin.core.log.Log
message: |
API governance completed successfully.
APIs processed:
{{ outputs.final_summary.vars.api_count }}
Overall verdict:
{{ outputs.final_summary.vars.overall_verdict }}
BLOCK:
{{ outputs.final_summary.vars.blocked_count }}
WARN:
{{ outputs.final_summary.vars.warning_count }}
PASS:
{{ outputs.final_summary.vars.passed_count }}
UNCHANGED:
{{ outputs.final_summary.vars.unchanged_count }}
Abstract
API Evolution Watchtower is a consumer-side API governance workflow for teams
that depend on third-party APIs they do not control. It monitors OpenAPI
specifications, detects contract changes with oasdiff, filters findings to
operations the consumer actually uses, and determines whether a revision should
be accepted or blocked.
The accepted API specification is persisted in the Kestra KV store, allowing future executions to compare new revisions against the last accepted baseline instead of repeatedly comparing against the original bootstrap specification.
How it works
- A scheduled trigger starts the Watchtower.
- The
Loopprocesses every configured API independently. - The baseline and candidate OpenAPI specifications are downloaded.
- SHA-256 hashes are calculated so unchanged specifications can skip expensive change analysis.
oasdiffgenerates a machine-readable changelog for changed specifications.- Findings are filtered against the consumer's configured operations.
- Changes are classified as PASS, WARN, or BLOCK according to the governance policy.
- PASS and WARN revisions are promoted to the persistent KV baseline.
- BLOCK revisions leave the previously accepted baseline frozen.
- A final summary aggregates the governance result across all monitored APIs.
What it does
- Monitors multiple third-party OpenAPI specifications in one flow.
- Detects changed specifications using SHA-256 hashing.
- Runs
oasdiffonly when a candidate differs from the accepted baseline. - Filters findings to operations actually used by the consumer.
- Applies consumer-side PASS / WARN / BLOCK governance policy.
- Persists accepted OpenAPI baselines in Kestra KV.
- Promotes PASS and WARN revisions and freezes the baseline on BLOCK.
- Produces per-API Markdown reports and an aggregate JSON governance summary.
Prerequisites
- Kestra with Docker task runner access.
- No secrets are required for the included public-spec demo.
- Internet access for the public OpenAPI fixtures and the oasdiff repository.
- Docker access for the included
python:3.12-slimandtufin/oasdiff:v1.32.1task containers.
Quick start
- Import the Blueprint into Kestra.
- Run the flow manually with the included example API list.
- Review the per-API governance results and final summary.
- Replace the example API definitions with your own baseline URL, revision URL, and consumer-used operations.
- Enable the schedule after reviewing the baseline and governance policy.
Inputs
apis: Array of API definitions. Each item contains:name: unique API identifier.baseline_url: bootstrap OpenAPI specification URL.revision_url: candidate OpenAPI specification URL.used_operations: consumer operations such asGET /users/*.
Outputs
outputs.final_summary.outputFiles['governance_summary.json']: aggregate PASS/WARN/BLOCK results for all monitored APIs.outputs.evaluate_policy.outputFiles['report.md']: detailed per-API governance report.
Links
Orchestrate with Kestra
Orchestrate Git with KestraCreate a Blueprint
Go to the developer platformMore Infrastructure Blueprints
More Business Blueprints
