Monitor MCP tool contract drift

id: mcp-tool-contract-monitor
namespace: company.team
description: Monitor the agent-facing contract of MCP-enabled flows in a namespace.

inputs:
  - id: monitored_namespace
    type: STRING
    defaults: company.team
    description: Namespace containing the MCP-enabled flows to monitor. The monitor
      may live elsewhere.
  - id: accept_current_contracts
    type: BOOL
    defaults: false
    description: Explicitly accept the current contracts after reviewing the drift report.
  - id: fail_on_breaking
    type: BOOL
    defaults: false
    description: Fail the execution when unaccepted breaking contract changes are found.
  - id: notify_on_drift
    type: BOOL
    defaults: false
    description: Send a Slack alert for unaccepted drift. Requires the
      SLACK_WEBHOOK_URL secret and Slack plugin.

triggers:
  - id: periodic_contract_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check contracts every six hours. Enable after setting monitored_namespace.
    cron: "0 */6 * * *"
    disabled: true

concurrency:
  behavior: QUEUE
  limit: 1

tasks:
  - id: collect_state
    type: io.kestra.plugin.core.flow.Parallel
    description: Read current flow definitions and the accepted baseline concurrently.
    tasks:
      - id: list_flows
        type: io.kestra.plugin.kestra.flows.List
        namespace: "{{ inputs.monitored_namespace }}"
      - id: export_flows
        type: io.kestra.plugin.kestra.flows.Export
        description: Export authoritative YAML sources because the typed List response
          omits plugin-specific trigger fields.
        namespace: "{{ inputs.monitored_namespace }}"
      - id: get_baseline
        type: io.kestra.plugin.core.kv.Get
        key: "mcp-contract-{{ inputs.monitored_namespace | slugify }}"
        errorOnMissing: false

  - id: analyze_contracts
    type: io.kestra.plugin.scripts.python.Script
    description: Normalize MCP contracts, compare the accepted baseline, and write
      both reports.
    containerImage: python:3.13-slim
    beforeCommands:
      - pip install kestra PyYAML
    executionContext: true
    inputFiles:
      flows.zip: "{{ outputs.export_flows.flowsZip }}"
    outputFiles:
      - mcp-contract-report.json
      - mcp-contract-report.md
    script: |
      import json
      import re
      import zipfile
      from datetime import datetime, timezone
      from pathlib import Path

      import yaml


      TRIGGER_TYPE = "io.kestra.plugin.core.trigger.McpToolTrigger"
      ANNOTATION_DEFAULTS = {
          "readOnly": False,
          "destructive": True,
          "openWorld": True,
          "idempotent": False,
          "returnDirect": False,
      }
      INPUT_FIELDS = (
          "defaults", "prefill", "description", "displayName", "itemType", "values",
          "jsonSchema", "min", "max", "validator", "allowedFileExtensions",
          "after", "before", "minLength", "maxLength", "pattern", "dependsOn",
      )
      OUTPUT_FIELDS = ("displayName", "description")


      def clean(value):
          if isinstance(value, dict):
              return {key: clean(value[key]) for key in sorted(value)}
          if isinstance(value, list):
              return [clean(item) for item in value]
          if isinstance(value, str) and "secret(" in value.lower():
              return "<secret reference>"
          return value


      def normalize_input(definition):
          result = {
              "id": definition["id"],
              "type": definition["type"],
              "required": definition.get("required") is not False,
          }
          for field in INPUT_FIELDS:
              if field not in definition or definition[field] is None:
                  continue
              if definition["type"] == "SECRET" and field in ("defaults", "prefill"):
                  result["has" + field.title()] = True
              else:
                  result[field] = clean(definition[field])
          return result


      def normalize_output(definition):
          result = {
              "id": definition["id"], "type": definition["type"],
              "required": definition.get("required") is not False,
          }
          for field in OUTPUT_FIELDS:
              if field in definition and definition[field] is not None:
                  result[field] = clean(definition[field])
          return result


      def expand_inputs(definitions):
          for definition in definitions:
              if definition.get("type") == "FORM":
                  for child in definition.get("inputs") or []:
                      leaf = dict(child)
                      leaf["id"] = definition["id"] + "." + child["id"]
                      yield normalize_input(leaf)
              else:
                  yield normalize_input(definition)


      def normalize_tool(flow, trigger):
          annotations = trigger.get("annotations") or {}
          if not isinstance(annotations, dict):
              raise ValueError("MCP annotations must be an object")
          inputs = flow.get("inputs") or []
          outputs = flow.get("outputs") or []
          if not isinstance(inputs, list) or not isinstance(outputs, list):
              raise ValueError("Flow inputs and outputs must be lists")
          return {
              "namespace": flow["namespace"],
              "flowId": flow["id"],
              "triggerId": trigger["id"],
              "toolName": trigger["toolName"],
              "title": trigger["title"],
              "toolDescription": trigger["toolDescription"],
              "mcpServer": trigger.get("mcpServer") or "default",
              "annotations": {
                  key: annotations.get(key) if annotations.get(key) is not None else default
                  for key, default in ANNOTATION_DEFAULTS.items()
              },
              "inputs": sorted(expand_inputs(inputs), key=lambda item: item["id"]),
              "outputs": sorted((normalize_output(item) for item in outputs), key=lambda item: item["id"]),
          }


      def load_exported_flows(path):
          flows = []
          try:
              with zipfile.ZipFile(path) as archive:
                  members = sorted(name for name in archive.namelist() if name.endswith((".yaml", ".yml")))
                  for name in members:
                      source = archive.read(name).decode("utf-8")
                      definition = yaml.safe_load(source)
                      if not isinstance(definition, dict):
                          raise ValueError(f"exported file '{name}' is not a flow object")
                      flows.append(definition)
          except (OSError, UnicodeDecodeError, zipfile.BadZipFile, yaml.YAMLError) as error:
              raise ValueError(f"Cannot read the exported flow definitions: {error}") from error
          return flows


      def validate_inventory(listed_flows, exported_flows, namespace):
          if not isinstance(listed_flows, list):
              raise ValueError("outputs.list_flows.flows must be a list")
          listed = {
              (flow.get("namespace"), flow.get("id"))
              for flow in listed_flows
              if isinstance(flow, dict) and flow.get("namespace") == namespace and flow.get("disabled") is not True
          }
          exported = {
              (flow.get("namespace"), flow.get("id"))
              for flow in exported_flows
              if flow.get("namespace") == namespace and flow.get("disabled") is not True
          }
          if listed != exported:
              missing = sorted(listed - exported)
              extra = sorted(exported - listed)
              raise ValueError(f"List/export flow inventory mismatch; missing={missing}, extra={extra}")


      def build_snapshot(flows, namespace):
          if not isinstance(flows, list):
              raise ValueError("outputs.list_flows.flows must be a list")
          tools = {}
          for flow in flows:
              if not isinstance(flow, dict) or "id" not in flow or "namespace" not in flow:
                  raise ValueError("flows.List returned a malformed flow definition")
              if flow["namespace"] != namespace or flow.get("disabled") is True:
                  continue
              triggers = flow.get("triggers") or []
              if not isinstance(triggers, list):
                  raise ValueError("Flow triggers must be a list")
              for trigger in triggers:
                  if trigger.get("type") != TRIGGER_TYPE or trigger.get("disabled") is True:
                      continue
                  tool = normalize_tool(flow, trigger)
                  key = "/".join((flow["namespace"], flow["id"], trigger["id"]))
                  if key in tools:
                      raise ValueError("Duplicate MCP tool identity: " + key)
                  tools[key] = tool
          return {
              "snapshotVersion": 1,
              "monitoredNamespace": namespace,
              "generatedAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
              "tools": {key: tools[key] for key in sorted(tools)},
          }


      def coerce_baseline(value, key, namespace):
          if value is None:
              return None
          try:
              baseline = json.loads(value) if isinstance(value, str) else value
              if not isinstance(baseline, dict):
                  raise ValueError("value is not a JSON object")
              if baseline.get("snapshotVersion") != 1:
                  raise ValueError("unsupported snapshotVersion")
              if baseline.get("monitoredNamespace") != namespace:
                  raise ValueError("monitoredNamespace does not match the requested namespace")
              tools = baseline.get("tools")
              if not isinstance(tools, dict):
                  raise ValueError("tools is not an object")
              for tool_id, tool in tools.items():
                  if not isinstance(tool_id, str) or not isinstance(tool, dict):
                      raise ValueError("tool entries must be named objects")
                  for field in ("namespace", "flowId", "triggerId", "toolName", "title", "toolDescription", "mcpServer"):
                      if not isinstance(tool.get(field), str):
                          raise ValueError(f"tool '{tool_id}' has invalid {field}")
                  if not isinstance(tool.get("annotations"), dict):
                      raise ValueError("tool annotations must be objects")
                  if not isinstance(tool.get("inputs"), list) or not isinstance(tool.get("outputs"), list):
                      raise ValueError("tool inputs and outputs must be lists")
                  for member in tool["inputs"] + tool["outputs"]:
                      if not isinstance(member, dict) or not isinstance(member.get("id"), str) or not isinstance(member.get("type"), str):
                          raise ValueError(f"tool '{tool_id}' has a malformed input or output")
              return baseline
          except (TypeError, ValueError, json.JSONDecodeError) as error:
              raise ValueError(
                  f"Baseline KV key '{key}' is malformed: {error}. "
                  "Delete or reset that KV entry manually only if replacing the accepted baseline is intentional."
              ) from error


      def change(tool, category, path, before, after, severity, breaking, message):
          return {
              "tool": tool, "category": category, "path": path,
              "before": before, "after": after, "severity": severity,
              "breaking": breaking, "message": message,
          }


      def diff_fields(tool_id, old, new):
          changes = []
          tool_fields = {
              "toolName": ("TOOL_NAME_CHANGED", "MEDIUM", True),
              "title": ("TOOL_TITLE_CHANGED", "INFO", False),
              "toolDescription": ("TOOL_DESCRIPTION_CHANGED", "MEDIUM", False),
              "mcpServer": ("MCP_SERVER_CHANGED", "HIGH", True),
          }
          for field, (category, severity, breaking) in tool_fields.items():
              if old.get(field) != new.get(field):
                  changes.append(change(tool_id, category, field, old.get(field), new.get(field),
                                        severity, breaking, f"{field} changed."))
          annotation_rules = {
              "readOnly": ("READ_ONLY_CHANGED", True, False),
              "destructive": ("DESTRUCTIVE_CHANGED", False, True),
              "openWorld": ("OPEN_WORLD_CHANGED", False, True),
              "idempotent": ("IDEMPOTENT_CHANGED", True, False),
              "returnDirect": ("RETURN_DIRECT_CHANGED", None, None),
          }
          for field, (category, risky_before, risky_after) in annotation_rules.items():
              before, after = old["annotations"].get(field), new["annotations"].get(field)
              if before != after:
                  risky = before is risky_before and after is risky_after if risky_before is not None else False
                  severity = "CRITICAL" if risky and field in ("readOnly", "destructive") else (
                      "HIGH" if risky else ("MEDIUM" if field == "returnDirect" else "INFO"))
                  changes.append(change(tool_id, category, f"annotations.{field}", before, after,
                                        severity, risky, f"Annotation '{field}' changed."))
          changes.extend(diff_members(tool_id, "inputs", old["inputs"], new["inputs"]))
          changes.extend(diff_members(tool_id, "outputs", old["outputs"], new["outputs"]))
          return changes


      def diff_members(tool_id, kind, old_items, new_items):
          old = {item["id"]: item for item in old_items}
          new = {item["id"]: item for item in new_items}
          changes = []
          for name in sorted(old.keys() | new.keys()):
              path = f"{kind}.{name}"
              if name not in old:
                  item = new[name]
                  required_without_default = kind == "inputs" and item["required"] and not usable_default(item)
                  category = "INPUT_ADDED" if kind == "inputs" else "OUTPUT_ADDED"
                  severity = "HIGH" if required_without_default else ("MEDIUM" if kind == "inputs" else "INFO")
                  changes.append(change(tool_id, category, path, None, item, severity,
                                        required_without_default, f"{kind[:-1].title()} '{name}' was added."))
                  continue
              if name not in new:
                  category = "INPUT_REMOVED" if kind == "inputs" else "OUTPUT_REMOVED"
                  changes.append(change(tool_id, category, path, old[name], None, "HIGH", True,
                                        f"{kind[:-1].title()} '{name}' was removed."))
                  continue
              for field in sorted(old[name].keys() | new[name].keys()):
                  if field == "id" or old[name].get(field) == new[name].get(field):
                      continue
                  before, after = old[name].get(field), new[name].get(field)
                  if kind == "inputs":
                      category, severity, breaking = classify_input(field, before, after, new[name])
                  else:
                      category, severity, breaking = classify_output(field, before, after)
                  changes.append(change(tool_id, category, f"{path}.{field}", before, after,
                                        severity, breaking, f"{kind[:-1].title()} '{name}' {field} changed."))
          return changes


      def usable_default(item):
          value = item.get("defaults")
          if value is None or value == "" or value == "<secret reference>":
              return False
          if isinstance(value, str) and chr(123) * 2 in value:
              return False
          if isinstance(value, dict) and ("expression" in value or "secret" in value):
              return False
          return True


      def classify_input(field, before, after, current):
          if field == "type":
              return "INPUT_TYPE_CHANGED", "HIGH", True
          if field == "required":
              breaking = before is False and after is True and not usable_default(current)
              return "INPUT_REQUIRED_CHANGED", "HIGH" if breaking else "INFO", breaking
          if field in ("defaults", "hasDefaults"):
              breaking = after is None and current["required"]
              return "INPUT_DEFAULT_CHANGED", "HIGH" if breaking else "MEDIUM", breaking
          if field in ("prefill", "hasPrefill"):
              return "INPUT_PREFILL_CHANGED", "INFO", False
          if field == "description":
              return "INPUT_DESCRIPTION_CHANGED", "MEDIUM", False
          if field == "displayName":
              return "INPUT_DISPLAY_NAME_CHANGED", "INFO", False
          if field in ("jsonSchema", "itemType", "values"):
              return "INPUT_SCHEMA_CHANGED", "HIGH", True
          return "INPUT_CONSTRAINT_CHANGED", "HIGH", True


      def classify_output(field, before, after):
          if field == "type":
              return "OUTPUT_TYPE_CHANGED", "HIGH", True
          if field == "displayName":
              return "OUTPUT_DISPLAY_NAME_CHANGED", "INFO", False
          if field == "description":
              return "OUTPUT_DESCRIPTION_CHANGED", "MEDIUM", False
          return "OUTPUT_SCHEMA_CHANGED", "HIGH", True


      def diff_snapshots(previous, current):
          old, new = previous["tools"], current["tools"]
          changes = []
          for tool_id in sorted(old.keys() | new.keys()):
              if tool_id not in old:
                  changes.append(change(tool_id, "TOOL_ADDED", "tool", None, new[tool_id],
                                        "INFO", False, "MCP tool was added."))
              elif tool_id not in new:
                  changes.append(change(tool_id, "TOOL_REMOVED", "tool", old[tool_id], None,
                                        "CRITICAL", True, "MCP tool was removed."))
              else:
                  changes.extend(diff_fields(tool_id, old[tool_id], new[tool_id]))
          return changes


      def render_markdown(report):
          counts = report["severityCounts"]
          lines = [
              "# MCP Tool Contract Report", "",
              f"Namespace: `{report['monitoredNamespace']}`  ",
              f"Status: **{report['status']}**", "", "## Summary", "",
              f"- Tools monitored: {report['toolCount']}",
              f"- Changes: {report['changeCount']}",
              f"- Breaking changes: {report['breakingCount']}",
              f"- Critical: {counts['critical']}", f"- High: {counts['high']}",
              f"- Medium: {counts['medium']}", f"- Info: {counts['info']}", "",
          ]
          if not report["changed"]:
              lines.extend(["No MCP contract drift was detected.", ""])
          else:
              for heading, breaking in (("Breaking Changes", True), ("Non-Breaking Drift", False)):
                  subset = [item for item in report["changes"] if item["breaking"] is breaking]
                  if not subset:
                      continue
                  lines.extend([f"## {heading}", ""])
                  for tool_id in sorted({item["tool"] for item in subset}):
                      lines.extend([f"### `{tool_id}`", ""])
                      for item in subset:
                          if item["tool"] == tool_id:
                              detail = item["message"]
                              if item["path"] != "tool":
                                  before = json.dumps(item["before"], ensure_ascii=False, sort_keys=True).replace("`", "'")
                                  after = json.dumps(item["after"], ensure_ascii=False, sort_keys=True).replace("`", "'")
                                  detail = f"`{before}` → `{after}`"
                              lines.append(f"- {item['severity']}, `{item['path']}`: {detail}")
                      lines.append("")
          lines.extend(["## Baseline", ""])
          if report["firstRun"]:
              lines.append("The current contracts were stored as the initial baseline.")
          elif report["status"] == "DRIFT_ACCEPTED":
              lines.append("The drift above was reported before the current contracts were accepted as the new baseline.")
          else:
              lines.append("The previous baseline was preserved.")
              if report["changed"]:
                  lines.extend(["", "To accept these contracts, rerun with `accept_current_contracts=true`."])
          return "\n".join(lines) + "\n"


      def analyze(flows, baseline_value, namespace, accepted, baseline_key):
          current = build_snapshot(flows, namespace)
          previous = coerce_baseline(baseline_value, baseline_key, namespace)
          first_run = previous is None
          changes = [] if first_run else diff_snapshots(previous, current)
          changed = bool(changes)
          status = ("BASELINE_INITIALIZED" if first_run else
                    "DRIFT_ACCEPTED" if changed and accepted else
                    "DRIFT_DETECTED" if changed else "NO_DRIFT")
          counts = {severity.lower(): sum(item["severity"] == severity for item in changes)
                    for severity in ("CRITICAL", "HIGH", "MEDIUM", "INFO")}
          report = {
              "snapshotVersion": 1, "status": status, "monitoredNamespace": namespace,
              "generatedAt": current["generatedAt"], "toolCount": len(current["tools"]),
              "firstRun": first_run, "changed": changed, "changeCount": len(changes),
              "breakingCount": sum(item["breaking"] for item in changes),
              "severityCounts": counts, "changes": changes, "currentSnapshot": current,
          }
          return report


      def main():
          from kestra import Kestra

          context = json.loads(Path(".kestra-execution-context.json").read_text())
          namespace = context["inputs"]["monitored_namespace"]
          accepted = context["inputs"]["accept_current_contracts"]
          listed_flows = context["outputs"]["list_flows"]["flows"]
          flows = load_exported_flows("flows.zip")
          validate_inventory(listed_flows, flows, namespace)
          baseline = context["outputs"].get("get_baseline", {}).get("value")
          slug = re.sub(r"[^\w-]", "", namespace).replace("_", "-").lower()
          baseline_key = "mcp-contract-" + re.sub(r"-+", "-", slug).strip("-")
          report = analyze(flows, baseline, namespace, accepted, baseline_key)
          Path("mcp-contract-report.json").write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
          Path("mcp-contract-report.md").write_text(render_markdown(report))
          Kestra.outputs({
              "status": report["status"], "first_run": report["firstRun"],
              "changed": report["changed"], "tool_count": report["toolCount"],
              "change_count": report["changeCount"], "breaking_count": report["breakingCount"],
              "critical_count": report["severityCounts"]["critical"],
              "high_count": report["severityCounts"]["high"],
              "medium_count": report["severityCounts"]["medium"],
              "info_count": report["severityCounts"]["info"],
              "snapshot": report["currentSnapshot"],
              "summary": f"{report['changeCount']} changes, {report['breakingCount']} breaking across {report['toolCount']} tools",
              "alert_summary": "\n".join(
                  f"- {item['severity']}: {item['tool']} ({item['path']})"
                  for item in report["changes"][:5]
              ) + (f"\n- ...and {len(report['changes']) - 5} more changes" if len(report["changes"]) > 5 else ""),
          })


      if __name__ == "__main__":
          main()

  - id: report_drift
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.analyze_contracts.vars.changed }}"
    then:
      - id: log_drift
        type: io.kestra.plugin.core.log.Log
        message: "MCP contract drift: {{ outputs.analyze_contracts.vars.change_count }}
          changes, {{ outputs.analyze_contracts.vars.breaking_count }} breaking.
          Reports: {{
          outputs.analyze_contracts.outputFiles['mcp-contract-report.md'] }} and
          {{ outputs.analyze_contracts.outputFiles['mcp-contract-report.json']
          }}."
    else:
      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        message: "MCP contract check complete: {{
          outputs.analyze_contracts.vars.tool_count }} tools, no drift
          detected."

  - id: alert_on_drift
    type: io.kestra.plugin.core.flow.If
    description: Notify the team only about drift that still needs review.
    condition: "{{ inputs.notify_on_drift and outputs.analyze_contracts.vars.changed
      and not inputs.accept_current_contracts }}"
    then:
      - id: send_slack_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Send a short drift digest and point reviewers to the full execution
          reports.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          MCP tool contract drift in {{ inputs.monitored_namespace }} (execution {{ execution.id }}).
          {{ outputs.analyze_contracts.vars.change_count }} changes, {{ outputs.analyze_contracts.vars.breaking_count }} breaking; {{ outputs.analyze_contracts.vars.critical_count }} critical and {{ outputs.analyze_contracts.vars.high_count }} high.
          {{ outputs.analyze_contracts.vars.alert_summary }}
          Open the execution outputs for the Markdown and JSON reports. The accepted baseline has not changed.

  - id: persist_baseline
    type: io.kestra.plugin.core.flow.If
    description: Write only on first run or after the operator explicitly accepts
      current contracts.
    condition: "{{ outputs.analyze_contracts.vars.first_run or
      inputs.accept_current_contracts }}"
    then:
      - id: set_baseline
        type: io.kestra.plugin.core.kv.Set
        key: "mcp-contract-{{ inputs.monitored_namespace | slugify }}"
        kvType: JSON
        overwrite: true
        value: "{{ outputs.analyze_contracts.vars.snapshot | toJson }}"
      - id: log_baseline
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.analyze_contracts.vars.changed and
          inputs.accept_current_contracts }}"
        then:
          - id: log_acceptance
            type: io.kestra.plugin.core.log.Log
            message: "MCP contract drift reviewed and current contracts accepted as the new
              baseline."
        else:
          - id: log_initial_baseline
            type: io.kestra.plugin.core.log.Log
            message: "MCP contract baseline stored for {{ inputs.monitored_namespace }}."

  - id: fail_breaking_gate
    type: io.kestra.plugin.core.execution.Fail
    condition: "{{ inputs.fail_on_breaking and
      outputs.analyze_contracts.vars.breaking_count > 0 and not
      inputs.accept_current_contracts }}"
    errorMessage: "{{ outputs.analyze_contracts.vars.breaking_count }} breaking MCP
      contract changes detected. The previous baseline was preserved. Review the
      Markdown and JSON reports, then rerun with accept_current_contracts=true
      if the new contract is intentional."

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.analyze_contracts.vars.status }}"
  - id: changed
    type: BOOL
    value: "{{ outputs.analyze_contracts.vars.changed }}"
  - id: breakingChanges
    type: INT
    value: "{{ outputs.analyze_contracts.vars.breaking_count }}"
  - id: totalChanges
    type: INT
    value: "{{ outputs.analyze_contracts.vars.change_count }}"
  - id: toolCount
    type: INT
    value: "{{ outputs.analyze_contracts.vars.tool_count }}"
  - id: jsonReport
    type: STRING
    value: "{{ outputs.analyze_contracts.outputFiles['mcp-contract-report.json'] }}"
  - id: markdownReport
    type: STRING
    value: "{{ outputs.analyze_contracts.outputFiles['mcp-contract-report.md'] }}"