Pipekit allows you to manage your workflows at scale. The control plane configures Argo Workflows for you in your infrastructure, enabling you to optimize multi-cluster workloads while reducing your cloud spend. The team at Pipekit is also happy to support you through your Argo Workflows journey via commercial support.

Pipekit Python SDK

Installation

pip install pipekit-sdk

Usage

# The Pipekit SDK interacts with Hera Workflows classes
from hera.workflows import Container, Step, Steps, Workflow, script
from pipekit_sdk.service import PipekitService

# Create a Pipekit service that is used to talk to the Pipekit API
pipekit = PipekitService(token="<token>")

# List clusters and Pipes
clusters = pipekit.list_clusters()
pipes = pipekit.list_pipes()

@script()
def flip_coin() -> None:
    import random

    result = "heads" if random.randint(0, 1) == 0 else "tails"
    print(result)

# Create a Workflow using Hera
with Workflow(
    generate_name="coinflip-",
    annotations={
        "workflows.argoproj.io/description": (
            "This is an example of coin flip defined as a sequence of conditional steps."
        ),
    },
    entrypoint="coinflip",
    namespace="argo",
    service_account_name="argo",
) as w:
    heads = Container(
        name="heads",
        image="alpine:3.6",
        command=["sh", "-c"],
        args=['echo "it was heads"'],
    )
    tails = Container(
        name="tails",
        image="alpine:3.6",
        command=["sh", "-c"],
        args=['echo "it was tails"'],
    )

    with Steps(name="coinflip") as s:
        fc: Step = flip_coin()

        with s.parallel():
            heads(when=f"{fc.result} == heads")
            tails(when=f"{fc.result} == tails")

# Submit the Workflow to Pipekit
pipekit.submit(w, "<cluster-name>")

# Tail the logs
pipekit.print_logs(pipe_run.uuid)

Connecting to Pipekit

pipekit_url is the single base URL for the ID, Users, and UI APIs. This is correct when they sit behind one gateway, which is the default (https://api.pipekit.io).

When the services are reachable on separate hosts, set id_url and users_url (or the PIPEKIT_ID_URL and PIPEKIT_USERS_URL env vars). The SDK logs in against id_url and makes every other call against users_url.

pipekit = PipekitService(
    username="<user>",
    password="<password>",
    id_url="http://id.internal:8080",
    users_url="http://users.internal:8080",
)

insecure=True (or PIPEKIT_INSECURE=true) skips TLS verification. Use it only for testing against a cluster with a self-signed certificate, never in production.

timeout (or PIPEKIT_TIMEOUT) sets the per-request timeout in seconds, default 10. Raise it for slow links or for the cron lifecycle calls, which can take up to the server's notification timeout to return. The log stream is exempt and stays unbounded.

Managing CronWorkflows

You can create, update, and delete a CronWorkflow from Python. The namespace must match the one in the manifest on every call (the platform default is argo). A wrong namespace makes the cron look missing.

from hera.workflows import Container, CronWorkflow
from pipekit_sdk.service import PipekitService

pipekit = PipekitService(token="<token>")

with CronWorkflow(
    name="daily-demand-forecast",
    namespace="argo",
    entrypoint="main",
    # Argo Workflows 3.6 deprecated the singular spec.schedule. Use schedules.
    schedules=["*/5 * * * *"],
    service_account_name="argo",
) as cron:
    Container(name="main", image="alpine", command=["sh", "-c", "echo hello"])

# Create
pipekit.create(cron, "<cluster-name>")

# Update: the namespace is taken from the manifest when not passed
updated = pipekit.update_cron(cron, "<cluster-name>")

# Suspend / resume scheduling
pipekit.suspend_cron("<cluster-name>", "argo", "daily-demand-forecast")
pipekit.resume_cron("<cluster-name>", "argo", "daily-demand-forecast")

# Get the current state
current = pipekit.get_cron("<cluster-name>", "argo", "daily-demand-forecast")

# Delete
pipekit.delete_cron("<cluster-name>", "argo", "daily-demand-forecast")

Further help

Please refer to the Pipekit Documentation for more information.

Metadata

Release files for pipekit-sdk 7.5.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for pipekit-sdk 7.5.0
File	Size	Uploaded
pipekit_sdk-7.5.0.tar.gz	50.0 kB	Aug 25, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for pipekit-sdk 7.5.0
File	Interpreter	ABI	Platform	Reset
pipekit_sdk-7.5.0-py3-none-any.whl 52.0 kB Aug 25, 2026	Python 3	none	any	Details

Total release size: 102.0 kB

Release files / pipekit_sdk-7.5.0.tar.gz

Download URL	pipekit_sdk-7.5.0.tar.gz
Size	50.0 kB
Tags	Source
SHA-256 checksum How to use checksums	`02bbb2ade9ce3887aade4cb374e9c86af0fec3747da8ae9a032e686a6f11991f`
BLAKE2b-256 checksum How to use checksums	`c8d295001794a6a138ee5dae04647dc7d8fbcc82add3ba7de388dd7475beac60`
Upload date	Aug 25, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.12.3`

Release files / pipekit_sdk-7.5.0-py3-none-any.whl

Download URL	pipekit_sdk-7.5.0-py3-none-any.whl
Size	52.0 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`3dc918f18f1d91cfa830fd8d97e02491b23386033d2ee91aa60a1af3e2311bc7`
BLAKE2b-256 checksum How to use checksums	`abacb5fdf0a97513dd9f5b627c73dc930d8901caca9c7ff4c14f023d077f7958`
Upload date	Aug 25, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.12.3`