osp-provider-sdk

A thin, optional authoring layer for OSP providers. Declare actions and lists, plan before effects, and report progress through the existing Provider protocol. The package layers are contracts → runtime → SDK → providers. The orchestrator never installs the SDK; existing providers can keep their own Provider implementations.

pip install osp-provider-sdk

One-file provider

This is examples/buckets.py. Replace backend with your team's adapter. The included backend is in-memory and makes no network calls; real backends must persist the task's client token for recovery across processes. Do not adopt a bucket merely because its name matches.

"""S3-style bucket provider on the OSP SDK, with a replaceable backend adapter."""

from contextlib import asynccontextmanager
from pathlib import Path
from typing import Literal

from osp_provider_contracts.canonical_digest import canonical_digest
from pydantic import BaseModel, ConfigDict, Field

from osp_provider_sdk import App, Done, Page, Plan, Resource, ResourceOwnership, Violation

from . import backend


@asynccontextmanager
async def lifespan():
    async with backend.Client.from_env() as client:
        yield client


app = App("s3", policy=Path(__file__).with_name("policy.yaml"), lifespan=lifespan)


class Create(BaseModel):
    model_config = ConfigDict(frozen=True)
    quota_gib: int = Field(10, ge=1, le=1000, title="Quota (GiB)")
    region: Literal["osl", "bgo"] = "osl"


class Resize(BaseModel):
    model_config = ConfigDict(frozen=True)
    quota_gib: int = Field(ge=1, le=1000, title="Quota (GiB)")


class Row(BaseModel):
    quota_gib: int = Field(title="Quota (GiB)")
    region: str


def _row(b) -> Resource:
    return Resource(
        target_ref=b.id,
        display=b.name,
        ownership=ResourceOwnership(contacts=b.contacts),
        fields=Row(quota_gib=b.quota_gib, region=b.region).model_dump(),
        created_at=b.created,
    )


@app.resource("bucket", fields=Row, filters=["region"])
async def buckets(q, access) -> Page:
    page = await app.state.list(region=q.filters.get("region"), cursor=q.cursor, limit=q.limit)
    return Page(
        tuple(_row(b) for b in page.items if access.allows(b.contacts)),
        page.cursor,
        observed_at=page.observed_at,
    )


@buckets.get
async def bucket(ref, access) -> Resource | None:
    b = await app.state.get(ref)
    return _row(b) if b and access.allows(b.contacts) else None


@app.action("create", inputs=Create, stages=["s3.create", "s3.tag"])
async def create(req, ctx) -> Done:
    with ctx.step(
        "s3.create",
        f"Creating bucket {req.name}: {req.inputs.quota_gib} GiB in {req.inputs.region}",
    ):
        b = await app.state.create(
            req.name, req.inputs.region, req.inputs.quota_gib, client_token=ctx.task_ref
        )
        ctx.created(b.id, req.name)
    with ctx.step("s3.tag", f"Recording owners {', '.join(ctx.contacts)} on {req.name}"):
        await app.state.tag(b.id, contacts=ctx.contacts)
    return Done(f"Created bucket {req.name} ({b.id})")


@create.plan
async def plan_create(req, ctx) -> Plan:
    if not req.name:
        return Plan(
            "Missing bucket name",
            violations=(Violation.missing_required("name", "A bucket name is required"),),
        )
    taken = await app.state.get_by_name(req.name)
    if taken and taken.client_token != ctx.task_ref:
        return Plan(
            f"bucket {req.name}: name taken",
            violations=(
                Violation.invalid_value("name", f"Bucket {req.name} already exists ({taken.id})."),
            ),
        )
    return Plan(
        f'create bucket "{req.name}" in {req.inputs.region}, quota {req.inputs.quota_gib} GiB, '
        f"owners {', '.join(sorted(ctx.contacts))}"
    )


@app.action("resize", target="resource", inputs=Resize)
async def resize(req, ctx) -> Done:
    with ctx.step(
        "s3.quota", f"Setting {req.target_ref} quota {ctx.plan.data} → {req.inputs.quota_gib} GiB"
    ):
        await app.state.set_quota(req.target_ref, req.inputs.quota_gib)
    return Done(f"Bucket {req.target_ref} quota is {req.inputs.quota_gib} GiB")


