`frankctl` — CLI Reference

Terminal-first client for the Frank Low-Code Pipeline platform. The CLI wraps the same REST API as the dashboard, plus a declarative apply path for git-managed provisioning.

This guide covers every shipped frankctl verb. Run frankctl <cmd> --help for option details.

Authoring a feed end to end? See From External Source to Ontology for the full recipe (preview → author → sandbox-validate → apply → verify), with a real worked example.

Install

bash

cd frank-low-code-pipeline/frank-cli
npm install
npm run build
npm link
frankctl --help

Configuration

Variable	Purpose
`FRANKCTL_API_URL`	Frank API base URL. Default `http://localhost:8000`.
`FRANKCTL_KEYCLOAK_URL`	Keycloak base URL.
`FRANKCTL_KEYCLOAK_REALM`	Keycloak realm. Default `frank`.
`FRANKCTL_CLIENT_ID`	OAuth client id.
`FRANKCTL_PROFILE`	Active config profile.
`FRANK_DEV_MODE=true` + `FRANKCTL_TENANT_ID=<uuid>`	Forward `X-Tenant-ID` header instead of an OIDC token. Local dev only.

frankctl auth login runs the device-code flow against the configured Keycloak realm and caches the token under $HOME/.frankctl/tokens.json.

Resources

`frankctl sources`

CRUD for data sources.

bash

frankctl sources list
frankctl sources get <id>
frankctl sources create -f source.yaml
frankctl sources update <id> -f patch.yaml
frankctl sources delete <id> --yes
frankctl sources discover <id>
frankctl sources sync <id>
frankctl sources logs <id> <run-id>
frankctl sources history <id>

