Plugin SDK Cookbook

Astrolift ships built-in provider plugins for AWS, GCP, Azure, and vanilla Kubernetes. This guide shows how to write and publish a third-party plugin that adds new cloud targets or new managed-service kinds.

What a plugin is

A plugin is a Python package that:

• Implements one or more driver protocols from astrolift-providers/_sdk/.
• Declares a ProviderPlugin manifest that maps driver roles to your implementations.
• Registers the manifest as a Python entry point under the astrolift.providers group.

The control plane loads every installed entry point at startup and makes the plugins available for cluster binding in the UI.

Project layout

my-astrolift-plugin/
├── pyproject.toml
├── my_plugin/
│   ├── __init__.py
│   ├── plugin.py          # ProviderPlugin manifest
│   ├── cluster.py         # ClusterDriver implementation
│   ├── managed/
│   │   └── redis_mine.py  # ManagedServiceDriver implementation
│   └── secrets.py         # SecretsBackend implementation
└── tests/
    └── test_plugin.py

pyproject.toml

[project]
name = "my-astrolift-plugin"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    # Pin the same SDK version range you tested against.
    # astrolift-providers ships the _sdk package.
    "astrolift-providers>=0.2,<1",
    "my-cloud-sdk>=1.0",
]

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

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

# This is the entry point that registers your plugin.
[project.entry-points."astrolift.providers"]
my_plugin = "my_plugin.plugin:PLUGIN"

The entry point key (my_plugin) becomes the plugin's id. It must be unique across all installed plugins. Use your package name or a namespaced identifier.

The ProviderPlugin manifest

# my_plugin/plugin.py
from _sdk.base import ProviderPlugin

from my_plugin.cluster import MyClusterDriver
from my_plugin.managed.redis_mine import MyRedisDriver
from my_plugin.secrets import MySecretsBackend

PLUGIN = ProviderPlugin(
    id="my_plugin",
    display_name="My Cloud Platform",
    drivers={
        "cluster": MyClusterDriver,
        "secrets": MySecretsBackend,
        # Other roles you implement: "ingress", "dns", "tls",
        # "identity", "registry", "notification".
        # Roles you omit are simply not offered by this plugin.
    },
    managed_service_drivers={
        # Maps (kind, variant) to a ManagedServiceDriver class.
        ("redis", "my_redis"): MyRedisDriver,
    },
    config_schema={
        "type": "object",
        "properties": {
            "api_endpoint": {"type": "string"},
            "region": {"type": "string"},
        },
        "required": ["api_endpoint", "region"],
    },
)

Not every driver role must be provided. The control plane validates at cluster-registration time that the cluster's required capabilities are covered by its assigned plugin.

Implementing ClusterDriver

ClusterDriver is the core protocol. It applies Kubernetes manifests to a target cluster, deletes them, and streams workload describe output.

# my_plugin/cluster.py
from __future__ import annotations

from collections.abc import AsyncIterator
from typing import Any

from _sdk.cluster import ApplyError, ApplyResult, DeleteResult, classify_apply_error