@resize.plan
async def plan_resize(req, ctx) -> Plan:
    b = await app.state.get(req.target_ref)
    if b is None:
        return Plan(
            "Bucket not found",
            violations=(Violation.invalid_value("target_ref", "Bucket does not exist"),),
        )
    if b.quota_gib == req.inputs.quota_gib:
        return Plan.nothing_to_do(f'bucket {b.id} "{b.name}": quota already {b.quota_gib} GiB')
    return Plan(
        f'bucket {b.id} "{b.name}": quota {b.quota_gib} → {req.inputs.quota_gib} GiB',
        data=b.quota_gib,
    )


@app.action(
    "destroy",
    target="resource",
    destructive=True,
    cancellable=True,
    stages=["s3.empty", "s3.delete"],
)
async def destroy(req, ctx) -> Done:
    with ctx.step("s3.empty", f"Emptying bucket {req.target_ref}"):
        async for done, total in app.state.empty(req.target_ref, objects=ctx.plan.data):
            ctx.progress(done, total, f"Emptied {done} of {total} prefixes in {req.target_ref}")
            ctx.checkpoint()  # The previous prefix delete has settled.
    with ctx.step("s3.delete", f"Deleting bucket {req.target_ref}"):
        await app.state.delete(req.target_ref)
    return Done(f"Deleted bucket {req.target_ref}")


@destroy.plan
async def plan_destroy(req, ctx) -> Plan:
    b = await app.state.get(req.target_ref)
    if b is None:
        return Plan.nothing_to_do(f"bucket {req.target_ref}: already absent")
    objects = frozenset(b.objects)
    scope = canonical_digest(sorted(objects))
    return Plan(
        f'delete bucket {b.id} "{b.name}" and {len(objects)} object versions (scope {scope})',
        data=objects,
    )


if __name__ == "__main__":
    app.run()

The matching policy.yaml is loaded with yaml.safe_load and validated against the generated manifest before entering lifespan:

request_policy:
  version: 1
  rules:
    - id: team_access
      match: {kind: bucket, actions: [create, resize, destroy], instance: dev}
      check: {netgroup_in: [team-storage]}
      level: gate
      approval: {kind: maintainer, required_approvals: 1}
      message: This team needs a storage operator's approval.
    - id: quota_warning
      match: {kind: bucket, actions: [create, resize]}
      check: {input: quota_gib, max: 80}
      level: warn
      message: This is a large bucket quota.
    - id: quota_approval
      match: {kind: bucket, actions: [create, resize]}
      check: {input: quota_gib, max: 100}
      level: gate
      approval: {kind: maintainer, required_approvals: 1}
      message: Quotas above 100 GiB need storage operator approval.

The default destructive-action Gate requires one Maintainer approval for APPLY (a CHECK deletes nothing; assessments still show the Gate). To change it provider-wide, add delete_approval: {kind: admin, required_approvals: 1} under request_policy. A destructive action that owners may run after typing its phrase, such as deleting their own snapshot, declares delete_approval=False; its planner then owns any Gate. peer is also supported. Rules belong to contracts; backend-specific decisions stay in the custom read-only planner.

Invocation and lifetime

app.run() reads RABBITMQ_URL, PROVIDER_INSTANCE, PROVIDER_INSTANCE_ID and CONCURRENCY (default 1); keyword arguments override any runtime option. Queue names come from ProviderIdentity. For example, PROVIDER_INSTANCE=dev python -m examples.buckets starts the example's RabbitMQ runner when RABBITMQ_URL is provided.

A provider's __main__.py is just app.run(). The SDK command covers the rest without an orchestrator; it imports the app and enters lifespan only for an online doctor and for plan:

python -m osp_provider_sdk manifest examples.buckets:app [--check [PATH]]
python -m osp_provider_sdk doctor examples.buckets:app [--offline]
python -m osp_provider_sdk plan examples.buckets:app create '{"quota_gib": 150}' --name logs

