Using with Hera

Pair the Pipekit CLI with Hera workflows and the Pipekit Python SDK.

Hera is a Python framework for authoring Argo Workflows. The Pipekit CLI integrates with Hera by issuing tokens you pass into the Pipekit Python SDK, which submits your Hera-authored workflows to Pipekit.

Generate a token

pipekit login          # one-time
pipekit hera           # prints a token to stdout

If you want only the token value (no banner), use -r for raw output:

pipekit hera -r

The token is a bearer token, the same format the REST API and the MCP Server use.

Use the token in Python

from pipekit_sdk.service import PipekitService
import os

pipekit = PipekitService(token=os.environ["PIPEKIT_TOKEN"])

Then author your workflow with Hera and submit it via the SDK:

from hera.workflows import DAG, Workflow, script

@script(image="python:3.12")
def echo(message):
    print(message)

with Workflow(
    generate_name="dag-diamond-",
    entrypoint="diamond",
    namespace="argo",
    service_account_name="argo-workflow",
) as w:
    with DAG(name="diamond"):
        A = echo(name="A", arguments={"message": "A"})
        B = echo(name="B", arguments={"message": "B"})
        C = echo(name="C", arguments={"message": "C"})
        D = echo(name="D", arguments={"message": "D"})
        A >> [B, C] >> D

pipe_run = pipekit.submit(w, "<cluster-name>")
print(pipe_run["uuid"])

For worked examples, see:

• Python SDK > Jupyter Notebooks

• Python SDK > Python Scripts

CronWorkflows with Hera

Hera and the SDK also support the full CronWorkflow lifecycle. See Python SDK > CronWorkflows.

Token lifetime

The token Pipekit issues for Hera carries the same TTL as your CLI session. For long-running Python scripts that exceed the session length, refresh the token by running pipekit login and re-issuing pipekit hera.