The Kestrel Python SDK provides a typed, fluent API for building and managing workflows programmatically. It supports both synchronous and asynchronous usage with API key authentication.

Installation

pip install kestrel-workflows

Authentication

Create an API key in the Kestrel platform under Workflows > API Keys, then use it to authenticate:

from kestrel import KestrelClient

client = KestrelClient(api_key="kestrel_sk_...")

For async usage:

from kestrel import AsyncKestrelClient

async with AsyncKestrelClient(api_key="kestrel_sk_...") as client:
    workflows = await client.workflows.list()

If you’re logged in via the CLI, the SDK can reuse your session:

client = KestrelClient.from_config()

Workflow Builder

Build workflows using typed classes with full IDE autocompletion:

from kestrel import KestrelClient
from kestrel.workflows import Workflow, Trigger, Action, Condition, Approval, PollUntil

wf = (
    Workflow("Pod Crash RCA + Jira")
    .description("Run RCA on pod crash, create Jira ticket")
    .trigger(
        Trigger.k8s_pod_status()
        .reasons("CrashLoopBackOff")
        .cluster("my-cluster-id")
        .namespace("production")
    )
    .cooldown(hours=24)
    .then(Action.kestrel_trigger_rca().label("Run RCA"))
    .then(Action.jira_create_ticket()
        .project("KAN")
        .title("{{incident.title}}")
        .priority("High")
        .assignee("Raman Varma")
        .label("Create Jira Ticket")
    )
    .then(Action.confluence_publish_rca()
        .space_key("SD")
        .title("RCA: {{incident.title}}")
        .label("Publish RCA")
    )
    .alert_on_failure(channel="alerts-channel-id")
)

client = KestrelClient(api_key="kestrel_sk_...")
created = client.workflows.deploy(wf, activate=True)
print(f"Deployed: {created.id} ({created.status})")

Triggers

Create triggers for different signal sources:

# Kubernetes — Pod Status
Trigger.k8s_pod_status().reasons("CrashLoopBackOff")
Trigger.k8s_pod_status().reasons("ImagePullBackOff", "ErrImagePull")
Trigger.k8s_pod_status().reasons("OOMKilled")  # uses terminated_reasons
Trigger.k8s_pod_status().reasons("Error")  # pod exits non-zero
Trigger.k8s_pod_status().restart_threshold(5)  # pods restarting excessively

# Kubernetes — Workload Rollout
Trigger.k8s_rollout_status().reasons("ProgressDeadlineExceeded").resource_kinds("Deployment")
Trigger.k8s_rollout_status().reasons("ReplicasNotReady").resource_kinds("StatefulSet")
Trigger.k8s_rollout_status().reasons("PodsUnavailable").resource_kinds("DaemonSet")

# Kubernetes — Node Conditions
Trigger.k8s_node_condition().conditions("NotReady")
Trigger.k8s_node_condition().conditions("MemoryPressure")
Trigger.k8s_node_condition().conditions("DiskPressure")

# Kubernetes — Catch-all
Trigger.k8s_any()

# AWS — CloudTrail (infrastructure events)
Trigger.aws_cloudtrail().service_names("iam", "sts")
Trigger.aws_cloudtrail().service_names("s3")
Trigger.aws_cloudtrail().service_names("ec2")
Trigger.aws_cloudtrail().service_names("rds")
Trigger.aws_cloudtrail().service_names("lambda")
Trigger.aws_cloudtrail().service_names("dynamodb")
Trigger.aws_cloudtrail().service_names("kms")
Trigger.aws_cloudtrail().service_names("secretsmanager")
Trigger.aws_cloudtrail().incident_types("RootAccountActivity")

# AWS — Application Errors
Trigger.aws_cloudwatch_log().incident_types("ApplicationLogErrors").service_names("ec2")

# AWS — CloudWatch Alarms
Trigger.aws_cloudwatch_metric()

# AWS — Security & Compliance
Trigger.aws_security_hub().severities("CRITICAL", "HIGH")
Trigger.aws_config_rule()
Trigger.aws_service_health()

# AWS — Cost & Budget
Trigger.aws_cost_anomaly()
Trigger.aws_budget_alert()
Trigger.aws_forecast_overrun()            # forecast projects a budget overrun
Trigger.aws_spend_spike()                 # day-over-day spend spike
Trigger.aws_idle_resource()               # daily idle-resource scan hit

# AWS — Catch-all
Trigger.aws_any()

# Schedule (recurring workflows — no external event needed)
Trigger.schedule_hourly()                                  # fires every hour
Trigger.schedule_daily("09:00")                            # daily at 09:00 UTC
Trigger.schedule_weekly(day_of_week=1, time_utc="09:00")   # Mondays 09:00 UTC (0=Sunday)
Trigger.schedule_monthly(day_of_month=1, time_utc="06:00") # 1st of month 06:00 UTC

# PagerDuty
Trigger.pagerduty_triggered()
Trigger.pagerduty_acknowledged()
Trigger.pagerduty_resolved()
Trigger.pagerduty_any()
Trigger.pagerduty_triggered().urgency("high")

