Skip to content

Contract-First Route Design

Every @app.telemetry, @app.command, and @app.device registration is also a contract declaration — a machine-readable description of what a device produces and what it accepts. Adding contract metadata turns main.py into an auditable, declarative interface document that humans and AI coding assistants can inspect without reading implementation code.

The pattern is directly analogous to FastAPI's route decorators: just as @app.get("/items", response_model=Item, summary="List items") declares both the route and its schema, a cosalette registration declares both the MQTT topic wiring and the data contract.

Declaring Contract Metadata

All three registration decorators accept optional contract fields:

Parameter Type Applies to Description
summary str telemetry, command, device Human-readable description
state_model type telemetry, command Pydantic model or dataclass for state
payload_model type command, triggerable telemetry Expected inbound payload type
behavior list[str] telemetry, command, device Ordered description of what the handler does
effects list[str] telemetry, command, device Side effects and mutations

None of these fields have runtime enforcement — they are informational metadata surfaced by the manifest and MCP tools.

Telemetry with Full Metadata

main.py
from pydantic import BaseModel
import cosalette

class SensorReading(BaseModel):
    celsius: float
    humidity: float

class RefreshCommand(BaseModel):
    days: int = 7

@app.telemetry(
    "climate",
    interval=cosalette.setting_ref("poll_interval"),
    triggerable=True,
    summary="Temperature and humidity from the I2C sensor",
    state_model=SensorReading,
    payload_model=RefreshCommand,       # accepted on /set when triggerable
    behavior=["reads I2C bus", "applies PT1 low-pass filter"],
    effects=["updates HA dashboard state"],
)
async def climate(ctx: cosalette.DeviceContext) -> dict[str, object]:
    sensor = ctx.adapter(ClimatePort)
    return {"celsius": sensor.read_temp(), "humidity": sensor.read_rh()}

Command with Full Metadata

main.py
from pydantic import BaseModel

class ValveCommand(BaseModel):
    position: int  # 0–100

class ValveState(BaseModel):
    position: int
    flow_lpm: float

@app.command(
    "valve",
    summary="Opens or closes the irrigation valve",
    payload_model=ValveCommand,
    state_model=ValveState,
    behavior=["validates position range", "logs to audit trail"],
    effects=["mutates valve position", "triggers flow sensor update"],
)
async def handle_valve(
    payload: ValveCommand, ctx: cosalette.DeviceContext
) -> dict[str, object]:
    driver = ctx.adapter(ValvePort)
    await driver.set_position(payload.position)
    return {"position": payload.position, "flow_lpm": await driver.read_flow()}

Device with Metadata

@app.device supports summary, behavior, and effects. It does not accept state_model or payload_model because device handlers manage their own publishing loop rather than returning a typed state snapshot.

main.py
@app.device(
    "receiver",
    summary="Read sensor frames from serial port and publish per-sensor state",
    behavior=[
        "opens serial port at startup",
        "reads LaCrosse protocol frames in a loop",
        "publishes per-sensor state through a sub-entity per discovered sensor",
    ],
    effects=["publishes to {name}/{sensor_id}/state for each discovered sensor"],
)
async def receiver(ctx: cosalette.DeviceContext) -> None:
    port = ctx.adapter(SerialPort)
    async for frame in port.read_frames():
        await ctx.sub_entity(frame.sensor_id).publish_state(frame.to_state())

Inspectable Settings Bindings

Using a raw lambda for interval hides the setting name from the manifest:

# Opaque — manifest shows "<deferred>", tooling cannot resolve the field name
@app.telemetry("sensor", interval=lambda s: s.poll_interval)
async def sensor() -> dict[str, object]: ...

setting_ref("field_name") wraps the same callable but preserves the field name so it appears in the manifest output:

# Inspectable — manifest shows interval: poll_interval (field name)
@app.telemetry("sensor", interval=cosalette.setting_ref("poll_interval"))
async def sensor() -> dict[str, object]: ...

setting_ref also works for enabled:

@app.telemetry(
    "magnetometer",
    interval=cosalette.setting_ref("poll_interval"),
    enabled=cosalette.setting_ref("enable_magnetometer"),
)
async def magnetometer() -> dict[str, object]: ...

The SettingRef type is exported from cosalette — use it for type annotations if you build tooling around the registry snapshot.

The Read/Write Split Pattern

A telemetry registration and a command registration can share the same device name. They use different MQTT topic suffixes (/state vs /set), and the framework creates a shared DeviceContext for both.

This is the canonical way to model a resource with separate read and write paths:

main.py
import cosalette

app = cosalette.App(name="gas2mqtt", version="1.0.0")


@app.telemetry(
    "gas_counter",
    interval=cosalette.setting_ref("poll_interval"),
    triggerable=True,
    summary="Current gas meter impulse count",
    state_model=GasCounterState,
)
async def read_gas_counter(ctx: cosalette.DeviceContext) -> dict[str, object]:
    """Poll impulse count; also fires immediately on /set trigger."""
    meter = ctx.adapter(GasMeterPort)
    return {"impulses": meter.read_impulses()}


@app.command(
    "gas_counter",               # same name — distinct MQTT suffix
    summary="Reset or adjust the impulse counter",
    payload_model=GasCounterCommand,
    state_model=GasCounterState,
    behavior=["validates offset bounds", "writes to non-volatile storage"],
    effects=["mutates persisted counter value"],
)
async def write_gas_counter(
    payload: GasCounterCommand, ctx: cosalette.DeviceContext
) -> dict[str, object]:
    """Accept counter mutations — reset or offset adjustment."""
    meter = ctx.adapter(GasMeterPort)
    await meter.set_offset(payload.offset)
    return {"impulses": meter.read_impulses()}


app.run()

Topic layout for this pair:

Topic Direction Handler
gas2mqtt/gas_counter/state outbound telemetry publishes
gas2mqtt/gas_counter/set inbound command subscribes

Triggerable vs. Read/Write Split

These are different patterns — do not conflate them:

Pattern What it does
triggerable=True on @app.telemetry A message on /set re-fires the read handler immediately — the value returned is still produced by the telemetry function. No mutation.
@app.telemetry + @app.command sharing a name The telemetry handler reads state; the command handler writes state. Different code paths, distinct contracts.

Use triggerable=True when the client wants a fresh reading on demand. Use the read/write split when the client wants to mutate the resource.

Viewing the Manifest

The cosalette manifest command prints the resolved registration surface of an app without running it:

# JSON output — full contract metadata
cosalette manifest myapp.main:app

# Human-readable table
cosalette manifest myapp.main:app --table

JSON output contains one entry per registered device:

{
  "name": "gas_counter",
  "type": "telemetry",
  "interval": "poll_interval",
  "triggerable": true,
  "summary": "Current gas meter impulse count",
  "state_model": "GasCounterState",
  "payload_model": null,
  "behavior": [],
  "effects": []
}

Module-level code runs

cosalette manifest imports the app module to resolve registrations. Any code at module level (outside functions) runs at import time — the same behaviour as cosalette_inspect_app in the MCP server.

When setting_ref() is used, the interval field shows the settings field name ("poll_interval") rather than "<deferred>". Raw lambdas show "<deferred>".

MCP Integration

AI coding assistants that use the cosalette MCP server can call cosalette_manifest to inspect the app's registration surface programmatically:

cosalette_manifest("myapp.main:app")

The tool returns the same JSON structure as cosalette manifest. Use it to answer questions like "what topics does this app subscribe to?" or "what payload does the valve command expect?" without reading implementation code.

See Also