SDK Reference

The plugin SDK is the boundary layer between your plugin code and the FreeSDN backend. It gives you typed, permission-gated access to devices, alerts, event bus, encrypted settings, and outbound HTTP - and blocks everything else.

This page covers every SDK class, every method signature, every limit, and every security constraint in the SDK surface. All behaviour described here is grounded in backend/app/plugins/sdk.py (the authoritative runtime implementation) and the published dev stub in freesdn-sdk/sdk/src/freesdn_sdk/.

---

How the SDK works at runtime

When your plugin class loads, PluginLoader builds a PluginContext for each organisation that has the plugin enabled and attaches it to the plugin instance as self.ctx. You reach every SDK object through that context:

self.ctx.devices    # DeviceSDK
self.ctx.sites      # SiteSDK
self.ctx.alerts     # AlertSDK
self.ctx.events     # EventSDK
self.ctx.settings   # PluginSettingsSDK
self.ctx.http       # PluginHTTPClient
self.ctx.logger     # logging.Logger named freesdn.plugin.<your-id>

Two codebases, one plugin binary. The published freesdn-sdk package (pip install freesdn-sdk) ships stub classes that raise NotImplementedError outside the runtime. At load time the runtime replaces every stub with the real implementation via sdk_alias.py. Your plugin code imports from freesdn_sdk and works unchanged in both environments.

The published SDK now matches the runtime

freesdn-sdk validate used to be looser than the runtime in two ways that only showed up at install time: it accepted loose dependency specifiers such as >=2.28 where the runtime demands exact == pins, and it silently dropped public_routes. Both are fixed. A manifest that passes validate is now validated by the same rules the runtime applies.

tests/test_sdk_parity.py in the SDK feeds a corpus of manifests to both validators and asserts they return the same verdict, so the two cannot drift apart again without a test failing and naming the rule.

The SDK still omits register_device, which is documented as runtime-only.

---

PluginContext

sdk.py:745-774 - a dataclass, not a class you instantiate. The loader builds it and attaches it to your plugin.

Field	Type	Purpose
plugin_id	str	Your plugin’s declared id from plugin.yaml.
organization_id	UUID	The organisation this runtime instance serves.
devices	DeviceSDK	Device read + inventory-write access.
alerts	AlertSDK	Alert read/create/resolve access.
metrics	MetricsSDK	Read-only time-series: device status history and anything else the platform records.
sites	SiteSDK	Site read access, subject to the caller’s site grants.
events	EventSDK	Event bus emit + subscription validation.
settings	PluginSettingsSDK	Per-org key/value store with encrypted secret support.
http	PluginHTTPClient	SSRF-protected outbound HTTP.
logger	logging.Logger	Named freesdn.plugin.<plugin_id>. Use this instead of print.

self.ctx is available after on_start completes. If you access it before that point you get None.

---

Permission model

Every SDK method that touches platform data requires a capability declared in your plugin.yaml permissions list. Two checks happen on every call (sdk.py:93-131):

1. Manifest check. The capability (e.g. devices.read) must appear in permissions[].code in your manifest. If it does not, the call raises PermissionError immediately.
2. Confused-deputy check (CAN-015). If an authenticated user triggered the call by hitting one of your REST endpoints, the plugin may only exercise capabilities that the calling user could exercise directly. A viewer-role user hitting your endpoint cannot cause the plugin to write alerts even if the plugin declared alerts.write. This check does not apply on public/HMAC routes, automation triggers, AI tool calls, or scheduled work - in those contexts the plugin acts with its own declared authority only.

The capability-to-core-permission map:

Capability code	Core permission required
devices.read	device:read
devices.register	device:write
devices.stage	device:write, device:update or config:push
sites.read	site:read
alerts.read	alert:read
metrics.read	analytics:read
alerts.write	alert:create or alert:update

Declare every capability you use. Under-declaring causes runtime PermissionError; over-declaring exposes authority you do not use.

---

DeviceSDK

sdk.py:134-369 - reached as self.ctx.devices.

list

