Bootstrapping a service

Python SDK

From zero to a running integration service — the receive endpoint, the webhooks, and the sync engine — in a few commands. Then swap the demo ERP for yours.

1. aiotic init

mkdir acme-aiotic && cd acme-aiotic
python -m venv .venv && source .venv/bin/activate && pip install "aiotic-sdk[all]"
aiotic init

Writes:

• .env — base URL, integration key, an empty sync key, a generated AIOTIC_ERP_RECEIVE_KEY (give this value to AIOTIC together with your endpoint URL; the service refuses to start without it), an empty webhook key, an empty events key.
• service.py — three lines that build the app with the in-memory demo ERP.

2. aiotic doctor

Checks the tenant is reachable, the keys work, and master data is present. Run it after every configuration change and in your deployment health checks.

3. aiotic serve

Runs service.py with uvicorn on :9000:

Route	Purpose
POST /aiotic/orders	ERP receive endpoint
POST /aiotic/processing	processing webhook (when AIOTIC_WEBHOOK_KEY is set)
POST /erp/events	ERP change events → sync engine (only when a sync_engine is passed; key AIOTIC_ERP_EVENTS_KEY, falling back to the receive key)
GET /healthz	liveness
GET /docs	OpenAPI UI of your service

4. Replace the ERP

service.py, grown up:

import pyodbc
from aiotic import AioticClient
from aiotic.erp.data_api import DataApiAdapter
from aiotic.pipeline import Pipeline, sanitizers as S, validators as V, rules as R
from aiotic.receive import SqliteStore
from aiotic.service import build_app
from aiotic.sync import HashStateStore, SyncEngine

erp = DataApiAdapter(lambda: pyodbc.connect(DSN), paramstyle="qmark")      # or FunctionalApiAdapter(...)
store = SqliteStore("integration.db")
pipeline = Pipeline(
    sanitizers=[S.StripWhitespace(), S.NormalizeCountryCodes(), S.NormalizeCurrency(), S.MapUnits({"doos": "BOX"}), S.FillShippingFromCustomer()],
    validators=[V.RequiredFields(), V.CustomerResolved(erp), V.ArticlesInCatalog(erp), V.PositiveQuantities(), V.CurrencyAllowed(["EUR"])],
    rules=[R.CustomerNotBlocked(erp), R.NoDuplicateOrder(store), R.ShipToAddressComplete()],
)
sync_engine = SyncEngine(AioticClient(), state=HashStateStore("sync-state.db"))

app = build_app(erp=erp, pipeline=pipeline, store=store, sync_engine=sync_engine)

build_app accepts any ErpPort; when the adapter also implements CatalogPort / CustomerPort (both templates do), the default pipeline picks them up automatically, so the explicit pipeline= above is optional.

5. Add the sync side

• Events: point your ERP's webhooks / outbox worker at POST /erp/events (format), or call sync_engine.apply(...) from your own code.
• Polling: a PollingChangeSource in a background thread or a separate process.
• Reconciliation: a scheduled aiotic sync reconcile … or sync_engine.reconcile(...).

6. Deploy

• Run behind TLS (a reverse proxy or your platform's ingress). The receive endpoint must be reachable by the tenant.
• One instance is enough for most volumes; if you scale out, replace SqliteStore / HashStateStore / SqliteWatchState with your database (each is a ~20-line protocol).
• Secrets from the environment; never bake .env into an image.
• Health: GET /healthz for liveness; aiotic doctor for readiness in CI.

A Dockerfile for the service:

FROM python:3.12-slim
WORKDIR /app
RUN pip install --no-cache-dir "aiotic-sdk[all]" pyodbc
COPY service.py .
CMD ["uvicorn", "service:app", "--host", "0.0.0.0", "--port", "9000"]