Webhooks & receivers

Python SDK

Three inbound HTTP surfaces live in an integration service. aiotic.receive and aiotic.webhooks implement them framework-free, with FastAPI routers on top.

Surface	Direction	SDK
ERP receive endpoint	AIOTIC → you	ErpReceiver, create_receive_router
Processing webhook	AIOTIC → you	ProcessingWebhookReceiver, create_webhook_routers(processing=…)
ERP change events	your ERP → you	parse_change_events, create_webhook_routers(on_change_events=…)

ERP receive endpoint

from aiotic.receive import ErpReceiver, SqliteStore, create_receive_router

receiver = ErpReceiver(erp, api_key=RECEIVE_KEY, pipeline=pipeline, store=SqliteStore("integration.db"),
                       on_accepted=lambda req, result, verdict: metrics.inc("booked"),
                       on_rejected=lambda req, resp, verdict: metrics.inc("rejected", reason=resp.error))
app.include_router(create_receive_router(receiver, path="/aiotic/orders"))

handle(body, api_key_header=…) -> ReceiveOutcome does everything in order: key check (constant time) → parse → idempotency (store, then adapter lookup) → pipeline → create_sales_order → remember → respond. Business problems never raise; unexpected exceptions become success: false with a generic message and are logged with the traceback.

Using another framework:

# Flask
@app.post("/aiotic/orders")
def receive():
    out = receiver.handle(request.get_data(), api_key_header=request.headers.get("X-API-KEY"))
    return jsonify(out.response.model_dump(exclude_none=True)), out.http_status

Processing webhook

from aiotic.webhooks import ProcessingWebhookReceiver, create_webhook_routers

hook = ProcessingWebhookReceiver(WEBHOOK_KEY, handler=lambda req: drafts.create(req.request_id, req.purchase_order))
app.include_router(create_webhook_routers(processing=hook))          # POST /aiotic/processing

The handler's exceptions are logged and acknowledged anyway — AIOTIC does not retry, so failing loudly would only lose the signal.

ERP change events

app.include_router(create_webhook_routers(on_change_events=engine.apply_many, erp_events_key=EVENTS_KEY))   # POST /erp/events

Accepts the generic format described in Event-driven sync. Protect it with its own key (X-API-KEY) — it is your ERP calling, not AIOTIC.

Signed webhooks (future)

verify_hmac_signature(secret, body, header) implements the scheme in the proposal (X-AIOTIC-Signature: t=<unix>,v1=<hmac-sha256>), with a 5-minute tolerance. Nothing in AIOTIC sends it yet; it is here so your endpoint can be ready.

build_app() wires all three

aiotic.service.build_app(erp=…, pipeline=…, store=…, sync_engine=…) mounts the receive endpoint, the processing webhook (when AIOTIC_WEBHOOK_KEY is set), the change-event endpoint (protected with AIOTIC_ERP_RECEIVE_KEY by default — give it its own key in production) and /healthz.