Connecting directly with CentralClient

Use CentralClient when an application already running on the device calls Central directly. It needs only httpx and pydantic — no agent, local queue or SQLite — so install it without extras.

pip install geo-mlops-sdk

Methods#

Each method pairs with one call of Central's device API (/api/v1/edge/…). They are all async, and the token is attached in the X-Edge-Token header.

Request and response bodies are Pydantic models in geo_mlops_sdk.contracts (edge, records, uploads, inference, commands, models). Unknown fields are ignored, so clients already deployed do not break when Central adds fields. The full signatures are in the Edge SDK API reference.

Example: register → heartbeat → records → upload → commands#

import asyncio
import hashlib
import os
import uuid
from datetime import datetime, timezone
from pathlib import Path

from geo_mlops_sdk.client import AuthError, CentralClient, OfflineError, RetryPolicy
from geo_mlops_sdk.contracts.commands import AckStatus, CommandAck
from geo_mlops_sdk.contracts.edge import HeartbeatBody
from geo_mlops_sdk.contracts.records import Record
from geo_mlops_sdk.contracts.uploads import UploadInit

BASE_URL = os.environ["GEO_EDGE_CENTRAL__BASE_URL"]
TOKEN = os.environ["GEO_EDGE_CENTRAL__TOKEN"]


async def main() -> None:
    retry = RetryPolicy(attempts=5, jitter=0.5)
    async with CentralClient(BASE_URL, TOKEN, retry=retry) as client:
        # 1. Register — safe to call on every boot (idempotent)
        device = await client.register(location="bench")
        print("registered:", device.id, device.status)

        # 2. Heartbeat
        beat = await client.send_heartbeat(HeartbeatBody(cpu=12.5, mem=40.0, disk=55.0))
        print("heartbeat ok, policy_revision =", beat.policy_revision)

        # 3. Records — the id is the idempotency key. Resending the same batch counts as duplicates
        records = [
            Record(id=str(uuid.uuid4()), kind="sensor",
                   ts=datetime.now(timezone.utc), payload={"temp": 21.5})
        ]
        first = await client.send_records(records)
        again = await client.send_records(records)
        print("records:", first.accepted, "accepted /", again.duplicates, "duplicate")

        # 4. File upload — init → chunk → complete → status
        data = Path("frame-0001.jpg").read_bytes()
        created = await client.upload_init(UploadInit(
            filename="frame-0001.jpg", size=len(data),
            sha256=hashlib.sha256(data).hexdigest(), chunk_size=8 * 1024 * 1024))
        size = created.chunk_size  # split by the value the server allowed
        for index in range(0, max(1, -(-len(data) // size))):
            await client.upload_chunk(created.upload_id, index,
                                      data[index * size:(index + 1) * size])
        await client.upload_complete(created.upload_id)
        status = await client.upload_status(created.upload_id)
        print("upload:", status.state.value)

        # 5. Commands — receive by long poll, and always ack the outcome
        listed = await client.poll_commands(wait_s=5)
        for command in listed.items:
            print("command:", command.type, command.args)
            await client.ack_command(command.id, CommandAck(status=AckStatus.OK,
                                                            result={"handled": True}))
        print("commands handled:", len(listed.items))


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except AuthError as exc:
        print("token rejected:", exc.status_code, exc.message)
    except OfflineError as exc:
        print("central unreachable:", exc)

Output (with one ping command queued on Central beforehand):

registered: edge-bench-01 ACTIVE
heartbeat ok, policy_revision = 0
records: 1 accepted / 1 duplicate
upload: assembling
command: ping {}
commands handled: 1

Things to know:

• Upload completion is asynchronous. Right after upload_complete the state is assembling, and it becomes done after Central joins the pieces. If you need the outcome, call upload_status again after a short wait.
• Use the server's chunk size. Split by the chunk_size in the response, not by the value you requested in upload_init. The server shrinks it to fit the front proxy's limit.
• Resuming: when you restart after an interruption, upload_status(upload_id).received holds the numbers of the pieces the server has already received. Sending the same number again is harmless.
• UploadInit(dataset_id=...) registers the assembled file as a file of that dataset. Without it the file is kept under Collected data.
• Token rotation is done with client.set_token(new_token), without creating a new connection.

Error hierarchy#

Status codes become exceptions, not return values. This prevents carrying on with a body that never arrived.

ApiError has status_code, code, message, detail, error_id, method, url and retry_after. A wrong token and an unreachable address give this:

token rejected: 401 invalid or expired edge token
central unreachable: POST /api/v1/edge/register: All connection attempts failed

On the edge, you must tell the two apart. OfflineError means "keep buffering and try later"; ApiError means "something needs fixing".

Retries: RetryPolicy#

Only 429, 5xx and transport failures (OfflineError) are retried. A 4xx fails the same way when resent, so it is not retried.

from geo_mlops_sdk.client import NO_RETRY, CentralClient, RetryPolicy

client = CentralClient(url, token, retry=RetryPolicy(attempts=5, jitter=0.5))
await client.send_heartbeat(body, retry=NO_RETRY)   # turn off retries for a single call

• If the server sends Retry-After, that value is followed, capped at max_backoff_s, and jitter is applied only upward.
• Jitter keeps devices that went down together from reconnecting all at once and overloading the server.
• health() and poll_commands() are not retried, because the caller is already a loop.

Other options#

CentralClient(
    base_url,                 # https://mlops.example.com — scheme and host required
    token="",
    timeout=10.0,             # seconds
    retry=RetryPolicy(),
    verify=True,              # TLS verification
    user_agent=None,          # default geo-mlops-sdk/0.2.0
)