# GitHub
Trigger.github_pr_opened()
Trigger.github_pr_merged()
Trigger.github_pr_approved()
Trigger.github_action_completed()
Trigger.github_action_failed()
Trigger.github_push()                                                          # New commits pushed to branch

# PostHog
Trigger.posthog_exception()
Trigger.posthog_rage_click()
Trigger.posthog_log_alert()
Trigger.posthog_console_error()
Trigger.posthog_any()

# Vercel
Trigger.vercel_deployment_failed()
Trigger.vercel_deployment_ready()
Trigger.vercel_deployment_created()
Trigger.vercel_checks_failed()
Trigger.vercel_rollback()
Trigger.vercel_firewall_attack()
Trigger.vercel_error_anomaly()
Trigger.vercel_usage_anomaly()
Trigger.vercel_domain_issue()

# Railway
Trigger.railway_deployment_failed()
Trigger.railway_deployment_crashed()
Trigger.railway_deployment_succeeded()
Trigger.railway_volume_alert()
Trigger.railway_resource_alert()
Trigger.railway_any()

# Fly.io (poll-based)
Trigger.flyio_machine_crashed()
Trigger.flyio_machine_stopped()
Trigger.flyio_machine_started()
Trigger.flyio_app_down()
Trigger.flyio_any()

# Nebius AI Cloud (poll-based)
Trigger.nebius_gpu_error()
Trigger.nebius_maintenance_scheduled()
Trigger.nebius_node_not_ready()
Trigger.nebius_instance_stopped()
Trigger.nebius_any()

# Jenkins (webhook via notification plugin or post-build step)
Trigger.jenkins_build_failed()
Trigger.jenkins_build_unstable()          # e.g. test failures
Trigger.jenkins_build_succeeded()
Trigger.jenkins_build_completed()         # any result
Trigger.jenkins_build_started()
Trigger.jenkins_any()

# CircleCI (native outbound webhooks, HMAC-signed)
Trigger.circleci_workflow_failed()        # failed or error
Trigger.circleci_workflow_succeeded()
Trigger.circleci_workflow_completed()     # any status
Trigger.circleci_job_failed()
Trigger.circleci_any()

# Terraform Cloud (webhook via workspace notification configurations)
Trigger.terraform_run_created()
Trigger.terraform_run_needs_attention()   # plan finished, awaiting confirmation
Trigger.terraform_run_completed()
Trigger.terraform_run_errored()
Trigger.terraform_drift_detected()        # health assessment found drift
Trigger.terraform_check_failed()
Trigger.terraform_any()

# Pulumi Cloud (webhook via organization or stack webhooks)
Trigger.pulumi_update_succeeded()
Trigger.pulumi_update_failed()
Trigger.pulumi_preview_failed()
Trigger.pulumi_destroy_succeeded()
Trigger.pulumi_deployment_started()
Trigger.pulumi_deployment_succeeded()
Trigger.pulumi_deployment_failed()
Trigger.pulumi_drift_detected()           # scheduled drift run found drift
Trigger.pulumi_policy_violation()         # mandatory policy pack blocked an update
Trigger.pulumi_stack_created()
Trigger.pulumi_stack_deleted()
Trigger.pulumi_any()

# Developer Requests (on-demand via Slack /kestrel-workflow, Kestrel Platform UI, or CLI)
Trigger.request_kubernetes()    # Kubernetes operations
Trigger.request_cloud()         # AWS / Cloud operations
Trigger.request_cloudflare()    # Cloudflare operations
Trigger.request_pagerduty()     # PagerDuty operations
Trigger.request_datadog()       # Datadog operations
Trigger.request_argocd()        # ArgoCD operations
Trigger.request_github()        # GitHub operations
Trigger.request_gitlab()        # GitLab operations
Trigger.request_helm()          # Helm operations
Trigger.request_vercel()        # Vercel operations
Trigger.request_jenkins()       # Jenkins operations
Trigger.request_circleci()      # CircleCI operations
Trigger.request_terraform()     # Terraform Cloud operations
Trigger.request_pulumi()        # Pulumi Cloud operations
Trigger.request_general()       # General (catch-all)

Each request trigger can be scoped further:

from kestrel.workflows import Trigger

# Kubernetes requests — only create operations from Slack
Trigger.request_kubernetes().config(
    request_sources=["slack"],
    request_types=["create"],
)

# Cloud requests — only from Platform UI and CLI
Trigger.request_cloud().config(
    request_sources=["platform", "cli"],
)

Submitting Requests Programmatically

# Submit a request (like /kestrel-workflow in Slack)
result = client.workflows.request("create a configmap for my-app in production")

# If clarification is needed:
if result.status == "awaiting_params":
    reply = client.workflows.request_reply(result.id, "use the prod-cluster, namespace=payments")

# If no workflow matches, the request is held as pending_confirmation.
# Confirm to send it to the platform team, or dismiss it:
if result.status == "pending_confirmation":
    client.workflows.request_confirm(result.id, True)   # file for the platform team
    # client.workflows.request_confirm(result.id, False)  # dismiss

