AI Flow Generation and Compiler Validation for Kestra

id: ai-flow-architect-and-self-healing-compiler
namespace: company.ai

description: |
  Generate Kestra 2.0 flow YAML from a text description with Google Gemini, validate it against
  Kestra's validation API, automatically fix compiler errors if any are returned, and optionally deploy
  the flow behind a Pause approval gate.

inputs:
  - id: workflow_requirement
    type: STRING
    displayName: Workflow Requirement Specification
    description: Natural language specification describing the desired orchestration
      pipeline.
    defaults: "Build an automated pipeline that runs daily at 08:00 UTC, polls a
      REST endpoint for active orders, checks record counts in DuckDB, and
      dispatches a Slack alert on failure."

  - id: target_namespace
    type: STRING
    displayName: Target Namespace
    description: Namespace where the generated workflow will be deployed upon approval.
    defaults: company.analytics

  - id: kestra_url
    type: STRING
    displayName: Kestra Webserver URL
    description: Base URL of the Kestra cluster used for compiler validation and deployment.
    defaults: "http://localhost:8080"

  - id: api_token
    type: STRING
    displayName: Kestra API Token / Authorization Header
    description: Optional bearer token or credentials for authenticating against
      secured Kestra clusters. Leave empty for unauthenticated local instances.
    defaults: ""

  - id: model_name
    type: STRING
    displayName: Gemini Foundational Model
    description: Foundational model used for flow architecture synthesis and
      reflection repair.
    defaults: gemini-2.5-flash

  - id: auto_deploy
    type: BOOL
    displayName: Enable Cluster Deployment
    description: When true, prompts for human sign-off via Pause before creating the
      verified flow in the target namespace.
    defaults: false