class MyClusterDriver:
    """ClusterDriver targeting My Cloud Platform's managed Kubernetes service."""

    def __init__(self, config: dict[str, Any]) -> None:
        self._endpoint = config["api_endpoint"]
        self._region = config["region"]
        # Initialise your cloud SDK client here.

    def apply_manifests(
        self,
        *,
        namespace: str,
        manifests: list[dict[str, Any]],
        prune_label_selector: str | None = None,
    ) -> ApplyResult:
        """Server-side apply a list of Kubernetes resource dicts.

        Each manifest is a standard Kubernetes resource dict
        (apiVersion, kind, metadata, spec). Apply is idempotent —
        existing resources are patched; new ones are created.
        """
        created: list[str] = []
        updated: list[str] = []
        unchanged: list[str] = []
        errors: list[ApplyError] = []

        for manifest in manifests:
            kind = manifest.get("kind", "Unknown")
            name = manifest.get("metadata", {}).get("name", "unknown")
            try:
                result = self._apply_one(namespace, manifest)
                if result == "created":
                    created.append(f"{kind}/{name}")
                elif result == "updated":
                    updated.append(f"{kind}/{name}")
                else:
                    unchanged.append(f"{kind}/{name}")
            except Exception as exc:
                errors.append(
                    ApplyError(
                        kind=kind,
                        name=name,
                        namespace=namespace,
                        exception_type=type(exc).__name__,
                        exception_message=str(exc),
                        is_retryable=classify_apply_error(exc),
                    )
                )

        return ApplyResult(
            created=created,
            updated=updated,
            unchanged=unchanged,
            errors=errors,
        )

    def delete_manifests(
        self,
        *,
        namespace: str,
        manifests: list[dict[str, Any]],
    ) -> DeleteResult:
        deleted: list[str] = []
        not_found: list[str] = []
        errors: list[str] = []

        for manifest in manifests:
            kind = manifest.get("kind", "Unknown")
            name = manifest.get("metadata", {}).get("name", "unknown")
            try:
                found = self._delete_one(namespace, manifest)
                if found:
                    deleted.append(f"{kind}/{name}")
                else:
                    not_found.append(f"{kind}/{name}")
            except Exception as exc:
                errors.append(f"{kind}/{name}: {exc}")

        return DeleteResult(deleted=deleted, not_found=not_found, errors=errors)

    async def describe_workload(
        self,
        *,
        namespace: str,
        pod_label_selector: str,
    ) -> AsyncIterator[str]:
        """Stream describe output for matching pods."""
        pods = self._list_pods(namespace, pod_label_selector)
        for pod in pods:
            yield f"=== {pod['name']} ===\n"
            yield self._describe_pod(namespace, pod["name"])

    # --- private helpers ---

    def _apply_one(self, namespace: str, manifest: dict[str, Any]) -> str:
        # Call your cloud's Kubernetes API here.
        # Return "created", "updated", or "unchanged".
        raise NotImplementedError

    def _delete_one(self, namespace: str, manifest: dict[str, Any]) -> bool:
        # Return True if the resource existed and was deleted.
        raise NotImplementedError

    def _list_pods(self, namespace: str, selector: str) -> list[dict[str, Any]]:
        raise NotImplementedError

    def _describe_pod(self, namespace: str, name: str) -> str:
        raise NotImplementedError

Implementing ManagedServiceDriver — Redis example

ManagedServiceDriver provisions and manages a backing service (databases, caches, queues). Here is a complete Redis driver.

# my_plugin/managed/redis_mine.py
from __future__ import annotations

from typing import Any

from _sdk.managed_service import (
    Binding,
    BindingSchema,
    DeprovisionResult,
    DeprovisionSpec,
    ProvisionResult,
    ProvisionSpec,
    ServiceHandle,
    ServiceStatus,
    SnapshotHandle,
    UpdateResult,
    UpdateSpec,
    ValueRef,
)


