Kubeflow Pipelines Pipeline Specification

Common errors

• ERROR: pip's dependency resolver does not currently take into account all the packages that are installed. This behaviour is the source of the following dependency conflicts. kfp-pipeline-spec X.Y.Z requires protobuf<4,>=3.13.0, but you have protobuf A.B.C which is incompatible.

  cause This error occurs due to conflicting `protobuf` package versions required by `kfp-pipeline-spec` and other installed libraries, leading to an incompatible dependency resolution.
  fix Downgrade or upgrade the `protobuf` package to a version compatible with `kfp-pipeline-spec` (e.g., `pip install protobuf==3.20.0`) or ensure all `kfp` related packages are installed with compatible versions, often by using a fresh virtual environment and pinning `kfp` to a specific version (e.g., `pip install kfp==2.7.0 kfp-pipeline-spec==0.3.0`).

• AttributeError: module 'kfp_server_api' has no attribute 'ApiPipelineSpec'

  cause This `AttributeError` typically indicates a version mismatch or breaking API change between the `kfp` SDK, `kfp-pipeline-spec`, and `kfp-server-api` packages, where a newer version of the SDK or `kfp-pipeline-spec` expects an `ApiPipelineSpec` attribute that is not present or has been renamed in the installed `kfp_server_api` version.
  fix Ensure all Kubeflow Pipelines SDK components (`kfp`, `kfp-pipeline-spec`, `kfp-server-api`) are compatible by installing specific, known-good versions together, for example, `pip install kfp==2.0.0b12 kfp-pipeline-spec==0.1.16 kfp-server-api==2.0.0b0` or upgrading all KFP-related packages to their latest compatible versions (`pip install --upgrade kfp kfp-pipeline-spec kfp-server-api`).

• TypeError: Artifacts must have both a schema_title and a schema_version, separated by '@'. Got: float.

  cause This `TypeError` arises in KFP v2 pipelines when a primitive type (like `float` or `str`) is used as an output from a component where an artifact type with a defined schema (`schema_title@schema_version`) is expected, or when the `__future__.annotations` import is used without proper handling for artifact types.
  fix Explicitly define the schema for artifact outputs using `Output[ArtifactType]` with a `schema_title` and `schema_version` (e.g., `Output[Artifact(schema_title='my.Schema', schema_version='1.0.0')]`) or, if applicable for simple parameters, ensure the `from __future__ import annotations` import is not causing type annotation misinterpretations, potentially by removing it if not strictly needed. For list and dict inputs/outputs, ensure KFP version supports them or remove explicit type hints like `list[int]` to `list`.

• NotImplementedError: Input argument supports only the following types: PipelineParam, str, int, float, bool, dict, and list. Got: "None".

  cause This `NotImplementedError` occurs during pipeline compilation in KFP v2 when a component or pipeline is called with an argument that evaluates to `None`, or an unsupported type, where a specific primitive type or a `PipelineParam` is expected by the `kfp-pipeline-spec` schema.
  fix Ensure all pipeline and component input arguments are explicitly set to a supported type (string, int, float, bool, dict, list, or PipelineParam) and do not default to or receive `None` unless specifically handled. Check pipeline function definitions and component calls for missing or unset parameters.

Warnings

• breaking The pipeline specification format underwent significant changes between Kubeflow Pipelines v1 and v2. `kfp-pipeline-spec` is designed for KFP v2 and is not compatible with pipeline definitions or compilers from KFP v1.
  Fix: Ensure your KFP SDK and backend are both on version 2.x or higher when using `kfp-pipeline-spec`. Old v1 pipelines need to be migrated or re-authored using KFP v2 SDK.
• gotcha `kfp-pipeline-spec` is primarily a definitions library for the KFP v2 pipeline schema. Most pipeline authors will interact with this schema indirectly through the `kfp` SDK's `kfp.dsl` and `kfp.compiler` modules, rather than directly constructing protobuf objects from `kfp_pipeline_spec`.
  Fix: For authoring pipelines, use the high-level `kfp` SDK (e.g., `@dsl.pipeline`, `@dsl.component`). Directly manipulating `kfp_pipeline_spec` is typically reserved for advanced use cases like custom compilers, introspection, or validation tools.
• gotcha The KFP ecosystem consists of several related Python packages (`kfp`, `kfp-pipeline-spec`, `kfp-server-api`, `kfp-kubernetes`). It's a common mistake to confuse which package provides specific classes or functionalities. For example, the `Compiler` is in `kfp`, while `PipelineSpec` definition is in `kfp-pipeline-spec`.
  Fix: Always refer to the official KFP documentation for correct import paths and module responsibilities. Generally, `kfp` is for pipeline authoring, `kfp-pipeline-spec` for schema definitions, and `kfp-server-api` for interacting with the KFP API server.

Install

• pip install kfp-pipeline-spec==2.16.0 Install specific version

Imports

• PipelineSpec
  wrong:

  from kfp.v2.compiler.pipeline_spec_pb2 import PipelineSpec

  correct:

  from kfp_pipeline_spec.pipeline_spec_pb2 import PipelineSpec

  The v2 compiler path was common in older pre-GA KFP v2 releases; for direct access to the spec protobufs, import from `kfp_pipeline_spec`.
• ContainerSpec

  from kfp_pipeline_spec.pipeline_spec_pb2 import ContainerSpec

Quickstart

This quickstart demonstrates how to programmatically construct a barebones `PipelineSpec` object using the protobuf definitions from `kfp-pipeline-spec`. While most users generate this spec via the `kfp` SDK's compiler, this shows direct interaction with the underlying data model. It then serializes the spec to JSON, which is a common output format for compiled KFP pipelines.

import kfp_pipeline_spec.pipeline_spec_pb2 as pb
from google.protobuf import json_format

# Create a minimal KFP PipelineSpec object directly using the protobuf definitions
pipeline_spec = pb.PipelineSpec(
    pipeline_info=pb.PipelineInfo(name="my-direct-pipeline"),
    root=pb.PipelineTaskSpec(
        executor_label="my-executor",
        inputs=pb.TaskInputsSpec(),
        outputs=pb.TaskOutputsSpec(),
    ),
    components={
        "my-executor": pb.ComponentSpec(
            executor_label="my-executor",
            input_definitions=pb.ComponentInputsSpec(),
            output_definitions=pb.ComponentOutputsSpec(),
            implementation=pb.ComponentSpec.ContainerImplementation(
                container=pb.ContainerSpec(
                    image="python:3.9-slim",
                    command=["python", "-c", "print('Hello from kfp-pipeline-spec!')"]
                )
            )
        )
    }
)

# Serialize the PipelineSpec to JSON (as it would be in a compiled YAML file)
json_output = json_format.MessageToJson(pipeline_spec, indent=2)
print(json_output)

# To verify it's a valid spec object
assert isinstance(pipeline_spec, pb.PipelineSpec)
print(f"Pipeline name: {pipeline_spec.pipeline_info.name}")