Edge SDK API | MLOps Central Platform 매뉴얼

geo-mlops-sdk 0.3.0 의 공개 API. 패키지의 시그니처와 docstring(영어 원문)에서 자동 생성. 사용법은 Edge SDK 장 참고.

CentralClient: 중앙 서버 클라이언트

장비 토큰으로 중앙 서버의 엣지 API(/api/v1/edge)를 호출하는 비동기 클라이언트. async with 로 사용하거나 사용 후 aclose() 호출.

CentralClient(base_url: str, token: str = '', *, timeout: float = 10.0, retry: RetryPolicy = RetryPolicy(attempts=3, initial_backoff_s=0.5, max_backoff_s=30.0, multiplier=2.0, jitter=0.25), verify: bool = True, transport: Optional[httpx.AsyncBaseTransport] = None, user_agent: Optional[str] = None) -> None

HTTP client bound to one Central deployment and one device token.

set_token

def set_token(token: str) -> None

Adopt a rotated token without rebuilding the connection pool.

aclose

async def aclose() -> None

health

async def health() -> dict

Liveness probe. Never retried: the caller is the retry loop.

register

async def register(*, os: Optional[str] = None, sdk_version: Optional[str] = None, location: Optional[str] = None) -> RegisterResult

Self-register this device (IF-E1).

send_heartbeat

async def send_heartbeat(body: HeartbeatBody, *, retry: Optional[RetryPolicy] = None) -> HeartbeatResult

Report resources and runtime status (IF-E2).

list_models

async def list_models() -> ModelListResult

Models this device's tenant has registered (IF-E6).

model_versions

async def model_versions(name: str) -> ModelVersionsResult

Version history of one model (IF-E6).

resolve_container

async def resolve_container(model: str, version: str) -> ContainerRef

Registry reference for a model version's serving image (IF-E5).

send_records

async def send_records(records: Sequence[Record]) -> BatchResult

Hand over a telemetry batch (IF-E3).

send_inference

async def send_inference(records: Sequence[InferenceRecord]) -> BatchResult

Hand over an inference-result batch (IF-E4). Same idempotency rule.

upload_init

async def upload_init(request: UploadInit) -> UploadCreated

Open a resumable transfer.

upload_chunk

async def upload_chunk(upload_id: str, index: int, data: bytes) -> ChunkAccepted

Send one chunk. Re-sending an index already stored is a no-op.

upload_status

async def upload_status(upload_id: str) -> UploadStatus

Where a restarted transfer should pick up.

upload_complete

async def upload_complete(upload_id: str) -> UploadStatus

Close the transfer. Assembly happens server-side, so poll the status.

poll_commands

async def poll_commands(wait_s: float = 25.0) -> CommandList

Long poll for commands.

ack_command

async def ack_command(command_id: str, ack: CommandAck) -> CommandAckResult

Report what happened to one command.

fetch_policy

async def fetch_policy() -> DevicePolicy

Retrieve the policy in force for this device.

download_model

async def download_model(name: str, version: str, dest: Union[str, Path], *, progress: Optional[ProgressCallback] = None, retry: Optional[RetryPolicy] = None) -> Path

Stream a model version's artifact to dest (IF-E6).

오류

모든 오류는 SdkError 를 상속. 재시도 대상은 429·5xx·전송 오류뿐.

예외	부모	설명
`SdkError`	`Exception`	Base class for every error raised by this package.
`OfflineError`	`SdkError`	The request never got an answer (DNS, refused, reset, timeout).
`ApiError`	`SdkError`	Central answered with a non-2xx status.
`AuthError`	`ApiError`	401: the device token is unknown, expired or revoked.
`ForbiddenError`	`ApiError`	403: the token is valid but lacks the scope for this call.
`NotFoundError`	`ApiError`	404.
`ConflictError`	`ApiError`	409: e.g. completing an upload whose chunks are not all in.
`PayloadTooLargeError`	`ApiError`	413: the body exceeds a server or proxy ceiling. Retrying cannot help.
`UnprocessableError`	`ApiError`	422: the body did not validate.
`RateLimitedError`	`ApiError`	429: back off, honouring `Retry-After` when present.
`ServerError`	`ApiError`	5xx: Central's problem, and worth retrying.

재시도 정책

RetryPolicy(attempts: int = 3, initial_backoff_s: float = 0.5, max_backoff_s: float = 30.0, multiplier: float = 2.0, jitter: float = 0.25) -> None