doctor --offline checks declarations and policy. Online it also connects to RABBITMQ_URL and runs the provider's own checks, which may use app.state and return problem strings:

@app.doctor
def data_dir() -> list[str]:
    return [] if DATA_DIR.is_dir() else [f"{DATA_DIR} is missing"]

It prints ok (exit 0) or one problem per line (exit 1). plan prints the planner result an assessment would return, against the configured backend.

Offline invocation uses with app.running():. This validates declarations, enters lifespan, and owns one event loop if any callback or lifespan is async. All async backend clients are created and closed on that loop. Runtime workers block for their handler's result, so runtime concurrency still bounds work. Async handlers must move blocking backend calls into asyncio.to_thread. Sync handlers returning awaitables fail with an instruction to use async def.

from osp_provider_contracts import ProviderRequest, RequestContext
from examples.buckets import app

with app.running():
    result = app.execute(
        ProviderRequest("create", name="logs"),
        RequestContext("local-task", netgroup="team-storage", contacts=("storage@example.org",)),
    )
    print(result.produced_resources)

Planning and approval

Assessment, CHECK and APPLY use one planner: model validation → shared rules → custom plan → combination. Assessment always describes APPLY intent and has no cleared approvals. It returns structured rejection, incomplete or approval facts without calling apply. CHECK and no-op plans stop before apply and return {"plan": <planner result>}; an App subclass that must keep an existing CHECK shape overrides plan_result(req, plan) and copies only public facts.

Gated actions must declare .plan and author a concrete summary. The Gate key names a requirement; its fingerprint covers every frozen Gate field except the key, including summary and effective inputs. Use native IDs, before/after values and material destructive scope. Bind destructive work to a content revision or immutable object-version scope, including its fingerprint in public evidence. Apply must enforce that scope with conditional writes and refuse a nonempty bucket delete. Leave out secrets, timestamps and unrelated volatile counts. If the planned work changes, approval must be obtained again. Private Plan.data never enters the public assessment; the checked frozen input model and the same Plan reach apply.

Custom plans must be read-only. Apply cannot request approval: a late ApprovalRequired becomes late_gate_contract_violation with uncertain state. Keep live-state-dependent authorization wholly in the custom plan when shared rules over model defaults would check different values from those executed.

Progress, cancellation and results

Stages use <system>.<verb> keys. Declared stages determine 1-based position; repeat and skip them freely. Progress counts are monotonic within a stage. The first and final counts are sent, intermediate updates are limited to one every two seconds, and the latest suppressed count is flushed on stage exit. Evidence is a JSON-safe mapping under 2 KiB without reserved progress keys. Offline calls log progress. Delivery failures do not fail backend work.

cancellable=True is cooperative. cancel_task accepts a request; only ctx.checkpoint() stops work. Call it only after prior backend operations have settled and the current state is safe to leave. Stage entry and progress never stop work. A request after the last checkpoint can finish normally.

Use Done for ordinary success or return a full ProviderResult for advanced outcomes (pending_retry, provenance, mixed resources). Advanced results retain all their fields. Record each survivor with ctx.created, specifying kind= for mixed kinds, and call ctx.removed only after confirmed cleanup. Survivors are merged into success, typed errors, unexpected failures and cancellation results. There is no rollback or durable SDK journal. A retried task must rediscover its resources through provider-owned backend identity.

Manifest

Fields come from frozen Pydantic models: str, int, float, bool, scalar Literal (an exhaustive enum), list[...] and dict[...] (list/object), and T | None = None for an optional input without a default. The model states name (Field(alias=...) for dotted wire names), type, required, default, integer ge/le, string pattern, title (label) and description (help). json_schema_extra adds the other contract field keys: unit, group, semantic, accepts and choices. Choices on a Literal relabel or annotate exactly its values (meta, hidden, aliases, …). Choices on any other type are suggestions: values must validate against the field, and type: "enum" or choices_exhaustive: False may be stated, never True. Annotate ResourceRef for references. Validators, serializers, nested models and other constraints fail startup.

