## Define pipelines in Python — steps, composition, compute, and upload — with no separate YAML to maintain

The Clarifai pipeline DSL lets you define entire pipelines in Python. Steps are decorated functions; pipelines are composed inside a `Pipeline` context; parameters come from function signatures. The DSL compiles to the same underlying pipeline bundle used by the YAML / scaffold-directory flow, but you never have to write or maintain the YAML yourself.

This page is the complete reference for the DSL. For a runnable end-to-end walkthrough, see the [Quick Start](https://docs.clarifai.com/compute/pipelines/#quick-start). For a more elaborate example with mixed local and remote steps, see [`examples/pipeline_dsl_text_pipeline.py`](https://github.com/Clarifai/clarifai-python/blob/master/examples/pipeline_dsl_text_pipeline.py) in the `clarifai-python` repo.

## Public API

The DSL is exported from `clarifai.runners.pipelines`:

```python
from clarifai.runners.pipelines import (
    ComputeInfo,
    Pipeline,
    step,
    step_ref,
    load_pipeline_from_file,  # advanced: load a Pipeline instance from a .py path
)
```

## Defining Steps with `@step`

Use the `@step` decorator to turn a Python function into a pipeline step. The function signature becomes the step's input parameters; the return value becomes the step's output.

```python
from clarifai.runners.pipelines import ComputeInfo, step

@step(
    id="prepare-text",                                # optional; defaults to function name with _→-
    requirements=["transformers>=4.0"],               # optional; pip requirements for the step
    assets=["./text_utils.py"],                       # optional; files copied alongside step code
    compute=ComputeInfo(cpu_limit="500m", cpu_memory="500Mi"),  # optional; compute requirements
    python_version="3.12",                            # optional; defaults to "3.12"
    secrets={"OPENAI_API_KEY": "users/me/secrets/openai-key"},  # optional; step-level secrets
)
def prepare_text(input_text: str) -> str:
    """Normalize text before downstream processing."""
    return input_text.strip().lower()
```

### `@step` Parameters

| Parameter | Type | Description |
| --- | --- | --- |
| `id` | `str` | The step's identifier. Defaults to the function name with underscores converted to hyphens (e.g. `prepare_text` → `prepare-text`). Must be a valid Clarifai resource ID. |
| `requirements` | `list[str]` or `str` | Pip requirements for the step. Accepts a list of requirement strings, a single requirement string, or a path to a `requirements.txt` file. |
| `assets` | `list[str]` | Files or directories to bundle alongside the step's code. Paths are resolved relative to the source file containing the step. |
| `compute` | `ComputeInfo` | Compute resources for the step (CPU, memory, accelerators). |
| `python_version` | `str` | Python version to use inside the step's container. Defaults to `"3.12"`. |
| `base_image` | `Optional[str]` | Custom base Docker image for the step's container — e.g., a CUDA-enabled image for GPU steps. Defaults to Clarifai's standard base image. |
| `platform` | `Optional[str]` | Build platform for the step's container image (e.g., `"linux/amd64"`). Defaults to multi-architecture build. |
| `secrets` | `dict[str, str]` | Environment variable name → secret resource path. Secrets are mounted as environment variables at step runtime. |

### Step Input Parameters

A step's input parameters are inferred from its function signature. Type annotations become parameter type hints; default values become parameter defaults.

```python
@step
def train(
    dataset_path: str,                              # required parameter
    base_model: str = "google/vit-base-patch16-224", # optional with default
    epochs: int = 10,
    learning_rate: float = 0.001,
) -> str:
    ...
```

No separate `pipeline_step_input_params` YAML block to maintain — the function signature is the source of truth.

### Calling a Step Function Directly

Step functions can be called directly outside of a `Pipeline` context. This is useful for local testing:

```python
result = prepare_text("Some Input")           # runs the underlying function locally
# or, equivalently:
result = prepare_text.test(input_text="Some Input")
```

Inside a `Pipeline` context, the same call adds a node to the DAG instead of executing the function.

## Compute Requirements

`ComputeInfo` declares the compute resources for a step:

```python
from clarifai.runners.pipelines import ComputeInfo

ComputeInfo(
    cpu_limit="500m",         # 500 millicores = 0.5 CPU cores
    cpu_memory="500Mi",       # 500 MiB of memory
    num_accelerators=0,       # number of GPUs/accelerators
    accelerator_type=["NVIDIA-A10G"],  # optional accelerator type constraints
)
```

`ComputeInfo` is the same proto type used in `config.yaml`'s `pipeline_step_compute_info` block.

## Composing Pipelines with `Pipeline(...)`

A `Pipeline` instance is the container for your DAG. Use it as a context manager (`with Pipeline(...) as pipeline:`) so step calls register nodes against it:

```python
from clarifai.runners.pipelines import Pipeline

with Pipeline(id="my-pipeline") as pipeline:
    raw = pipeline.input("input_text", default="hello world")

    prepared = prepare_text(input_text=raw)
    summary  = summarize(input_text=prepared.output())

    prepared >> summary
```

### `Pipeline` Constructor

| Parameter | Type | Description |
| --- | --- | --- |
| `id` | `str` | The pipeline's identifier. Required. |
| `user_id` | `Optional[str]` | User ID owning the pipeline. Defaults to the user from your `clarifai login` context. |
| `app_id` | `Optional[str]` | App ID containing the pipeline. Defaults to the app from your `clarifai login` context. |
| `visibility` | `str` | Visibility setting; defaults to `"PRIVATE"`. |

If neither explicit args nor `clarifai login` context provide `user_id` and `app_id`, the constructor raises a `ValueError` pointing you to run `clarifai login` first.

When the `with` block exits, the pipeline is validated automatically — missing dependencies and DAG cycles raise a `ValueError` before upload.

### Workflow Inputs

`pipeline.input(name, default=None)` declares a workflow-level input parameter — a value provided at runtime when you `clarifai pipeline run`. The return value is a reference you can pass into step calls:

```python
with Pipeline(id="my-pipeline") as pipeline:
    raw = pipeline.input("input_text", default="hello")
    cleaned = prepare_text(input_text=raw)
```

### Connecting Steps

There are two ways to express dependencies between steps:

**1. Pass one step's output to another step's input.** This creates an implicit dependency:

```python
prepared = prepare_text(input_text=raw)
summary  = summarize(input_text=prepared.output())   # summary depends on prepared
```

**2. Use the `>>` operator for explicit ordering.** This is useful when you need a dependency that isn't expressed by data flow, or for fan-in/fan-out:

```python
# Linear: a → b → c
a >> b >> c

# Fan-out: prepared feeds two parallel branches
prepared >> [summary, sentiment]

# Fan-in (diamond DAG): both branches feed the final report step
prepared >> [summary, sentiment] >> report
```

## Referencing Existing Remote Steps with `step_ref`

To compose pipelines that include already-uploaded pipeline steps (your own or shared from another app), use `step_ref`. The referenced step is pinned to a specific version and is not re-uploaded with the pipeline:

```python
from clarifai.runners.pipelines import step_ref

# By IDs:
classify_sentiment = step_ref(
    id="classify-sentiment",
    user_id="demo-user",
    app_id="shared-app",
    version_id="sentiment-v3",
)

# By URL:
summarize = step_ref.from_url(
    "https://api.clarifai.com/v2/users/demo-user/apps/shared-app/pipeline_steps/summarize/versions/summary-v1",
    secrets={"OPENAI_API_KEY": "users/demo-user/secrets/openai-key"},
)
```

You can mix locally-defined steps and `step_ref` steps freely in the same pipeline.

### Step-Level Secrets

Both `@step` and `step_ref` accept a `secrets` dictionary mapping environment variable names to Clarifai secret resource paths:

```python
@step(
    id="call-openai",
    secrets={"OPENAI_API_KEY": "users/me/secrets/openai-key"},
)
def call_openai(prompt: str) -> str:
    import os
    api_key = os.environ["OPENAI_API_KEY"]
    ...
```

Secrets are injected as environment variables at step runtime. They are not visible in your pipeline source or in the generated bundle.

## Upload

Upload a pipeline directly from its `.py` file:

```bash
clarifai pipeline upload my_pipeline.py
```

The CLI loads the `Pipeline` instance from the file, generates the underlying step bundles into a temporary `generated-<pipeline-id>/` directory next to the file, and uploads. You can also upload programmatically:

```python
pipeline_version_id = pipeline.upload(no_lockfile=False)
```

The pipeline's source file must contain exactly one `Pipeline` instance. If you want multiple pipelines in the same project, put them in separate files.

## Generating the YAML Bundle (Codegen)

To produce the underlying scaffold-directory bundle without uploading — useful for inspection, version control, or handing off to the YAML/CLI flow — there are two options.

**From the CLI:**

```bash
clarifai pipeline compile my_pipeline.py --output-dir ./generated-bundle
```

**From Python:**

```python
pipeline.generate("./generated-bundle")
```

The generated directory has the same shape as a scaffold-directory pipeline: a root `config.yaml`, per-step subdirectories with their own `config.yaml`, `requirements.txt`, and a `1/pipeline_step.py` containing an `argparse`-wrapped version of your `@step` function. You can upload it the same way as a scaffold-directory pipeline:

```bash
clarifai pipeline upload ./generated-bundle
```

## Programmatic Run

You can run a pipeline from Python without going through the CLI:

```python
result = pipeline.run(
    inputs={"input_text": "hello world"},
    timeout=3600,
    monitor_interval=10,
    nodepool_id="...",
    compute_cluster_id="...",
)
```

If the pipeline hasn't been uploaded yet, `run()` uploads first.

## Validation

Pipelines are validated automatically when the `with Pipeline(...) as pipeline:` block exits. Validation checks for:

- **Missing dependencies** — any `>>` reference to a step not defined in the pipeline raises a `ValueError`.
- **DAG cycles** — circular dependencies between steps raise a `ValueError` with the offending node.