CentralClient 로 직접 붙이기

이미 장비에서 돌고 있는 애플리케이션이 있고 거기서 직접 중앙을 부르고 싶다면 CentralClient 를 씁니다. 에이전트·로컬 큐·SQLite 없이 httpx 와 pydantic 만 필요하므로 엑스트라 없이 설치합니다.

pip install geo-mlops-sdk

메서드#

메서드는 중앙의 장비 API(/api/v1/edge/…)와 하나씩 짝을 이룹니다. 모두 async 이고, 토큰은 X-Edge-Token 헤더로 붙습니다.

요청·응답 본문은 geo_mlops_sdk.contracts 의 Pydantic 모델입니다(edge, records, uploads, inference, commands, models). 모르는 필드는 무시하므로 중앙이 필드를 늘려도 이미 배포된 클라이언트가 깨지지 않습니다. 시그니처 전체는 Edge SDK API 레퍼런스에 있습니다.

예시: 등록 → 하트비트 → 레코드 → 업로드 → 명령#

아래 코드는 촬영용 스택에 대고 실제로 실행한 것입니다.

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. 등록. 부팅마다 불러도 된다(멱등)
        device = await client.register(location="bench")
        print("registered:", device.id, device.status)

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

        # 3. 레코드. id 가 멱등 키다. 같은 배치를 다시 보내면 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. 파일 업로드: 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  # 서버가 허용한 값으로 자른다
        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. 명령. 롱폴로 받고, 처리 결과를 반드시 ack 한다
        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)

출력(중앙에서 ping 명령을 하나 넣어 둔 상태):

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

알아 둘 점:

• 업로드 완료는 비동기입니다. upload_complete 직후 상태는 assembling 이고, 중앙이 조각을 합친 뒤 done 이 됩니다. 결과가 필요하면 upload_status 를 잠시 간격을 두고 다시 부릅니다.
• 청크 크기는 서버 값을 씁니다. upload_init 에 요청한 값이 아니라 응답의 chunk_size 로 잘라야 합니다. 서버가 앞단 프록시 한도에 맞춰 줄여서 돌려줍니다.
• 이어 올리기: 끊겼다가 다시 시작할 때는 upload_status(upload_id).received 에 이미 받은 조각 번호가 있습니다. 같은 번호를 다시 보내도 문제없습니다.
• UploadInit(dataset_id=...) 를 주면 조립이 끝난 파일이 그 데이터셋 파일로 등록됩니다.
• 토큰 교체는 연결을 새로 만들지 않고 client.set_token(new_token) 으로 합니다.

오류 계층#

상태 코드는 반환값이 아니라 예외가 됩니다. 도착하지 않은 본문으로 계속 진행하는 실수를 막기 위해서입니다.

SdkError
├── OfflineError                  응답 자체가 없음(DNS, 연결 거부, 끊김, 타임아웃)
└── ApiError                      서버가 2xx 가 아닌 답을 함
    ├── AuthError            401  토큰이 없거나 만료·폐기됨
    ├── ForbiddenError       403  토큰은 유효하지만 이 호출의 스코프가 없음
    ├── NotFoundError        404
    ├── ConflictError        409  예: 조각이 다 오지 않았는데 complete
    ├── PayloadTooLargeError 413  서버·프록시 한도 초과. 다시 보내도 안 됨
    ├── UnprocessableError   422  본문 검증 실패
    ├── RateLimitedError     429  잠시 쉬기. Retry-After 를 따름
    └── ServerError          5xx  중앙 쪽 문제. 재시도할 가치 있음

ApiError 에는 status_code, code, message, detail, error_id, method, url, retry_after 가 있습니다. 실제로 틀린 토큰과 닿지 않는 주소로 돌려 보면 이렇게 나옵니다.

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

엣지에서 둘을 가르는 것이 중요합니다. OfflineError 는 "계속 쌓아 두고 나중에", ApiError 는 "무언가를 고쳐야 한다"입니다.

재시도: RetryPolicy#

429, 5xx, 전송 실패(OfflineError)만 다시 시도합니다. 4xx 는 다시 보내도 똑같이 실패하므로 재시도하지 않습니다.

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)   # 호출 하나만 재시도 끄기

• 서버가 Retry-After 를 주면 그 값을 따릅니다. 이때 jitter(대기 시간을 무작위로 조금씩 바꾸는 것)는 대기를 늘리는 쪽으로만 적용합니다.
• jitter 는 생각보다 중요합니다. 함께 끊겼던 장비들이 동시에 돌아와 한꺼번에 다시 접속하면 서버가 버티지 못합니다.
• health() 와 poll_commands() 는 호출하는 쪽이 이미 루프이므로 재시도하지 않습니다.

그 밖의 옵션#

CentralClient(
    base_url,                 # 예: https://mlops.example.com (스킴과 호스트 필수)
    token="",
    timeout=10.0,             # 초
    retry=RetryPolicy(),
    verify=True,              # TLS 검증
    user_agent=None,          # 기본 geo-mlops-sdk/0.2.0
)