# flowx
Orchestrator-to-Databricks Lakeflow Jobs translator, delivered as agent skills.
flowx converts a source orchestrator's pipelines — **Azure Data Factory (ADF)** or **Apache
Airflow** — into Databricks Lakeflow Jobs packaged as Declarative Automation Bundles (DABs). It
deterministically translates known activity/operator types and falls back to agentic (LLM-assisted)
translation for complex or rare types. flowx runs as a set of [agent skills](skills/) usable from
Databricks Genie Code, Claude Code, or any tool that supports the Agent Skills standard.
Both sources emit the same source-neutral Pipeline IR, so the convert-configuration and package
phases are shared; only discovery and translation are source-specific. Pick the source with
`--source {adf,airflow}` (required for discover/convert; package is source-independent).
## Architecture
```
flowx Pipeline
==================
ADF ARM/JSON (UC Volumes / Workspace) | Airflow DAG .py files
\ | /
v v v
+---------------------------------------------------------------+
| 1. DISCOVER sources/<adf|airflow>/ -> metadata/inventory.json
| (ADF: ARM/JSON parse; Airflow: static ast parse)
+---------------------------------------------------------------+
|
v
+---------------------------------------------------------------+
| 2. CONVERT sources/<adf|airflow>/ -> shared Pipeline IR
| (deterministic mappings + agentic gaps)
+---------------------------------------------------------------+
|
v
+---------------------------------------------------------------+
| 3. PACKAGE bundler/dab_writer.py (source-independent)
| IR -> DAB YAML + notebooks + setup scripts
+---------------------------------------------------------------+
|
v
databricks bundle validate / deploy
```
The phases are exposed two ways: as **skills** the agent runs directly (via a Python virtual
environment locally), and as a single **MCP tool** hosted on a Databricks App (for Genie Code). See
[Running flowx as an MCP server](#running-flowx-as-an-mcp-server).
## Installation
flowx installs in one of two shapes depending on where your agent runs. Full, step-by-step
instructions for both are in the [installation docs](docs/content/docs/installation.mdx); the
summary:
### Databricks Genie Code
The phases run as an MCP server (a Databricks App). Clone the repo into **`/Workspace/Shared`**
(so the app's service principal can read the source), copy `skills/` into your skills folder, then
run the setup skill:
```
@flowx-setup
```
On Databricks, `flowx-setup` deploys the `mcp-flowx` app for you. You can also deploy it directly by
running the **`app/deploy_app.py`** notebook (SDK-based, works on serverless), or `app/deploy.sh`
from a workspace web terminal. Then add the app under Genie Code **Settings → MCP Servers → Add
Server → Custom MCP server**.
### Claude Code (and other local agent harnesses)
flowx is a Claude Code plugin distributed through its marketplace:
```
/plugin marketplace add databricks-solutions/flowx
/plugin install flowx@flowx
```
Run `/reload-plugins`, then set up the local runtime once:
```
/flowx:flowx-setup
```
This provisions a Python virtual environment (via `scripts/bootstrap.sh`) and writes a
`.migration-venv` marker the phase skills read. No `uv` is required for plugin users.
## Usage
Run the end-to-end migration:
```
/flowx:flowx-migrate
```
Or run individual phases:
```
/flowx:flowx-extract-metadata # (ADF) Get the factory's definitions: ADF Studio export, Git, or live Azure API
/flowx:flowx-discover # Parse the source (ADF JSON / Airflow DAGs), produce inventory + complexity report
/flowx:flowx-convert # Deterministic + agentic translation
/flowx:flowx-package # Generate DABs project
/flowx:flowx-profile # (ADF, optional) Estimate current ADF spend from a live Azure scan
```
(In Genie Code, invoke the same skills with the `@` prefix, e.g. `@flowx-migrate`.)
## Setup
`/flowx:flowx-setup` keys off the `DATABRICKS_RUNTIME_VERSION` environment variable
(the same signal the rest of the plugin uses to detect Databricks) and prepares one
of two execution paths:
- **Local / Claude Code (virtual environment).** The phases run from the plugin's
CLI. Setup runs `scripts/bootstrap.sh`, which creates a `.venv`, installs
`requirements.txt`, and writes the resolved interpreter path to a
`.migration-venv` marker file that the phase skills read. Optionally, a local
(stdio) MCP server can be registered to drive the phases through MCP tools
instead of the CLI.
- **Databricks Genie Code (MCP server, no virtual environment).** The phases run as
a single `flowx` MCP tool hosted on a Databricks App. Setup runs `app/deploy.sh`,
which stages a self-contained bundle, syncs it to `/Workspace/Shared/mcp-flowx`,
and deploys the `mcp-flowx` app. You then grant app/data access and register the
app under Genie Code **Settings → MCP Servers**. No venv is created on this path.
Run setup once before any other flowx skill, or again whenever the environment is
missing.
## Supported ADF Activity Types
### Deterministic (16 types)
| ADF Activity | Databricks Task | Category |
|---|---|---|
| Copy | Notebook task | Data movement |
| DatabricksNotebook | Notebook task | Compute |
| DatabricksSparkJar | Spark JAR task | Compute |
| DatabricksSparkPython | Spark Python task | Compute |
| ForEach | for_each_task | Control flow |
| IfCondition | if_else_task | Control flow |
| Switch | if_else_task chain | Control flow |
| SetVariable | run_job_task | Control flow |
| AppendVariable | run_job_task | Control flow |
| Filter | Notebook task | Control flow |
| Wait | Notebook task (sleep) | Control flow |
| Lookup | Notebook task | Data access |
| WebActivity | Notebook task | External |
| Delete | Notebook task | Data management |
| ExecutePipeline | run_job_task | Orchestration |
| DatabricksJob | run_job_task | Compute |
### Agentic Fallback (12 types)
Activities with complex semantics, or without a direct Databricks equivalent, are translated by the
agent using LLM-assisted reasoning from the activity's ARM JSON.
| ADF Activity | Strategy |
|---|---|
| ExecuteDataFlow | LLM-assisted (agentic) |
| SqlServerStoredProcedure | LLM-assisted (agentic) |
| AzureFunction | LLM-assisted (agentic) |
| WebHook | LLM-assisted (agentic) |
| Custom | LLM-assisted (agentic) |
| ExecuteSSISPackage | LLM-assisted (agentic) |
| AzureMLExecutePipeline | LLM-assisted (agentic) |
| GetMetadata | LLM-assisted (agentic) |
| Validation | LLM-assisted (agentic) |
| Fail | LLM-assisted (agentic) |
| Script | LLM-assisted (agentic) |
| Until | LLM-assisted (agentic) |
## Supported Airflow Operators
The Airflow source parses DAG `.py` modules **statically** (via `ast`, no Airflow install or DAG execution) and maps the constructs below to the shared IR — 74 operator/sensor types in the parser registry, plus the dbt CLI, Cosmos, and TaskFlow constructs.
Unlike the ADF path, an Airflow operator's path is decided **per instance** by whether its inputs are statically resolvable. A recognized operator lowers to a real Databricks task (**deterministic**), or — when a value can't be resolved from source alone — to a failing placeholder + `gaps.json` entry (**agentic**). A few constructs are always agentic because their behavior is inherently runtime.
| Category | Operator(s) | Databricks target | Path |
|---|---|---|---|
| Compute / scripts | `PythonOperator` | Notebook task (callable + transitive deps) | Deterministic † |
| Compute / scripts | `PythonVirtualenvOperator`, `ExternalPythonOperator` | Notebook task + `%pip install` | Deterministic |
| Compute / scripts | `BashOperator`, `SSHOperator` | `%sh` notebook (`spark-submit` lifted when option arity is known) | Deterministic † |
| Compute / scripts | `SparkSubmitOperator` | Spark JAR / Python task | Deterministic |
| Teradata | `BteqOperator`, `TeradataOperator`, and Bash/SSH commands that run `bteq`, TPT (`tbuild`, `tdload`), or the load/export utilities | Placeholder + gap carrying the script | Agentic |
| Databricks provider | `DatabricksSubmitRunOperator` (+ `…DeferrableOperator`) | Notebook / run task | Deterministic |
| Databricks provider | `DatabricksRunNowOperator` (+ `…DeferrableOperator`) | `run_job_task` | Deterministic |
| Databricks provider | `DatabricksNotebookOperator` | Notebook task | Deterministic |
| SQL | `DatabricksSqlOperator`, `DatabricksSQLStatementsOperator`, `SQLExecuteQueryOperator`, `PostgresOperator`, `MySqlOperator`, `HiveOperator` | `sql_task` | Deterministic † |
| SQL | `DatabricksCopyIntoOperator` | `sql_task` (COPY INTO) | Deterministic |
| Orchestration | `TriggerDagRunOperator` | `run_job_task` (target DAG by sanitized job name) | Deterministic |
| Control flow | `BranchPythonOperator`, `ShortCircuitOperator` | Placeholder + gap | Agentic |
| Control flow | `DummyOperator`, `EmptyOperator` | — (dropped, dependencies rewired) | Dropped |
| Notification | `EmailOperator` | Placeholder + gap | Agentic |
| Sensors — file | `S3KeySensor`, `GCSObjectExistenceSensor`, `FileSensor`, `HdfsSensor`, `WebHdfsSensor` | `file_arrival` trigger (root, no schedule) or `dbutils.fs` polling task | Deterministic † |
| Sensors — table/SQL | `DatabricksPartitionSensor`, `DatabricksSqlSensor`, `DatabricksSQLStatementsSensor`, `SqlSensor` | `table_update` trigger (root literal table, no schedule) or `spark.sql` polling task | Deterministic † |
| Sensors — poll | `HttpSensor`, `PythonSensor`, `DateTimeSensor` (+ `…Async`) | Polling notebook task | Deterministic † |
| Sensors — cross-DAG | `ExternalTaskSensor` (+ `…Async`) | Placeholder + gap | Agentic |
| Sensors — time | `TimeSensor`, `TimeDeltaSensor` | Placeholder + gap | Agentic |
| dbt | `DbtRun/Test/Seed/Snapshot/Build/DepsOperator`, Cosmos `DbtDag` / `DbtTaskGroup` | single `DbtFactoryActivity` (static explosion, or PyDABs via `--dbt-mode pydabs`) | Deterministic |
| Dataproc / Managed Spark | `DataprocSubmitJobOperator`, `DataprocCreateBatchOperator` (+ `ManagedSpark…` aliases) with `pyspark_job` / `pyspark_batch` | `spark_python_task` | Deterministic † |
| Dataproc / Managed Spark | same, with `spark_job` / `spark_batch` | `spark_jar_task` (JARs as task libraries) | Deterministic † |
| Dataproc / Managed Spark | same, with inline `spark_sql_job` | `sql_task` | Deterministic † |
| Dataproc / Managed Spark | same, with `spark_r_*`, `pyspark_notebook_batch`, `hive_job`, `hadoop_job`, `pig_job`, `flink_job`, `presto_job`, `trino_job` | Placeholder + gap | Agentic |
| Dataproc / Managed Spark | cluster create / delete / start / stop, `DataprocJobSensor`, `DataprocBatchSensor` | — (absorbed into Jobs compute / native task completion, dependencies rewired) | Dropped ‡ |
| Dataproc / Managed Spark | cluster update / scale / diagnose, workflow templates, batch control, cancel operation | Placeholder + gap | Agentic |
| TaskFlow API | `@dag`, `@task`, `@task.virtualenv` | task; implicit XCom → `dbutils.jobs.taskValues` | Deterministic † |
| TaskFlow API | `@task.expand([literal])` | `for_each_task` | Deterministic † |
| TaskFlow API | `@task.branch`, `@task.short_circuit`, `@task_group` (decorator form) | Placeholder + gap | Agentic |
**†** Deterministic when the relevant values are statically resolvable; otherwise the instance routes
to an agentic gap — e.g. a non-literal command / SQL / endpoint / path, a SQL template file (`.sql` / `.hql`) that is not next to the DAG or in a static `template_searchpath`, a callable that reads Airflow
task context or XCom, an unsafe inline template context, or a non-literal `.expand()` /
`.partial().expand()`.
**Agentic** emits a failing placeholder + `gaps.json` entry (eligible for the leaf-gap
resolver) because the construct's behavior is inherently runtime.
**‡** Dropped only when provably safe: every job on the cluster migrates, no other task reads the removed
task's output, the sensor pairs with an upstream asynchronous submission (or matching `batch_id`) (its timing is reported, not turned into a task timeout), and rewiring does
not change trigger-rule behavior. Otherwise the task stays as a placeholder with the reason attached.
**Scheduling & semantics** (source-level, applied alongside the operators above): cron → Quartz,
`timedelta` → a `start_date`-anchored Quartz cron when its length divides an hour or a day (else periodic), `@continuous` → continuous mode, Airflow-3 Asset/Dataset lists →
`ALL_UPDATED` / `ANY_UPDATED` table triggers, `trigger_rule` → `run_if`, `params={...}` / `Param` →
job parameters, and `>>` / `<<` / `set_upstream` / `set_downstream` / TaskGroup edges → task
dependencies. Interval macros (`{{ ds }}`, `{{ ts }}`, `{{ data_interval_start }}`, ...) are computed per run
by a generated first task that reproduces Airflow's rendering: the previous schedule tick for an Airflow 2
data-interval timetable, the fire time for an Airflow 3 raw cron, and the trigger time for a manual run.
Operators without a deterministic mapping become a failing placeholder and are recorded in
`gaps.json` for review. Eligible leaf gaps can use the fingerprint-bound resolver backed by the pinned [`airflow-to-dabs`](https://github.com/park-peter/airflow-to-dabs/tree/main/providers/flowx-gap-resolver) provider profile; flowx retains ownership of parsing, graph identity, policy, IR, and packaging. Full matrix:
[`skills/flowx-convert/sources/airflow-coverage.md`](skills/flowx-convert/sources/airflow-coverage.md).
Airflow discovery independently audits DAG declarations, task candidates, dependency declarations,
DAG settings, mapped calls, and operator arguments before comparing them with captured IR. An
included DAG is `verified` when every audited construct has a proven translation,
`verified_with_gaps` when every unsupported construct is linked to a runnable-failure placeholder,
or `failed` when reconciliation finds unexplained loss. Failed reconciliation exits nonzero and
blocks package writes. `--exclude-dag <dag_id>` is repeatable; excluded DAGs emit no Job but remain
visible with zero translated activities in inventory and coverage reporting. This guarantee applies
to the supported static subset; flowx never imports or executes DAG modules.
## How It Works
### Phase 1: Discover
Parses the source into typed nodes and classifies each activity/operator as deterministic, agentic, or unsupported — ADF JSON from Unity Catalog volumes (or a `/Workspace` Git folder, normalizing ARM template format), or Airflow DAG `.py` modules read statically with `ast`. Airflow inventory includes audited/deterministic/agentic/failed/excluded counts, reconciliation status, stable finding fingerprints, translation-path coverage, and deterministic coverage. Produces `metadata/inventory.json` and a per-pipeline complexity report at `metadata/profile_report.csv`.
### Phase 2: Convert
Applies deterministic translators (ADF activity registry / Airflow operator mapping), resolves dependencies, and records unresolved gaps. ADF supports its guided agentic translation workflow. Airflow supports a fingerprint-bound, explicitly reviewed leaf-gap workflow whose constrained provider output is replayed against an immutable deterministic baseline before packaging. Produces the shared Pipeline IR consumed unchanged by the package phase.
### Phase 3: Package
Converts Pipeline IR into a deployable DABs project: `databricks.yml`, per-job YAML resource files, generated Python notebooks, and setup scripts for UC volumes, secrets, and connections. A multi-pipeline factory can be laid out into bundles with `--packaging-mode` (`per-pipeline` default / `single` / `per-group`); a top-level `DEPLOY.md` records the suggested callees-first deploy order, and `python -m flowx.adapter deploy` deploys the bundles in that order, wiring cross-bundle job ids automatically.
## Output Format
All three phases write into one shared output directory (default `./flowx_output`):
```
flowx_output/
databricks.yml # Bundle configuration (package)
resources/
jobs/
<pipeline_name>.yml # One Job per included ADF pipeline or Airflow DAG
src/
notebooks/
<pipeline_name>/
<activity_name>.py # Generated notebooks per activity
setup/
create_volumes.py # UC volume setup
create_secrets.py # Secret scope setup
create_connections.py # Connection setup
SETUP.md # Setup instructions (package)
metadata/
inventory.json # discover: activity inventory
profile_report.csv # discover: per-pipeline complexity report
<pipeline>.arm.json # discover: verbatim original ADF/ARM source
configuration.json # modify: collected configuration answers
.work/ # transient intermediates (translation report, IR, gaps.json); pruned by package
```
## Running flowx as an MCP server
The phases are also packaged as [Model Context Protocol](https://modelcontextprotocol.io) tools (in
[`src/flowx/mcp/`](src/flowx/mcp)) so an agent can invoke them directly instead of shelling out to
the CLI. The server exposes a single `flowx(command, parameters)` tool to stay under host tool
limits. For Databricks Genie Code it runs as a Databricks App; see the [app README](app/README.md)
for deployment (SDK notebook or CLI script) and Genie Code registration.
## Development
```bash
make dev # Install dependencies (uses uv)
make test # Run unit tests
make integration # Run integration tests (excludes the live-Azure suite; gates CI)
make integration-live # Also run tests marked integration (live Azure; none today)
make fmt # Format + lint (ruff + mypy)
make clean # Remove build artifacts
```
### Prerequisites
- Python 3.12+
- [uv](https://docs.astral.sh/uv/) package manager
These prerequisites are for contributing to the flowx project. Plugin *users* do not need `uv` —
`flowx-setup` provisions the runtime (a pip-based `.venv` locally, or the MCP server on Databricks).
## Contributing
1. Fork the repository
2. Create a feature branch (`git checkout -b feature/my-feature`)
3. Follow the [adding a new translator](CLAUDE.md#adding-a-new-deterministic-translator) guide
4. Run `make fmt && make test` before committing
5. Open a pull request