The async client mirrors this via client.requests.confirm(request_id, confirm).

Custom Webhook

Trigger.custom_webhook(“my-event-type”)

Datadog Monitor

Trigger.datadog_monitor_alert() Trigger.datadog_monitor_warn() Trigger.datadog_monitor_recovered() Trigger.datadog_monitor_no_data() Trigger.datadog_monitor_any() Trigger.datadog_monitor_alert().monitor_names(“High CPU”).datadog_tags(“env:production”)

All triggers support filter chaining:

```python
trigger = (
    Trigger.k8s_pod_status()
    .reasons("CrashLoopBackOff")
    .cluster("cluster-id-1", "cluster-id-2")
    .namespace("production", "staging")
    .workload("api-server", "worker")
)

Fly.io triggers are poll-based — set the apps, regions, and poll cadence to control how often Kestrel checks the Fly Machines API:

trigger = (
    Trigger.flyio_machine_crashed()
    .fly_apps("my-app", "another-app")   # omit for all apps the token can see
    .fly_regions("iad", "sjc")
    .fly_poll_interval("1m")             # 1m | 5m | 15m | 30m
)

Nebius triggers are also poll-based — scope to projects/clusters and set the poll cadence to control how often Kestrel checks the Nebius compute and Managed Kubernetes APIs:

trigger = (
    Trigger.nebius_gpu_error()
    .nebius_projects("project-abc")      # omit for all projects the credentials can see
    .nebius_clusters("cluster-xyz")      # optional: restrict to specific clusters
    .nebius_poll_interval("5m")          # 1m | 5m | 15m | 30m
)

Jenkins and CircleCI triggers are webhook-based — scope to jobs/projects and optionally statuses:

trigger = (
    Trigger.jenkins_build_failed()
    .jenkins_jobs("platform/deploy-api")       # full path for folder-nested jobs; omit for all
    .jenkins_build_statuses("FAILURE")         # optional: SUCCESS | FAILURE | UNSTABLE | ABORTED
)

trigger = (
    Trigger.circleci_workflow_failed()
    .circleci_projects("gh/acme/api")          # project slugs; omit for all followed projects
    .circleci_branches("main")                 # optional: filter by branch
    .circleci_statuses("failed", "error")      # optional: filter by status
)

Terraform Cloud triggers are webhook-based — scope to workspaces and optionally run statuses:

trigger = (
    Trigger.terraform_run_errored()
    .terraform_workspaces("prod-vpc", "prod-eks")   # omit or use "*" for all workspaces
    .terraform_run_statuses("errored")              # optional: filter by run status
)

Pulumi Cloud triggers are webhook-based — scope to stacks (project/stack) and optionally projects:

trigger = (
    Trigger.pulumi_update_failed()
    .pulumi_stacks("my-project/prod")    # project/stack refs or bare stack names; omit or "*" for all
    .pulumi_projects("my-project")       # optional: filter by Pulumi project name
)

Actions

Every integration has typed factory methods:

# Kestrel — RCA & Investigation
Action.kestrel_trigger_rca()
Action.kestrel_trigger_cloud_rca()
Action.kestrel_generate_runbook()  # Distill RCA + fixes into a reusable runbook (requires an upstream RCA step)
Action.kestrel_investigate_k8s().query("Why is {{incident.workload_name}} crashing?")
Action.kestrel_investigate_cloud().query("Investigate IAM event")
Action.kestrel_find_causal_prs()
Action.kestrel_find_service_deps().cluster_id("...").ns("default").workload_name("api")
Action.kestrel_analyze_costs().analysis_type("anomaly")

# Kestrel — K8s Operations
Action.kestrel_apply_yaml_fix()
Action.kestrel_restart_workload().workload_name("{{incident.workload_name}}").namespace("{{incident.namespace}}")
Action.kestrel_scale_workload().workload_name("{{incident.workload_name}}").namespace("{{incident.namespace}}").replicas(0)
Action.kubectl_execute("get pods -n production")                           # Run any kubectl command
Action.generate_kubectl_command("Find pods in CrashLoopBackOff in production")  # AI generates a kubectl command
Action.kestrel_generate_k8s_manifest().resource_type("Deployment").query("Create a deployment for nginx")
Action.kestrel_generate_helm_values().chart_name("nginx-ingress")
Action.kestrel_apply_k8s_manifest()
Action.kestrel_create_gitops_pr().repo("org/k8s-manifests")

# Kestrel — Platform Engineering
Action.kestrel_generate_cloud_resource().cloud_service("rds").output_format("terraform")
Action.kestrel_execute_cloud_cli()
Action.kestrel_create_iac_pr().repo("org/infra-repo")

# Kestrel — Flow Control
Action.kestrel_wait().duration(5).unit("minutes")
PollUntil(Action.daytona_get_sandbox().sandbox_id("{{signal.sandbox_id}}"),
          Condition.equals("sandbox_state", "stopped")).every(seconds=60).timeout(minutes=30)

# Slack
Action.slack_send_message().channel("incidents").message("{{incident.title}}").include_rca(True)
Action.slack_update_message()
Action.slack_request_justification().channel("approvals")

# Jira
Action.jira_create_ticket().project("KAN").title("{{incident.title}}").priority("High")
Action.jira_add_comment().ticket_key("KAN-123").comment("Update: {{rca_result.root_cause}}")
Action.jira_transition_ticket().ticket_key("KAN-123").status("Done")

# Linear
Action.linear_create_issue().team("ENG").title("{{incident.title}}").priority("High")
Action.linear_add_comment().issue_identifier("ENG-123").body("Update: {{rca_result.root_cause}}")
Action.linear_update_issue().issue_identifier("ENG-123").status("Done")
Action.linear_search_issues().query("{{incident.title}}").team("ENG")

# GitHub
Action.github_create_pr().repo("org/repo").title("Fix: {{incident.title}}")
Action.github_create_issue().repo("org/repo").title("Bug: {{incident.title}}")
Action.github_investigate_code().query("{{rca_result.root_cause}}").repo("org/repo")
Action.github_analyze_push().repo("org/repo").head_sha("{{signal.commit_sha}}")     # Determine affected services from push
Action.github_generate_code_fix().context("{{rca_result.root_cause}}").repo("org/repo")
Action.github_trigger_workflow().repo("org/repo").workflow_file("deploy.yml")
Action.github_wait_for_run().repo("org/repo").workflow_file("deploy.yml")
Action.github_get_run_status().repo("org/repo").workflow_file("ci.yml")
Action.github_read_file().repo("org/repo").path("src/config.yaml")
Action.github_search_code().repo("org/repo").query("handleError")
Action.github_wait_pr_approval()
Action.github_wait_pr_merge()
Action.github_investigate_action_failure().repo("org/repo").run_id("{{signal.run_id}}")

# GitLab
Action.gitlab_create_mr().repo("org/repo").title("Fix: {{incident.title}}")
Action.gitlab_create_issue().repo("org/repo").title("Bug: {{incident.title}}")
Action.gitlab_trigger_pipeline().repo("org/repo").ref("main")
Action.gitlab_wait_for_pipeline().repo("org/repo")
Action.gitlab_get_pipeline_status().repo("org/repo")
Action.gitlab_wait_mr_approval()
Action.gitlab_wait_mr_merge()

# Confluence
Action.confluence_publish_rca().space_key("ENG")
Action.confluence_publish_postmortem().space_key("ENG")
Action.confluence_publish_runbook().space_key("ENG")
Action.confluence_update_page().page_id("123456")

# PagerDuty
Action.pagerduty_create_alert().service("service-id").severity("critical")
Action.pagerduty_acknowledge_alert()
Action.pagerduty_resolve_alert()
Action.pagerduty_add_note().note("RCA: {{rca_result.root_cause}}")
Action.pagerduty_escalate()

# Datadog
Action.datadog_query_metrics().cluster_id("...").query("avg:kubernetes.cpu.usage.total{*}")
Action.datadog_create_monitor().cluster_id("...").name("High CPU")
Action.datadog_send_event().cluster_id("...").title("Incident resolved")
Action.datadog_mute_monitor().cluster_id("...").monitor_id("12345")

# ArgoCD
Action.argocd_sync().cluster_id("...").app_name("my-app")
Action.argocd_wait_sync().cluster_id("...").app_name("my-app")
Action.argocd_get_status().cluster_id("...").app_name("my-app")
Action.argocd_rollback().cluster_id("...").app_name("my-app")
Action.argocd_find_app().cluster_id("...").workload_name("{{incident.workload_name}}").namespace("{{incident.namespace}}")

# Helm
Action.helm_upgrade().cluster_id("...").release_name("my-app").namespace("production").chart("oci://ghcr.io/org/app").values_set("image.tag={{signal.commit_sha}}")
Action.helm_install().cluster_id("...").release_name("my-app").namespace("production").chart("bitnami/nginx").create_namespace(True)
Action.helm_rollback().cluster_id("...").release_name("my-app").namespace("production")
Action.helm_uninstall().cluster_id("...").release_name("my-app").namespace("production")
Action.helm_status().cluster_id("...").release_name("my-app").namespace("production")

# AWS Cost — insights (read-only)
Action.aws_query_cost_explorer().aws_account("123456789012").time_range("30d")
Action.aws_get_cost_anomalies().aws_account("123456789012")
Action.aws_get_cost_forecast().aws_account("123456789012")
Action.aws_get_budget_status().aws_account("123456789012")
Action.aws_get_rightsizing_recommendations().aws_account("123456789012")
Action.aws_get_savings_plans_recommendations().aws_account("123456789012")
Action.aws_get_reservation_recommendations().aws_account("123456789012")
Action.aws_get_commitment_utilization().aws_account("123456789012")
Action.aws_compare_cost_periods().aws_account("123456789012")
Action.aws_find_idle_resources().aws_account("123456789012")
Action.aws_get_compute_optimizer_recommendations().aws_account("123456789012")
Action.aws_get_trusted_advisor_cost_checks().aws_account("123456789012")

# AWS Cost — remediation (destructive; place behind an Approval block.
# Resources tagged kestrel:protected are always skipped.)
Action.aws_stop_ec2_instances().aws_account("123456789012") \
    .config("instance_ids", "{{step_outputs.scan.idle_instance_ids}}").config("region", "us-east-1")
Action.aws_delete_unattached_ebs_volumes().aws_account("123456789012") \
    .config("volume_ids", "{{step_outputs.scan.idle_volume_ids}}").config("region", "us-east-1") \
    .config("snapshot_first", True)
Action.aws_release_elastic_ips().aws_account("123456789012") \
    .config("allocation_ids", "{{step_outputs.scan.idle_eip_allocation_ids}}").config("region", "us-east-1")
Action.aws_delete_old_snapshots().aws_account("123456789012") \
    .config("region", "us-east-1").config("age_days", 90)

# PostHog
Action.posthog_get_session_summary().session_ids("{{signal.session_id}}")
Action.posthog_get_session_recording().session_id("{{signal.session_id}}")
Action.posthog_query_events().query("SELECT event FROM events WHERE ...")
Action.posthog_list_session_recordings().limit(10)
Action.posthog_get_error_issue().issue_id("{{signal.properties.$exception_issue_id}}")

# Vercel
Action.vercel_get_deployment().deployment_id("{{signal.deployment_id}}")
Action.vercel_get_build_logs().deployment_id("{{signal.deployment_id}}")
Action.vercel_rollback().project_id("{{signal.project_id}}")
Action.vercel_promote().deployment_id("{{signal.deployment_id}}")
Action.vercel_list_deployments().project_id("{{signal.project_id}}")
Action.vercel_investigate().query("Why did the build fail?")

# Railway
Action.railway_get_deployment().deployment_id("{{signal.deployment_id}}")
Action.railway_get_deployment_logs().deployment_id("{{signal.deployment_id}}")
Action.railway_rollback().config("service_id", "{{signal.service_id}}").config("environment_id", "{{signal.environment_id}}")
Action.railway_redeploy().deployment_id("{{signal.deployment_id}}")
Action.railway_restart().deployment_id("{{signal.deployment_id}}")
Action.railway_list_deployments().config("service_id", "{{signal.service_id}}").config("status", "FAILED")
Action.railway_set_variables().config("service_id", "{{signal.service_id}}").config("environment_id", "{{signal.environment_id}}").config("variables", {"LOG_LEVEL": "debug"})
Action.railway_investigate().config("query", "Why did the service crash?")

# Fly.io (machine-scoped: require app_name + machine_id)
Action.flyio_restart_machine().app_name("{{signal.app_name}}").machine_id("{{signal.machine_id}}")
Action.flyio_start_machine().app_name("{{signal.app_name}}").machine_id("{{signal.machine_id}}")
Action.flyio_stop_machine().app_name("{{signal.app_name}}").machine_id("{{signal.machine_id}}")
Action.flyio_suspend_machine().app_name("{{signal.app_name}}").machine_id("{{signal.machine_id}}")
Action.flyio_cordon_machine().app_name("{{signal.app_name}}").machine_id("{{signal.machine_id}}")
Action.flyio_uncordon_machine().app_name("{{signal.app_name}}").machine_id("{{signal.machine_id}}")
Action.flyio_get_machine().app_name("{{signal.app_name}}").machine_id("{{signal.machine_id}}")
Action.flyio_get_machine_events().app_name("{{signal.app_name}}").machine_id("{{signal.machine_id}}")
Action.flyio_list_machines().app_name("{{signal.app_name}}").config("state", "started")
Action.flyio_set_secrets().app_name("{{signal.app_name}}").config("secrets", {"API_KEY": "..."})
Action.flyio_investigate().config("query", "Why did the machine crash?")

# Nebius AI Cloud (compute: project_id + instance_id; mk8s: cluster_id + node_group_id)
Action.nebius_get_instance().project_id("{{signal.project_id}}").instance_id("{{signal.instance_id}}")
Action.nebius_start_instance().project_id("{{signal.project_id}}").instance_id("{{signal.instance_id}}")
Action.nebius_stop_instance().project_id("{{signal.project_id}}").instance_id("{{signal.instance_id}}")
Action.nebius_restart_instance().project_id("{{signal.project_id}}").instance_id("{{signal.instance_id}}")
Action.nebius_list_instances().project_id("{{signal.project_id}}")
Action.nebius_list_clusters().project_id("{{signal.project_id}}")
Action.nebius_list_node_groups().cluster_id("{{signal.cluster_id}}")
Action.nebius_scale_node_group().cluster_id("{{signal.cluster_id}}").node_group_id("{{signal.node_group_id}}").size(3)
Action.nebius_investigate().config("query", "Why did the GPU node error?")

# Jenkins (job-scoped: .job(); build-scoped: .build_number())
Action.jenkins_trigger_build().job("platform/deploy-api").parameters("ENV=staging\nVERSION=1.2.3")
Action.jenkins_wait_for_build().job("{{signal.job_name}}").build_number("{{signal.build_number}}").timeout_minutes(30)
Action.jenkins_get_build_status().job("{{signal.job_name}}").build_number("{{signal.build_number}}")
Action.jenkins_stop_build().job("{{signal.job_name}}").build_number("{{signal.build_number}}")   # gate behind approval
Action.jenkins_get_console_log().job("{{signal.job_name}}").build_number("{{signal.build_number}}").max_lines(200)
Action.jenkins_investigate().query("Why did the nightly build fail?").job("{{signal.job_name}}")

# CircleCI (project-scoped: .project_slug(); workflow-scoped: .workflow_id())
Action.circleci_trigger_pipeline().project_slug("gh/acme/api").branch("main").parameters("deploy_env=staging")
Action.circleci_wait_for_pipeline().project_slug("{{signal.project_slug}}").pipeline_id("{{signal.pipeline_id}}").timeout_minutes(30)
Action.circleci_get_workflow_status().project_slug("{{signal.project_slug}}").pipeline_id("{{signal.pipeline_id}}")
Action.circleci_rerun_workflow().workflow_id("{{signal.workflow_id}}").from_failed(True)
Action.circleci_cancel_workflow().workflow_id("{{signal.workflow_id}}")
Action.circleci_approve_job().workflow_id("{{signal.workflow_id}}").job_name("hold-production")   # gate behind approval
Action.circleci_get_job_tests().project_slug("{{signal.project_slug}}").job_number("{{signal.job_number}}")
Action.circleci_investigate().query("Why did the main-branch workflow fail?").project_slug("{{signal.project_slug}}")

# Terraform Cloud (workspace-scoped: .workspace(); run-scoped: .run_id())
Action.terraform_list_workspaces()
Action.terraform_get_workspace().workspace("{{signal.workspace}}")
Action.terraform_lock_workspace().workspace("{{signal.workspace}}").reason("Incident freeze")
Action.terraform_unlock_workspace().workspace("{{signal.workspace}}")
Action.terraform_force_unlock_workspace().workspace("{{signal.workspace}}")   # gate behind approval
Action.terraform_list_runs().workspace("{{signal.workspace}}").limit(10)
Action.terraform_get_run().workspace("{{signal.workspace}}").run_id("{{signal.run_id}}")
Action.terraform_create_run().workspace("prod-vpc").run_message("Queued by Kestrel").auto_apply(False)
Action.terraform_create_destroy_run().workspace("staging-sandbox")           # gate behind approval
Action.terraform_apply_run().workspace("{{signal.workspace}}").run_id("{{signal.run_id}}").comment("Approved via Kestrel")  # gate behind approval
Action.terraform_discard_run().workspace("{{signal.workspace}}").run_id("{{signal.run_id}}")
Action.terraform_cancel_run().workspace("{{signal.workspace}}").run_id("{{signal.run_id}}")
Action.terraform_wait_for_run().workspace("{{signal.workspace}}").run_id("{{signal.run_id}}").timeout_minutes(30)
Action.terraform_get_state_outputs().workspace("prod-vpc")
Action.terraform_list_variables().workspace("prod-vpc")
Action.terraform_set_variable().workspace("prod-vpc").key("instance_count").value("3").category("terraform")  # gate behind approval
Action.terraform_get_drift().workspace("prod-vpc")
Action.terraform_investigate().query("Why did the latest run fail?").workspace("{{signal.workspace}}")

# Pulumi Cloud (stack-scoped: .stack() in project/stack format; deployment-scoped: .deployment_id())
Action.pulumi_list_stacks()
Action.pulumi_get_stack().stack("{{signal.stack}}")
Action.pulumi_list_updates().stack("{{signal.stack}}").limit(10)
Action.pulumi_get_update().stack("{{signal.stack}}").update_version("{{signal.update_version}}")
Action.pulumi_run_deployment().stack("my-project/prod").operation("update")   # gate destroy/remediate-drift behind approval
Action.pulumi_get_deployment().stack("{{signal.stack}}").deployment_id("{{signal.deployment_id}}")
Action.pulumi_wait_for_deployment().stack("{{signal.stack}}").deployment_id("{{signal.deployment_id}}").timeout_minutes(30)
Action.pulumi_cancel_deployment().stack("{{signal.stack}}").deployment_id("{{signal.deployment_id}}")
Action.pulumi_pause_deployments().stack("my-project/prod")     # incident/maintenance freeze
Action.pulumi_resume_deployments().stack("my-project/prod")
Action.pulumi_get_stack_outputs().stack("my-project/prod")
Action.pulumi_get_drift().stack("my-project/prod")
Action.pulumi_set_stack_tag().stack("{{signal.stack}}").tag_name("kestrel:quarantined").tag_value("true")
Action.pulumi_delete_stack_tag().stack("{{signal.stack}}").tag_name("kestrel:quarantined")
Action.pulumi_investigate().query("Why did the update fail?").stack("{{signal.stack}}")

# Approval Gates
Action.approval_manual()
Action.approval_slack().channel("approvals")
Action.approval_pr()

Conditions (Branching)

wf = (
    Workflow("Conditional RCA Response")
    .trigger(Trigger.k8s_pod_status().reasons("CrashLoopBackOff"))
    .then(Action.kestrel_trigger_rca())
    .then(Condition.equals("rca_result.is_application_level_failure", "true"))
    .on_true(Action.github_create_pr().repo("org/repo").label("Create Fix PR"))
    .on_false(Action.slack_send_message().channel("infra").label("Notify Infra Team"))
)

Available operators: equals, not_equals, contains, not_contains, exists, not_exists. The value-comparing factories accept multiple candidate values — equals/contains are met when the field matches any value, not_equals/not_contains only when it matches none:

Condition.equals("sandbox_state", "stopped", "error")       # met if stopped OR error
Condition.not_equals("phase", "Failed", "Degraded")         # met only if neither

Poll Until (Loops)

PollUntil builds a self-looping node that repeatedly executes ONE embedded catalog action at a fixed interval and exits on the met branch when the condition holds against that iteration’s output, or on the timeout branch when the timeout elapses. Wire the branches with .on_met() / .on_timeout():

wf = (
    Workflow("Snapshot stopped sandbox")
    .trigger(Trigger.custom_webhook("sandbox.watch"))
    .then(
        PollUntil(
            Action.daytona_get_sandbox().sandbox_id("{{signal.sandbox_id}}"),
            Condition.equals("sandbox_state", "stopped", "error"),
        )
        .every(seconds=60)
        .timeout(minutes=30)
        .label("Poll until sandbox stops")
    )
    .on_met(Action.daytona_create_snapshot().name("post-stop-{{signal.sandbox_id}}"))
    .on_timeout(Action.slack_send_message().channel("ops").message("Sandbox never stopped"))
)

.every(seconds=..., minutes=...) — polling interval (default 60s, minimum 30s).
.timeout(minutes=..., hours=...) — loop timeout (default 60 minutes).
Downstream steps on the met branch can reference the final iteration’s output fields via {{step_outputs.<loop-id>.<field>}}.

Polling is durable across server restarts: an in-flight interval resumes its countdown rather than restarting.

For Each (Fan-Out)

ForEach builds a fan-out node that resolves a list from an upstream step’s output at runtime and executes ONE embedded catalog action per element, sequentially. Per-item config templates may reference {{item}} (the current element), {{item.<field>}} (a field of an object element), and {{item_index}} (the zero-based index):

from kestrel.workflows import Workflow, Trigger, Action, ForEach

wf = (
    Workflow("Ticket per audit finding")
    .trigger(Trigger.schedule_weekly())
    # Script writes an array of finding objects to outputs.json:
    .then(Action("kestrel", "kestrel-execute-script")
          .config("script", AUDIT_SCRIPT)
          .label("Audit cluster"))
    .then(
        ForEach(
            "{{step_outputs.action-1.outputs.new_findings}}",
            Action.jira_create_ticket()
            .project("SEC")
            .title("[Audit] {{item.title}}")
            .body("{{item.description}}"),
        )
        .max_items(50)
        .continue_on_error()
        .label("Create ticket per finding")
    )
    # Runs once, after ALL items have been processed:
    .then(Action.slack_send_message().channel("security")
          .message("Filed {{step_outputs.foreach-1.succeeded}} tickets"))
)

The items expression must resolve to a list: an array output, a JSON-encoded array string, or a comma-separated string.
.max_items(n) — caps the fan-out (server default 25, hard cap 100).
.continue_on_error() — a failed item doesn’t fail the node; remaining items still run.
The node emits one aggregated output (items_total, items_processed, succeeded, failed, per-item results, and a flat summary), referenced as {{step_outputs.<foreach-id>.<field>}}.
ForEach runs items once each, immediately. To repeatedly re-run an action until a condition holds, use PollUntil instead.

Approvals

wf = (
    Workflow("Approved Deployment Fix")
    .trigger(Trigger.k8s_rollout_status())
    .then(Action.kestrel_trigger_rca())
    .then(Approval.slack("#approvals").message("Apply the generated fix?"))
    .on_approved(Action.kestrel_apply_yaml_fix())
    .on_rejected(Action.slack_send_message().channel("ops").message("Fix rejected"))
)

Types: Approval.manual(), Approval.slack(channel), Approval.pr_approval(), Approval.pr_merge(). Approval.refine() builds a single self-looping gate that presents the upstream RCA + fixes for review. When the approver requests changes with free-text guidance, the same upstream RCA agent re-runs with that feedback (accumulated across rounds) and re-requests approval — looping until approved/rejected or max_rounds is reached. It must be placed after a kestrel_trigger_rca() or kestrel_trigger_cloud_rca() step.

wf = (
    Workflow("RCA with human refinement")
    .trigger(Trigger.k8s_rollout_status())
    .then(Action.kestrel_trigger_rca().label("Run RCA"))
    .then(Action.kestrel_generate_runbook().label("Draft runbook"))
    # Reviewer can iteratively refine the RCA before it's applied:
    .then(Approval.refine().max_rounds(3).message("Review the RCA + fix"))
    .on_approved(Action.kestrel_apply_yaml_fix())
    .on_approved(Action.confluence_publish_runbook())  # publishes runbook_html
    .on_rejected(Action.slack_send_message().channel("ops").message("RCA rejected"))
)

Use Approval.refine("slack") to request the refinement approval in Slack instead of the Kestrel UI. max_rounds defaults to 5; once exhausted the loop advances on the approved branch.

Workflow Management

List, Get, Update, Delete

# List all workflows
workflows = client.workflows.list()
active = client.workflows.list(status="active")

# Get a specific workflow
wf = client.workflows.get("workflow-id")

# Activate / Pause
client.workflows.activate("workflow-id")
client.workflows.pause("workflow-id")

# Delete
client.workflows.delete("workflow-id")

Generate from Natural Language

result = client.workflows.generate(
    "When a PagerDuty alert fires, run K8s RCA, post to Slack, and create a Jira ticket"
)
print(result.name, result.explanation)

Execution Management

List and Inspect Executions

execs = client.workflows.executions("workflow-id", page=1, page_size=10)
for ex in execs.executions:
    print(f"{ex.id[:8]}... {ex.status}")

# Get a specific execution
execution = client.executions.get("execution-id")

Test and Wait

Trigger a test execution and wait for it to complete:

execution = client.workflows.test("workflow-id")
result = client.executions.wait(execution.id, timeout=120.0)
print(f"Final status: {result.status}")

Async SDK

The AsyncKestrelClient provides the same API surface with async/await:

import asyncio
from kestrel import AsyncKestrelClient
from kestrel.workflows import Workflow, Trigger, Action

async def main():
    async with AsyncKestrelClient(api_key="kestrel_sk_...") as client:
        # List workflows
        workflows = await client.workflows.list()

        # Deploy a workflow
        wf = (
            Workflow("Async Example")
            .trigger(Trigger.k8s_pod_status().reasons("OOMKilled"))
            .then(Action.kestrel_trigger_rca())
            .then(Action.slack_send_message().channel("incidents"))
        )
        created = await client.workflows.deploy(wf, activate=True)

        # Test and wait for result
        execution = await client.workflows.test(created.id)
        result = await client.executions.wait(execution.id, timeout=60.0)
        print(f"Result: {result.status}")

asyncio.run(main())

Approvals

# List pending approvals
pending = client.approvals.list_pending()

# Approve or reject
client.approvals.approve("approval-id", justification="Looks good")
client.approvals.reject("approval-id")

Version History & Rollback

# List version history
versions = client.workflows.list_versions("workflow-id")
for v in versions["versions"]:
    print(f"v{v['version_number']} — {v['change_summary']} ({v['created_at']})")

# Roll back to a previous version
client.workflows.rollback("workflow-id", version=3)

Execution Replay

Replay failed executions from the beginning or from the failed step:

# Replay from beginning (re-triggers with same signal data)
new_exec = client.executions.replay("execution-id", mode="full")

# Replay from failed step (preserves outputs from prior steps)
new_exec = client.executions.replay("execution-id", mode="from_failed")

# Wait for replay to complete
result = client.executions.wait(new_exec.id)
print(f"Replay status: {result.status}")

Error Handling

from kestrel import KestrelError, AuthError, NotFoundError

try:
    wf = client.workflows.get("nonexistent")
except NotFoundError:
    print("Workflow not found")
except AuthError:
    print("Authentication failed — check your API key")
except KestrelError as e:
    print(f"API error: {e}")

API Key Scopes

API keys can be scoped to specific permissions:

Scope	Description
`*`	Full access (all permissions)
`workflows:read`	List and view workflows
`workflows:write`	Create, update, test, duplicate workflows
`workflows:activate`	Activate and pause workflows
`workflows:delete`	Delete workflows
`executions:read`	View execution history
`executions:cancel`	Cancel running executions
`approvals:read`	View pending approvals
`approvals:manage`	Approve or reject gates
`requests:read`	View workflow requests
`requests:manage`	Approve or reject requests
`catalog:read`	View signals, actions, and integrations

Create scoped keys in the platform under Workflows > API Keys.

​Installation

​Authentication

​Workflow Builder

​Triggers

​Submitting Requests Programmatically

​Custom Webhook

​Datadog Monitor

​Actions

​Conditions (Branching)

​Poll Until (Loops)

​For Each (Fan-Out)

​Approvals

​Iterative RCA Refinement (human-in-the-loop)

​Workflow Management

​List, Get, Update, Delete

​Generate from Natural Language

​Execution Management

​List and Inspect Executions

​Test and Wait

​Async SDK

​Approvals

​Version History & Rollback

​Execution Replay

​Error Handling

​API Key Scopes

Installation

Authentication

Workflow Builder

Triggers

Submitting Requests Programmatically

Custom Webhook

Datadog Monitor

Actions

Conditions (Branching)

Poll Until (Loops)

For Each (Fan-Out)

Approvals

Iterative RCA Refinement (human-in-the-loop)

Workflow Management

List, Get, Update, Delete

Generate from Natural Language

Execution Management

List and Inspect Executions

Test and Wait

Async SDK

Approvals

Version History & Rollback

Execution Replay

Error Handling

API Key Scopes