Kestra

Kestra é uma plataforma open-source de orquestração de workflows que permite definir, executar e monitorar pipelines complexos usando YAML simples. Fundada em 2021 e com suporte a 600+ integrações, é usada por empresas como Apple, Toyota, Bloomberg e JPMorgan Chase.

O grande diferencial é resolver a escolha falsa entre plataformas sem-código (inflexíveis) e frameworks em-código (complexos). No Kestra, você escreve YAML — qualquer engenheiro entende, qualquer linguagem funciona.

Problemas que resolve

GitHub & Comunidade

Fluxo de Execução

4 Pilares de Infraestrutura

Stack Tecnológico

3.1 Docker (teste rápido — sem persistência)

# Iniciar Kestra (banco H2 em memória — reiniciar apaga tudo)
docker run --pull=always --rm -it \
  -p 8080:8080 \
  --user=root \
  -v /var/run/docker.sock:/var/run/docker.sock \
  -v /tmp/kestra-storage:/app/storage \
  kestra/kestra:latest server local

# Acessar UI
open http://localhost:8080

3.2 Docker Compose com PostgreSQL (recomendado para dev)

# docker-compose.yml
version: '3.8'
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: kestra
      POSTGRES_USER: kestra
      POSTGRES_PASSWORD: kestra
    volumes:
      - postgres-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -d kestra -U kestra"]
      interval: 5s

  kestra:
    image: kestra/kestra:latest
    entrypoint: /bin/bash
    command: -c "kestra server standalone"
    depends_on:
      postgres:
        condition: service_healthy
    environment:
      KESTRA_CONFIGURATION: |
        datasources:
          postgres:
            url: jdbc:postgresql://postgres:5432/kestra
            username: kestra
            password: kestra
        kestra:
          repository:
            type: postgres
          storage:
            type: local
            local:
              base-path: /app/storage
    ports:
      - "8080:8080"
      - "8081:8081"
    volumes:
      - kestra-data:/app/storage
      - /var/run/docker.sock:/var/run/docker.sock

volumes:
  postgres-data:
  kestra-data:

# Subir
docker compose up -d

# Verificar logs
docker compose logs -f kestra

# Parar
docker compose down

3.3 Kubernetes com Helm (produção)

# Adicionar repositório
helm repo add kestra https://helm.kestra.io
helm repo update

# Deploy em desenvolvimento (inclui PostgreSQL + MinIO)
helm install kestra kestra/kestra-starter

# Deploy em produção (configure seu próprio PostgreSQL)
helm install kestra kestra/kestra \
  --set kestra.configuration.datasources.postgres.url="jdbc:postgresql://seu-postgres:5432/kestra" \
  --set kestra.configuration.datasources.postgres.username="kestra" \
  --set kestra.configuration.datasources.postgres.password="sua-senha" \
  --set kestra.configuration.kestra.storage.type="s3" \
  --set kestra.configuration.kestra.storage.s3.bucket="seu-bucket"

3.4 Variáveis de Ambiente Importantes

4.1 Estrutura Mínima

id: hello_world                  # identificador único dentro do namespace
namespace: company.team          # organização hierárquica (imutável após criação)
description: |
  Meu primeiro workflow no Kestra

tasks:
  - id: log_mensagem
    type: io.kestra.plugin.core.log.Log
    message: "Hello, Kestra! 🚀"

4.2 Inputs, Triggers, Outputs e Error Handling

id: pipeline_completo
namespace: company.data

# Inputs fortemente tipados (STRING, INTEGER, BOOLEAN, FILE, DATE...)
inputs:
  - id: data_ref
    type: STRING
    defaults: "{{ now | date('yyyy-MM-dd') }}"
  - id: batch_size
    type: INTEGER
    defaults: 1000

# Triggers: agendamento, webhook, evento de arquivo, mensageria
triggers:
  - id: execucao_diaria
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 2 * * *"           # toda noite às 2h