async def list(
    status: str | None = None,
    site_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[dict]

Permission: devices.read

One page of devices scoped to the plugin’s organisation. Results include id, name, type, status, ip, mac, site_id and controller_id.

limit is clamped to 1-500 regardless of what you pass, and results are ordered by id so offset paging is stable.

Do not use list() to scan an estate

Because limit is clamped, list(limit=2000) returns 500 devices and no indication there are more. A drift report or an inventory sync built on that is wrong above 500 devices and correct below it, which is the hardest kind of bug to notice: it passes every test on a lab estate and is silently wrong in production.

Use iter_all() for a full scan.

iter_all

async def iter_all(
    status: str | None = None,
    site_id: str | None = None,
    page_size: int = 500,
) -> AsyncIterator[dict]

Permission: devices.read

Every device the plugin can see, a page at a time:

async for device in self.ctx.devices.iter_all():
    ...

# or collect them, if the estate is small enough to hold in memory
devices = [d async for d in self.ctx.devices.iter_all()]

Raises RuntimeError rather than truncating if the inventory exceeds 50,000 devices - a ceiling that silently stops is the same bug one level up. Narrow with status= or site_id= if you hit it.

SiteSDK has the same pair: list(limit, offset) and iter_all(page_size).

Passing status filters by device status string. Passing site_id restricts to that site; the site must belong to the plugin’s organisation or the result set is empty.

get

async def get(device_id: str) -> dict | None

Permission: devices.read

Returns a single device dict. Adds model and firmware fields on top of the list shape. Returns None if the device does not exist or does not belong to the plugin’s organisation. Import failures in the underlying model degrade to None rather than raising.

register_device

async def register_device(device_data: dict) -> dict

Permission: devices.register

Runtime-only - not in the published stub

register_device is absent from the DeviceSDK stub in freesdn-sdk. It exists only inside the FreeSDN runtime. You cannot call or test it through the published dev package; use create_test_context and mock the call in unit tests.

Upserts a device into the core inventory. The device_data dict must satisfy these constraints:

Field	Rule
external_id	Must start with plugin.<your-plugin-id>:. Rejected otherwise.
name	Human-readable label. Truncated to 255 characters. Defaults to "Unknown" if omitted.
device_type	Device category string (e.g. "switch", "camera"). Defaults to "other".
ip_address	Must be a public routable IP. Internal / loopback / link-local / RFC-1918 addresses are SSRF-blocked. Omit the field to skip the IP check.
site_id	Must exist and belong to the plugin’s organisation. Required.
manufacturer	Optional vendor string.
model	Optional model string.
firmware_version	Optional firmware version string.
mac_address	Optional MAC address string.
serial_number	Optional serial number string.
status	Device status string. Defaults to "unknown".
metadata	Validated for size and nesting depth.

Additional hard limits:

• 1,000 devices per plugin. Attempting to register beyond this cap raises an error.
• Writes an audit row with actor_type="plugin".
• Uses DeviceSyncService.upsert_single atomically.

Returns {id, external_id, name} on success.

get_ports

async def get_ports(device_id: str) -> list[dict]

Permission: devices.read

Returns the port list for a device. Each entry contains id, name, port_number, status, speed, and poe_enabled. The device must belong to the plugin’s organisation; an empty list is returned otherwise.

stage_change

async def stage_change(
    *,
    controller_id: UUID,
    feature: str,
    operation: str,          # "create" | "update" | "delete"
    payload: dict,
    site_id: UUID | None = None,
    target_id: str | None = None,
    notes: str | None = None,
) -> dict

Permission: devices.stage

Proposes a configuration change. It records intent in the staging table and returns the staged row (id, status, feature, operation, site_id). It does not touch a device.

This is the capability that lifts a plugin above being an observer, and the reason it is safe to grant is that it does not shorten the path to a device by one step:

• the change reaches hardware only when an operator applies it, behind ADAPTER_READ_ONLY=false and an explicit per-request apply;
• there is deliberately no SDK method that can apply one, and a test fails the build if a method that could ever appears;
• the per-user site grant, the controller-containment rule and the create|update|delete whitelist are enforced inside the shared chokepoint, not re-implemented here;
• the staged row’s notes always begin with [plugin:<your-id>]. The prefix is prepended and cannot be substituted or spoofed, so the operator approving the change sees where the proposal came from. Your own note is appended after it.

staged = await self.ctx.devices.stage_change(
    controller_id=controller.id,
    feature="omada.vlan",
    operation="create",
    payload={"name": "guest", "vlan_id": 40},
    site_id=site.id,
    notes="guest VLAN missing at this site",
)
# staged["status"] == "pending" - an operator applies it, not you

---

SiteSDK

Reached as self.ctx.sites.

Every device carries a site_id and nothing else about the site, so without this the only options were showing an operator a raw UUID or keeping a private copy of the site list that goes stale.

Scoping matches DeviceSDK exactly: the plugin’s organisation first, then the bound caller’s per-user site grants. A site-limited user driving a plugin route sees the sites they are granted and no others, exactly as they would through the product.

list

async def list(limit: int = 100) -> list[dict]

Permission: sites.read

Sites in the plugin’s organisation that the caller may see. Each entry contains id, name, description, address, city, country, timezone, and is_active.

get

async def get(site_id: UUID) -> dict | None

Permission: sites.read

One site, or None. It returns None both for a site that does not exist and for one the caller may not see, so a plugin cannot be used to probe whether a site exists outside the caller’s grants.

---

AlertSDK

sdk.py:372-507 - reached as self.ctx.alerts.

list

async def list(
    severity: str | None = None,
    limit: int = 50,
) -> list[dict]

Permission: alerts.read

Returns alerts scoped to the plugin’s organisation. Each entry contains id, title, message, severity, status, and device_id. The limit argument is clamped to 1-200. Pass severity to filter by severity string.

create

async def create(
    title: str,
    message: str,
    severity: str = "warning",
    device_id: str | None = None,
) -> dict

Permission: alerts.write

Creates a new alert in the platform. Rules:

• severity must be one of info, warning, error, or critical. Any other value raises ValueError before touching the database.
• title is truncated to 200 characters; message is truncated to 2,000 characters.
• The runtime automatically creates or reuses a system AlertRule named __plugin_alerts_<plugin_id> to own the alert. You do not need to create a rule manually.
• Deduplication fingerprint: sha256("plugin:<id>:<title>:<severity>")[:64]. Alerts with the same title and severity from the same plugin fingerprint to the same slot.

Returns {id, title, severity}.

resolve

async def resolve(alert_id: str, resolution: str = "") -> None

Permission: alerts.write

Marks an alert as resolved and records the optional resolution string (truncated to 2,000 characters if present). The alert must belong to the plugin’s organisation; the call is a no-op if the ID does not exist within scope.

---

MetricsSDK

Reached as self.ctx.metrics. Requires the metrics.read capability. Reads the time-series FreeSDN already records, so a plugin can ask questions about time rather than only about right now - has this AP been flapping, is that uplink degrading, which site got worse this week.

Scoped by site, not by organisation

This is deliberate, and the reason is worth knowing because it affects what you get back.

metric_data carries organization_id, site_id and device_id. The collector recorded the last two and left the first NULL on every row ever written - and the platform’s own /analytics/metrics/query filters on organization_id for tenant isolation, so the filter was correct and matched nothing. Twenty-five thousand rows in the hypertable and an empty result for every operator, always. The collector is fixed, but rows written before the fix still have NULL.

Scoping this SDK by site is correct for old and new rows alike, and it composes with per-user site grants for free: a site-limited operator driving your plugin route sees the series for their sites and no others.

available

async def available() -> bool

Is the time-series tier configured on this deployment?

Call it before reporting “no data”. On a lite deployment with no LogDB, empty and absent look identical from inside a plugin, and telling an operator “no metrics in the last 24 hours” when the database was never configured is a wrong answer delivered confidently.

metric_names

async def metric_names(*, hours: float = 24.0) -> list[str]

What is actually being recorded, for the sites this caller can see. Guessing a metric name gets you an empty list that looks exactly like “nothing happened”; this is how you discover what exists instead of hard-coding a name a future release renames.

query

async def query(
    metric_name: str,
    *,
    hours: float = 24.0,
    device_id: UUID | None = None,
    site_id: UUID | None = None,
    limit: int = 1000,
) -> list[dict]

Raw points, newest first. Each is {time, metric_name, value, labels, site_id, device_id}.

device_id and site_id narrow, they never widen. Your site scope still applies, so a device or site id handed to you by a caller returns nothing rather than another tenant’s data - which means a route that accepts a site parameter cannot be turned into a cross-tenant read.

latest

async def latest(metric_name: str, *, device_id: UUID | None = None,
                 hours: float = 24.0) -> dict | None

The most recent point, or None. None rather than an exception because “this device has no recent data” is an ordinary answer, and a plugin that wraps every read in a try block writes worse code.

series

async def series(
    metric_name: str,
    *,
    hours: float = 24.0,
    bucket_minutes: int = 60,
    device_id: UUID | None = None,
    site_id: UUID | None = None,
    aggregation: str = "avg",   # avg | min | max | sum | count
) -> list[dict]

Bucketed points, oldest first: {time, value, samples}. This is what you want for a chart or a trend.

Use it rather than reducing raw points yourself. A week of per-minute data for one site is already past the row cap, so query would answer with a truncated slice - right for the last few hours, wrong for the week you asked for, and nothing in the result to tell you which. series aggregates in the database.

aggregation is an allowlist; anything else raises ValueError.

# "which of my sites got worse this week?"
week = await self.ctx.metrics.series(
    "device.status", hours=24 * 7, bucket_minutes=360, aggregation="avg"
)
if week and week[-1]["value"] < week[0]["value"] - 0.05:
    await self.ctx.alerts.create(
        title="Device availability is trending down",
        message=f"{week[0]['value']:.1%} -> {week[-1]['value']:.1%} over seven days",
        severity="warning",
    )

Limits

Limit	Value	Why
MAX_POINTS	5,000 rows per call	A metrics table is the largest thing a plugin can ask for by a wide margin. An unbounded query over a hypertable is how one plugin takes the API process down with it.
MAX_RANGE_DAYS	90 days	Same reason. Clamped, not rejected: asking for more history than exists should give you what there is.
bucket_minutes	1 to 1440

Both are clamped silently rather than raised, because a plugin asking for too much should still work.

No writes

There is no metrics.write. A plugin cannot record metrics, only read them. Emit an event or raise an alert instead - both are first-class, both are attributable to your plugin, and neither can corrupt the series the rest of the platform charts.

---

EventSDK

sdk.py:510-559 - reached as self.ctx.events. EventSDK has no capability code; it is gated by the event_subscriptions list in your manifest and by automatic namespace enforcement.

emit

async def emit(event_type: str, payload: dict) -> None

Publishes an event on the platform event bus. The full event type is always plugin.<plugin_id>.<event_type> - the prefix is added by the runtime regardless of what you pass. You cannot emit a bare core event type; the namespace is enforced to prevent spoofing.

The event carries source="plugin:<plugin_id>" and the bound organization_id.

subscribe

def subscribe(event_pattern: str) -> None

Synchronous validation call. Confirms that event_pattern is declared in your manifest’s event_subscriptions list. Raises PermissionError if it is not. Bare wildcard patterns (*, #) are rejected.

You do not need to call this directly in most cases. The loader calls bind_event_subscriptions() automatically after on_start, which subscribes every declared pattern and wires it to your on_event handler, filtered to your organisation’s events.

Org isolation

Event subscriptions are scoped by organization_id. Your on_event handler only receives events from the organisation whose runtime is running.

---

PluginSettingsSDK

sdk.py:562-631 - reached as self.ctx.settings. Settings are scoped per (plugin_id, organization_id, key) in the core.plugin_settings table.

get

async def get(key: str, default=None) -> Any

Reads the JSONB value for key. Returns default if the key does not exist.

set

async def set(key: str, value: Any) -> None

Upserts a JSONB value. The total settings blob for a plugin/org pair is capped at 32 KiB by the management API (enforced on PUT /plugins/{id}/settings).

get_secret

async def get_secret(key: str) -> str | None

Reads the encrypted value stored at {key}:encrypted and returns the decrypted plaintext. Returns None if the key does not exist. Encryption is Fernet (AES-128-CBC + HMAC-SHA256). The key is derived from SECRET_KEY via PBKDF2-HMAC-SHA256 (260 000 iterations). This is a separate key class from the ENCRYPTION_SALT-based credentials used for device and controller passwords; rotating ENCRYPTION_SALT does not re-key plugin secrets stored with set_secret.

set_secret

async def set_secret(key: str, value: str) -> None

Encrypts value and stores it at {key}:encrypted. Use this for API keys, webhook secrets, credentials, or any value that should not appear in plaintext in the database.

Use secrets for credentials

Store any third-party API key or credential with set_secret / get_secret, not set / get. Values stored with set are JSONB and readable by anyone with database access. Fernet encryption means secrets at rest are protected by the platform key material.

---

PluginHTTPClient

sdk.py:634-737 - reached as self.ctx.http. Every outbound HTTP call from a plugin must go through this client. Direct use of socket, http, urllib, and webbrowser (and their submodules) is blocked at load time by the import hygiene layer. requests, httpx, and aiohttp are not individually listed in BLOCKED_MODULES; if those packages are already loaded by the backend process they remain accessible. Always use self.ctx.http for outbound calls - it is the only path with SSRF protection.

All four methods share identical signatures and all funnel through a single internal _request method:

async def get(url: str, **kwargs) -> httpx.Response
async def post(url: str, **kwargs) -> httpx.Response
async def put(url: str, **kwargs) -> httpx.Response
async def delete(url: str, **kwargs) -> httpx.Response

What _request enforces

Timeout cap. The constructor timeout argument is capped at MAX_TIMEOUT = 60.0 seconds. You cannot exceed this regardless of what you pass in kwargs.

Response size cap. The body is read and its length checked after the request completes. If the response body exceeds MAX_RESPONSE_SIZE = 10 MB, the call raises ValueError. You cannot stream a large payload around this limit.

kwargs allowlist. Only these keyword arguments are passed to the underlying request:

Allowed kwarg	Purpose
json	JSON request body
data	Form or raw body
params	Query string parameters
content	Raw bytes body
cookies	Request cookies
headers	Custom request headers (blocklist-filtered; see below)

All other keyword arguments - including anything that could inject a custom transport, proxy, auth handler, or SSL context - are silently dropped. headers is handled separately: it is extracted outside the allowlist filter, each header name is checked against the blocklist below, and the surviving headers are injected into the request alongside the forced User-Agent.

Header blocklist. The following headers are stripped from any headers dict you supply:

Authorization, Host, X-Forwarded-For, X-Forwarded-Host, X-Forwarded-Proto, X-Real-IP, Proxy-Authorization, Cookie, Set-Cookie, Transfer-Encoding

The User-Agent is forced to FreeSDN-Plugin/<plugin_id>/<version> regardless of what you set.

SSRF protection. The actual request is made through app.core.security_utils.safe_http_request, which:

• Resolves the hostname to an IP once.
• Validates the resolved IP is not private, loopback, link-local, CGNAT, or IPv4-mapped private.
• Pins the connection to that IP for the life of the request (defeats DNS-rebind TOCTOU).
• Disables redirect following entirely (follow_redirects=False).

There is no way to reach an internal platform address through PluginHTTPClient.

Do not try to bypass the HTTP client

The import hygiene layer blocks socket (and _socket), http (and submodules), urllib (and submodules), and webbrowser at load time. A plugin that tries to import any of these modules fails to load.

Higher-level HTTP libraries - httpx, requests, aiohttp - are not individually blocked. They are already in sys.modules as backend dependencies, so a plugin can technically name them. However, self.ctx.http is the only path that carries SSRF protection (RFC-1918 blocklist, DNS-rebind pin, redirect disable, timeout cap). Using a raw HTTP library bypasses all of those protections and is a policy violation that auditors will flag.

---

PluginCoordinator

One shared fetch for everything in your plugin that needs the same remote data.

Modelled on Home Assistant’s DataUpdateCoordinator, and here for the reason it is there: without it every consumer polls independently, and four shipped sync plugins each hand-rolled the same retry loop, the same transient-versus-permanent classification and the same “did the last fetch work” bookkeeping.

from freesdn_sdk import PluginCoordinator

def _coordinator(self) -> PluginCoordinator:
    if self._devices is None:
        self._devices = PluginCoordinator(
            self.ctx, "librenms-devices", self._fetch_devices, update_interval=3600,
        )
    return self._devices

devices = await self._coordinator().async_get()

Member	Purpose
async_get()	Cached data, fetching only if stale. Concurrent callers share one fetch.
async_refresh(force=False)	Refresh now. Returns a bool; it does not raise.
data	Last good data, kept through a failed refresh.
last_update_success	Whether the most recent fetch worked.
snapshot()	{name, last_update_success, age_seconds, stale, consecutive_failures, last_error} for a status route.

What it gives you beyond a cache: concurrent callers share one in-flight request; a failed refresh keeps the last good data rather than blanking a page; and repeated failure backs off, because a coordinator on a 30-second schedule would otherwise retry a dead remote every 30 seconds for ever.

PluginNotReady is the one exception it re-raises, so the supervisor still sees it.

---

PluginNotReady

Say “the remote is down”, not “I am broken”.

from freesdn_sdk import PluginNotReady

ok, body = await self._request("get", url, token, "/devices")
if not ok:
    raise PluginNotReady(f"NetBox at {url} is not reachable: {body}")

Supervision quarantines a plugin that fails repeatedly, which is right for code that crashes and wrong for a NetBox that is down for ten minutes. PluginNotReady is recorded, surfaced and retried with its own gentler backoff, and never counts toward quarantine. Anything else you raise is still treated as a fault - the right default, since a plugin that does not know why it failed should not get to claim it was somebody else’s problem.

Pass retry_after=120 when the remote told you when to come back. A plugin route that hits it answers 503 with Retry-After rather than 500.

---

async_validate_config

Check the settings when they are saved, not six hours later.

async def async_validate_config(self) -> str | None:
    config = await self._config()
    if config is None:
        return "netbox_url and netbox_token are both required"
    try:
        await self._get(*config, "/dcim/sites/", limit=1)
    except Exception as exc:
        return f"NetBox did not accept the request: {exc}"
    return None

Return None when everything works, or a short sentence naming what is wrong. The result comes back on the save and is shown next to the form.

settings_schema validates the SHAPE of what an operator typed; it cannot tell them the token is wrong. Without this they see “Saved” and learn the truth from the first scheduled run - or never.

Worth reporting the connected but unusable case too, which is the one most likely to waste an afternoon:

if not await self.ctx.settings.get("host_group_id"):
    return "Connected. host_group_id is still required before a sync can run."

Advisory, not blocking

The settings are committed before your validator runs, and a validator that raises, hangs or times out is logged and ignored. An operator half-way through configuring a plugin must not be blocked by a remote that happens to be down, and a plugin whose validator is wrong must not be able to lock its own configuration.

See Lifecycle and Resilience for how these three fit together.

---

FreeSDNPlugin base class and lifecycle

sdk.py:782-1152 - the class your plugin must extend. Override these methods; do not override manifest.

Method	Signature	When called
on_install(db)	async	Once, on first install. Create database tables or seed data. Exceptions are logged but non-fatal - the install proceeds.
on_start(organization_id, db=None)	async	Per-organisation, every time the plugin is enabled or the backend starts. You MUST call await super().on_start(...) to initialise self.ctx.
on_upgrade(from_version, db)	async	After re-install on upgrade. Run schema migrations here.
on_uninstall(db)	async	Just before the plugin directory is removed. Clean up any database rows, files, or external registrations you created.
on_event(event)	async	Called for each event matching your declared event_subscriptions. Default implementation logs and returns. Override to react.
on_stop(organization_id, db=None)	async	Per-organisation stop. Unbinds event subscriptions and clears ctx.
get_router()	→ APIRouter	Once per load event - at first install and again at every server startup. Return an APIRouter to expose REST endpoints. Default returns an empty router.
get_models()	→ list[type]	Return a list of SQLAlchemy model classes your plugin defines. Default returns [].
health_check()	async → dict	Returns {"status": "ok", "organization_id": ...} when active, {"status": "inactive"} otherwise. Surfaced by GET /api/v1/plugins/{id}/health.

Minimal implementation

from freesdn_sdk import FreeSDNPlugin, PluginContext
from fastapi import APIRouter, Depends

class MyPlugin(FreeSDNPlugin):

    async def on_start(self, organization_id, db=None):
        await super().on_start(organization_id, db)
        # self.ctx is now available
        self.ctx.logger.info("started for org %s", organization_id)

    def get_router(self) -> APIRouter:
        router = APIRouter()

        @router.get("/status")
        async def status():
            return {"plugin": self.ctx.plugin_id}

        return router

Emitting events and registering with automation

Call these helpers from on_start:

async def on_start(self, organization_id, db=None):
    await super().on_start(organization_id, db)

    # Emit a namespaced event (becomes plugin.<id>.device_offline)
    # await self.ctx.events.emit("device_offline", {"device_id": "..."})

    # Register an automation trigger
    self.register_automation_trigger(
        trigger_type="device_offline",
        description="Fires when a monitored device goes offline",
        schema={"type": "object", "properties": {"device_id": {"type": "string"}}},
    )

    # Register an automation action
    self.register_automation_action(
        action_type="send_notification",
        handler=self._send_notification,
        description="Send a notification via the plugin",
        params_schema={"type": "object", "properties": {"message": {"type": "string"}}},
    )

    # Register an AI tool (permission required  -  see PS-11 below)
    self.register_ai_tool(
        name="device_summary",
        description="Summarise device status for an organisation",
        parameters={"type": "object", "properties": {}},
        handler=self._ai_device_summary,
        permission="device:read",
    )

async def _send_notification(self, params: dict) -> dict:
    return {"ok": True}

async def _ai_device_summary(self, user, db, **kwargs) -> dict:
    return {"summary": "all devices nominal"}

---

Automation and AI bridges

register_automation_trigger

def register_automation_trigger(
    trigger_type: str,
    description: str,
    schema: dict,
) -> None

Registers a trigger in the platform automation engine as plugin.<plugin_id>.<trigger_type>. The trigger fires when your plugin emits a matching event via self.ctx.events.emit.

Limits:

• trigger_type must match ^[a-z0-9][a-z0-9_-]{0,98}[a-z0-9]$.
• Maximum 50 triggers per plugin. Exceeding the cap raises an error.
• description is truncated to 500 characters.
• Registering a duplicate full type is silently idempotent.

register_automation_action

def register_automation_action(
    action_type: str,
    handler: Callable,  # async (params: dict) -> dict
    description: str,
    params_schema: dict,
) -> None

Registers an action handler as plugin.<plugin_id>.<action_type>. The runtime wires handler into the automation engine’s action dispatch table. A plugin cannot overwrite an action key registered by another plugin - the second registration is silently dropped.

Limits:

• Same name regex as triggers.
• Maximum 50 actions per plugin.

register_ai_tool

def register_ai_tool(
    name: str,
    description: str,
    parameters: dict,  # JSON Schema
    handler: Callable,  # async (user, db, **kwargs) -> dict
    permission: str | None = None,
) -> None

Registers an AI tool available to the platform’s AI Assistant. The tool name is auto-prefixed to plugin_<plugin_id>_<name>.

PS-11: always declare a permission

If permission is None or empty, the runtime coerces it to the sentinel plugin:undeclared_ai_tool, which no role is granted. The tool becomes effectively super_admin-only. This is intentional and fail-closed - an ungated AI tool is a privilege-escalation path.

Always pass a real permission string (e.g. "device:read"). The AI service double-checks at execution time and returns “Permission denied” for any plugin_* tool with no declared permission.

AI-tool handlers receive the calling user’s raw user and db objects, not the org-scoped PluginContext. Your handler must not assume it has plugin-SDK authority; it has whatever the calling user is allowed to do.

Limits:

• Maximum 20 tools per plugin.
• Handler return value is capped at 256 KB serialized. Larger results are replaced with {"error": "Plugin tool result too large", "truncated": true}.
• Non-serializable returns are replaced with an error dict.
• A plugin cannot overwrite an existing built-in or other plugin’s tool of the same prefixed name.

---

Hard limits reference

These limits are enforced by the runtime regardless of what your plugin requests. They are also exported as PLUGIN_LIMITS from the published SDK (types.py).

Limit	Value
ZIP archive (compressed)	50 MB
ZIP archive (uncompressed)	200 MB
python_dependencies entries	50
Automation triggers per plugin	50
Automation actions per plugin	50
AI tools per plugin	20
AI tool result size	256 KB
Devices registered per plugin	1,000
HTTP request timeout	60 seconds
HTTP response body	10 MB
Settings blob (management API)	32 KiB
HMAC timestamp skew (public routes)	300 seconds
HMAC nonce length	16-128 characters

---

Security model - what the SDK is and is not

The plugin loader is not a security sandbox

backend/app/plugins/sandbox.py performs load-time import hygiene. It is explicitly not a security boundary. A loaded plugin runs in the same Python process as the backend with full interpreter access. A hostile plugin can escape via ().__class__.__base__.__subclasses__() or gc.get_objects().

The threat model is cooperative plugins from reviewed, trusted authors. Installing a plugin is equivalent to installing a Python package directly into your production backend. Treat it with the same scrutiny.

Real subprocess isolation is not available in this release. Do not frame the hygiene layer as a sandbox to your users.

What the hygiene layer does enforce at load time:

• Blocked module imports. The following categories are blocked via a MetaPathFinder active during plugin exec: os/filesystem (os, sys, shutil, pathlib, io, tempfile), process execution (subprocess, asyncio.subprocess), raw network (socket, http*, urllib*, webbrowser), dynamic loading (importlib*, pkgutil, runpy), FFI (ctypes, cffi), serialization (pickle, marshal, shelve), concurrency (multiprocessing, threading), introspection (inspect, gc), and others. Attempting to import any blocked module at load time causes the plugin to fail to load.
• Restricted builtins. exec, eval, compile, open, and bare __import__ are replaced with restricted versions or removed from the plugin module’s builtins.

Install-time isolation (separate venv). If your plugin declares python_dependencies, the install pipeline (_install_python_deps in loader.py) creates a per-plugin .venv and installs the hash-pinned deps there. When the plugin is later loaded, _load_plugin_class appends the venv’s site-packages to sys.path (never prepends), so plugin deps never shadow core packages. This is handled by the install pipeline and the plugin loader - it is not part of sandbox.py or the load-time hygiene layer.

What is NOT blocked:

• Already-cached third-party modules that were imported before the plugin loaded.
• Introspection via ().__class__...__subclasses__().
• traceback module.

Defense-in-depth layers that do hold independently:

• SSRF protection in PluginHTTPClient and register_device.
• CAN-015 confused-deputy authority intersection on authenticated routes.
• PS-11 AI-tool permission fail-closed.
• ==-only dependency pinning plus hash-pinned lockfile.
• Marketplace Ed25519 catalog signing (see Marketplace and the env var reference below).
• All install / uninstall / enable / disable / upgrade / secret-rotate operations are audited with tag ["plugin", "supply-chain"].

---

Environment variables

These variables affect plugin behaviour at runtime. They are read from the environment at startup and are not in the web UI.

Variable	Default	Purpose
PLUGIN_DIR	/data/plugins	Directory where plugin files are stored.
PLUGIN_ENABLE_DIRECT_URL_INSTALLS	false	Enable POST /api/v1/plugins/install-url. Off by default.
PLUGIN_ALLOWED_DOMAINS	(empty - blocks all URL installs)	Comma-separated list of hostnames allowed for URL installs. Empty means all URL installs are blocked even when PLUGIN_ENABLE_DIRECT_URL_INSTALLS=true.
PLUGIN_ALLOW_RUNTIME_PYTHON_DEPS	false	Allow plugins that declare python_dependencies to install them at install time. Off by default.
PLUGIN_PYPI_INDEX_URL	https://pypi.org/simple/	PyPI index used for hash-pinned dependency installs.
MARKETPLACE_REGISTRY_URL	https://registry.freesdn.org/plugins.json	Remote catalog URL used by POST /api/v1/marketplace/plugins/sync. Ed25519-signed.
MARKETPLACE_PUBLISHER_PUBLIC_KEY	(empty)	Hex Ed25519 public key. When set, the catalog must carry a valid signature or sync is refused.
MARKETPLACE_ALLOW_UNSIGNED	false	When true and no publisher key is set, unsigned catalogs are accepted with a loud warning. When neither this nor a key is set (the default), unsigned sync returns 403.

---

Management API - quick reference

These endpoints manage plugin lifecycle. Full detail is in backend/app/api/v1/endpoints/plugins.py.

Plugin management (/api/v1/plugins)

Method	Path	Purpose	Minimum role
GET	/api/v1/plugins	List installed plugins with org-effective status	org_admin
GET	/api/v1/plugins/{plugin_id}	Plugin detail including cached manifest	org_admin
POST	/api/v1/plugins/install	Install from uploaded ZIP (201)	super_admin
POST	/api/v1/plugins/install-url	Install by URL (requires env gates)	super_admin
DELETE	/api/v1/plugins/{plugin_id}	Uninstall; removes files (204)	super_admin
POST	/api/v1/plugins/{plugin_id}/enable	Enable globally (super_admin) or per-org	org_admin
POST	/api/v1/plugins/{plugin_id}/disable	Disable globally or per-org	org_admin
POST	/api/v1/plugins/{plugin_id}/upgrade	Upgrade via new ZIP; restarts everywhere	super_admin
GET	/api/v1/plugins/{plugin_id}/settings	Read org-scoped settings map	org_admin
PUT	/api/v1/plugins/{plugin_id}/settings	Upsert org-scoped settings (≤32 KiB)	org_admin
GET	/api/v1/plugins/{plugin_id}/health	Runtime health for caller’s org	org_admin
GET	/api/v1/plugins/{plugin_id}/public-auth	Public-route HMAC auth status	org_admin
POST	/api/v1/plugins/{plugin_id}/public-auth/rotate-secret	Rotate org HMAC secret; returns plaintext once	org_admin

Marketplace (/api/v1/marketplace/plugins)

Method	Path	Purpose	Auth
GET	/api/v1/marketplace/plugins	Browse published plugins (paginated)	public
GET	/api/v1/marketplace/plugins/featured	Up to 6 featured plugins	public
GET	/api/v1/marketplace/plugins/categories	Category list with counts	public
POST	/api/v1/marketplace/plugins/{slug}/upgrade	Update an installed plugin to the catalog version	super_admin
GET	/api/v1/marketplace/plugins/{slug}	Plugin detail by slug	public
GET	/api/v1/marketplace/plugins/{slug}/versions	Version history	public
POST	/api/v1/marketplace/plugins/{slug}/install	Download, verify SHA-256, install (201)	super_admin
GET	/api/v1/marketplace/plugins/{slug}/reviews	Paginated reviews	public
POST	/api/v1/marketplace/plugins/{slug}/reviews	Submit a review (one per user)	active user
POST	/api/v1/marketplace/plugins/sync	Sync catalog from registry	super_admin

Enable / disable semantics

A globally disabled plugin (super_admin, no org context) sets InstalledPlugin.is_active = false. Per-org disable writes a PluginOrganizationState row with is_enabled = false; deleting that row re-enables. A globally disabled plugin returns 409 if you attempt an org-level enable. Both paths are wrapped in the per-plugin lifecycle lock.

When a plugin is disabled its REST routes remain registered in the FastAPI router (routes cannot be removed at runtime without a restart), but the route guard returns 410 Gone for all requests to disabled plugin routes.

---

Public routes and inbound webhooks

If your plugin needs to receive unauthenticated inbound HTTP (webhooks from a third-party service, for example), declare the routes in plugin.yaml under public_routes and authenticate them with the platform’s HMAC scheme.

Required request headers:

Header	Value
X-FreeSDN-Plugin-Org	The organisation UUID
X-FreeSDN-Plugin-Timestamp	Unix timestamp (seconds, string)
X-FreeSDN-Plugin-Nonce	Random string, 16-128 characters
X-FreeSDN-Plugin-Signature	sha256=<HMAC-SHA256 hex>

The canonical message signed is the newline-joined concatenation of: timestamp, nonce, METHOD, path, query string, org_id, and sha256(body).

Replay protection runs via Redis SET NX EX 300 keyed on plugin:public:nonce:{plugin}:{org}:{nonce}. A replayed nonce returns 401. If Redis is unavailable the check fails closed with 503.

Timestamp skew is limited to ±300 seconds. The comparison is constant-time.

Rotate the per-org HMAC secret with POST /api/v1/plugins/{plugin_id}/public-auth/rotate-secret. The plaintext secret is returned exactly once in the response and is never stored unencrypted.

Only POST, PUT, PATCH, and DELETE methods are allowed for public routes. GET is intentionally excluded.

---

dev SDK - CLI tools

pip install freesdn-sdk gives you the freesdn-sdk command. Use it throughout development; run the runtime install as the final gate.

freesdn-sdk init <name> [--author --description --api-prefix --output-dir]

Scaffolds plugin.yaml, plugin.py, tests/test_plugin.py, README.md, and requirements.txt. Validates the id against reserved names and the api_prefix format.

freesdn-sdk validate [path]

Parses the manifest, checks the entry_point file exists inside the directory, confirms the declared class is present via AST inspection, and warns on missing description/author/permissions. Note the divergence caveat - this does not catch ==-only pinning violations.

freesdn-sdk package [path] [-o out.zip]

Builds the installable ZIP. Skips junk directories, symlinks, and .env files. Enforces the 50 MB compressed / 200 MB uncompressed limits from PLUGIN_LIMITS. Default output name: {id}-{version}.zip.

freesdn-sdk check [path]

Static AST scan for blocked module imports and dangerous builtins (exec, eval, compile, __import__, open, breakpoint, globals, locals, vars). Exit 1 on any finding.

Testing with mock context

from freesdn_sdk.testing import create_test_context

ctx = create_test_context(
    plugin_id="my-plugin",
    organization_id="00000000-0000-0000-0000-000000000001",
    devices=[{"id": "dev-1", "name": "Switch A", "status": "online"}],
    settings={"api_key": "test-key"},
)

# ctx.devices is MockDeviceSDK
# ctx.alerts is MockAlertSDK
# ctx.events is MockEventSDK  → ctx.events.assert_emitted("device_offline")
# ctx.settings is MockPluginSettingsSDK
# ctx.http is MockPluginHTTPClient  → ctx.http.mock_response(url, json={...})

The mock context raises NotImplementedError on any method not covered by the test doubles, making missed SDK calls visible in your test suite rather than silently passing.

---

Next steps

• Manifest Reference - every plugin.yaml field with validation rules and the SDK-vs-runtime divergences in full detail.
• Getting Started - scaffold, implement, and install your first plugin end-to-end.
• Plugins Overview - the two-tier extensibility model, trust contract, and what SDK plugins cannot do compared to native modules.