Kestra Python SDK – Client Setup and Examples icon Kestra Python SDK – Client Setup and Examples

OVERVIEW

Created by

Kestra Core Team

Managed by

Kestra Core Team

Interact with Kestra’s API via Python SDK.

Use the Kestra Python SDK programmatically

Install the Python SDK

This guide demonstrates how to use the Kestra Python SDK to create and execute flows programmatically. Before starting, make sure your Kestra instance is running and accessible via the KESTRA_HOST environment variable. You can store credentials in an .env file:

KESTRA_HOST=http://localhost:8080
KESTRA_USERNAME=admin@kestra.io
KESTRA_PASSWORD=Admin1234

Set up your environment

Create a virtual environment and install the Kestra Python SDK. kestrapy is the core package.

uv venv
source .venv/bin/activate
uv pip install kestrapy
uv pip install python-dotenv  # Optional: for loading .env variables automatically

Configure the client

Import and initialize the client with your Kestra credentials:

from kestrapy import Configuration, KestraClient


configuration = Configuration(
    host="http://localhost:8080",
    username="root@root.com",
    password="Root!1234"
)


kestra_client = KestraClient(configuration)

Create a flow

Use the following Python script to create a simple flow with a Sleep task. This example uses the create_flow method.

def create_flow():
    tenant = "main"
    body = """
    id: my_flow
    namespace: my_namespace


    tasks:
      - id: hello
        type: io.kestra.plugin.core.flow.Sleep
        duration: PT1S
    """
    created = kestra_client.flows.create_flow(tenant=tenant, body=body)
    print(f"Flow created: {created.id}")

Update a flow

Use the following Python script to update an existing flow.

def update_flow():
    tenant = "main"
    body = """
    id: my_flow
    namespace: my_namespace


    tasks:
      - id: hello
        type: io.kestra.plugin.core.log.Log
        message: "Updated message!"
    """
    updated = kestra_client.flows.update_flow(
        id="my_flow",
        namespace="my_namespace",
        tenant=tenant,
        body=body
    )
    print(f"Flow updated: {updated.id}")

Execute a flow

Execute flows programmatically using the create_execution method.

def create_execution():
    tenant = "main"
    execution = kestra_client.executions.create_execution(
        tenant=tenant,
        namespace="my_namespace",
        flow_id="my_flow",
        wait=True,
        inputs={"input_id": "value"}
    )
    print(f"Execution started: {execution.id}")

Follow an execution

You can stream live execution updates using the follow_execution method.

def follow_execution():
    tenant = "main"
    execution = kestra_client.executions.create_execution(
        namespace="my_namespace",
        id="my_flow",
        wait=False,
        tenant=tenant
    )


    for event in kestra_client.executions.follow_execution(
        execution_id=execution.id,
        tenant=tenant
    ):
        print(event.state.current)

Best practices

  • Reuse your client: Initialize one KestraClient per application and share it.
  • Avoid hardcoding credentials: Use .env or environment variables.
  • Validate YAML before submission: Invalid syntax causes 422 responses.
  • Automate your workflows: Combine create_flow and create_execution for full CI/CD automation.

Was this page helpful?