piphi-runtime-kit-python

Small Python helpers for building PiPhi runtime integrations.

This package is intentionally thin. It removes repetitive PiPhi runtime plumbing without hiding the HTTP contract behind a large framework. The goal is to help developers move faster while still understanding what their runtime is doing.

Version 0.6.0 is the current documented baseline.

New to PiPhi? Start with First 15 Minutes, then read The Golden Path, then compare your code to the example app.

Safety note: every ID, token, hostname, and UUID shown in this README is placeholder example data. Do not copy real internal tokens, production hosts, or live container IDs into code samples, docs, or tests.

Quick Navigation

• Who this is for
• Install
• First 15 Minutes
• Route Contract At A Glance
• UI Config Endpoints
• One Complete Mini Example
• The Golden Path
• The IDs You Need To Understand
• Telemetry vs Events
• Plain-Language Concepts
• When To Use Which Helper
• Troubleshooting
• What A Real Integration Still Needs

Reading Paths

• New developer Read First 15 Minutes, One Complete Mini Example, The Golden Path, and Plain-Language Concepts.
• Experienced developer Read Route Contract At A Glance, When To Use Which Helper, and What A Real Integration Still Needs.
• Debugging a runtime Jump to Troubleshooting, Common Mistakes, and Clear Error Handling.

Who this is for

This SDK is for developers building PiPhi integrations in Python.

It is a good fit if you are using:

• FastAPI
• Starlette
• another async Python HTTP framework
• plain request/response handlers where you still want PiPhi helpers

If you are brand new to PiPhi, start with the golden path in this README and then compare your code to the example app.

What the SDK handles

The SDK is meant to own the shared PiPhi runtime plumbing:

• runtime auth context
• request/header auth helpers
• optional FastAPI request helpers
• process state
• background task tracking
• telemetry delivery to PiPhi Core
• event delivery to PiPhi Core
• config sync helpers
• typed config validation helpers
• discovery normalization and response helpers
• lifecycle bootstrap helpers
• runtime health and diagnostics helpers
• in-memory runtime registry for active entries, state, and recent events
• clearer PiPhi-specific delivery errors

What stays in your integration

Your integration still owns vendor-specific behavior:

• how to discover devices
• how to talk to the vendor API or device
• how often to poll
• how to transform vendor data into PiPhi entities or telemetry
• which runtime events matter for that integration

This boundary is important. The SDK should make runtime plumbing easier, not hide vendor logic behind an overly magical abstraction.

Telemetry vs Events

Use telemetry for measurements and current state: temperature, humidity, power, signal strength, online status, or whether a switch is on. PiPhi Core uses these readings to derive capability and state changes, so integrations do not need to emit a second event for every changed metric.

Use an explicit event for a discrete semantic occurrence that cannot be reliably derived from state: a button press, vendor alarm, pairing result, or command result.

For new integrations, build a device reference from the Core-managed runtime registry entry. This keeps the Core config_id, stable vendor device_id, and runtime scope together and produces an actionable error when required identity is missing:

from piphi_runtime_kit_python import RuntimeDeviceRef, TelemetryReading

entry = starter.registry.primary_entry()
device = RuntimeDeviceRef.from_value(entry)

await starter.telemetry_client.send_device_readings(
    device=device,
    readings=[
        TelemetryReading("temperature_c", 21.4, "C"),
        TelemetryReading("power", True),
    ],
    auth_context=starter.runtime.auth,
)

Send only meaningful, non-derivable occurrences as events:

await starter.event_client.send_device_event(
    device=device,
    event_type="device.button_pressed",
    payload={"button": "top", "press_type": "long"},
    auth_context=starter.runtime.auth,
)

Identity rules:

• always retain the config_id supplied by PiPhi Core
• use a stable vendor-native device_id; do not generate a new ID on each poll
• keep integration_id on registry entries that can publish events
• prefer send_device_readings(...) and send_device_event(...) for new code
• use the lower-level dict APIs when an integration genuinely needs explicit control over the wire fields

The SDK remains intentionally thin: it validates and sends the data, while PiPhi Core owns state-change derivation, persistence policy, and automation semantics.

Install

The package currently supports Python >=3.11.

Install from PyPI:

pdm add piphi-runtime-kit-python

If you need the optional FastAPI helpers:

