ai-platform-toolbox / clients /python /coreweave_cks.py
AnthropicallyHuggingFaces's picture
Add CoreWeave CKS client
2d158b5 verified
Raw History Blame Contribute Delete
965 Bytes
"""CoreWeave Kubernetes Service (CKS) client - api.coreweave.com."""
import os
import requests
BASE = os.getenv("COREWEAVE_API_BASE", "https://api.coreweave.com")
TOKEN = os.getenv("COREWEAVE_API_TOKEN")
CKS = f"{BASE}/v1beta1/cks/clusters"
def _h():
if not TOKEN:
raise RuntimeError("COREWEAVE_API_TOKEN not set")
return {"Authorization": f"Bearer {TOKEN}"}
def list_clusters():
return requests.get(CKS, headers=_h(), timeout=60).json()
def get_cluster(cluster_id: str):
return requests.get(f"{CKS}/{cluster_id}", headers=_h(), timeout=60).json()
def create_cluster(payload: dict):
return requests.post(CKS, headers=_h(), json=payload, timeout=120).json()
def patch_cluster(cluster_id: str, payload: dict):
return requests.patch(f"{CKS}/{cluster_id}", headers=_h(), json=payload, timeout=60).json()
def delete_cluster(cluster_id: str):
return requests.delete(f"{CKS}/{cluster_id}", headers=_h(), timeout=60).json()