DocsApi ReferenceKestra Sdk

Kestra Python SDK​Kestra ​Python ​S​D​K

Release: 1.0.0

Interact with Kestra's API via Python SDK.

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:

text
KESTRA_HOST=http://localhost:8080
[email protected]
KESTRA_PASSWORD=Admin1234

Set up your environment

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

shell
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:

python
from kestrapy import Configuration, KestraClient

configuration = Configuration(
    host="http://localhost:8080",
    username="[email protected]",
    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.

python
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.

python
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.

python
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.

python
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?