pdm add "piphi-runtime-kit-python[fastapi]"

If you prefer pip:

pip install piphi-runtime-kit-python

Package page:

• https://pypi.org/project/piphi-runtime-kit-python/

For release gates, capacity limits, delivery guarantees, and the current production-readiness assessment, see Production readiness.

Releasing the SDK

SDK releases are built and published by .github/workflows/release.yml when a GitHub Release is published. The workflow requires the release tag to exactly match v<project.version>, runs the full test and lint suites, verifies that the wheel matches the checked-out source, and publishes only that verified artifact.

PyPI trusted publishing must authorize the following GitHub identity:

• owner: PiPhi-io
• repository: piphi-runtime-kit-python
• workflow: release.yml
• environment: pypi

The publishing job uses GitHub OIDC credentials, so no long-lived PyPI token is stored in the repository. Where the repository plan supports deployment protection rules, protect the pypi environment with a required reviewer. The workflow intentionally has no manual-dispatch trigger and only accepts a formal GitHub Release.

First 15 Minutes

If you want the shortest path to a working runtime, do this:

1. install the SDK
2. create a starter with create_runtime_starter(...)
3. define one typed config model
4. add GET /health
5. add POST /config
6. add one telemetry example route
7. compare your result to the example app

Minimal success checklist:

• the runtime starts
• /health returns 200
• /config stores one device in the registry
• a route can queue telemetry without crashing

If those four things work, you already have a real PiPhi runtime foundation.

Route Contract At A Glance

These are the most common runtime routes and what they are for.

Modeling /entities

For simple integrations, /entities can still be a plain list of generic entities.

For smart-home and multi-device integrations, PiPhi now recommends a richer runtime-owned entity shape that ties each entity to a real saved config or device. That lets Core make better device-first dashboard suggestions, and it gives the frontend enough context to render better widget defaults for plugs, bulbs, thermostats, and other device-specific entities.

Recommended fields:

• id: stable runtime entity id
• name: user-facing label
• capabilities: actual capabilities for that specific device
• config_id / configId: PiPhi config UUID when available
• device_id / deviceId: integration-native device id
• device_type / device_class: values like plug, bulb, sensor, climate
• entity_type: values like switch, light, sensor, media
• dashboard.allowed_widgets, dashboard.default_widget, dashboard.recommended_widgets: optional UI hints

The SDK now includes RuntimeEntityResponse, RuntimeEntitiesResponse, build_entities_response(...), and starter.entities_response(...) to make that payload easier to return.

The helper returns the standard wrapper shape:

• entities: the runtime-owned list you generated
• capabilities: optional manifest capability metadata
• commands: optional manifest command metadata

from piphi_runtime_kit_python import build_entities_response

@app.get("/entities")
async def entities() -> dict[str, Any]:
    return build_entities_response(
        entities=[
            {
                "id": "office-plug",
                "name": "Office Plug",
                "config_id": "core-config-uuid",
                "device_id": "office-plug",
                "device_class": "plug",
                "entity_type": "switch",
                "capabilities": ["switch", "power", "energy_today"],
                "dashboard": {
                    "allowed_widgets": ["tile", "button", "stat"],
                    "default_widget": "tile",
                },
            }
        ],
        capabilities=manifest["capabilities"],
        commands=manifest.get("commands", {}),
    ).model_dump(exclude_none=True)

If you are already using the starter object, the same response can be built with starter.entities_response(...) instead of calling the standalone helper directly.

UI Config Endpoints

Many integrations expose /ui or /ui-config so the PiPhi frontend knows how to render a configuration form.

The important thing to know is:

• this is still just plain JSON
• the runtime SDK does not require a special wrapper for it
• your integration can return schema data directly

The usual pattern is to return:

• a JSON Schema object under schema
• a UI customization object under uiSchema

Simple example:

@app.get("/ui-config")
async def ui_config() -> dict[str, Any]:
    return {
        "schema": {
            "title": "Demo Device Setup",
            "type": "object",
            "required": ["host"],
            "properties": {
                "host": {
                    "type": "string",
                    "title": "Host",
                },
                "alias": {
                    "type": "string",
                    "title": "Alias",
                },
            },
        },
        "uiSchema": {
            "host": {
                "placeholder": "192.168.1.50",
            },
            "alias": {
                "placeholder": "Office Sensor",
            },
        },
    }

