Skip to content

Provider plugin SDK cookbook

Astrolift's AWS, GCP, Azure, and Kubernetes-native providers are Python plugin packages discovered through the astrolift.providers entry-point group. A plugin supplies typed driver implementations and a ProviderPlugin manifest.

Distribution status

The provider SDK and official plugins now live in astrolift-app/backend/providers; the former standalone repository is archived. The Astrolift release workflow does not currently publish a stable third-party SDK to PyPI or provide a supported marketplace/install API.

The supported contribution path is an in-tree provider change. An operator can experiment with an out-of-tree wheel only by baking the compatible SDK, plugin, and cloud dependencies into a custom control-plane image, then restarting and bootstrapping its provider catalogue. Treat that as a custom Astrolift distribution that the operator owns and tests.

Repository layout

backend/providers/
├── _sdk/                  # Protocols, data classes, catalogue, parity checks
├── aws/
│   └── plugin.py          # exports PLUGIN
├── gcp/
├── azure/
├── k8s_native/
├── tests/
├── pyproject.toml
└── Makefile

The SDK imports no provider implementation. Provider packages import _sdk. The Django control plane loads installed entry points once during application startup and adapts each manifest into its in-process driver registry.

There are two distinct catalogues:

  • the in-process registry holds implementation classes loaded from entry points; and
  • database ProviderPlugin rows hold operator-visible configuration.

Installing a wheel does not create the database row during apps.ready(). After a compatible package is present, run the catalogue bootstrap management command as part of deployment.

Minimal plugin package

my-astrolift-provider/
├── pyproject.toml
├── my_provider/
│   ├── __init__.py
│   ├── plugin.py
│   ├── cluster.py
│   └── secrets.py
└── tests/
    └── test_plugin.py

A syntactically valid minimal pyproject.toml is:

[project]
name = "my-astrolift-provider"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
  "my-cloud-sdk>=1,<2",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["my_provider"]

[project.entry-points."astrolift.providers"]
my_provider = "my_provider.plugin:PLUGIN"

The example deliberately does not claim an astrolift-providers PyPI dependency. The custom control-plane build must install the exact in-tree SDK version it was tested against before installing this wheel. If your organization publishes an internal SDK wheel, add its exact compatible version to dependencies.

The entry-point key is the runtime plugin ID. Keep it identical to PLUGIN.id. The current loader warns on a mismatch and uses the entry-point key.

Plugin manifest

from _sdk.base import ProviderPlugin

from my_provider.cluster import MyClusterDriver
from my_provider.secrets import MySecretsBackend

PLUGIN = ProviderPlugin(
    id="my_provider",
    display_name="My Provider",
    drivers={
        "cluster": MyClusterDriver,
        "secrets": MySecretsBackend,
    },
    managed_service_drivers={},
    config_schema={
        "type": "object",
        "properties": {
            "api_endpoint": {"type": "string"},
            "region": {"type": "string"},
        },
        "required": ["api_endpoint", "region"],
    },
)

Canonical driver roles include cluster, ingress, dns, tls, identity, registry, secrets, notification, object_store, log_stream, and metrics. Register only roles that are implemented. Cluster binding fails when required roles are absent and no supported cross-plugin composition supplies them.

Implementing a protocol

Implement the protocol from the SDK shipped in the target Astrolift commit. Do not copy a method list from an older guide: these protocols expand as the control plane gains operational capabilities.

For example, the current ClusterDriver begins with these positional signatures:

def apply_manifests(
    self,
    cluster: str,
    namespace: str,
    manifests: list[dict],
    *,
    dry_run: bool = False,
) -> ApplyResult: ...

def delete_manifests(
    self,
    cluster: str,
    namespace: str,
    manifests: list[dict],
) -> DeleteResult: ...

def get_namespace(self, cluster: str, name: str) -> NamespaceState | None: ...

def ensure_namespace(
    self,
    cluster: str,
    name: str,
    labels: dict[str, str],
    annotations: dict[str, str],
) -> Namespace: ...

It also covers namespace deletion, workload status/rollout, exec, port-forwarding, pods, logs, metrics, events, resource inventory, drift, certificate discovery, managed-model identity, and other operations. A class that implements only apply/delete is not a current ClusterDriver.

Start from the closest official driver and replace its cloud client boundary. Preserve these guarantees:

  • mutation methods are idempotent;
  • status values are normalized to SDK data classes;
  • authentication and provider errors remain typed/actionable;
  • tenant cluster configuration, namespace, and selectors are never ignored;
  • dry-run performs no mutation; and
  • a production client factory never returns a stub whose public methods raise NotImplementedError.

The parity harness checks method presence, but it cannot prove semantics, credentials, tenancy, or idempotency. Those require tests against the provider.

Secrets backend

The current secrets protocol includes explicit reveal metadata:

from __future__ import annotations