Exponential backoff with jitter.

계약 모델 (요청·응답)

중앙 서버와 주고받는 Pydantic 모델. 알 수 없는 필드는 무시하므로 서버에 필드가 추가되어도 호환성 유지.

Record

One structured observation.

필드	형식	기본값
`id`	`str`	(필수)
`kind`	`str`	(필수)
`ts`	`datetime`	(필수)
`priority`	`int`	`50`
`payload`	`dict[str, Any]`	`{}`

RecordBatch

Request body of POST /api/v1/edge/telemetry.

필드	형식	기본값	설명
`records`	`list[Record]`	`[]`

RejectedRecord

One record Central refused, with the reason it refused it.

필드	형식	기본값	설명
`id`	`str`	(필수)
`reason`	`str`	(필수)

BatchResult

Response to a telemetry or inference batch.

필드	형식	기본값
`accepted`	`int`	`0`
`duplicates`	`int`	`0`
`rejected`	`list[RejectedRecord]`	`[]`

UploadInit

Request body of POST /api/v1/edge/uploads.

필드	형식	기본값
`filename`	`str`	(필수)
`size`	`int`	(필수)
`sha256`	`str`	(필수)
`chunk_size`	`int`	(필수)
`kind`	`str`	`blob`
`dataset_id`	`str \| None`	`None`
`meta`	`dict[str, Any]`	`{}`

UploadCreated

Response to upload init.

필드	형식	기본값
`upload_id`	`str`	(필수)
`chunk_size`	`int`	(필수)
`received`	`list[int]`	`[]`

ChunkAccepted

Response to a chunk PUT: the indices the server now holds.

필드	형식	기본값	설명
`received`	`list[int]`	`[]`

UploadStatus

Response to GET /api/v1/edge/uploads/{id}: the resume point.

필드	형식	기본값
`upload_id`	`str`	(필수)
`state`	`UploadState`	`uploading`
`received`	`list[int]`	`[]`
`size`	`int`	`0`
`chunk_size`	`int`	`0`
`storage_uri`	`str \| None`	`None`
`error`	`str \| None`	`None`

Detection

One detected instance.

필드	형식	기본값
`cls`	`int`	(필수)
`name`	`str`	`""`
`conf`	`float`	`0.0`
`bbox`	`list[float]`	`[]`
`polygon`	`list[list[float]]`	`[]`

InferenceOutput

Result of one predict call.

필드	형식	기본값
`task`	`str`	`""`
`width`	`int`	`0`
`height`	`int`	`0`
`detections`	`list[Detection]`	`[]`

ModelRef

Registry coordinates of the model that produced a result.

필드	형식	기본값	설명
`name`	`str`	(필수)
`version`	`str`	(필수)

InferenceRecord

One inference result queued for Central (IF-E4).

필드	형식	기본값
`id`	`str`	(필수)
`ts`	`datetime`	(필수)
`model`	`ModelRef`	(필수)
`input_ref`	`str \| None`	`None`
`output`	`dict[str, Any]`	`{}`
`latency_ms`	`float`	`0.0`
`priority`	`int`	`50`

InferenceBatch

Request body of POST /api/v1/edge/inference.

필드	형식	기본값	설명
`records`	`list[InferenceRecord]`	`[]`

RegisterRequest

Request body of POST /api/v1/edge/register (IF-E1).

필드	형식	기본값
`os`	`str`	(필수)
`sdk_version`	`str`	(필수)
`location`	`str \| None`	`None`

RegisterResult

Response to registration.

필드	형식	기본값	설명
`id`	`str`	(필수)
`status`	`str`	`ACTIVE`

BacklogStatus

What is waiting in the local queue.

필드	형식	기본값
`count`	`int`	`0`
`bytes`	`int`	`0`
`oldest_ts`	`datetime \| None`	`None`
`evicted_24h`	`int`	`0`
`by_kind`	`dict[str, int]`	`{}`

SyncStatus

Uploader state as reported to the fleet.

필드	형식	기본값
`state`	`SyncState`	`idle`
`last_ok_at`	`datetime \| None`	`None`
`last_error`	`str \| None`	`None`
`rate_bps`	`float`	`0.0`
`in_flight`	`int`	`0`
`denied`	`str \| None`	`None`

ModelStatus

One model present in the local cache.