tasks:
  - id: buscar_dados
    type: io.kestra.plugin.core.http.Request
    uri: "https://api.exemplo.com/dados?data={{ inputs.data_ref }}"
    method: GET

  - id: transformar
    type: io.kestra.plugin.scripts.python.Script
    script: |
      import json
      dados = json.loads('{{ outputs.buscar_dados.body }}')
      print(json.dumps([{**item, "processado": True} for item in dados]))

  - id: salvar_banco
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: "{{ secret('DATABASE_URL') }}"
    sql: |
      INSERT INTO dados_processados (data, conteudo)
      VALUES ('{{ inputs.data_ref }}', '{{ outputs.transformar.vars.result }}')

# Output do flow
outputs:
  - id: resultado
    type: STRING
    value: "Concluído em {{ now }}"

# Error handling: executa se qualquer task falhar
errors:
  - id: notificar_falha
    type: io.kestra.plugin.slack.IncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {"text": "❌ Pipeline falhou: {{ exception.message }}"}

4.3 Paralelismo, ForEach e Condicional

id: processamento_paralelo
namespace: company.analytics

tasks:
  # Execução paralela de múltiplas tasks
  - id: paralelo
    type: io.kestra.plugin.core.flow.Parallel
    concurrencyLimit: 3              # máximo 3 simultâneas (0 = ilimitado)
    tasks:
      - id: fonte_a
        type: io.kestra.plugin.scripts.python.Script
        script: "print('processando fonte A')"
      - id: fonte_b
        type: io.kestra.plugin.scripts.python.Script
        script: "print('processando fonte B')"

  # Iteração (ForEach) com lista dinâmica
  - id: para_cada_item
    type: io.kestra.plugin.core.flow.ForEach
    values: "{{ outputs.fonte_a.vars.lista | json }}"
    concurrencyLimit: 5
    tasks:
      - id: processar_item
        type: io.kestra.plugin.core.log.Log
        message: "Processando: {{ item }}"

  # Condicional: só executa se condição for verdadeira
  - id: condicional
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.fonte_a.vars.count > 0 }}"
    then:
      - id: sucesso
        type: io.kestra.plugin.core.log.Log
        message: "Dados encontrados!"
    else:
      - id: sem_dados
        type: io.kestra.plugin.core.log.Log
        message: "Nenhum dado para processar"

4.4 Retry, Timeout e Tolerância a Falhas

tasks:
  - id: api_instavel
    type: io.kestra.plugin.core.http.Request
    uri: "https://api-instavel.exemplo.com/dados"
    timeout: PT30S                    # 30 segundos por tentativa
    retry:
      type: EXPONENTIAL               # CONSTANT ou EXPONENTIAL
      interval: PT5S                  # intervalo inicial
      maxInterval: PT5M              # intervalo máximo
      maxAttempts: 5
      multiplier: 2.0               # dobra a cada tentativa

  # Permite falha sem parar o flow
  - id: step_opcional
    type: io.kestra.plugin.core.flow.AllowFailure
    tasks:
      - id: tarefa_opcional
        type: io.kestra.plugin.core.log.Log
        message: "Esta task pode falhar sem problemas"

4.5 Subflows (Modularidade)

# flow principal: chama subflow reutilizável
id: pipeline_principal
namespace: company.pipelines

tasks:
  - id: notificar
    type: io.kestra.plugin.core.flow.Subflow
    namespace: company.helpers
    flowId: enviar_notificacao
    inputs:
      canal: "#equipe-dados"
      mensagem: "Pipeline concluído! {{ execution.id }}"
    wait: true                       # aguarda subflow finalizar

4.6 Webhook Trigger

id: api_trigger
namespace: company.webhooks

triggers:
  - id: receber_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: "{{ secret('WEBHOOK_SECRET') }}"

tasks:
  - id: processar
    type: io.kestra.plugin.core.log.Log
    message: "Webhook recebido: {{ trigger.body | json }}"

