| name | migrate-airbyte-connector |
| description | Migrate an existing Airbyte source connector to a Fivetran Connector SDK connector. Use when the user has an Airbyte Python CDK source, low-code YAML manifest, source connector directory, configured catalog, spec/config/state files, or Airbyte protocol output they want to port to CSDK. |
| argument-hint | Path or description of the Airbyte source connector to migrate |
Context: This plugin is for the Fivetran Connector SDK (CSDK). "CSDK" is shorthand for "Connector SDK".
Migrate an Airbyte Source Connector to CSDK
FIRST: Read sdk-reference.md from the plugin directory to load SDK rules, operation names, schema constraints, configuration rules, and testing requirements.
Use this workflow to convert Airbyte source connector logic into a Connector SDK project. Airbyte migration references:
When validating the migration workflow itself, use representative public connectors such as Airbyte's source-github, which exercises:
- Python CDK source structure with
source.py, stream classes, schema files, utilities, and tests.
- Nested credential objects with
oneOf auth modes and airbyte_secret fields.
- Array settings such as
repositories and branches, plus deprecated string aliases such as repository and branch.
- Wildcard repository expansion and config preflight validation.
- Parent-child streams, stream slices, per-stream state, semi-incremental streams, GraphQL/REST pagination, rate-limit token rotation, and stream-level failure handling.
Step 1: Locate the Source and Target
If the user did not provide a source path, ask for one of:
- An Airbyte source connector directory, such as
source-<name>/.
- A Python CDK source package with
source.py, stream classes, schemas, and tests.
- A low-code connector manifest YAML file.
- Airbyte artifacts:
spec.json, config.json, catalog.json, configured_catalog.json, state.json, and sample Airbyte protocol output.
Find or create the target CSDK project:
- If the current directory already has
connector.py, configuration.json, and requirements.txt, migrate in place.
- If no CSDK project exists, ask for the target directory name, then scaffold it with
fivetran init before editing files.
- Do not overwrite unrelated files.
Scope the migration:
- Migrate Airbyte sources only.
- Do not port Airbyte destinations, normalization, workspace/job orchestration, Docker packaging, or Airbyte platform metadata into
connector.py.
- For Airbyte platform resources that correspond to Fivetran platform objects, create a follow-up plan using
fivetran-cli instead of embedding that behavior in connector code.
Step 2: Inventory the Airbyte Source
Read source files and identify:
- Connector type: Python CDK source, low-code YAML manifest, Java/custom source, or protocol artifacts only.
- Source entry points:
spec, check, discover, and read command implementations.
- Configuration schema from
spec.json, connectionSpecification, manifest spec, or CDK spec().
- Credential fields from
airbyte_secret, OAuth/authenticator config, environment variables, or docs.
- Config migrations and deprecated aliases, such as old root-level credentials or string fields that are normalized into newer nested/array config shapes.
- Selected streams from
configured_catalog.json, connection catalog, low-code manifest, or user request.
- Stream schema from Airbyte catalog
json_schema, manifest schemas, schema files, CDK stream classes, or sample CATALOG/RECORD messages.
- Primary keys from
source_defined_primary_key, configured stream primary keys, stream class fields, or docs.
- Sync mode per stream:
incremental or full_refresh.
- Destination sync mode per stream:
append, append_dedup, or overwrite.
- Cursor field resolution from source-defined cursor, configured
cursor_field, default cursor field, manifest cursor, or CDK stream cursor.
- State shape from
state.json, Airbyte STATE messages, stream state, global state, or CDK state methods.
- Request logic: base URL, paths, request parameters/body/headers, authenticators, pagination, stream slices/partitions, record selector, transformations, retries, and error handling.
- Config preflight logic that expands wildcards, validates resource existence, or normalizes user inputs before stream construction.
- Parent-child stream dependencies. Some selected child streams require parent stream API calls for context even when the parent stream is not selected for output.
- Stream-level failure policy, including connectors that continue syncing other streams after one stream fails or is unavailable.
- Airbyte platform intent from destinations, connections, schedules, normalization, and jobs. Record this separately as potential
fivetran-cli follow-up work.
Do not ask the user to paste credentials in chat. Do not print config.json, .env, or OAuth credentials. Replace real values with placeholders if they appear in source files.
Step 3: Map Airbyte Concepts to CSDK
Use this mapping:
| Airbyte concept | Connector SDK concept |
|---|
| Airbyte source connector | One CSDK connector, or one scoped connector per source if the repo contains multiple unrelated sources |
spec.json / connectionSpecification / manifest spec | Flat string fields in configuration.json |
airbyte_secret: true | Sensitive configuration field; placeholder only, entered through secure config tool |
| Airbyte array/object config fields | Prefer separate connector deployments for multi-entity sync; use JSON-encoded string fields parsed with json.loads() only when unavoidable for source-connector parity |
| Airbyte config migrations / deprecated aliases | Backward-compatible parsing or documented renamed fields |
check command | validate_configuration(configuration) or a lightweight authenticated probe |
discover output / Airbyte catalog stream | CSDK schema(configuration) table entry |
| Airbyte JSON Schema properties | Optional CSDK columns with SDK data types |
| Configured/selected streams | Tables and fields implemented by schema() and update() |
| Parent streams used only for context | Internal helper/API calls; do not emit the parent table unless selected |
Airbyte RECORD message data | op.upsert(table=stream, data=record) |
Airbyte STATE message | op.checkpoint(state=new_state) |
Airbyte LOG and TRACE messages | SDK logging and exceptions; do not emit as rows |
| Airbyte per-stream tolerated errors | Catch/log documented nonfatal stream errors and continue only when the source connector did so |
incremental sync mode | Cursor logic inside update(configuration, state) |
full_refresh sync mode | Snapshot logic with explicit truncate/upsert decision |
Destination sync mode overwrite | op.truncate(table=...) only when full snapshot replacement is intended |
Destination sync mode append_dedup | op.upsert(...) with declared primary key |
Destination sync mode append | Requires explicit primary-key strategy before CSDK migration |
Map JSON Schema types conservatively:
| JSON Schema | CSDK column type |
|---|
boolean | BOOLEAN |
integer | LONG |
number | DOUBLE |
string with format: date-time | UTC_DATETIME |
string with format: date | NAIVE_DATE |
string | STRING |
object or array | JSON |
multiple types including null | Use the non-null type |
| unknown or mixed type | Omit the column type or use JSON when the field is structured |
Airbyte append-only streams need explicit review. CSDK tables should have stable primary keys. If an Airbyte stream has no primary key because the old destination used append mode, ask the user for a stable key strategy or document a generated deterministic key based on immutable source fields. Do not invent a lossy key silently.
Deletion behavior needs explicit review. If the Airbyte source emits deletion markers or CDC fields, map confirmed row deletes to op.delete(table=..., keys={...}) using primary-key fields only. Do not infer deletes from missing records unless the stream is explicitly full-refresh overwrite and the migration intentionally uses op.truncate.
Step 4: Port the Implementation
Edit the CSDK project files:
connector.py
- Keep the scaffolded structure where present:
validate_configuration(), schema(), update(), global connector = Connector(...), and if __name__ == "__main__": connector.debug().
- Port extraction logic directly into Python helper functions.
- For Python CDK sources, reuse small pure request/parse helpers where practical, but remove Airbyte command dispatch, Docker assumptions, protocol message generation, and CDK runtime dependencies.
- For low-code YAML manifests, translate manifest components into Python: requester/authenticator, pagination, record selector, stream slices, transformations, and cursor logic.
- If only Airbyte artifacts exist and no source code is available, treat catalog/spec/sample output as a behavior specification and reimplement from the source API/database/file protocol.
- Replace Airbyte config access with
configuration.
- Replace Airbyte state with the CSDK
state dict. Preserve the original state shape when it makes parity easier.
- Emit rows using SDK operations directly. Do not
yield operations and do not print Airbyte JSON messages.
- Call
op.checkpoint(state=state) after each safe page, slice, batch, or stream boundary.
- Preserve configured stream selection; do not silently add streams that were disabled in Airbyte.
- Preserve incremental cursor semantics, including start-date behavior for empty state.
- Preserve preflight normalization and validation that affects stream construction, such as wildcard expansion, deprecated config aliases, repository/org discovery, and branch normalization.
- Preserve documented stream-level failure behavior. If the Airbyte connector continued after a stream-specific 403/404/409/422/5xx, implement the same scoped skip/log behavior; otherwise raise.
- Add retry handling for HTTP 429 and transient 5xx responses.
- Keep type hints simple: use
dict and list; do not import Dict, Any, or use op.Operation.
configuration.json
- Keep flat string key/value pairs only.
- Include fields needed by the connector, using obvious placeholders only.
- Convert nested Airbyte settings to clear flat names, and document any rename.
- Prefer separate connector deployments for multi-entity configs (see
sdk-reference.md). Only when the Airbyte source truly requires multi-item selection, convert Airbyte arrays or objects to JSON-encoded string placeholders, such as "streams": "[\"users\"]" or "credentials": "{\"auth_type\":\"token\"}", then parse and validate them with json.loads() in connector.py.
- Preserve support for deprecated Airbyte config aliases when source code has explicit migration logic, or document that the CSDK connector only supports the new field names.
- Do not include real credentials from
config.json, .env, or Airbyte UI exports.
- Do not use arrays or nested objects.
requirements.txt
- Include only source-specific dependencies not already available in the CSDK environment.
- Do not include Airbyte CDK, Airbyte platform packages, Docker tooling, destination packages, or
fivetran_connector_sdk.
- Do not include normalization, dbt, orchestration, or test-only dependencies.
README.md
- Explain that this connector was migrated from an Airbyte source connector.
- Document source connector name, selected streams, primary keys, cursor fields, and sync modes.
- Document configuration fields with placeholders only.
- Document any behavior changes, especially append-only primary-key strategy, full-refresh overwrite/truncate behavior, delete/CDC handling, stream selection changes, and removed Airbyte platform behavior.
- Document any platform resources that were intentionally left out of connector code and should be handled with
fivetran-cli, such as destination/connection setup, scheduling, or transformation resources.
- Direct users to
tools/enter_configuration.py for secure configuration entry.
fivetran-cli follow-up notes
- If the Airbyte deployment includes destination or connection configuration that maps to Fivetran destinations or connections, document the intended Fivetran CLI resource commands to run after connector migration.
- If the Airbyte deployment includes schedules or transformation resources that map to Fivetran platform resources, document the intended Fivetran CLI follow-up.
- Do not execute account-changing
fivetran-cli commands unless the user explicitly asks for deployment/setup and provides the required non-secret identifiers. Never ask the user to paste API keys in chat.
Step 5: Validate the Migration
Check behavior before testing:
- Every selected stream has a CSDK schema entry and an explicit primary-key strategy.
- Disabled streams are not emitted as tables. If a disabled parent stream is needed for child context, it is used only as an internal helper path.
- Every Airbyte
RECORD stream maps to op.upsert.
- Every incremental stream checkpoints after a safe cursor boundary.
- Airbyte
STATE shape is preserved or deliberately mapped and documented.
- Array/object configuration fields are documented as JSON-encoded strings and parsed safely.
- Deprecated config aliases and migration behavior are either supported or explicitly documented as changed.
- Preflight expansion/validation behavior that affects stream construction is preserved.
- Stream-level failure behavior matches the source connector's documented behavior.
- Full-refresh/overwrite streams have an explicit documented decision: upsert-only snapshot or
op.truncate plus reload.
- Delete/CDC behavior is either implemented with
op.delete using primary-key fields or explicitly documented as not present in the source connector.
- JSON Schema to CSDK column type mapping is conservative and does not invent unsupported SDK types.
- No Airbyte protocol stdout message generation remains.
- No Airbyte destination, normalization, Docker, schedule, or workspace code is required for local CSDK execution.
- Any Airbyte platform resources that should survive migration are listed as
fivetran-cli follow-up work instead of being embedded in connector code.
- No real credentials are present in source files, README, or chat.
Then follow the secure test flow from test-connector:
- Run the secure runner, not
fivetran debug directly.
- If configuration values need to be entered or refreshed, direct the user to run
tools/enter_configuration.py in their own terminal.
- Do not inspect or print configuration values.
Step 6: Report Results
Summarize:
- Source Airbyte connector migrated.
- Connector type: Python CDK, low-code YAML, custom, or protocol-artifact migration.
- Streams migrated, primary keys, sync modes, and cursor fields.
- Configuration fields created or renamed.
- State mapping.
- Append/full-refresh/delete behavior decisions.
- Airbyte pieces intentionally not migrated into connector code, such as destinations, normalization, Docker, schedules, and workspace metadata.
fivetran-cli follow-up plan for platform resources, if applicable.- Remaining manual checks, especially API credentials, endpoint access, rate-limit behavior, and data parity against the original Airbyte sync.