tasks:
  - id: architect_initial_flow
    type: io.kestra.plugin.ai.completion.ChatCompletion
    description: Uses Google Gemini to generate a Kestra 2.0 flow from the
      requirement prompt.
    provider:
      type: io.kestra.plugin.ai.provider.GoogleGemini
      modelName: "{{ inputs.model_name }}"
      apiKey: "{{ secret('GEMINI_API_KEY') }}"
    messages:
      - type: USER
        content: |
          Generate a complete Kestra 2.0 flow in YAML for this requirement:
          "{{ inputs.workflow_requirement }}"

          Follow these requirements:
          1. Flow namespace must be "{{ inputs.target_namespace }}".
          2. Flow id must be a lowercase, hyphenated slug derived from the requirement (e.g. daily-order-sync).
          3. For iterations, use 'io.kestra.plugin.core.flow.Loop' with '{{ item.value }}'.
          4. For conditions, use 'io.kestra.plugin.core.flow.If'.
          5. Use '{{ secret("SECRET_NAME") }}' for credentials.
          6. Return ONLY the raw YAML without markdown code fences.

  - id: clean_draft_yaml
    type: io.kestra.plugin.core.output.OutputValues
    description: Strips markdown code fences from the generated YAML if present.
    values:
      flow_yaml: >-
        {{
          outputs.architect_initial_flow.textOutput
          | replace('```yaml\n', '')
          | replace('```yaml', '')
          | replace('```\n', '')
          | replace('```', '')
          | trim
        }}

  - id: validate_draft_compiler_syntax
    type: io.kestra.plugin.core.http.Request
    description: Submits the generated flow YAML to Kestra's official compiler
      validation API.
    uri: "{{ inputs.kestra_url }}/api/v1/flows/validate"
    method: POST
    contentType: application/x-yaml
    body: "{{ outputs.clean_draft_yaml.values.flow_yaml }}"
    headers: "{{ (inputs.api_token is not empty) ? {'Authorization':
      (inputs.api_token contains ' ' ? inputs.api_token : 'Bearer ' ~
      inputs.api_token)} : {} }}"

  - id: evaluate_validation_diagnostics
    type: io.kestra.plugin.core.output.OutputValues
    description: Checks whether Kestra's validation API returned any errors.
    values:
      has_errors: >-
        {{
          (outputs.validate_draft_compiler_syntax.body | jq('[.[]? | select(.constraints != null or .errors != null)]') | length) > 0
        }}
      diagnostic_summary: >-
        {{
          outputs.validate_draft_compiler_syntax.body | jq('[.[]? | .constraints // .errors // empty] | join("; ")')
        }}

  - id: handle_compiler_reflection
    type: io.kestra.plugin.core.flow.If
    description: If validation errors were returned, asks Gemini to fix them and
      revalidates.
    condition: "{{ outputs.evaluate_validation_diagnostics.values.has_errors == 'true' }}"
    then:
      - id: reflect_and_repair_flow
        type: io.kestra.plugin.ai.completion.ChatCompletion
        description: Sends validation errors back to Gemini to fix the YAML.
        provider:
          type: io.kestra.plugin.ai.provider.GoogleGemini
          modelName: "{{ inputs.model_name }}"
          apiKey: "{{ secret('GEMINI_API_KEY') }}"
        messages:
          - type: USER
            content: |
              The Kestra validation API returned these errors:
              "{{ outputs.evaluate_validation_diagnostics.values.diagnostic_summary }}"

              Here is the rejected YAML:
              ---
              {{ outputs.clean_draft_yaml.values.flow_yaml }}
              ---

              Fix all errors so the YAML passes validation. Correct any invalid task classes, property names, or Pebble syntax.
              Return ONLY the corrected YAML without markdown code fences.

      - id: clean_repaired_yaml
        type: io.kestra.plugin.core.output.OutputValues
        description: Strips code fences from the repaired flow YAML.
        values:
          flow_yaml: >-
            {{
              outputs.reflect_and_repair_flow.textOutput
              | replace('```yaml\n', '')
              | replace('```yaml', '')
              | replace('```\n', '')
              | replace('```', '')
              | trim
            }}

      - id: revalidate_repaired_flow
        type: io.kestra.plugin.core.http.Request
        description: Re-submits repaired YAML to Kestra's validation API to verify the
          fixes.
        uri: "{{ inputs.kestra_url }}/api/v1/flows/validate"
        method: POST
        contentType: application/x-yaml
        body: "{{ outputs.clean_repaired_yaml.values.flow_yaml }}"
        headers: "{{ (inputs.api_token is not empty) ? {'Authorization':
          (inputs.api_token contains ' ' ? inputs.api_token : 'Bearer ' ~
          inputs.api_token)} : {} }}"
    else:
      - id: log_first_pass_clean
        type: io.kestra.plugin.core.log.Log
        description: Logs when the initial draft passed validation with zero errors.
        message: "Generated flow passed validation on the first attempt."

  - id: resolve_final_yaml
    type: io.kestra.plugin.core.output.OutputValues
    description: Selects the repaired YAML if errors were fixed, or the initial
      draft if it was clean.
    values:
      final_yaml: "{{ outputs.clean_repaired_yaml.values.flow_yaml ??
        outputs.clean_draft_yaml.values.flow_yaml }}"

  - id: notify_slack_preview
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Posts a preview of the validated flow to Slack.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :robot_face: *Kestra Flow Generation Summary*
      *Target Namespace:* `{{ inputs.target_namespace }}`
      *Compiler Status:* :white_check_mark: Clean
      *Repair Step:* `{{ outputs.evaluate_validation_diagnostics.values.has_errors == 'true' ? 'Fixed via compiler feedback' : 'Clean on first attempt' }}`
      *Specification:*
      >{{ inputs.workflow_requirement }}

      *YAML Preview:*
      ```yaml
      {{ outputs.resolve_final_yaml.values.final_yaml }}
      ```

  - id: evaluate_deployment_intent
    type: io.kestra.plugin.core.flow.If
    description: If auto_deploy is enabled, pauses for approval and saves the flow.
    condition: "{{ inputs.auto_deploy }}"
    then:
      - id: approve_production_deployment
        type: io.kestra.plugin.core.flow.Pause
        description: Pauses for approval before saving the flow to the cluster.
        onResume:
          - id: approved
            type: BOOL
            description: Approve creating this flow in the target namespace.
            defaults: true

      - id: deploy_flow_to_cluster
        type: io.kestra.plugin.core.http.Request
        description: Saves the validated flow to Kestra via POST /api/v1/flows.
        uri: "{{ inputs.kestra_url }}/api/v1/flows"
        method: POST
        contentType: application/x-yaml
        body: "{{ outputs.resolve_final_yaml.values.final_yaml }}"
        headers: "{{ (inputs.api_token is not empty) ? {'Authorization':
          (inputs.api_token contains ' ' ? inputs.api_token : 'Bearer ' ~
          inputs.api_token)} : {} }}"

      - id: notify_deployment_success
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Sends a confirmation to Slack that the flow was created.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          :rocket: *Kestra Flow Created*
          *Namespace:* `{{ inputs.target_namespace }}`
          *Status:* Flow created in cluster.
          Open the Kestra UI to review or run it.

triggers:
  - id: intake_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Webhook trigger for generating flows from external tools or webhooks.
    key: "{{ secret('AI_ARCHITECT_WEBHOOK_KEY') }}"

errors:
  - id: alert_architect_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Sends an alert to Slack if flow generation fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :x: *Flow Generation Failed*
      *Execution ID:* `{{ execution.id }}`
      *Flow:* `{{ flow.id }}`
      Check execution logs in Kestra UI for details.

Asking an LLM to write a workflow often produces YAML with small errors: an unquoted Pebble expression, a misspelled task property, or a missing required field. When someone copies that YAML straight into Kestra, it fails validation and requires manual troubleshooting.

This blueprint generates Kestra 2.0 YAML from a natural language requirement using Google Gemini, posts the draft directly to Kestra's /api/v1/flows/validate endpoint, and if any compiler violations are returned, feeds those errors back to the model to fix the YAML before posting a preview to Slack.