URL gerada automaticamente: https://seu-kestra/api/v1/executions/webhook/{namespace}/{flowId}/{key}

Cada plugin expõe tasks e triggers que se integram diretamente ao YAML. Disponíveis no Kestra Plugin Marketplace.

Exemplo: Script Python com dependências

- id: analise_python
  type: io.kestra.plugin.scripts.python.Script
  beforeCommands:
    - pip install pandas numpy scikit-learn -q    # instala no container isolado
  script: |
    import pandas as pd
    import numpy as np

    df = pd.read_csv("{{ inputs.arquivo }}")
    resultado = df.groupby('categoria').agg({'valor': 'sum'})
    print(resultado.to_json())

Kestra 1.0 introduz suporte nativo a AI Agents — workflows dinâmicos que raciocinam, usam ferramentas e se adaptam autonomamente. Integração com 8+ providers de LLM diretamente no YAML.

6.1 Provedores LLM Suportados

6.2 Chat Completion Simples

id: resumo_com_claude
namespace: company.ai

inputs:
  - id: texto
    type: STRING

tasks:
  - id: chamar_claude
    type: io.kestra.plugin.ai.anthropic.ChatCompletion
    model: claude-3-5-sonnet-20241022
    apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
    systemMessage: |
      Você é um especialista em resumos concisos.
      Resuma em no máximo 3 bullet points.
    userMessage: |
      Resuma este texto:
      {{ inputs.texto }}
    maxTokens: 500
    temperature: 0.3

outputs:
  - id: resumo
    type: STRING
    value: "{{ outputs.chamar_claude.message }}"

6.3 AI Agent com Ferramentas (Autonomo)

id: agente_pesquisador
namespace: company.ai

inputs:
  - id: topico
    type: STRING

tasks:
  - id: agente
    type: io.kestra.plugin.ai.agent.Agent
    llmProvider: anthropic
    model: claude-3-5-sonnet-20241022
    apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
    systemMessage: |
      Você é um pesquisador. Busque informações sobre o tópico,
      analise os resultados e produza um relatório estruturado.
    userMessage: "Pesquise sobre: {{ inputs.topico }}"
    tools:
      - type: WebSearch            # busca na web
      - type: CodeExecution        # executa código Python
        language: python
      - type: FlowCaller           # chama outros flows
        namespace: company.helpers
      - type: McpClient            # conecta a MCP servers
        serverUrl: "http://meu-mcp:3000"
    maxIterations: 10          # limite de iterações do agente

6.4 Pipeline Multi-Agente

id: pipeline_multi_agente
namespace: company.ai

tasks:
  # Agente 1: coleta e resume dados
  - id: agente_coletor
    type: io.kestra.plugin.ai.agent.Agent
    llmProvider: anthropic
    model: claude-3-5-haiku-20241022  # modelo mais rápido/barato
    apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
    systemMessage: "Colete e resuma os dados brutos"
    userMessage: "Processe: {{ inputs.dados_brutos }}"
    tools:
      - type: CodeExecution

  # Agente 2: analisa e decide (modelo mais poderoso)
  - id: agente_analista
    type: io.kestra.plugin.ai.agent.Agent
    llmProvider: anthropic
    model: claude-3-5-sonnet-20241022
    apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
    systemMessage: "Analise o resumo e tome decisões"
    userMessage: "Análise: {{ outputs.agente_coletor.output }}"

  # Notificar resultado
  - id: notificar
    type: io.kestra.plugin.slack.IncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {"text": "{{ outputs.agente_analista.output }}"}

6.5 AI Copilot (IDE)

O AI Copilot integrado converte linguagem natural em YAML:

• Descreva em inglês/português → gera o flow YAML completo
• Auto-complete inteligente para plugins e configurações
• Servidor MCP oficial para integração com Claude Code IDE
• Busca de documentação alimentada por IA diretamente na UI

