> For the complete documentation index, see [llms.txt](https://docs.pipekit.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.pipekit.io/reference/python-sdk.md).

# Python SDK

The [Pipekit Python SDK](https://pypi.org/project/pipekit-sdk/) lets you talk to the Pipekit API from Python. It's the right surface for use in [Jupyter Notebooks](/reference/python-sdk/examples-jupyter.md), inside [Python scripts](/reference/python-sdk/examples-scripts.md), and for building automation against Pipekit programmatically.

The SDK wraps the [REST API](/reference/rest-api.md) and is designed to pair well with [Hera](https://github.com/argoproj-labs/hera) for authoring workflows in Python.

## Install

```bash
pip install pipekit-sdk
```

The SDK targets Python 3.10+.

## Authenticate

The SDK uses a bearer token, which you generate with the [Pipekit CLI](/reference/cli.md):

```bash
pipekit login          # one-time, opens browser for SSO or prompts for password
pipekit hera           # prints a token to stdout
```

Pass the token to a `PipekitService`:

```python
from pipekit_sdk.service import PipekitService

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

For longer-lived programs, set `PIPEKIT_TOKEN` and pull it from the environment:

```python
import os
from pipekit_sdk.service import PipekitService

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

## Submit a Workflow

```python
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"])
```

`submit` returns the `pipe_run` dictionary including the Run's UUID.

## List Runs

```python
runs = pipekit.list_runs(cluster_name="<cluster-name>")
for r in runs:
    print(r["uuid"], r["status"])
```

Filter by Pipe, namespace, or status. See the SDK source for the full parameter list.

## Stop, terminate, or restart a Run

```python
pipekit.stop(run_uuid="<run-uuid>")
pipekit.terminate(run_uuid="<run-uuid>")
pipekit.restart(run_uuid="<run-uuid>")
```

`stop` runs exit handlers; `terminate` does not. `restart` resubmits with the same parameters and returns a new `pipe_run`.

## Stream logs

```python
pipekit.print_logs(pipe_run["uuid"])
```

For non-blocking access, use the lower-level log fetcher (see SDK source).

## CronWorkflows

The SDK supports the full CronWorkflow lifecycle:

```python
from hera.workflows import Container, CronWorkflow

with CronWorkflow(
    name="cron-wf-example",
    namespace="argo",
    entrypoint="main",
    schedule="*/5 * * * *",
    concurrency_policy="Replace",
    service_account_name="argo-workflow",
) as cw:
    main = Container(
        name="main",
        image="alpine",
        command=["sh", "-c", "echo I am a CronWorkflow"],
    )

pipekit.create(cw, "<cluster-name>")
# Later:
pipekit.suspend(cw)
pipekit.resume(cw)
pipekit.update(cw)
pipekit.delete(cw)
```

## What's next

* [Jupyter Notebooks](/reference/python-sdk/examples-jupyter.md): a worked example with log streaming.
* [Python Scripts](/reference/python-sdk/examples-scripts.md): a worked example of a coinflip workflow.
* [REST API](/reference/rest-api.md): if you'd rather call the API directly.
* [CLI > Using with Hera](/reference/cli/using-with-hera.md): token generation flow for Hera users.