Langflow Control Plane Python SDK

langflow-control-plane-sdk is the generated async Python 3.10+ client for the Langflow Control Plane OpenAPI contract. The import package is langflow_control_plane_sdk, which is intentionally distinct from the langflow-sdk package for Langflow's main REST API.

Install

pip install langflow-control-plane-sdk

Normal installs select stable releases. Use pip install --pre langflow-control-plane-sdk to include development and release-candidate builds, or install an exact version such as langflow-control-plane-sdk==0.1.0rc1.

Configure

Set the Control Plane base URL and the bearer token the authoring plane minted for the caller:

import langflow_control_plane_sdk as control_plane

configuration = control_plane.Configuration(
    host="https://control-plane.example",
    access_token="replace-me",
)

The token is sent as Authorization: Bearer ...; do not log it. The Control Plane verifies it against the one JWKS issuer it trusts, and calls Langflow with its own service credential rather than forwarding anything the client supplied. An authentication change affects SDK users and is treated as semver-significant.

Pass-through planes

A Control Plane started with LANGFLOW_CONTROL_PLANE_AUTH_MODE=passthrough — the local development mode — does not verify a bearer token at all. It requires a Langflow API key in x-api-key and forwards that key to Langflow unchanged. The contract declares only the bearer scheme, so the generated Configuration has no field for it; set it as a default header on the client:

client = control_plane.ApiClient(configuration)
client.default_headers["x-api-key"] = "your-langflow-api-key"

Without that header a pass-through plane answers 401 UNAUTHENTICATED no matter what access_token is set to, which is otherwise an unexplained rejection. Do not send the key to a plane running in idp mode: there it is ignored, and the bearer token is the only credential read.

Async CRUD

from uuid import uuid4

import langflow_control_plane_sdk as control_plane

flow_id = uuid4()
flow = control_plane.Flow(
    flow_id=flow_id,
    name="hello-world",
    payload=control_plane.FlowPayload(
        data=control_plane.FlowPayloadData(nodes=[], edges=[]),
    ),
)

async with control_plane.ApiClient(configuration) as client:
    deployments = control_plane.DeploymentsApi(client)
    created = await deployments.create_deployment(
        control_plane.CreateDeployment(project_id=uuid4(), slug="hello-world", flows=[flow]),
    )
    fetched = await deployments.get_deployment(created.id)
    updated = await deployments.replace_deployment(
        created.id,
        control_plane.ReplaceDeployment(description="updated", flows=[flow]),
    )
    patched = await deployments.patch_deployment(
        created.id,
        control_plane.PatchDeployment(description="patched"),
    )
    await deployments.delete_deployment(created.id)

List responses always contain next_cursor; it is None on the last page. Pass a non-null cursor back unchanged:

async with control_plane.ApiClient(configuration) as client:
    deployments = control_plane.DeploymentsApi(client)
    page = await deployments.list_deployments(limit=50)
    while page.next_cursor is not None:
        page = await deployments.list_deployments(
            cursor=page.next_cursor,
            limit=50,
        )

Unsuccessful responses raise ApiException. For documented responses, exception.data is a typed ErrorEnvelope:

try:
    await deployments.get_deployment(deployment_id)
except control_plane.ApiException as exception:
    if isinstance(exception.data, control_plane.ErrorEnvelope):
        print(exception.data.error.code, exception.data.error.message)
    raise

Use the top-level error.code for stable classification. Optional error.details may support diagnostics or optional handling, but any use must tolerate an absent object and unknown or changed fields and values. When a service detail omits code, the top-level error.code applies to that service; a service-scoped code may appear when its classification differs. Optional error.warning contains a secondary diagnostic and does not replace the primary error. The warning may change shape; clients must tolerate an absent object and unknown or changed fields and values. The generated ErrorCode is a closed enum for the OpenAPI version that produced the SDK, so keep a general exception/status fallback for a primary code introduced by a newer server. Returned Deployment and Flow IDs are authoritative opaque Control Plane identifiers; do not derive them from create-request values. See the repository's API compatibility policy.

Regenerate and test

Generation requires Python 3.10+, Java 11+, and the pinned generator. If Java is unavailable, jdk4py can optionally provide a local JVM.

make openapi-install
make sdk-py-install
make openapi-test
make sdk-py-generate
make sdk-py-test
make sdk-py-build

When using jdk4py, install it in the codegen environment with sdk/openapi-generator/.venv/bin/python -m pip install jdk4py.

Generated package files must not be edited by hand. Change api/openapi.yaml, the generator options in the root Makefile, or the templates, then regenerate them.

The checked-in SDK uses the base version 0.0.0. Release builds apply the exact v-prefixed Git tag version to a temporary build copy; the OpenAPI info.version independently identifies the API contract line.

See the repository's release guide for shared service and SDK versioning and tag formats.