class MyRedisDriver:
    """ManagedServiceDriver for a Redis-compatible cache on My Cloud Platform.

    Registered as ("redis", "my_redis") in the plugin manifest.
    The kind contract (from managed_service_kinds.py) requires
    the binding to emit REDIS_HOST and REDIS_PORT env vars.
    REDIS_PASSWORD and REDIS_TLS are optional.
    """

    def __init__(self, config: dict[str, Any]) -> None:
        self._region = config["region"]
        # Initialise SDK client.

    def provision(self, spec: ProvisionSpec) -> ProvisionResult:
        """Create a new Redis cache instance.

        ``spec.size`` is one of: small | medium | large | xlarge | custom.
        ``spec.config`` may carry driver-specific keys declared by config_schema().
        ``spec.isolation`` is "shared" or "dedicated".
        """
        handle_name = (
            f"redis-{spec.organization_slug}-{spec.app_slug}-{spec.environment_name}"
        )
        try:
            self._create_cache(
                name=handle_name,
                size=spec.size,
                region=self._region,
                tags={
                    "astrolift.io/binding": spec.binding_id,
                    "astrolift.io/managed_service_id": spec.managed_service_id,
                    **spec.tags,
                },
            )
        except Exception as exc:
            return ProvisionResult(
                ok=False,
                handle=handle_name,
                message=str(exc),
                errors=[str(exc)],
            )
        return ProvisionResult(ok=True, handle=handle_name, message="provisioned")

    def update(self, spec: UpdateSpec) -> UpdateResult:
        try:
            if spec.size:
                self._resize_cache(spec.handle, spec.size)
        except Exception as exc:
            return UpdateResult(ok=False, handle=spec.handle, message=str(exc))
        return UpdateResult(ok=True, handle=spec.handle, message="updated")

    def deprovision(
        self,
        spec: DeprovisionSpec,
        *,
        delete_data: bool = False,
        force_destroy: bool = False,
    ) -> DeprovisionResult:
        try:
            if not delete_data:
                self._export_backup(spec.handle)
            self._delete_cache(spec.handle, force=force_destroy)
        except Exception as exc:
            return DeprovisionResult(ok=False, handle=spec.handle, message=str(exc))
        return DeprovisionResult(ok=True, handle=spec.handle, message="deprovisioned")

    def status(self, handle: ServiceHandle) -> ServiceStatus:
        raw = self._describe_cache(handle.handle)
        # Map your cloud's status string to one of:
        # provisioning | available | updating | error | deprovisioning | deprovisioned
        state_map = {
            "creating": "provisioning",
            "available": "available",
            "modifying": "updating",
            "deleting": "deprovisioning",
            "deleted": "deprovisioned",
        }
        return ServiceStatus(
            handle=handle.handle,
            state=state_map.get(raw.get("status", ""), "error"),
            message=raw.get("message", ""),
        )

    def binding(
        self,
        handle: ServiceHandle,
        config: dict[str, Any] | None = None,
    ) -> Binding:
        """Return connection env vars for this cache.

        The redis kind contract requires REDIS_HOST and REDIS_PORT.
        """
        endpoint = self._get_endpoint(handle.handle)
        env: dict[str, ValueRef] = {
            "REDIS_HOST": ValueRef(literal=endpoint["host"]),
            "REDIS_PORT": ValueRef(literal=str(endpoint["port"])),
        }
        if endpoint.get("auth_token_secret_arn"):
            env["REDIS_PASSWORD"] = ValueRef(secret_ref=endpoint["auth_token_secret_arn"])
        if endpoint.get("tls"):
            env["REDIS_TLS"] = ValueRef(literal="true")
        return Binding(env_vars=env)

    def snapshot(self, handle: ServiceHandle) -> SnapshotHandle:
        snapshot_id = self._create_snapshot(handle.handle)
        return SnapshotHandle(
            handle=handle.handle,
            snapshot_id=snapshot_id,
            created_at="",  # fill from API response
        )

    def restore(self, snapshot: SnapshotHandle, target: ProvisionSpec) -> ProvisionResult:
        try:
            self._restore_from_snapshot(snapshot.snapshot_id, target.service_handle_hint)
        except Exception as exc:
            return ProvisionResult(ok=False, handle=snapshot.handle, message=str(exc))
        return ProvisionResult(ok=True, handle=snapshot.handle, message="restored")

    def config_schema(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "engine_version": {"type": "string", "default": "7.0"},
                "enable_tls": {"type": "boolean", "default": True},
            },
        }

    def binding_schema(self) -> BindingSchema:
        return BindingSchema(
            env_vars={
                "REDIS_HOST": "Cache endpoint hostname",
                "REDIS_PORT": "Cache port (default 6379)",
                "REDIS_PASSWORD": "Auth token (optional, from secrets store)",
                "REDIS_TLS": "Set to 'true' when TLS is enabled",
            }
        )

    def editable_fields(self) -> list[str]:
        # Only size and enable_tls can be changed without full reprovision.
        return ["size", "enable_tls"]

    # --- private helpers ---

    def _create_cache(self, name: str, size: str, region: str, tags: dict) -> None:
        raise NotImplementedError

    def _resize_cache(self, handle: str, size: str) -> None:
        raise NotImplementedError

    def _delete_cache(self, handle: str, force: bool) -> None:
        raise NotImplementedError

    def _describe_cache(self, handle: str) -> dict[str, Any]:
        raise NotImplementedError

    def _get_endpoint(self, handle: str) -> dict[str, Any]:
        raise NotImplementedError

    def _export_backup(self, handle: str) -> None:
        raise NotImplementedError

    def _create_snapshot(self, handle: str) -> str:
        raise NotImplementedError

    def _restore_from_snapshot(self, snapshot_id: str, target_handle: str) -> None:
        raise NotImplementedError

