Analitiq Data Sync Engine

Move data between any two systems — just tell Claude where it lives and where it should go. No coding, no config files, no complexity. Open source.

Move data from/to any system:

• API -> Database
• Database -> Database
• API -> API
• SFTP -> Database
• and more

---

Why Analitiq Data Sync Engine?

• Just describe what you need — tell Claude in plain English where your data lives and where it should go. No code to write, no config files to learn.
• Connect to almost anything — APIs, databases, files, SFTP — if a connector exists in the registry, it just works. Need one that doesn't exist yet? Claude can build it for you.
• Only sync what changed — after the first run, the engine only picks up new and updated records, saving time and reducing load on your systems.
• Never lose data — if a sync is interrupted (network issue, server restart, anything), it picks up exactly where it left off. Failed records are saved for review, not silently dropped.
• Data arrives correctly — field types, formats, and mappings are handled automatically. Your dates stay dates, your numbers stay numbers, your text stays text.
• No vendor lock-in — runs on your own machine with Docker. No cloud account required, no surprise bills, no data leaving your network unless you want it to.
• One image, zero complexity — a single Docker image handles everything. No cluster to manage, no services to coordinate, no infrastructure to maintain.

---

How It All Fits Together

A Claude Code plugin builds your pipelines by pulling connectors from the registry. This repo is the engine that runs those pipelines.

---
config:
  layout: fixed
---
flowchart LR
 subgraph Engine["Analitiq Engine (this repo)"]
    direction LR
        Destination(["Data Destination"])
        Pipeline[["DATA PIPELINE"]]
        Source(["Data Source"])
  end
    ConnectorPlugin["Claude Plugin<br>(Connector Builder)"] -. new connector .-> Registry["Analitiq DIP Registry<br>1000s of connectors"]
    Source --> Pipeline
    Pipeline --> Destination
    Registry --> Pipeline
    PipelinePlugin["Claude Plugin<br>(Pipeline Builder)"] -. builds .-> Pipeline

     ConnectorPlugin:::plugin
     PipelinePlugin:::plugin
    classDef plugin fill:#fdf6e3,stroke:#cb8a02,stroke-dasharray: 5 5,color:#1a1a2e
    style Destination fill:#e8f5e9,stroke:#43a047,color:#1a1a2e
    style Pipeline fill:#f0f4ff,stroke:#4a6cf7,color:#1a1a2e
    style Source fill:#e8f5e9,stroke:#43a047,color:#1a1a2e
    style Registry fill:#f0fff4,stroke:#38a169,color:#1a1a2e
    style Engine fill:#fff0f0,stroke:#e53e3e,color:#1a1a2e

Component	Repository	What it does
Connector Builder Plugin	ai-plugins-official	Claude Code plugin that creates a connector for a new API or database
Pipeline Builder Plugin	ai-plugins-official	Claude Code plugin that assembles a pipeline from existing connectors
Connector Registry	analitiq-dip-registry	Pre-built connectors for common systems (Pipedrive, Wise, Xero, PostgreSQL, ...)
Data Sync Engine	analitiq-core (this repo)	Runs the pipelines — extract, transform, load

The registry already covers the most common systems. Use the plugin to pick a connector and assemble a pipeline. This engine executes it. Or skip the setup and use Analitiq Cloud for a fully managed experience.

Don't see your system in the registry? The Connector Builder plugin can create one from API docs — but for most use cases, a pre-built connector is already available.

Learn more at analitiq.ai.

---

Getting Started

Prerequisites

• Docker
• Claude Code (recommended)
• A pipeline configuration (see below)

1. Build a pipeline

Install the analitiq-pipeline-builder plugin for Claude Code, then open this project directory and say:

"I need to move data from Pipedrive to PostgreSQL"

The plugin walks you through choosing connectors, entering credentials, and mapping fields. When it finishes, three directories appear:

connectors/    # connector definitions (downloaded from the registry)
connections/   # your connection configs and credentials
pipelines/     # pipeline and stream definitions

2. Run the pipeline

Open Claude Code in this project directory and say:

"Run the pipedrive-to-postgresql pipeline"

Claude reads the pipeline configuration and runs it in Docker for you.

Or run it manually:

cd docker && \
  PIPELINE_ID=<your-pipeline-id> \
  docker compose run --rm source_engine

