Pipelines (Python)

Pipelines are scheduled batch jobs that produce named datasets agents can read. They are written in Python with the kraken-ai SDK: a declarative manifest plus an async run function.

#Overview

A pipeline is a Python module that declares a manifest and an async run function. The manifest names the pipeline, sets its schedule, and lists the datasets it produces; run does the work and writes those datasets out. The platform runs each pipeline on its schedule and makes the resulting data available to agents.

The SDK is the kraken-ai package on PyPI. It requires Python 3.11 or newer and is built on Pydantic 2. It ships only the types you need to define a pipeline — there is no runtime to install or manage.

#Installation

Add kraken-ai to your pipeline project with your Python package manager of choice.

$ uv add kraken-ai

# or, with pip
$ pip install kraken-ai

Note

The PyPI distribution is named kraken-ai; the import package is kraken_ai. The TypeScript Agent and Platform SDKs are separate packages — see the Agent SDK.

#PipelineManifest

PipelineManifest is the declarative definition of a pipeline. Assign it to a module-level manifest variable so the platform can discover it.

namestrRequired

The pipeline’s identity.

descriptionstrOptional

What the pipeline produces. Defaults to an empty string.

schedulestrRequired

A cron expression that sets how often the pipeline runs — for example "0 6,18 * * *" for 06:00 and 18:00 daily.

outputslist[Output]Optional

The named datasets this pipeline produces. Defaults to an empty list.

#Output

An Output declares one named dataset the pipeline writes. List every dataset your pipeline produces in the manifest’s outputs so agents and operators know what is available.

namestrRequired

The dataset’s name.

descriptionstrOptional

What the dataset contains. Defaults to an empty string.

#PipelineContext

PipelineContext is injected into run(ctx) by the platform runner — you never construct it yourself. It tells the run where to write its data and how to emit structured progress.

run_idstrRequired

Unique identifier for this run.

pipeline_namestrRequired

The name of the pipeline being run.

output_pathPathRequired

Directory the run writes its datasets into.

metadatadict[str, Any]Optional

Run metadata supplied by the platform. Empty by default.

log(message)(str) -> NoneOptional

Emit a structured log event to the platform. Use it to report progress and milestones.

#Writing a pipeline

A pipeline module assigns manifest and defines an async run(ctx) that writes its datasets to ctx.output_path and reports progress with ctx.log().

python
import csv

from kraken_ai import Output, PipelineContext, PipelineManifest

manifest = PipelineManifest(
    name="daily-prices",
    description="Fair-market price bands for tracked SKUs",
    schedule="0 6,18 * * *",
    outputs=[
        Output(name="prices", description="One row per SKU with a price band"),
    ],
)


async def run(ctx: PipelineContext) -> None:
    ctx.log(f"Starting {ctx.pipeline_name} ({ctx.run_id})")

    rows = await fetch_price_bands()

    out_file = ctx.output_path / "prices.csv"
    with out_file.open("w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["sku", "low", "high"])
        for row in rows:
            writer.writerow([row.sku, row.low, row.high])

    ctx.log(f"Wrote {len(rows)} rows to prices.csv")

Note

run writes datasets under ctx.output_path and reports progress with ctx.log(). The platform collects what you write there and exposes it as the pipeline’s named outputs.

#Scheduling & deployment

The manifest’s schedule is a cron expression: the platform runs the pipeline on that cadence, each run isolated with its own run_id and output_path. A cron schedule is one of the trigger types the platform supports — see Triggers.

Pipelines deploy the same way agents do: push the project to a connected repository and the platform builds and deploys it automatically. There is nothing to provision by hand. See Deployment for the repository-intake flow.

#Next steps

  • Triggers — Cron, event, and webhook triggers — how scheduled runs start and how they are configured.
  • Deployment — Connect a repository and let every push build and deploy your agents and pipelines.
  • Platform API — Read pipelines and their data programmatically from your own code over the Platform API.