class MySecretsBackend:
    provider_id = "my_provider"
    supports_value_reveal = True
    value_reveal_limitation: str | None = None

    def get(self, path: str) -> dict[str, str] | None:
        raise NotImplementedError

    def upsert(self, path: str, kvs: dict[str, str]) -> None:
        raise NotImplementedError

    def delete(self, path: str) -> None:
        raise NotImplementedError

    def list(self, prefix: str) -> list[str]:
        raise NotImplementedError

    def list_keys(self, path: str) -> list[str]:
        payload = self.get(path)
        return sorted(payload) if payload else []

    def ensure_initialized(self) -> dict | None:
        raise NotImplementedError

get() is the value-reading operation. Set supports_value_reveal=False and a plain-language value_reveal_limitation for write-only stores; the dashboard must not offer reveal merely because an older structural fallback exists. list_keys() returns names only and backs Secret Bundle known-key refresh.

If the store needs no initialization, ensure_initialized() should raise NotImplementedError; the cluster-bring workflow treats that as a successful no-op. A backend that provisions a KMS key or secrets engine may return relevant non-secret metadata.

Never log secret payloads, provider responses containing values, or exception objects that embed request bodies.

Managed-service driver

A managed-service driver is registered by (kind, variant):

PLUGIN = ProviderPlugin(
    id="my_provider",
    display_name="My Provider",
    managed_service_drivers={
        ("redis", "my_cache"): MyCacheDriver,
    },
)

The current ManagedServiceDriver implements this lifecycle:

def provision(self, spec: ProvisionSpec) -> ProvisionResult: ...
def update(self, spec: UpdateSpec) -> UpdateResult: ...
def deprovision(
    self,
    spec: DeprovisionSpec,
    *,
    delete_data: bool = False,
    force_destroy: bool = False,
) -> DeprovisionResult: ...
def status(self, handle: ServiceHandle) -> ServiceStatus: ...
def binding(
    self,
    handle: ServiceHandle,
    config: dict | None = None,
) -> Binding: ...
def snapshot(self, handle: ServiceHandle) -> SnapshotHandle: ...
def restore(
    self,
    snapshot: SnapshotHandle,
    target: ProvisionSpec,
) -> ProvisionResult: ...
def config_schema(self) -> dict: ...
def binding_schema(self) -> BindingSchema: ...
def editable_fields(self) -> list[str]: ...

All lifecycle methods are idempotent. delete_data controls destruction of persistent state; force_destroy controls provider safety guards. Do not merge those choices. The safe default preserves data and respects deletion protection.

Binding may emit literal environment values, secret references, identity grants, and volume mounts. It must emit every required environment name in the kind catalogue. Keep credentials as ValueRef(secret_ref=...), not literals.

Adding a managed-service kind or variant

For a new variant of an existing kind:

  1. implement the driver;
  2. add (kind, variant) to PLUGIN.managed_service_drivers;
  3. add the provider/kind/variant to _sdk/availability.py; and
  4. add protocol, binding, lifecycle, and real-provider tests.

For a new kind, also add its contract to _sdk/managed_service_kinds.py, including required/optional binding names and snapshot/replication semantics. Both directions of the matrix check must pass: every manifest capability must be in the matrix, and every matrix capability must have a loaded implementation.

The kind catalogue currently contains similar but distinct identifiers such as kv_store, key_value, and vector_index/vector_db across variants and historical code. Check the actual target release before choosing a name; adding a near-duplicate identifier increases API compatibility debt.

Discovery and catalogue bootstrap

Inside the exact control-plane environment:

python - <<'PY'
from importlib.metadata import entry_points

for entry_point in entry_points(group="astrolift.providers"):
    plugin = entry_point.load()
    print(entry_point.name, plugin.id, sorted(plugin.drivers))
PY

Then restart the Django processes so startup discovery runs, and reconcile the database catalogue:

python manage.py bootstrap_provider_plugins

Discovery failures are logged and skipped so one broken optional plugin does not take down the control plane. That resilience can hide a packaging problem; verify the plugin appears in the in-process registry and the operator catalogue before registering a cluster.

Test gates

From astrolift-app/backend/providers:

ruff check .
ruff format --check .
mypy _sdk/ aws/ gcp/ azure/ k8s_native/
pytest -x

For an added in-tree package, include it in the mypy command and in pyproject.toml's wheel packages and entry points. Run the full control-plane suite too, because loader, database catalogue, cluster binding, managed-service workflows, secret CRUD/reveal, and UI schema live outside the provider tree.

At minimum, test:

  • entry-point discovery and ID agreement;
  • manifest/availability matrix parity in both directions;
  • every protocol method and signature;
  • idempotent retry of provision/apply/delete;
  • safe deprovision defaults and all destructive-option combinations;
  • required binding names and absence of plaintext secret leakage;
  • invalid, expired, and cross-tenant credentials;
  • provider throttling and transient errors; and
  • one real smoke operation through the default (non-fake) client factory.

Do not publish an external wheel as “Astrolift compatible” solely because it loads as an entry point. Pin it to an Astrolift release, ship it in a custom image, and maintain a compatibility matrix until a versioned public SDK and installation lifecycle are released.