Connecting directly with CentralClient

If you already have an application running on the device and want it to call Central directly, use CentralClient. 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#

The code below was actually run against the screenshot stack.

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 already received. Sending the same number again is harmless.
• UploadInit(dataset_id=...) registers the assembled file as a file of that dataset.
• 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.

SdkError
├── OfflineError                  no response at all — DNS, connection refused, dropped, timeout
└── ApiError                      the server answered with a non-2xx
    ├── AuthError            401  token missing, expired or revoked
    ├── ForbiddenError       403  token valid but lacks the scope for this call
    ├── NotFoundError        404
    ├── ConflictError        409  e.g. complete before all pieces arrived
    ├── PayloadTooLargeError 413  server or proxy limit exceeded — resending will not help
    ├── UnprocessableError   422  body validation failed
    ├── RateLimitedError     429  back off for a while. Follows Retry-After
    └── ServerError          5xx  a problem on Central's side — worth retrying

ApiError has status_code, code, message, detail, error_id, method, url and retry_after. Running it with a wrong token and with an unreachable address actually gives this:

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

On the edge, telling the two apart matters. 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, and jitter is applied only upward.
• Jitter matters more than you might think. Devices that went down together come back together, and reconnecting all at once knocks the server over.
• 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
)