If your frontend uses svelte-jsonschema-form, the official docs are here:

• https://x0k.dev/svelte-jsonschema-form/

That library is a good fit when your frontend is already rendering JSON Schema forms and you want integrations to stay simple by returning plain schema data.

For now, the recommended SDK approach is:

• document /ui-config
• return plain JSON schema/uiSchema objects
• keep any frontend-specific rendering helpers outside the runtime SDK

Optional MQTT Source Topics

Some integrations work better with a shared source stream than direct runtime-to-runtime HTTP calls.

Examples:

• rtl_433 collectors
• packet capture helpers
• protocol bridges that many integrations may want to listen to

For those cases, the SDK now includes an optional MQTT helper with a small source-oriented topic contract.

Recommended topic layout:

• piphi/sources/<source>/packets
• piphi/sources/<source>/models/<model>/packets
• piphi/sources/<source>/status
• piphi/sources/<source>/errors

For rtl_433, the default shared packet topic is:

piphi/sources/rtl433/packets

The payload should carry detailed device hints in the JSON body so subscribers do not need complex topic parsing just to identify a packet.

One Complete Mini Example

The snippets in this README are useful, but sometimes it helps to see one small working shape in one place.

from fastapi import FastAPI, Request

from piphi_runtime_kit_python import (
    RuntimeConfig,
    build_config_apply_response,
    create_runtime_starter,
    schedule_telemetry_delivery,
    validate_typed_config,
)
from piphi_runtime_kit_python.fastapi import sync_runtime_auth_from_fastapi_payload


class DemoConfig(RuntimeConfig):
    host: str


starter = create_runtime_starter(
    integration_id="demo-runtime",
    integration_name="Demo Runtime",
    version="0.1.0",
)
app = FastAPI()


@app.get("/health")
async def health():
    return starter.health_response()


@app.post("/config")
async def config(payload: DemoConfig, request: Request):
    sync_runtime_auth_from_fastapi_payload(starter.runtime, request, payload)
    typed_payload = validate_typed_config(payload, DemoConfig)
    identity = build_runtime_identity(typed_payload)
    starter.registry.set(
        typed_payload.id,
        {
            **identity,
            "host": typed_payload.host,
        },
    )
    return build_config_apply_response(config_id=identity["config_id"])


@app.post("/telemetry/example")
async def telemetry_example():
    entry = starter.registry.primary_entry()
    if entry is None:
        return {"status": "skipped", "reason": "no configured device"}

    schedule_telemetry_delivery(
        process_state=starter.runtime.process_state,
        telemetry_client=starter.telemetry_client,
        auth_context=starter.runtime.auth,
        config_id=str(entry["config_id"]),
        device_id=str(entry["device_id"]),
        metrics={"temperature_c": 21.4},
        units={"temperature_c": "C"},
    )
    return {"status": "queued"}

That example is not production-ready, but it is enough to show the most important runtime ideas working together.

The Golden Path

If you only read one section, read this one. This is the intended beginner path.

1. Create a starter

Start with create_runtime_starter(...). It gives you one obvious object that already contains the most common pieces:

• shared runtime auth and process state
• an in-memory registry
• a telemetry client
• an event client
• a config sync coordinator

from piphi_runtime_kit_python import create_runtime_starter

starter = create_runtime_starter(
    integration_id="demo-runtime",
    integration_name="Demo Runtime",
    version="0.1.0",
)

runtime = starter.runtime
registry = starter.registry
telemetry = starter.telemetry_client
events = starter.event_client
config_sync = starter.config_sync

For most new integrations, this is the right place to begin.

2. Define a typed config model

Each integration should subclass RuntimeConfig with its own fields.

from piphi_runtime_kit_python import RuntimeConfig


class DemoDeviceConfig(RuntimeConfig):
    host: str
    alias: str | None = None
    poll_interval_seconds: int = 30

Use validate_typed_config(...) when accepting config payloads:

from piphi_runtime_kit_python import validate_typed_config

typed_payload = validate_typed_config(payload, DemoDeviceConfig)

3. Sync auth from each request

PiPhi runtimes receive auth and scope through headers. Your integration should sync that into the runtime context for every relevant request.

Framework-agnostic usage:

runtime.auth.sync_from_headers(request.headers, payload_container_id=payload.container_id)

FastAPI helper usage:

from piphi_runtime_kit_python.fastapi import sync_runtime_auth_from_fastapi_payload

parsed = sync_runtime_auth_from_fastapi_payload(runtime, request, payload)

4. Store active runtime entries in the registry

Use the runtime registry for the in-memory working set of active devices.

registry.set(
    typed_payload.id,
    {
        "device_id": typed_payload.id,
        "config_id": typed_payload.config_id or typed_payload.id,
        "integration_id": typed_payload.integration_id,
        "host": typed_payload.host,
    },
)

This is not the source of truth for configs. PiPhi Core is. The registry is just the runtime's active in-memory working set.

5. Send telemetry and events

For new integrations, prefer the typed device-scoped methods shown in Telemetry vs Events. They infer identity from a runtime registry entry and fail early with an actionable message when required scope is missing.

The lower-level clients remain available when you need direct control:

await starter.telemetry_client.send_metrics(
    auth_context=starter.runtime.auth,
    config_id="core-config-uuid",
    device_id="plug-1",
    metrics={"is_on": True, "current_power_w": 13.2},
)

Or queue lower-level delivery in the background:

from piphi_runtime_kit_python import (
    schedule_event_delivery,
    schedule_telemetry_delivery,
)

schedule_telemetry_delivery(
    process_state=runtime.process_state,
    telemetry_client=telemetry,
    auth_context=runtime.auth,
    config_id="core-config-uuid",
    device_id="plug-1",
    metrics={"is_on": True},
    container_id=runtime.auth.container_id,
)

schedule_event_delivery(
    process_state=runtime.process_state,
    event_client=events,
    auth_context=runtime.auth,
    event_type="device.turned_on",
    device={
        "device_id": "plug-1",
        "config_id": "core-config-uuid",
        "integration_id": "demo-runtime",
    },
    source="demo_runtime",
)

6. Expose the common runtime routes

Most runtimes should provide at least:

• /health
• /diagnostics
• /discover
• /config
• /configs/sync or /config/sync
• /deconfigure
• /events
• /state
• /entities

Some integrations also expose /ui or /ui-config.

7. Use the example app as your checklist

The example app shows the intended flow end to end:

• examples/minimal_fastapi_runtime/app.py
• examples/minimal_fastapi_runtime/README.md

The IDs You Need To Understand

These are the identifiers that show up most often:

• id This is the runtime's local config id in the payload your integration receives.
• config_id This is the real PiPhi Core config UUID.
• device_id This is the physical or logical device identifier used by the runtime.
• container_id This is the runtime/container scope used for Core auth.
• integration_id This is the installed integration id in Core.

The most common beginner mistake is confusing id with config_id.

A safe mental model is:

• id is local to the runtime payload shape
• config_id is the real Core identity for config-backed event flows

If you are sending events back to Core, config_id, container_id, and integration_id need to be correct.

Plain-Language Concepts

If you are new to the platform, these terms can feel more complicated than they really are. Here is the simple version.

• snapshot A snapshot is just PiPhi saying, "Here is the full list of configs you should have right now." You compare that list to what your runtime currently has. Then you add the missing ones and remove the stale ones. Example: await config_sync.apply_snapshot(snapshot=payload, active_config_ids=registry.ids(), apply_config=apply_config, remove_config=remove_config, get_active_config_ids=registry.ids)
• config sync Config sync is the process of making your runtime match the latest snapshot. Think of it like refreshing a shopping list and making sure your cart matches it. Example: typed_configs = validate_typed_configs(snapshot.configs, DemoDeviceConfig)
• registry The registry is the runtime's in-memory notebook. It keeps track of the devices and state that are active right now. Example: registry.set(config.id, {"device_id": config.device_id or config.id, "host": config.host})
• telemetry Telemetry is the stream of measurements, like temperature, humidity, power, or signal strength. Example: schedule_telemetry_delivery(process_state=runtime.process_state, telemetry_client=telemetry, auth_context=runtime.auth, config_id="core-config-uuid", device_id="plug-1", metrics={"temperature_c": 21.4}, units={"temperature_c": "C"})
• event An event is a meaningful thing that happened, like "device configured" or "device turned on." Example: registry.append_event(build_local_event_record(event_type="device.configured", device=entry, payload={"host": entry["host"]}, source="demo-runtime", severity="info"))
• container_id This is the identity of the running runtime process from Core's point of view. It helps Core know which runtime is talking to it. Example: runtime.auth.sync_from_headers(request.headers, payload_container_id=payload.container_id)
• config_id This is the real Core-side id for a config. If a route or event needs the official Core identity, this is the one that matters. Example: entry = {"config_id": payload.config_id or payload.id, "device_id": payload.device_id or payload.id}
• device_id This is the actual device or logical thing you are monitoring or controlling. One config often points at one device, but they are not always the same idea. Example: await telemetry.send_metrics(auth_context=runtime.auth, device_id="plug-1", metrics={"is_on": True})
• starter The starter is the beginner-friendly bundle that gives you the common SDK pieces in one place so you do not have to wire them up one by one. Example: starter = create_runtime_starter(integration_id="demo-runtime", integration_name="Demo Runtime", version="0.1.0")