필드	형식	기본값
`name`	`str`	(필수)
`version`	`str`	(필수)
`framework`	`str`	`""`
`active`	`bool`	`False`

CollectorStatus

One configured collector.

필드	형식	기본값
`name`	`str`	(필수)
`type`	`str`	`""`
`state`	`str`	`stopped`
`last_ts`	`datetime \| None`	`None`
`error`	`str \| None`	`None`

ContainerStatus

A container the edge reports running (populated by the host app).

필드	형식	기본값
`image`	`str`	(필수)
`version`	`str \| None`	`None`
`health`	`str \| None`	`None`

HeartbeatPayload

Free-form half of the heartbeat, given a shape by this SDK.

필드	형식	기본값
`agent_version`	`str`	`""`
`os`	`str`	`""`
`uptime_s`	`float`	`0.0`
`policy_revision`	`int`	`0`
`backlog`	`BacklogStatus`
`sync`	`SyncStatus`
`models`	`list[ModelStatus]`	`[]`
`collectors`	`list[CollectorStatus]`	`[]`
`containers`	`list[ContainerStatus] \| None`	`None`

HeartbeatBody

Request body of POST /api/v1/edge/heartbeat (IF-E2).

필드	형식	기본값
`cpu`	`float`	`0.0`
`gpu`	`float \| None`	`None`
`mem`	`float`	`0.0`
`disk`	`float`	`0.0`
`payload`	`HeartbeatPayload`

HeartbeatResult

Response to a heartbeat.

필드	형식	기본값	설명
`ok`	`bool`	`True`
`policy_revision`	`int`	`0`

RetentionPolicy

Local storage ceiling. Whichever bound trips first wins.

필드	형식	기본값
`max_bytes`	`int`	`53687091200`
`max_age_days`	`int`	`30`
`free_disk_min_bytes`	`int`	`5368709120`

SyncPolicy

How aggressively the uploader may work.

필드	형식	기본값
`batch_size`	`int`	`500`
`chunk_bytes`	`int`	`33554432`
`max_bytes_per_s`	`int`	`0`
`cpu_pause_percent`	`float`	`85.0`
`windows`	`list[str]`	`[]`
`concurrency`	`int`	`1`
`urgent_priority`	`int`	`90`

DevicePolicy

Response to GET /api/v1/edge/config.

필드	형식	기본값
`revision`	`int`	`0`
`heartbeat_interval_s`	`float`	`30.0`
`commands_poll_s`	`float`	`25.0`
`retention`	`RetentionPolicy`
`sync`	`SyncPolicy`

Command

One queued command.

필드	형식	기본값
`id`	`str`	(필수)
`type`	`str`	(필수)
`args`	`dict[str, Any]`	`{}`
`created_at`	`datetime \| None`	`None`

CommandList

Response to GET /api/v1/edge/commands (empty when the wait elapsed).

필드	형식	기본값	설명
`items`	`list[Command]`	`[]`

CommandAck

Request body of POST /api/v1/edge/commands/{id}:ack.

필드	형식	기본값	설명
`status`	`AckStatus`	`ok`
`result`	`dict[str, Any] \| None`	`None`

CommandAckResult

Response to an ack.

필드	형식	기본값	설명
`id`	`str`	(필수)
`state`	`str`	`""`

ModelInfo

One registered model.

필드	형식	기본값
`name`	`str`	(필수)
`stages`	`dict[str, str]`	`{}`
`tags`	`dict[str, str]`	`{}`

ModelListResult

Response to GET /api/v1/edge/models.

필드	형식	기본값	설명
`items`	`list[ModelInfo]`	`[]`
`available`	`bool`	`False`

ModelVersionInfo

One version of a model.

필드	형식	기본값
`version`	`str`	(필수)
`stage`	`str`	`""`
`status`	`str`	`""`
`run_id`	`str`	`""`
`creation_timestamp`	`int`	`0`

ModelVersionsResult

Response to GET /api/v1/edge/models/{name}/versions.

필드	형식	기본값	설명
`name`	`str`	(필수)
`versions`	`list[ModelVersionInfo]`	`[]`

ContainerRef

Response to GET /api/v1/edge/containers/pull (IF-E5).

필드	형식	기본값
`image`	`str`	(필수)
`model`	`str`	`""`
`version`	`str`	`""`

EdgeSettings: 에이전트 설정