O Kestra se encaixa no AIOS como a camada de orquestração de workflows de longa duração e pipelines — complementando o Claude Code (desenvolvimento) e n8n (automações simples).

Casos de Uso AIOS com Kestra

Integração com Ferramentas AIOS

Exemplo: Pipeline Lesson Analysis no Kestra

id: lesson_analysis_pipeline
namespace: aios.lesson_squad

inputs:
  - id: vtt_file_path
    type: STRING
  - id: tipo_sessao
    type: STRING       # AULA ou PS

tasks:
  # 5 agentes em paralelo
  - id: agentes_paralelos
    type: io.kestra.plugin.core.flow.Parallel
    tasks:
      - id: radar
        type: io.kestra.plugin.ai.anthropic.ChatCompletion
        model: claude-3-5-haiku-20241022
        apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
        systemMessage: "Você é o agente Radar. Analise sentimentos e emoções."
        userMessage: "{{ read(inputs.vtt_file_path) }}"
      - id: coletor
        type: io.kestra.plugin.ai.anthropic.ChatCompletion
        model: claude-3-5-haiku-20241022
        apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
        systemMessage: "Você é o agente Coletor. Extraia links e recursos."
        userMessage: "{{ read(inputs.vtt_file_path) }}"
      - id: didatico
        type: io.kestra.plugin.ai.anthropic.ChatCompletion
        model: claude-3-5-sonnet-20241022
        apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
        systemMessage: "Você é o agente Didático. Analise a qualidade pedagógica."
        userMessage: "{{ read(inputs.vtt_file_path) }}"
      - id: gps
        type: io.kestra.plugin.ai.anthropic.ChatCompletion
        model: claude-3-5-sonnet-20241022
        apiKey: "{{ secret('ANTHROPIC_API_KEY') }}"
        systemMessage: "Você é o agente GPS. Avalie os 10 aspectos GPS e 7 pilares ML."
        userMessage: "{{ read(inputs.vtt_file_path) }}"

  # Compilar resultados e gerar páginas HTML
  - id: gerar_html
    type: io.kestra.plugin.scripts.python.Script
    script: |
      # compilar outputs dos agentes e gerar HTML
      radar = """{{ outputs.radar.message }}"""
      gps   = """{{ outputs.gps.message }}"""
      # ... gera lesson-pages HTML ...

  # Deploy no GitHub Pages / Vercel
  - id: publicar
    type: io.kestra.plugin.scripts.shell.Commands
    commands:
      - git add . && git commit -m "feat: análise {{ now }}" && git push

Comparativo Rápido

Kestra vs n8n (foco AIOS)

Kestra vs Airflow

Recomendação para AIOS

Custo Estimado de Infraestrutura (Self-hosted)

Comandos Docker

Estrutura de um Flow (anatomia completa)

id: meu_flow              # obrigatório — único no namespace
namespace: empresa.equipe  # obrigatório — hierárquico com pontos
description: |            # markdown suportado
  Descrição do flow

labels:                   # tags para organização
  env: production
  team: data

inputs:                   # parâmetros de entrada tipados
  - id: nome
    type: STRING
    defaults: "valor_padrao"
    required: false

variables:                # variáveis estáticas do flow
  api_url: "https://api.exemplo.com"

triggers:                 # como o flow é ativado
  - id: agendamento
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"

tasks:                    # lista de tasks (executam sequencialmente por padrão)
  - id: task_1
    type: io.kestra.plugin.core.log.Log
    message: "Iniciando"

outputs:                  # outputs exportados pelo flow
  - id: resultado
    type: STRING
    value: "{{ outputs.task_1.value }}"

errors:                   # tasks executadas em caso de falha
  - id: alerta_falha
    type: io.kestra.plugin.core.log.Log
    message: "Falhou: {{ exception.message }}"

concurrency:             # controle de execuções simultâneas
  limit: 1              # apenas 1 execução por vez
  behavior: CANCEL      # CANCEL ou QUEUE a nova

Links Úteis