When To Use Which Helper

Some helpers look similar at first. This is the fast way to choose.

Typical Runtime Flow

Most integrations follow this shape:

1. PiPhi calls your runtime.
2. Your route syncs auth from headers.
3. You validate the config payload into a typed model.
4. You connect to the vendor API or local device.
5. You store the active runtime entry in the registry.
6. You begin polling or listening for updates.
7. You send telemetry to Core.
8. You emit meaningful events back to Core.
9. You expose health and diagnostics so the runtime can be supported.

The SDK is designed to make steps 2, 3, 5, 7, 8, and 9 easier.

Core Usage Patterns

Request auth helpers

The kit includes framework-agnostic helpers for extracting PiPhi runtime auth from request-like header mappings.

from piphi_runtime_kit_python import extract_runtime_auth_headers

parsed = extract_runtime_auth_headers(request.headers)
runtime.auth.sync_from_headers(request.headers, payload_container_id="runtime-123")

For FastAPI integrations, the optional adapter layer trims route boilerplate:

from piphi_runtime_kit_python import format_runtime_auth_sync_log
from piphi_runtime_kit_python.fastapi import sync_runtime_auth_from_fastapi_payload

parsed = sync_runtime_auth_from_fastapi_payload(runtime, request, payload)
logger.info(
    format_runtime_auth_sync_log(
        parsed,
        payload_container_id=payload.container_id,
    )
)

Discovery helpers

Use the discovery helpers to normalize input and return a consistent response:

from piphi_runtime_kit_python import (
    build_discovery_response,
    format_discovery_attempt_log,
    normalize_discovery_inputs,
)

inputs = normalize_discovery_inputs({"username": " user@example.com ", "password": " "})
logger.info(format_discovery_attempt_log(inputs=inputs))
response = build_discovery_response(devices)

Event helpers

Use the event helpers when you want consistent runtime event logging and Core event payload generation:

from piphi_runtime_kit_python import (
    build_core_event_payload,
    build_event_ingest_response,
    format_event_log,
)

logger.info(format_event_log(payload))
event_response = build_event_ingest_response(event)

core_event = build_core_event_payload(
    event_type="device.turned_on",
    integration_id="demo-runtime",
    config_id="core-config-uuid",
    container_id="runtime-123",
    device_id="plug-1",
    payload={"host": "10.0.0.227"},
)

Health and diagnostics helpers

The SDK can build consistent support endpoints:

from piphi_runtime_kit_python import (
    build_runtime_diagnostics_response,
    build_runtime_health_response,
)

health = build_runtime_health_response(
    runtime,
    integration={"id": "demo-runtime", "version": "0.1.0"},
)

diagnostics = build_runtime_diagnostics_response(
    runtime,
    integration={"id": "demo-runtime", "version": "0.1.0"},
    diagnostics={"configured_device_ids": ["plug-1"]},
)

These helpers include:

• pending task counts
• current config generation
• whether runtime auth is present
• whether a shared Core client is bound

Clear Error Handling

The SDK now classifies common delivery failures into PiPhi-specific errors.

Important examples:

• CoreUnavailableError PiPhi Core could not be reached at all.
• CoreTimeoutError PiPhi Core did not respond before the client timeout.
• CoreRouteNotFoundError The expected Core route is not mounted or the URL is wrong.
• CoreAuthError PiPhi Core rejected runtime auth.
• CoreServerError PiPhi Core returned a server-side failure.

