Python SDK

Deploy a model onto your own clusters from Python, and get an endpoint back.

Every method and type is generated from the platform's OpenAPI document, so the SDK says exactly what the API says. Three things are hand-written and deliberate: the client, its retry and timeout policy, and wait_until_serving.

Terminal
pip install "git+https://github.com/astropulseinc/astropulse-python"

Source: astropulseinc/astropulse-python.

Authenticate

For scripts and automation, create a service account key in the console under Service Accounts in the sidebar and export it. A key carries its organization, so nothing else is needed.

Terminal
export ASTROPULSE_API_KEY=...

Deploy a model

Python
from astropulse import Client
from astropulse.models import (
    ApplicationCreate, ApplicationRequest, Source, Model,
    ModelResources, ModelAccelerator, ModelGPU, ModelScaling, ExternalAccess,
    EndpointAccessPolicyReference,
)

ap = Client()

ap.applications.deploy_application(
    ApplicationCreate(
        api_version="platform.astropulse.io/v1",
        kind="Application",
        spec=ApplicationRequest(
            name="qwen-chat",
            profile_name="prod-inference",
            source=Source(
                type="Model",
                model=Model(
                    uri="hf://Qwen/Qwen3-8B",
                    resources=ModelResources(
                        accelerator=ModelAccelerator(gpu=ModelGPU(model="l4", count_per_replica=1)),
                    ),
                    scaling=ModelScaling(min_replicas=1, max_replicas=4),
                ),
            ),
            external_access=ExternalAccess(
                domain="acme.ai",
                access_policy_ref=EndpointAccessPolicyReference(name="partner-keys"),
            ),
        ),
    )
)

application = ap.wait_until_serving("qwen-chat")
print(application.model.external_endpoint)

Exposing a model requires access_policy_ref: the platform refuses a public endpoint that does not say who may call it. partner-keys is an access policy in your organization. Leave external_access out entirely to keep the model reachable only inside the cluster.

profile_name names an application profile, the cluster and resource tier the model runs on. Anything you leave out is the platform's default, not the SDK's guess: effective_spec on the application shows what was actually chosen.

deploy_application returns as soon as the platform accepts the work. wait_until_serving blocks until the application reports itself serving and has no deployment work still in flight, so redeploying something already running waits for the new revision rather than handing back the old one's endpoint. It raises DeploymentFailed if the deploy ends somewhere it cannot serve from, and WaitTimeout if it never settles; those are also a RuntimeError and a TimeoutError, so plain except TimeoutError works.

Gated models

Python
from astropulse.models import HuggingFaceIntegrationRequest

ap.integrations.connect_huggingface(HuggingFaceIntegrationRequest(token="hf_..."))

Call the model

The endpoint speaks the OpenAI API, so calling the model is not this SDK's job, use the openai package, pointed at the endpoint:

Python
from openai import OpenAI

chat = OpenAI(base_url=application.model.external_endpoint, api_key=endpoint_key)

chat.chat.completions.create(
    model=application.model.served_model_name,  # qwen-chat-a1b2c3
    messages=[{"role": "user", "content": "hello"}],
)

Every OpenAI request names a model, and the runtime answers to exactly one id, which the platform reports as served_model_name. It is neither the name the application was deployed under nor the model's repository id, send either of those and the runtime answers 404 The model ... does not exist. The endpoint's own models.list() reports the same value.

endpoint_key is a key from the access policy attached to the endpoint; an endpoint served without one needs no key. internal_endpoint is the same model over plain http for workloads on the same cluster, which should call it there rather than leaving and coming back.

Read what it is doing

Python
series = ap.applications.get_application_metrics_series(app_name="qwen-chat", window="15m").data
summary = series.inference_summary

print(summary.runtime_name, summary.request_count)
print(summary.request_latency_p50_seconds, summary.time_to_first_token_p50_seconds)
print(summary.requests_waiting, summary.kv_cache_utilization)

series.points carries the same figures per window, what the console charts. A window that served nothing reports its latency percentiles as absent rather than zero, so plot it as a gap. Windows are about two minutes wide, so these are trends; a brief spike can fall inside one.

Errors

Python
from astropulse import NotFoundException, OpenApiException

try:
    ap.applications.get_application("qwen-chat")
except NotFoundException as error:
    print(error.status)          # 404
    print(error.data.code)       # NotFound: branch on this, never on the message
    print(error.data.trace_id)   # quote this in a bug report
except OpenApiException as error:
    print(error)                 # every other refusal, including bad arguments

ApiException covers everything the API refused: BadRequestException, UnauthorizedException, ForbiddenException, NotFoundException, ConflictException, UnprocessableEntityException, ServiceException. OpenApiException covers those plus argument errors raised before anything is sent. Waiting raises DeploymentFailed and WaitTimeout.

Requests

Every request carries a deadline (10s to connect, 60s to read) and retries a 429 or a 5xx with backoff that honours Retry-After, but never a POST, since repeating one can create a second thing. Both are constructor arguments:

Python
ap = Client(timeout=(5, 30), retries=None)      # or timeout=None for no deadline

A Client is safe to share across threads and to keep for the life of a process. close() releases its connection pool; the client still works afterwards, it just reconnects.

Everything else the platform does

Clusters, application profiles, domains, add-ons, cloud accounts, the provider catalog, organizations, API keys, audit, the same client, the same shape, one resource per group in the API reference:

Python
for cluster in ap.clusters.list_clusters().data or []:
    print(cluster.cluster_name, cluster.state)

Authenticate​

Deploy a model​

Gated models​

Call the model​

Read what it is doing​

Errors​

Requests​

Everything else the platform does​