설정 파일(YAML)의 구조. 우선순위는 환경 변수 > 설정 파일 > 기본값이고, 환경 변수는 GEO_EDGE_ 접두사에 계층을 __ 로 연결. 예: central.token 은 GEO_EDGE_CENTRAL__TOKEN.

EdgeSettings

필드	형식	기본값	환경 변수
`central`	`CentralSettings`		`GEO_EDGE_CENTRAL`
`device`	`DeviceSettings`		`GEO_EDGE_DEVICE`
`data_dir`	`Path`	`PosixPath('/var/lib/geo-mlops-edge')`	`GEO_EDGE_DATA_DIR`
`disk_path`	`str`	`""`	`GEO_EDGE_DISK_PATH`
`retention`	`SizedRetentionPolicy`		`GEO_EDGE_RETENTION`
`sync`	`SizedSyncPolicy`		`GEO_EDGE_SYNC`
`link`	`LinkSettings`		`GEO_EDGE_LINK`
`heartbeat_interval_s`	`float`	`30.0`	`GEO_EDGE_HEARTBEAT_INTERVAL_S`
`commands_poll_s`	`float`	`25.0`	`GEO_EDGE_COMMANDS_POLL_S`
`api`	`ApiSettings`		`GEO_EDGE_API`
`collectors`	`list[CollectorSettings]`	`[]`	`GEO_EDGE_COLLECTORS`
`models`	`ModelSettings`		`GEO_EDGE_MODELS`
`policy_source`	`str`	`central`	`GEO_EDGE_POLICY_SOURCE`
`log_level`	`str`	`INFO`	`GEO_EDGE_LOG_LEVEL`

central (CentralSettings)

How to reach the platform.

필드	형식	기본값	환경 변수
`base_url`	`str`	`""`	`GEO_EDGE_CENTRAL__BASE_URL`
`token`	`str`	`""`	`GEO_EDGE_CENTRAL__TOKEN`
`timeout_s`	`float`	`10.0`	`GEO_EDGE_CENTRAL__TIMEOUT_S`
`verify_tls`	`bool`	`True`	`GEO_EDGE_CENTRAL__VERIFY_TLS`

device (DeviceSettings)

Identity overrides. Empty id means "use the hostname".

필드	형식	기본값	설명	환경 변수
`id`	`str`	`""`		`GEO_EDGE_DEVICE__ID`
`location`	`Optional[str]`	`None`		`GEO_EDGE_DEVICE__LOCATION`

retention (SizedRetentionPolicy)

Retention with human-readable sizes accepted from YAML.

필드	형식	기본값	환경 변수
`max_bytes`	`int`	`53687091200`	`GEO_EDGE_RETENTION__MAX_BYTES`
`max_age_days`	`int`	`30`	`GEO_EDGE_RETENTION__MAX_AGE_DAYS`
`free_disk_min_bytes`	`int`	`5368709120`	`GEO_EDGE_RETENTION__FREE_DISK_MIN_BYTES`

sync (SizedSyncPolicy)

Sync policy with human-readable sizes accepted from YAML.

필드	형식	기본값	환경 변수
`batch_size`	`int`	`500`	`GEO_EDGE_SYNC__BATCH_SIZE`
`chunk_bytes`	`int`	`33554432`	`GEO_EDGE_SYNC__CHUNK_BYTES`
`max_bytes_per_s`	`int`	`0`	`GEO_EDGE_SYNC__MAX_BYTES_PER_S`
`cpu_pause_percent`	`float`	`85.0`	`GEO_EDGE_SYNC__CPU_PAUSE_PERCENT`
`windows`	`list[str]`	`[]`	`GEO_EDGE_SYNC__WINDOWS`
`concurrency`	`int`	`1`	`GEO_EDGE_SYNC__CONCURRENCY`
`urgent_priority`	`int`	`90`	`GEO_EDGE_SYNC__URGENT_PRIORITY`

link (LinkSettings)

Connectivity probing.

필드	형식	기본값	환경 변수
`probe_interval_s`	`float`	`5.0`	`GEO_EDGE_LINK__PROBE_INTERVAL_S`
`backoff_max_s`	`float`	`60.0`	`GEO_EDGE_LINK__BACKOFF_MAX_S`
`online_after_ok`	`int`	`2`	`GEO_EDGE_LINK__ONLINE_AFTER_OK`
`offline_after_fail`	`int`	`3`	`GEO_EDGE_LINK__OFFLINE_AFTER_FAIL`