This is meant to make runtime logs easier to understand than raw httpx exceptions alone.

Common Mistakes

• Mistake: using id where config_id should be used. Wrong: {"config_id": payload.id} Right: {"config_id": payload.config_id or payload.id} Symptom: Core event delivery may fail or point at the wrong config identity.
• Mistake: forgetting to sync auth from request headers before sending telemetry. Wrong: await telemetry.send_metrics(auth_context=runtime.auth, device_id="plug-1", metrics={"is_on": True}) Right: runtime.auth.sync_from_headers(request.headers, payload_container_id=payload.container_id) Symptom: Core may reject or ignore the request because the runtime context has no valid scope.
• Mistake: sending events without integration_id, config_id, or container_id. Wrong: building event payloads with only device_id Right: include the full device/config/runtime scope whenever the event goes back to Core Symptom: Core event delivery may fail or become ambiguous.
• Mistake: treating the registry as the source of truth instead of Core. Wrong: storing config state only in the registry and assuming that is enough Right: treat Core as the source of truth and the registry as runtime working memory Symptom: config sync and rehydrate flows drift from what Core expects.
• Mistake: putting polling cadence and vendor logic into the SDK layer instead of the integration. Wrong: expecting the SDK to decide vendor polling behavior Right: keep vendor behavior in the integration and use the SDK for runtime plumbing Symptom: the integration becomes harder to reason about and the SDK becomes too magical.

Troubleshooting

Symptom: telemetry times out

Check:

• PiPhi Core is reachable
• the request timeout is long enough for your environment
• Core is not busy or restarting
• your route is using schedule_telemetry_delivery(...) if background delivery is acceptable

Symptom: event delivery returns 404

Check:

• the correct Core route exists
• the runtime is pointing at the right Core base URL
• config_id is the real Core config UUID
• container_id and integration_id are present

Symptom: Core is unavailable

Check:

• PiPhi Core is actually running
• the runtime can reach the host and port
• the SDK error is CoreUnavailableError and not a different failure class

Symptom: config sync removes devices unexpectedly

Check:

• the snapshot really contains the configs you expect
• your registry is storing the right active ids
• your get_active_config_ids callback matches what you actually applied
• you are not mixing local id and real config_id

Symptom: telemetry or events are rejected by Core

Check:

• auth was synced before delivery
• the runtime has a valid container_id
• the device/config scope is complete
• the typed delivery error explains whether this is auth, routing, timeout, or server failure

Example App

The example app is intentionally small, but it is meant to be a real reference:

• examples/minimal_fastapi_runtime/app.py
• examples/minimal_fastapi_runtime/README.md

Summary

If you are unsure where to begin:

1. create a starter
2. define a typed config model
3. sync auth in every route
4. store active entries in the registry
5. send telemetry and events through the SDK
6. compare your runtime to the example app

What belongs in the SDK

The SDK should own PiPhi-specific plumbing:

• runtime auth parsing and outbound Core headers
• config sync orchestration and typed config validation
• telemetry and Core event publishing
• health and diagnostics helpers
• background task tracking
• thin framework adapters

What stays in the integration

The integration should own vendor logic:

• device library calls and protocol handling
• discovery strategy specific to the vendor
• entity modeling and command behavior
• device-specific event semantics
• integration-specific UI schema and config fields

What A Real Integration Still Needs

Even with the SDK, a real integration still needs application code.

You still need to write:

• a vendor client or local device client
• discovery logic that makes sense for that vendor
• entity mapping for the PiPhi frontend and automation model
• polling or subscription logic
• command handling if the device supports actions
• integration-specific config fields and UI schema

The SDK is the runtime foundation, not the whole house.

Versioning and compatibility

See VERSIONING.md for:

• semver policy
• 0.x stability expectations
• Core compatibility guidance
• release checklist notes

Planned direction

The kit is intentionally small. It should cover PiPhi runtime plumbing, not device-specific integration logic.

Good candidates for future additions:

• framework adapters layered on top of the core runtime helpers
• more route-level helpers once the core abstractions stabilize

Included example

See examples/minimal_fastapi_runtime for a small reference runtime that shows how the kit fits together in a real app.