Safety

Validation gate: The flow checks the YAML with Kestra's validation API before anything is posted or saved.
Bounded repair: If the first draft has errors, it runs a single repair pass with the compiler output rather than looping indefinitely.
No automatic deployment: auto_deploy defaults to false. The flow only generates and tests the YAML.
Operator approval: When auto_deploy is enabled, a Pause task requires manual confirmation before creating the flow in the cluster.
Namespace boundary: Generated flows are placed into target_namespace.

How it works

architect_initial_flow (io.kestra.plugin.ai.completion.ChatCompletion) prompts Google Gemini to write a Kestra 2.0 flow for the requirement.
clean_draft_yaml (io.kestra.plugin.core.output.OutputValues) removes markdown code fences from the model output.
validate_draft_compiler_syntax (io.kestra.plugin.core.http.Request) sends the draft YAML to Kestra's /api/v1/flows/validate endpoint.
evaluate_validation_diagnostics (io.kestra.plugin.core.output.OutputValues) checks whether the compiler returned any constraint violations.
handle_compiler_reflection (io.kestra.plugin.core.flow.If) branches: if errors occurred, reflect_and_repair_flow sends the compiler message back to Gemini to fix the YAML, then revalidate_repaired_flow tests the result.
resolve_certified_yaml picks the repaired YAML if reflection ran, or the first draft if it was clean.
notify_slack_preview (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) posts the requirement and the final YAML to Slack.
evaluate_deployment_intent (io.kestra.plugin.core.flow.If) pauses for approval if auto_deploy is enabled, then saves the flow to Kestra via POST /api/v1/flows.

What you get

A draft Kestra 2.0 flow generated from a plain English prompt.
Automatic validation against Kestra's compiler API before human review.
An automatic repair step that uses the compiler error message to fix broken YAML.
A Slack preview showing the prompt, the validation outcome, and the final YAML.
An optional approval gate to deploy the validated flow directly into your cluster.

Who it's for

Platform Engineers building internal developer platforms and self-service workflow generators.
Data Engineers scaffolding pipelines from natural language requirements.
Teams looking for AI-assisted workflow generation with automated validation.

Why orchestrate this with Kestra

Kestra can validate flows using its own API and handle human approvals with native Pause tasks. Combining AI generation with native validation in a single flow gives you an automated, reviewable process without external wrapper scripts.

Pitfalls

Ambiguous prompts: Vague specifications like "sync data" will generate generic HTTP tasks. Provide concrete endpoints, schemas, or schedules in the prompt for best results.
Custom plugin availability: The flow generates standard Kestra core and official plugin tasks. Ensure any specialized plugins referenced in generated YAML are installed on your cluster.
API authentication: When validating or deploying against secured clusters, provide a valid token in api_token.

Prerequisites

Accessible Kestra 2.0+ cluster webserver API.
Google Gemini API key configured as a Kestra secret.
Slack incoming webhook URL for notifications.

Secrets

GEMINI_API_KEY: API key for Google Gemini completion tasks.
SLACK_WEBHOOK_URL: Slack incoming webhook URL for flow previews and alerts.
AI_ARCHITECT_WEBHOOK_KEY: Secret key for authenticating the inbound trigger webhook.

Quick start

Configure GEMINI_API_KEY and SLACK_WEBHOOK_URL in your Kestra namespace secrets.
Run the flow with auto_deploy: false to test generation, validation, and Slack preview.
Review the generated YAML delivered to Slack.
Enable auto_deploy: true and resume the Pause approval to create the flow in Kestra.

How to extend

Swap the GoogleGemini provider for OpenAI or Anthropic based on your organization's standards.
Add a GitHub pull request creation task using io.kestra.plugin.git to commit the validated YAML to a repository instead of direct cluster deployment.
Extend the validation gate with custom governance policies (e.g. enforcing mandatory tags or execution timeouts).

Generate and Validate Kestra Flows with AI and Compiler Reflection

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Pitfalls

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Generate and Validate Kestra Flows with AI and Compiler Reflection

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Pitfalls

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

Daily arXiv AI digest with keyword ranking

Generate a book chapter summary using Perplexity AI and email it.

AI Contract Clause Risk Analyzer and Legal Sign-Off Gate

Convert Natural Language Prompts to Structured Todoist Tasks with Deepseek AI

AI Website Accessibility Auditor

Summarize Text with an AI Agent, Generate Multilingual Summaries in Short, Medium, or Long Format

More Core Blueprints

AI Incident Post-Mortem and Root Cause Generator

Convert Natural Language Prompts to Structured Todoist Tasks with Deepseek AI

Route AI Requests to the Optimal Model Based on Query Classification

AI Multi-Provider Consensus Gate and Governance Adjudicator

Advanced Scheduling Automation, Run Tasks Only on Business Days by Country

AI Contract Clause Risk Analyzer and Legal Sign-Off Gate

New to Kestra?