Actions take description, target, destructive, stages, risk_hint and confirmation_phrase. Destructive actions always advertise a phrase, by default the upper-cased action name (snapshot_delete → SNAPSHOT DELETE); clients offer destructive actions only with one. Resources take list fields and filters; list fields carry no required, so views stay meaningful. Exact get is optional; without one the SDK walks list cursors until the native reference is found. Undeclared row fields are dropped while downloads, timestamps and other native metadata are retained. Page is the contracts ResourceListResponse: pass its required observed_at freshness timestamp.

The manifest version is the installed osp-provider-<family> distribution's version (family smoke_stack → osp-provider-smoke-stack) unless App(version=...) sets it; code that is not installed advertises 0.0.0.

Everything else

extra= on App, @app.action and @app.resource (merged into its list) adds any remaining key of the contracts ManifestV2, ActionSpec or ListSpec. Unknown keys and keys owned by an SDK argument fail at startup, and the whole document is then checked with validate_manifest_v2.

app = App("nrec", description="NREC provider", extra={"features": ["console.novnc"]})


@app.resource("vm", fields=VmRow, filters=["name"], extra={"page_size_max": 1000})
def vms(q, access) -> Page: ...


@app.action(
    "resize",
    target="resource",
    inputs=Resize,
    risk_hint="Resizing interrupts the VM.",
    extra={"inputs": {"soft_limits": {"flavor": {"allowed_without_force": ["m1.small"]}}}},
)
def resize(req, ctx) -> Done: ...

app.run(**runtime) overrides any RuntimeConfig field, including the environment and queue defaults: concurrency=5 replaces CONCURRENCY, and prefetch_count, control_concurrency or max_attempts work the same way.

Protocol methods the SDK does not declare stay plain methods on an App subclass: get_console, get_history and get_build_log are found by the runtime, and self.call(fn, *args) runs sync or async backend code on the lifespan's loop with self.state.

Hot-loaded data (reloaded choices, points, actions an operator switched off) overrides effective_manifest(), starting from super().effective_manifest(). The SDK validates its result once per call and uses that one snapshot for capabilities, action membership and list field/filter checks. It can hide declared actions and lists or add list fields and filters, never offer an undeclared handler; input models still validate the inputs themselves. Ownership policy rules, native backend cancellation and structured inputs are not part of the SDK.

Development and snapshots

Install the released contracts/runtime dependencies and SDK development tools:

uv venv --python 3.13
uv pip install -e '.[dev]'
.venv/bin/ruff check .
.venv/bin/ty check src tests
.venv/bin/pytest
.venv/bin/python -m osp_provider_sdk manifest examples.buckets:app --check examples/manifest.json
.venv/bin/python -m build
.venv/bin/twine check dist/*

Generate snapshots with python -m osp_provider_sdk manifest module:app. --check [path] compares exact canonical JSON, defaulting to manifest.json. Snapshot generation checks startup but never enters lifespan or connects a backend.

CI runs the SDK and shared provider conformance suite with lowest-direct and highest dependency resolution, checks the manifest, builds both artifacts, and imports the installed wheel outside the source directory. Tag CI attaches artifacts to a GitHub release; it does not publish to PyPI.

First release

Release the orchestrator fingerprint lifecycle first, then contracts 0.9.x, runtime 0.10.x, and SDK 0.1.0. Contracts must contain Plan/Gate, cleared fingerprints, resource access and request-policy exports; runtime must expose unexpected_provider_error. Re-run both dependency-resolution CI lanes with those released versions. Finish the plan's field-vocabulary prerequisite, NREC/tutorial pilot and Q1 audit before declaring adoption ready. Existing providers and cookiecutter tests stay in place.

The private repository's Actions policy must allow its configured actions. Set up the new distribution name and PyPI credentials or trusted publishing. After review, quality checks and signed release/tag verification, a maintainer can run bash scripts/publish.sh (or -r testpypi). --tag additionally creates, verifies and pushes a signed tag after upload; normal development never runs it. No release or live environment is needed for the local tests.