The engine reads from the source, transforms the data, and writes to the destination. State is checkpointed so interrupted runs resume where they left off.

Runtime archives

For environments where the generated configuration is prepared separately, you can package the normal runtime directories into a local archive and hydrate them before starting the engine.

Create an archive:

tar -czf runtime-config.tar.gz connectors connections pipelines state

Hydrate it into an empty working directory:

python src/runtime_archive.py hydrate runtime-config.tar.gz -C /tmp/analitiq-run

Hydrate and run the engine in one command:

PIPELINE_ID=<your-pipeline-id> \
python src/runtime_archive.py run runtime-config.tar.gz -C /tmp/analitiq-run -- \
  python -m src.main

The archive must contain pipelines/manifest.json. After hydration, the engine uses the same filesystem layout as a normal local run.

---

Architecture

graph LR
    subgraph Docker
        direction LR
        SRC[Source Engine]
        DST[Destination Service]
        SRC -- "gRPC stream" --> DST
    end

    EXT[(Source System)] --> SRC
    DST --> TGT[(Destination System)]

• Single Docker image — engine and destination run from the same image, toggled by RUN_MODE
• Engine performs extract, transform, and streams batches to the destination over gRPC
• Destination receives batches and writes idempotently (database upsert, file manifest, API POST)
• State is checkpointed per stream so interrupted runs resume automatically

---

Secrets

Secret references are resolved at connection time (late-binding), not at config load. The engine resolves from:

• an environment variable;
• a local file inside the connection directory;
• a sidecar credentials file beside the connection, the local-development default;
• an S3 / S3-compatible object, which lazily imports boto3 under pip install 'analitiq-core[s3]'.

The first three are built-in and cloud-free. An unresolvable reference (missing env var, missing file/object, missing sidecar entry, unsupported scheme) fails loud — the connection is never established with an empty secret.

Configuration and defaults

Every Python-side engine default and its environment-variable override is declared once, in src/config/settings.py. Per-pipeline overrides go in the pipeline document's runtime block; precedence and resolution order are specified in docs/config/settings-reference.md.

The table below is a quick-start subset for local development, not the complete environment-variable list — it deliberately excludes deployment-injected correlation IDs (RUN_ID, AWS_BATCH_JOB_ID, INVOCATION_ID, ORG_ID) that a cloud run supplies rather than an operator setting. The one complete list, kept in one place so it can't go stale in two, is docs/config/settings-reference.md for everything outside settings.py, and src/config/settings.py itself for everything in it.

Environment Variables

Variable	Description
PIPELINE_ID	Pipeline ID from manifest.json (required)
RUN_MODE	source or destination
LOG_LEVEL	Logging level until the pipeline config is read; a declared runtime.logging.log_level supersedes it
DESTINATION_GRPC_HOST / DESTINATION_GRPC_PORT	Destination service address (engine mode)
GRPC_PORT	gRPC listen port (destination mode)
DESTINATION_INDEX	Which destination from the pipeline config to serve
ANALITIQ_BATCH_SIZE / ANALITIQ_BUFFER_SIZE	Batch size and inter-stage queue depth
ANALITIQ_ERROR_STRATEGY / ANALITIQ_MAX_RETRIES / ANALITIQ_RETRY_DELAY_SECONDS	Exhausted-retry policy and backoff
AWS_ENDPOINT_URL_S3 (falls back to AWS_ENDPOINT_URL) / AWS_REGION (falls back to AWS_DEFAULT_REGION)	Endpoint and region for s3:// secret refs (S3-compatible stores, e.g. MinIO)
CONFIG_BUNDLE	Read by docker/entrypoint.sh, before Python starts: a config archive (local path or URL) to hydrate into the container. Unset for local runs that bind-mount config directly
CONNECTORS_DIR	Read by docker/entrypoint.sh; default /app/connectors

Defaults and every other settings.py-backed env var (gRPC, worker supervision, process role) are in src/config/settings.py; the handful that aren't are the ones named just above.

Durable incremental cursors resume from per-stream state/{pipeline_id}/{stream_id}.json checkpoint files rather than an environment variable: each stream writes its own committed cursor on every destination ACK, and the deployment delivers those same files in the config bundle on a fresh container. See engine-architecture.md.

---

Contributing

See CONTRIBUTING.md.

License

Apache License 2.0