First unit — Python

Guide for SDK 0.8.0 and Python ≥3.12. The cats example declares an exposition, composes its artifact and registers a handler. Platform operation requires a Control Plane, Dapr, an OIDC provider, a link key and a persistent identity directory. Cat data is synthetic. The example keeps it in memory and implements no writes.

Project and private index

Create the project directory and an empty cats/__init__.py. Replace the index address with the administrator-provided value. Mapping is explicit; do not use --extra-index-url.

# pyproject.toml
[project]
name = "cat-unit"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["seeraos-unit-sdk==0.8.0", "uvicorn"]

[[tool.uv.index]]
name = "seeraos"
url = "https://registry.example.com/pypi/simple/"
explicit = true

[tool.uv.sources]
seeraos-unit-sdk = { index = "seeraos" }

export UV_INDEX_SEERAOS_USERNAME='YOUR_USER'
export UV_INDEX_SEERAOS_PASSWORD='YOUR_READ_TOKEN'
uv sync

Declaration and manifest

Save the following file. domain defines the platform namespace and the event type prefix in seeraos.shelter.cat.created.v1. The manifest explicitly declares bindings and the capacity field.

# cats/contract.py
from seeraos.unit_sdk import (
    Binding,
    ConfigField,
    Effect,
    Event,
    Exposition,
    Field,
    FieldType,
    Manifest,
    ManifestExposition,
    ObjectType,
    Operation,
    Verb,
    Visibility,
)

Cat = ObjectType(
    name="Cat",
    identity="id",
    fields={
        "id": Field(type=FieldType.STRING, required=True),
        "name": Field(type=FieldType.STRING, required=True),
    },
)
GET_CAT = Operation(object=Cat, verb=Verb.GET, response=Cat)
CAT_CREATED = Event(object=Cat, operation="created", effect=Effect.ADD)

EXPOSITION = Exposition(
    name="cat", major=1, unit="shelter", operations=[GET_CAT], events=[CAT_CREATED]
)
MANIFEST = Manifest(
    expositions=[
        ManifestExposition(
            domain="shelter",
            name=EXPOSITION.name,
            version=EXPOSITION.major,
            bindings=frozenset({Binding.RPC, Binding.EVENT}),
            visibility=Visibility.INTERNAL,
        )
    ],
    config=[ConfigField(name="capacity", required=True)],
)

uv run python -m seeraos.unit_sdk compose cats.contract:EXPOSITION --output cats/cats.v1.json

Expected artifact: cats/cats.v1.json. The document belongs to the release and is not an independently maintained manual copy of the declaration. Recompose it after a declaration change.

Handler and host

EXPOSITION.published compares saved bytes with the declaration. Host construction requires handlers for declared operations. It opens no connections before running. The token_issuer address is a placeholder for your OIDC provider.

# cats/unit.py
from pathlib import Path

from seeraos.unit_sdk import (
    Configuration,
    JsonValue,
    OperationCall,
    UnitError,
    UnitHost,
)

from cats.contract import EXPOSITION, GET_CAT, MANIFEST

DOCUMENT = Path(__file__).with_name("cats.v1.json")


def create_unit(
    *, control_plane_url: str, identity_directory: Path, link_key: str | None = None
) -> UnitHost:
    cats: dict[str, JsonValue] = {}

    async def configure(configuration: Configuration) -> None:
        """Open what the unit connects to, with configuration.values."""

    async def get_cat(call: OperationCall) -> JsonValue:
        cat = cats.get(call.entity_id or "")
        if cat is None:
            raise UnitError.not_found(code="Cat.NotFound", description="No such cat.")
        return cat

    return UnitHost(
        documents=[EXPOSITION.published(DOCUMENT)],
        control_plane_url=control_plane_url,
        identity_directory=identity_directory,
        link_key=link_key,
        manifest=MANIFEST,
        token_issuer="https://idp.example.com/realms/seeraos",
        on_configure=configure,
        handlers={GET_CAT.key: get_cat},
    )

ASGI process

Save main.py. The factory import from the preceding file is included. Application variables identify the Control Plane, persistent identity directory and link key issued for the instance. SDK reads the Dapr HTTP port from the environment.

# main.py
import asyncio
import os
import sys
from pathlib import Path

import uvicorn
from cats.unit import create_unit

app = create_unit(
    control_plane_url=os.environ["SEERAOS_CONTROL_PLANE_URL"],
    identity_directory=Path(os.environ["SEERAOS_IDENTITY_DIRECTORY"]),
    link_key=os.environ.get("SEERAOS_LINK_KEY"),
)


async def main() -> int:
    server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=8080))
    serving = asyncio.create_task(server.serve())
    idle = asyncio.create_task(app.idle())
    await asyncio.wait({serving, idle}, return_when=asyncio.FIRST_COMPLETED)
    server.should_exit = True
    idle.cancel()
    await serving
    ended = any(slot.refusal is not None or slot.stopped for slot in app.status)
    return 1 if ended else 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))

uv run dapr run --app-id shelter --app-port 8080 -- python main.py

The server must forward lifespan: it starts enrollment, configuration and reporting. On first start the host uses the link key and saves identity; later starts restore it. An HTTP exposition response alone does not indicate an active instance. app.idle() and app.status support terminating the process after permanent refusal rather than leaving an apparently healthy listener.

configure is where configuration.values are used; required fields come from the manifest. Events are published within declarations and active instance identity. Declaring CAT_CREATED does not make the shown read handler emit it.

A missing cat returns Cat.NotFound; an unready tenant leads to Tenant.NotReady. Further troubleshooting separates installation, parity, enrollment and activation. The guide describes documented SDK behaviour; its author need not rerun the platform.