Implementing SecretsBackend

# my_plugin/secrets.py
from __future__ import annotations

from _sdk.secrets import SecretsBackend


class MySecretsBackend:
    """SecretsBackend storing secrets in My Cloud's key-value store."""

    def get(self, path: str) -> dict[str, str] | None:
        """Return all key-value pairs at path, or None if the path does not exist."""
        raise NotImplementedError

    def upsert(self, path: str, kvs: dict[str, str]) -> None:
        """Create or overwrite key-value pairs at path."""
        raise NotImplementedError

    def delete(self, path: str) -> None:
        """Delete the secret at path. No-op if it does not exist."""
        raise NotImplementedError

    def list(self, prefix: str) -> list[str]:
        """Return all paths with the given prefix."""
        raise NotImplementedError

    def ensure_initialized(self) -> dict | None:
        """Bootstrap the backend (create KMS key, enable secrets engine, etc.).

        Backends that require no setup should raise NotImplementedError —
        the control plane treats this as a no-op and stamps the
        provisioned_at timestamp.
        """
        raise NotImplementedError

Adding a new managed-service kind

If your plugin introduces a kind that does not yet exist in the catalog (e.g. vector_db for a proprietary vector store), register it before submitting:

1. Open a pull request to astrolift-providers/_sdk/managed_service_kinds.py adding a ManagedServiceKind entry with the required and optional binding env names.
2. Your driver's binding() method must emit every key in binding_envs_required or the matrix-check CI job will fail.

If you are shipping a new variant of an existing kind (e.g. ("redis", "my_redis")), no catalog change is needed — just register the (kind, variant) tuple in your plugin manifest.

Testing locally

Install the plugin in the same virtual environment as the Astrolift control plane using pip's editable mode:

# Inside the control plane's venv
pip install -e /path/to/my-astrolift-plugin

The control plane discovers plugins via importlib.metadata.entry_points(group="astrolift.providers") at startup, so no code changes are needed. Restart the control plane and the plugin will appear in the provider plugin list.

To verify discovery without starting the full stack:

python -c "
from importlib.metadata import entry_points
for ep in entry_points(group='astrolift.providers'):
    print(ep.name, '->', ep.load())
"

Unit testing your driver

Write tests against your driver directly without a running control plane. Use your cloud's mock SDK (e.g. moto for AWS) or a local stub.

# tests/test_plugin.py
import pytest
from my_plugin.managed.redis_mine import MyRedisDriver
from _sdk.managed_service import ProvisionSpec, ServiceHandle


@pytest.fixture()
def driver():
    return MyRedisDriver(config={"region": "us-east-1"})


def test_provision_returns_handle(driver, mock_cloud):
    spec = ProvisionSpec(
        organization_id="org-1",
        organization_slug="acme",
        app_id="app-1",
        app_slug="api",
        environment_id="env-1",
        environment_name="production",
        tenant_cluster_id="cluster-1",
        service_handle_hint="redis",
        size="small",
    )
    result = driver.provision(spec)
    assert result.ok
    assert result.handle.startswith("redis-acme-api")


def test_binding_emits_required_envs(driver, mock_cloud):
    from _sdk.managed_service_kinds import KINDS, validate_binding_envs
    handle = ServiceHandle(handle="redis-acme-api-production")
    binding = driver.binding(handle)
    missing = validate_binding_envs(
        kind="redis",
        emitted_envs=list(binding.env_vars.keys()),
    )
    assert missing == [], f"Binding is missing required env vars: {missing}"

Run with:

pytest tests/ -v

Publishing to PyPI

pip install build twine
python -m build
twine upload dist/*

After publishing, operators install your plugin on any Astrolift control plane:

pip install my-astrolift-plugin
# then restart the control plane

The plugin appears in Settings → Provider Plugins and can be assigned to new or existing clusters.

Versioning

Follow semantic versioning. The _sdk protocols are stable within a minor version of astrolift-providers. Breaking changes to the SDK increment the minor version and are listed in the providers changelog. Pin your astrolift-providers dependency to a compatible range:

"astrolift-providers>=0.2,<1"

Test against each SDK version in your CI matrix before releasing a new plugin version.