api (ApiSettings)

Local HTTP surface for the on-site UI.

필드	형식	기본값	환경 변수
`enabled`	`bool`	`True`	`GEO_EDGE_API__ENABLED`
`host`	`str`	`0.0.0.0`	`GEO_EDGE_API__HOST`
`port`	`int`	`8600`	`GEO_EDGE_API__PORT`
`token`	`str`	`""`	`GEO_EDGE_API__TOKEN`
`cors_origins`	`list[str]`	`['*']`	`GEO_EDGE_API__CORS_ORIGINS`
`max_body_bytes`	`int`	`2147483648`	`GEO_EDGE_API__MAX_BODY_BYTES`

collectors (CollectorSettings)

One configured collector. Type-specific keys stay in options.

필드	형식	기본값	환경 변수
`type`	`str`	(필수)	`GEO_EDGE_COLLECTORS__TYPE`
`name`	`str`	`""`	`GEO_EDGE_COLLECTORS__NAME`
`enabled`	`bool`	`True`	`GEO_EDGE_COLLECTORS__ENABLED`
`priority`	`int`	`50`	`GEO_EDGE_COLLECTORS__PRIORITY`
`options`	`dict[str, Any]`	`{}`	`GEO_EDGE_COLLECTORS__OPTIONS`

models (ModelSettings)

Local model cache behaviour.

필드	형식	기본값	설명	환경 변수
`auto_activate`	`str`	`Production`		`GEO_EDGE_MODELS__AUTO_ACTIVATE`
`keep_versions`	`int`	`2`		`GEO_EDGE_MODELS__KEEP_VERSIONS`

확장 지점

직접 만든 수집기·추론 러너 등록용 프로토콜과 함수.

Sink

What a collector is handed to publish through.

async def record(kind: str, payload: dict, *, priority: int = 50, ts: Optional[datetime] = None, meta: Optional[dict] = None, record_id: str = '') -> Any
async def blob(kind: str, source: Union[str, Path, bytes], *, filename: str = '', priority: int = 50, ts: Optional[datetime] = None, meta: Optional[dict] = None, move: bool = False) -> Any

Collector

A source of data attached to this edge.

async def start(sink: Sink) -> None
async def stop() -> None
def status() -> CollectorStatus

register_collector

def register_collector(type_: str, factory: CollectorFactory) -> None

Make type_ usable in configuration.

build_collector

def build_collector(type_: str, name: str, *, priority: int = 50, options: Optional[dict] = None) -> Collector

Instantiate one configured collector.

Runner

Loads one model and answers predictions for it.

def load(model: LocalModel) -> None
def predict(image: bytes, **params) -> InferenceOutput
def close() -> None

register_runner

def register_runner(framework: str, factory: Callable[[], Runner]) -> None

Teach the SDK about a framework it does not ship support for.

CentralClient: 중앙 서버 클라이언트#

set_token#

aclose#

health#

register#

send_heartbeat#

list_models#

model_versions#

resolve_container#

send_records#

send_inference#

upload_init#

upload_chunk#

upload_status#

upload_complete#

poll_commands#

ack_command#

fetch_policy#

download_model#

오류#

재시도 정책#

계약 모델 (요청·응답)#

Record#

RecordBatch#

RejectedRecord#

BatchResult#

UploadInit#

UploadCreated#

ChunkAccepted#

UploadStatus#

Detection#

InferenceOutput#

ModelRef#

InferenceRecord#

InferenceBatch#

RegisterRequest#

RegisterResult#

BacklogStatus#

SyncStatus#

ModelStatus#

CollectorStatus#

ContainerStatus#

HeartbeatPayload#

HeartbeatBody#

HeartbeatResult#

RetentionPolicy#

SyncPolicy#

DevicePolicy#

Command#

CommandList#

CommandAck#

CommandAckResult#

ModelInfo#

ModelListResult#

ModelVersionInfo#

ModelVersionsResult#

ContainerRef#

EdgeSettings: 에이전트 설정#

EdgeSettings#

central (CentralSettings)#

device (DeviceSettings)#

retention (SizedRetentionPolicy)#

sync (SizedSyncPolicy)#

link (LinkSettings)#

api (ApiSettings)#

collectors (CollectorSettings)#

models (ModelSettings)#

확장 지점#

Sink#

Collector#

register_collector#

build_collector#

Runner#

register_runner#