"""Thin Python SDK client for the ``/v1`` REST API. Capability: ``platform-sdk``. Uses ``httpx`` directly so the client can talk to any deployed cloud-runtime process over HTTP, or to a FastAPI ``TestClient`` instance in tests. """ from __future__ import annotations from typing import Any import httpx class CloudAPIError(httpx.HTTPStatusError): def __init__(self, response: httpx.Response, detail: str) -> None: super().__init__( f"cloud API request failed with status {response.status_code}: {detail}", request=response.request, response=response, ) self.status_code = response.status_code self.detail = detail class CloudAuthorizationError(CloudAPIError): pass class CloudClient: """A minimal Python wrapper for the platform SDK's ``/v1`` routes.""" def __init__( self, base_url: str, *, http_client: httpx.Client | Any = None, api_prefix: str = "/v1", token: str | None = None, auth: httpx.Auth | None = None, ) -> None: if token is not None and auth is not None: raise ValueError("token and auth are mutually exclusive") self._base_url = base_url.rstrip("/") self._api_prefix = api_prefix.rstrip("/") self._headers = {"Authorization": f"Bearer {token}"} if token else None self._auth = auth if http_client is None: self._http = httpx.Client(base_url=self._base_url) self._owns_client = True else: self._http = http_client self._owns_client = False def close(self) -> None: if self._owns_client: self._http.close() def __enter__(self) -> "CloudClient": return self def __exit__(self, *exc: object) -> None: self.close() # ------------------------------------------------------------------- tasks def submit_task( self, *, goal: str | None = None, workflow_definition_id: str | None = None, driver_type: str | None = None, capability_tags: list[str] | None = None, ) -> dict[str, Any]: payload: dict[str, Any] = { "goal": goal, "workflow_definition_id": workflow_definition_id, } if driver_type is not None or capability_tags is not None: payload["constraints"] = { "driver_type": driver_type, "capability_tags": list(capability_tags or []), } resp = self._request("POST", "/tasks", json=payload) return resp.json() def get_task_status(self, task_id: str) -> dict[str, Any]: resp = self._request("GET", f"/tasks/{task_id}") return resp.json() # ----------------------------------------------------------------- devices def list_devices(self) -> list[dict[str, Any]]: resp = self._request("GET", "/devices") return resp.json() def list_hosts(self) -> list[dict[str, Any]]: resp = self._request("GET", "/hosts") return resp.json() # ----------------------------------------------------------------- plugins def list_plugins(self) -> list[dict[str, Any]]: resp = self._request("GET", "/plugins") return resp.json() def register_plugin( self, *, name: str, version: str, entry_point_kind: str, target: str, ) -> dict[str, Any]: payload = { "name": name, "version": version, "entry_point_kind": entry_point_kind, "target": target, } resp = self._request("POST", "/plugins", json=payload) return resp.json() # ------------------------------------------------------------------ helpers def _url(self, path: str) -> str: return f"{self._base_url}{self._api_prefix}{path}" def _request( self, method: str, path: str, *, json: dict[str, Any] | None = None, ) -> httpx.Response: response = self._http.request( method, self._url(path), json=json, headers=self._headers, auth=self._auth, ) if response.is_success: return response try: payload = response.json() except ValueError: payload = {} detail = payload.get("detail") or "request rejected" error_type = ( CloudAuthorizationError if response.status_code in {401, 403} else CloudAPIError ) raise error_type(response, str(detail))