Stream-level type-drift ops (issue #469):

bash

frankctl sources streams diff-types <source-id> <stream-id>
frankctl sources streams set-types  <source-id> <stream-id> \
    --field dt=integer --field pm25=number

See § Bronze type drift at the bottom of this guide for the playbook.

`frankctl pipelines`

CRUD for multi-step pipelines plus the declarative apply path (see below).

bash

frankctl pipelines list
frankctl pipelines get <id>
frankctl pipelines validate <id>          # sandbox dry-run
frankctl pipelines delete <id> --yes      # delete (leaves silver tables)

# Declarative apply (Gap #5 in the provisioning spec)
frankctl pipelines apply -f vertical.yaml
frankctl pipelines export <id> -o yaml > vertical.yaml

# Scaffold a starter vertical from a synced source's physical bronze schema
frankctl pipelines scaffold --source <id-or-name> > vertical.yaml
frankctl pipelines scaffold --source my_source --stream obs -o vertical.yaml

pipelines scaffold reads a synced source's physical bronze schema — the real nested Iceberg ROW columns, not discovery's flattened logical names — and emits a starter Source + Pipeline + BackingDataset multi-doc YAML. The silver custom_sql step comes pre-filled with the correct Trino ROW references (e.g. o."properties"."idEstacao"), so you edit SQL expressions and ontology property names only, then pipelines apply -f --wait. The source must have synced to bronze first (the ROW structure only exists after a sync); scaffolding an un-synced source errors with that hint. --stream selects which stream when a source has more than one enabled.

`frankctl backing-datasets`

CRUD for backing datasets — Iceberg tables that bind to an ontology entity type. Aliased as frankctl bd.

bash

frankctl backing-datasets list
frankctl backing-datasets get <id>
frankctl backing-datasets create -f bd.yaml [--allow-deprecate]
frankctl backing-datasets update <id> -f patch.yaml [--allow-deprecate]
frankctl backing-datasets sync <id> [--no-wait]   # push silver → ontology, poll to terminal
frankctl backing-datasets entities <id> [--limit N]   # read entities back out of the ontology
frankctl backing-datasets delete <id> --yes

entities closes the authoring loop: after a sync it reads the actual entities back from the ontology (values, not just status), scoped by the BD's ontology_tenant_id — so verification never leaves the CLI.

When binding a BD to a pipeline, omit iceberg_namespace/iceberg_table: set pipeline: (or pipeline_id:) and the silver target is derived from the pipeline's terminal step.

sync triggers an ontology sync and polls until the BD reaches synced, error, or needs_remapping (exit 1 on the non-synced terminals). A BD with sync_mode: on_materialization also syncs automatically on each new silver snapshot; sync is for an explicit, on-demand push.

--allow-deprecate permits the ensure_schema: block to narrow the ontology entity type by deprecating fields. Without the flag, narrowing returns 409 (additive evolution only).

`frankctl ai compose-pipeline`

AI-assisted pipeline authoring. Takes a spec describing user intent + source tables + (optionally) the target shape, fires the frank_compose_pipeline Martha workflow, and returns either the raw JSON proposal or a YAML envelope that pipes straight into pipelines apply -f.

bash

# JSON proposal (default) — useful for inspection / piping into other tools
frankctl ai compose-pipeline -f intent.yaml > proposal.json

# YAML envelope — pipes directly into declarative apply
frankctl ai compose-pipeline -f intent.yaml --output yaml > vertical.yaml
frankctl pipelines apply -f vertical.yaml

Spec file shape (JSON or YAML):

yaml

user_intent: "Land OWM air-quality data into the air_quality_observed entity type"
pipeline_name: air_quality_observed       # required for --output yaml
source_tables: [bronze.cm_ave.air_pollution]
target_description: "Promote raw OWM payload into a typed dataset"
target_schema:                            # optional
  - { name: external_id,  type: string }
  - { name: dt,           type: integer }
  - { name: pm2p5,        type: number }
target_sdm_id: fiware:Environment/AirQualityObserved   # optional; flips target_type to "sdm"

--output yaml emits a single kind: Pipeline doc. Source and BackingDataset docs are not auto-generated — the compose endpoint takes already-existing source tables, and the BD shape (iceberg_namespace / property_mappings) isn't yet part of the AI's proposal. Operators add Source and BD docs by hand or via pipelines export after applying.

`frankctl ai suggest source-config`

Paste in raw upstream config (.env, vendor docs, API doc snippets) and get back an inferred pattern_id, redacted source_config, and a suggested name. The other half of the authoring on-ramp — compose-pipeline handles the Pipeline; this handles the Source.

bash

# From a file (JSON/YAML with {raw_text, available_pattern_ids?})
frankctl ai suggest source-config -f spec.yaml

# Inline (short snippets only)
frankctl ai suggest source-config --text "feed_url: https://example.com/rss"

# Via stdin (paste-anywhere workflow)
pbpaste | frankctl ai suggest source-config

# Pipe straight into a Source YAML the operator can `sources create -f`
frankctl ai suggest source-config -f spec.yaml --output yaml > source.yaml

Spec file shape:

yaml

raw_text: |
  REDIS_HOST=cache.example.com
  REDIS_PORT=6379
  REDIS_PASSWORD=secret
available_pattern_ids: [rest_api, kafka, postgres, redis]   # optional hint

Response (JSON, default):

json

{
  "pattern_id": "redis",
  "confidence": 0.9,
  "config": {"host": "cache.example.com", "port": 6379},
  "rationale": "Redis env vars detected",
  "detected_fields": ["host", "port"],
  "suggested_name": "redis_cache",
  "suggested_description": "Redis cache instance",
  "available": true
}

--output yaml emits the kind: Source envelope using suggested_name, pattern_id, and config. When the AI can't infer a pattern_id (confidence < threshold), --output yaml falls back to JSON and exits 5 with the rationale on stderr — the operator picks the pattern manually.

`frankctl transforms`, `frankctl runs`, `frankctl patterns`, `frankctl schedules`, `frankctl datasets`, `frankctl ai`

Domain-specific verbs. Run frankctl <noun> --help for the full list.

Diagnosing a transform that won't run

When a transform isn't materializing on upstream change, three commands tell you why without touching the database.

`frankctl transforms get <id>` — runnability at a glance

get surfaces the fields the scheduler actually gates on:

lifecycle_stage    ready
last_run_outcome   running        # "running" with no real run in flight = stuck
can_run_now        false          # false here means triggers will 400

can_run_now=false with last_run_outcome=running is the classic zombie: the cached outcome says a run is in flight when none is, so POST /schedule/trigger returns 400 forever and the source-update sensor can never fire it. Heal it with frankctl admin reconcile-runs (below).

`frankctl transforms gate-status` — the change-gate (ADL-016)

Eager transforms fire only when their upstream_version (a hash of their input tables' Iceberg snapshot ids) advances. This lists it per transform:

bash

frankctl transforms gate-status
# ID    NAME          UPSTREAM_VERSION   FIRES_ON_CHANGE
# t1    fires         97d2ec9566272863   yes
# t2    stuck                            NO (cold/unresolved)

upstream_version is empty (the sensor skips) when either the source has never synced (cold start) or an input table FQN doesn't resolve (e.g. a pre-#513 bronze.<ns>.<table> identifier). A whole vertical showing NO usually means the latter.

`frankctl admin reconcile-runs` — heal stuck runs & zombies

Syncs stuck backend runs to their real terminal state and clears zombie last_run_outcome=RUNNING markers (transforms with no run actually in flight), restoring can_run_now.

bash

# Preview only — lists what would be healed, changes nothing:
frankctl admin reconcile-runs --dry-run
# ID    CURRENT   WOULD_HEAL_TO
# t1    running   failed

# Heal (stale runs + zombies):
frankctl admin reconcile-runs

# Also re-check every active run, not just stale ones:
frankctl admin reconcile-runs --force

The periodic reconciliation sweep performs the same zombie heal automatically; this command is the on-demand operator path.

Declarative apply (`pipelines apply -f`)

frankctl pipelines apply -f <file> provisions a complete Frank vertical (Source + Pipeline + BackingDataset) from a single multi-doc YAML file. The file is the unit of truth: re-running apply against the same file is idempotent.

File format

Each document carries a kubectl-style envelope:

yaml

apiVersion: frank.platform/v1
kind: Source | Pipeline | BackingDataset
metadata:
  name: <string>           # lookup key on apply
  labels: {...}            # optional, free-form
spec:
  # …the existing JSON request body for the resource…

spec is exactly what you'd POST to the corresponding REST endpoint — the envelope is a YAML-only convention. The same JSON wire format still works for clients that don't use YAML.

Worked example

air_quality_observed.yaml:

yaml

apiVersion: frank.platform/v1
kind: Source
metadata: { name: owm_air_pollution }
spec:
  name: owm_air_pollution
  pattern_id: rest_api
  source_config:
    base_url: https://api.openweathermap.org/data/2.5
    endpoint: /air_pollution
    query_params: { lat: "41.38", lon: "-8.20" }
  schedule_type: cron
  schedule_value: "0 * * * *"
---
apiVersion: frank.platform/v1
kind: Pipeline
metadata: { name: air_quality_observed }
spec:
  name: air_quality_observed
  description: OWM air quality → silver
  # source_ids resolves Source names → UUIDs at apply time
  source_ids: [ owm_air_pollution ]
  schedule_config:
    type: cron
    value: "0 * * * *"
---
apiVersion: frank.platform/v1
kind: BackingDataset
metadata: { name: air_quality_observed }
spec:
  # `pipeline` references the Pipeline doc above by name; the CLI
  # resolves to pipeline_id before POSTing.
  pipeline: air_quality_observed
  iceberg_namespace: silver_air_quality
  iceberg_table: air_quality
  entity_type_id: air_quality_observed
  entity_type_name: Air Quality Observed
  ontology_tenant_id: ts_demo
  primary_key_column: external_id
  property_mappings:
    - { column: external_id,   property: external_id, is_primary_key: true, type: string }
    - { column: date_observed, property: date_observed, type: string }
    - { column: pm2p5,         property: pm2p5,         type: number }
    - { column: no2,           property: no2,           type: number }
  # ensure_schema: Frank manages the ontology entity-type schema for this BD.
  # Spec § Gap #4. Field keys must match ^[a-z_][a-z0-9_]*$ — snake_case only.
  ensure_schema:
    display_name: Air Quality Observed
    fields:
      - { field_key: external_id,   field_type: { type: string },   required: true, indexed: true }
      - { field_key: date_observed, field_type: { type: datetime }, required: true }
      - { field_key: pm2p5,         field_type: { type: number },   required: false }
      - { field_key: no2,           field_type: { type: number },   required: false }

Apply:

bash

frankctl pipelines apply -f air_quality_observed.yaml
# applying Source/owm_air_pollution...
# applying Pipeline/air_quality_observed...
# applying BackingDataset/air_quality_observed...
# applied 3 doc(s)

Apply semantics

Dependency order. Docs are sorted client-side: Source → Pipeline → BackingDataset. File order within a kind is preserved.
Name resolution. Cross-doc references use metadata.name. A Pipeline.spec.source_ids entry that's a name (not a UUID) is looked up against (a) Sources just POSTed in this apply, (b) Sources already in the calling tenant. A miss aborts the apply BEFORE any POST.

Idempotency. Each POST carries ?if-not-exists=true. The server handles existing rows per the matrix below:

Existing row state	Response
Missing at the lookup key	201 (create)
Exists, mutable fields all match	200, untouched (no PATCH fires)
Exists, mutable fields differ	200, PATCHed in place
Exists, immutable fields differ	409 with `{error: "immutable_diff", existing_id, fields: [{name, existing, requested}]}`

Lookup keys: (tenant_id, name) for Source/Pipeline, (tenant_id, entity_type_id, ontology_tenant_id) for BackingDataset (NULLS NOT DISTINCT). Re-running apply -f against an unchanged file is a true no-op — no row writes, no schedule churn, operational state (cursors, file ledgers, snapshot ids, run counters, last_sync_at) is preserved.

Stops on first 409, no partial-apply. A 409 (typically immutable_diff) aborts the apply. Docs that succeeded stay in the database — fix the failing doc (or pass --allow-recreate, below) and re-run.
Immutable fields per kind (a change here triggers immutable_diff):
- Source: pattern_id
- Pipeline: none beyond tenant_id (everything else PATCHes; step DAG changes go through /versions)
- BackingDataset: iceberg_namespace, iceberg_table, entity_type_name, schema_library_ref, schema_version, sync_mode, transform_id, pipeline_id
--allow-recreate opt-in escape hatch: on an immutable_diff 409, DELETE the existing row via the existing_id in the response body and re-POST. Operational state is lost on the DELETE — the recreated row starts from a clean cursor, empty file ledger, no run history, status DRAFT/PENDING. Default off; use only when an immutable field genuinely needs to change (e.g. swapping pattern_id on a Source).
--allow-deprecate is passed through to BackingDataset POSTs to permit ensure_schema: narrowing. Default off.
--dry-run prints the apply plan (sorted, with metadata names) without POSTing.

Adoption workflow (export → review → commit → re-apply)

For pipelines that started life in the wizard UI and need to move into git, the round-trip is:

bash

# 1. Export the live pipeline (and its Sources + BackingDatasets).
frankctl pipelines export <pipeline-id> -o yaml > vertical.yaml

# 2. Review the YAML. Edit credentials to use ${ENV_VAR} expansion,
#    add `ensure_schema:` blocks on BackingDatasets you want Frank to
#    manage going forward, add labels, etc.
$EDITOR vertical.yaml

# 3. Commit the file to your repo.
git add vertical.yaml && git commit -m "adopt air-quality vertical"

# 4. Re-apply. Idempotent — runtime state (cursors, run history) is
#    preserved, so adoption is non-destructive.
frankctl pipelines apply -f vertical.yaml

After step 4 the file is the source of truth. Subsequent edits land via the same apply -f re-run. Wizard edits and apply edits coexist as long as nobody changes the file out of band: the next apply will PATCH any mutable drift back to the YAML.

`pipelines export <id>`

Round-trips an existing Pipeline (and its Sources + BackingDatasets) back to multi-doc YAML, ready for adoption into git:

bash

frankctl pipelines export <pipeline-id> -o yaml > vertical.yaml

Server-set fields (id, created_at, runtime counters, status) are stripped. ensure_schema: is not included on export — the operator adds the block if they want Frank to manage the ontology schema going forward. See § Gap #6 in the provisioning spec.

Direct YAML on any apply-able route

The YAML envelope works on the bare REST routes too — the same shape the CLI emits:

bash

curl -X POST https://api.frank.example/api/v1/sources \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/yaml" \
  --data-binary @source.yaml

Request: Content-Type: application/yaml on POST/PUT/PATCH to /api/v1/{sources, pipelines, backing-datasets} is accepted and unwrapped before the route handler sees it.

Response: Accept: application/yaml or ?format=yaml on GET against the same routes returns a YAML-wrapped envelope. JSON is still the default for both directions — the envelope is opt-in.

Exit codes

Code	Meaning
0	Success
1	Generic error
2	Usage / config error
3	Authentication error
4	API / server error
5	Validation error (e.g. sandbox failed)
6	Config-file error

Bronze type drift

A sync that fails with Cannot change column type: <field>: <bronze-type> -> <new-type> means Iceberg's union_by_name rejected the latest batch because the extracted value's type doesn't match what the bronze column was first written with. This usually happens when:

A source synced for a while, established the bronze column type (e.g. long), and
The extract pipeline later started emitting a different type for the same field (e.g. double — see the defensive int → float promotion in backend/services/extraction/coercers.py).

Two cures, your pick depending on what you want bronze to look like going forward.

Cure A — declare the field type so the coercer matches bronze

Pin the field's JSON-Schema type on the stream so the extract-time coercer fires on the way in (backend/services/extraction/coercers.build_field_coercers). Pinning dt: integer keeps dt as int64 end-to-end and the write matches the existing long column.

bash

# 1. See where bronze and the stream disagree.
frankctl sources streams diff-types <source-id> <stream-id>

# bronze: bronze.tenant_..._<source>.<stream>  (exists=true)
# FIELD  BRONZE  DECLARED  STATUS
# dt     long    (none)    bronze_only       ← will fail next sync
# value  double  (none)    bronze_only

# 2. Declare types matching bronze.
frankctl sources streams set-types <source-id> <stream-id> \
    --field dt=integer

# 3. Verify.
frankctl sources streams diff-types <source-id> <stream-id>
# dt   long  integer  coerce              ← now safe

The PATCH is partial-merge: it only touches the properties you pass, so an existing 50-field schema from discovery stays intact when you pin one field.

Cure B — widen the bronze column (deferred, see #469 slice B)

When you actually want bronze to take the wider type going forward (e.g. an integer field truly needs to become a float), the right fix is a bronze-side widen via PyIceberg update_schema().promote(...) for legal Iceberg promotions or a CTAS-and-atomic-swap for the others. Not shipped yet — tracked in issue #469.

Cure C — preventive (deferred, see #469 slice C)

The defensive int → float promotion in coercers.py is an extraction-side default that prevents an unrelated PyArrow inference bug. Teaching it to skip the promotion when bronze already has a narrower type pinned would prevent this class of incident at the source. Also tracked in #469.

frankctl — CLI Reference ​

Install ​

Configuration ​

Resources ​

frankctl sources ​

frankctl pipelines ​

frankctl backing-datasets ​

frankctl ai compose-pipeline ​

frankctl ai suggest source-config ​

frankctl transforms, frankctl runs, frankctl patterns, frankctl schedules, frankctl datasets, frankctl ai ​

Diagnosing a transform that won't run ​

frankctl transforms get <id> — runnability at a glance ​

frankctl transforms gate-status — the change-gate (ADL-016) ​

frankctl admin reconcile-runs — heal stuck runs & zombies ​

Declarative apply (pipelines apply -f) ​

File format ​

Worked example ​

Apply semantics ​

Adoption workflow (export → review → commit → re-apply) ​

pipelines export <id> ​

Direct YAML on any apply-able route ​

Exit codes ​

See also ​

Bronze type drift ​

Cure A — declare the field type so the coercer matches bronze ​

Cure B — widen the bronze column (deferred, see #469 slice B) ​

Cure C — preventive (deferred, see #469 slice C) ​

`frankctl` — CLI Reference

Install

Configuration

Resources

`frankctl sources`

`frankctl pipelines`

`frankctl backing-datasets`

`frankctl ai compose-pipeline`

`frankctl ai suggest source-config`

`frankctl transforms`, `frankctl runs`, `frankctl patterns`, `frankctl schedules`, `frankctl datasets`, `frankctl ai`

Diagnosing a transform that won't run

`frankctl transforms get <id>` — runnability at a glance

`frankctl transforms gate-status` — the change-gate (ADL-016)

`frankctl admin reconcile-runs` — heal stuck runs & zombies

Declarative apply (`pipelines apply -f`)

File format

Worked example

Apply semantics

Adoption workflow (export → review → commit → re-apply)

`pipelines export <id>`

Direct YAML on any apply-able route

Exit codes

See also

Bronze type drift

Cure A — declare the field type so the coercer matches bronze

Cure B — widen the bronze column (deferred, see #469 slice B)

Cure C — preventive (deferred, see #469 slice C)