Run Convilyn AI workflows on the device.

The Edge SDK is for code that runs on the device — a Jetson, an AI-PC, a Snapdragon box, a register sidecar. Calling Convilyn from a server or app instead? That's the Consumer SDK. Extending the platform with your own tools? That's the Author SDK.

The three planes

Convilyn splits an edge AI product into three planes so the vertical logic stays removable and the SDK stays general:

Convilyn ships the SPI + a simulator + reference adapters only — never hardware drivers or real action connectors. Real OPOS/.NET, Zebra/Kotlin, serial / MQTT / camera adapters and any actuation beyond R0/R1 (R0/R1 — display-only and request-for-help actions) are integrator work. That boundary is what keeps the SDK a general-purpose substrate: delete any vertical pack and the SDK still works.

Where workflows come from — the shared Builder

A workflow is authored once in the Builder. Convilyn's chat-driven Builder turns a natural-language conversation into a compiled workflow (uw_…) stored in your account. The Builder is a shared, client-agnostic authoring capability — the same one behind the Convilyn web app — not part of the Edge SDK, and it bakes in no notion of "edge" vs "web" vs any product. Whatever client you author from, the result is the same generic workflow spec.

You don't have to bounce a user to the web app to author: the Builder is now callable from the Consumer SDK too (client.builder — build by chat, get the uw_ back), so a desktop app or an edge-deployment tool can drive the whole "describe it → deploy it" loop from its own surface. Building runs in the cloud (Pro tier). At run time, cloud placement runs the workflow server-side from any client; edge placement runs the workflow's model node on-device via a local model — offline-capable inference, re-grounded server-side.

Every client then runs that workflow through the one server-side workflow runtime:

• a server or app via the Consumer SDK — client.goals.run(user_workflow_id="uw_…")
• a device via this Edge SDK's ModelOperator (cloud placement wraps that same client.goals call; edge placement runs a local model instead)

So the split is build once (Builder) → invoke from any client: cloud placement runs the workflow server-side; edge placement runs its model node on-device. There is no portable run-anywhere workflow artifact — on-device workflows are composed with the Edge SDK's runtime primitives. The Edge SDK never builds a workflow — it consumes one, exactly like every other client.

Install

uv add --prerelease=allow convilyn-edge   # or: pip install --pre convilyn-edge

Python ≥ 3.10, zero runtime dependencies — the SPI is pure typed Protocols and frozen dataclasses, so adapters and packs depend on the interface, never on a runtime. Runnable examples ship in the package's examples/ directory.

The 7 primitives

Each is one narrow Protocol — depend on the interface, not a runtime (DIP/ISP).

Everything crosses the SDK inside one EventEnvelope (uniform id / schema version / correlation / ordering — the basis for dedup, replay, and audit).

Simulate with no hardware

A developer shouldn't need a real scanner to build a workflow. Replay a scenario through the built-in simulator — from the CLI or in code:

convilyn-edge simulate scenario.json --no-delay   # one wire-JSON envelope per event
convilyn-edge init adapter my-sensor              # scaffold a device adapter
convilyn-edge init workflow my-workflow           # scaffold a workflow

from convilyn_edge import EventSourceRef
from convilyn_edge.simulator import Scenario, ScenarioEvent, SimulatedSource
from convilyn_edge.spi.source import SourceContext
 
scenario = Scenario(
    source=EventSourceRef("dev-01", "sim-sensor", "0.1.0"),
    events=(
        ScenarioEvent(
            event_type="device.sensor.reading.received",
            event_schema="convilyn://schemas/sensor-reading/v1",
            data={"device_id": "dev-01", "metric": "motion", "value": 1},
        ),
    ),
)
source = SimulatedSource(scenario, no_delay=True)
async for envelope in source.start(SourceContext(device_id="dev-01")):
    print(envelope.to_wire())   # the exact camelCase JSON that crosses the boundary

Offline-first

The device keeps working when the cloud is unreachable — structured events buffer durably and flush exactly once on reconnect. Enqueue is idempotent (keyed by the envelope's unique event_id), and the reconcile key reproduces the server's content-addressed key byte-for-byte, so a retried flush is a no-op, never a duplicate.

from pathlib import Path
from convilyn_edge.offline import DurableQueue, EventEmitter, event_key
 
queue = DurableQueue(Path("edge-events.jsonl"), key_of=event_key)
emitter = EventEmitter(sink, queue)   # sink: your HTTP / MQTT transport
 
await emitter.emit(envelope)          # delivered, or durably buffered if offline
report = await emitter.flush()        # drain on reconnect; report.clean == True

On-device inference — the client-compute keystone

When a cloud workflow routes the extractor role to the device, it pauses with a client_compute interrupt and hands the device a content-free delegation request (files by reference only). The device runs a local model over its own copy of the file and returns grounded anchors; the server re-grounds them before trusting them. Any value that isn't a verbatim substring of the local source degrades to "Not specified" on the device — an ungrounded (possibly injected) string never crosses the boundary. The consumer SDK is injected behind a narrow Protocol, so convilyn-edge itself stays dependency-free.

A worked vertical: removable Solution Packs

A removable Solution Pack composes the SPI into offline-first workflows with a single model extension point — the rest of the workflow is deterministic and runs with no model at all. The flagship reference pack is pet-monitoring (see the moat & lock-in charter, edge_solution_pack_moat_and_lockin.md, for the full integrator journey).

Behind the one ModelOperator Protocol the model node takes either placement:

• cloud — call a workflow built in the shared Builder: client.goals.run(user_workflow_id="uw_…"), which runs on the server's workflow runtime runtime (re-grounded server-side).
• edge — run a local on-device SLM via the client_compute injection point (fully offline).

The two are substitutable behind the same Protocol (LSP); placement is data, not a branch (never if provider == …).

Design principles

• The device is never a second source of truth. The 7 safety checks are server-enforced, and the server re-grounds every device value; the edge SPI inherits that contract.
• No LLM in DeterministicOperator — a sync signature makes "no I/O, no model" a type-level guarantee. Scenario rules live in a removable pack.
• One envelope, one Result, one observability convention. No parallel transports; no if provider == ….

The litmus test

Delete the entire Solution Pack. Does the remaining SDK still let you build another IoT AI workflow?

If yes, this is a general SDK — not a vertical wearing an SDK costume.