Skip to content

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/.


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.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.


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

FieldTypePurpose
plugin_idstrYour plugin’s declared id from plugin.yaml.
organization_idUUIDThe organisation this runtime instance serves.
devicesDeviceSDKDevice read + inventory-write access.
alertsAlertSDKAlert read/create/resolve access.
eventsEventSDKEvent bus emit + subscription validation.
settingsPluginSettingsSDKPer-org key/value store with encrypted secret support.
httpPluginHTTPClientSSRF-protected outbound HTTP.
loggerlogging.LoggerNamed 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.


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 codeCore permission required
devices.readdevice:read
devices.writedevice:write
alerts.readalert:read
alerts.writealert:write

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


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

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

Permission: devices.read

Returns devices scoped to the plugin’s organisation. Results include id, name, type, status, ip, mac, and site_id. The limit argument is clamped to the range 1-500 regardless of what you pass.

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.

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.

async def register_device(device_data: dict) -> dict

Permission: devices.write

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

FieldRule
external_idMust start with plugin.<your-plugin-id>:. Rejected otherwise.
nameHuman-readable label. Truncated to 255 characters. Defaults to "Unknown" if omitted.
device_typeDevice category string (e.g. "switch", "camera"). Defaults to "other".
ip_addressMust be a public routable IP. Internal / loopback / link-local / RFC-1918 addresses are SSRF-blocked. Omit the field to skip the IP check.
site_idMust exist and belong to the plugin’s organisation. Required.
manufacturerOptional vendor string.
modelOptional model string.
firmware_versionOptional firmware version string.
mac_addressOptional MAC address string.
serial_numberOptional serial number string.
statusDevice status string. Defaults to "unknown".
metadataValidated 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.

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.


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

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.

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}.

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.


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.

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.

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.


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

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

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

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).

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.

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.


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

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 kwargPurpose
jsonJSON request body
dataForm or raw body
paramsQuery string parameters
contentRaw bytes body
cookiesRequest cookies
headersCustom 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.


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

MethodSignatureWhen called
on_install(db)asyncOnce, on first install. Create database tables or seed data. Exceptions are logged but non-fatal - the install proceeds.
on_start(organization_id, db=None)asyncPer-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)asyncAfter re-install on upgrade. Run schema migrations here.
on_uninstall(db)asyncJust before the plugin directory is removed. Clean up any database rows, files, or external registrations you created.
on_event(event)asyncCalled for each event matching your declared event_subscriptions. Default implementation logs and returns. Override to react.
on_stop(organization_id, db=None)asyncPer-organisation stop. Unbinds event subscriptions and clears ctx.
get_router()→ APIRouterOnce 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 → dictReturns {"status": "ok", "organization_id": ...} when active, {"status": "inactive"} otherwise. Surfaced by GET /api/v1/plugins/{id}/health.
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

Section titled “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"}

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.
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.
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>.

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.

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).

LimitValue
ZIP archive (compressed)50 MB
ZIP archive (uncompressed)200 MB
python_dependencies entries50
Automation triggers per plugin50
Automation actions per plugin50
AI tools per plugin20
AI tool result size256 KB
Devices registered per plugin1,000
HTTP request timeout60 seconds
HTTP response body10 MB
Settings blob (management API)32 KiB
HMAC timestamp skew (public routes)300 seconds
HMAC nonce length16-128 characters

Security model - what the SDK is and is not

Section titled “Security model - what the SDK is and is not”

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"].

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

VariableDefaultPurpose
PLUGIN_DIR/data/pluginsDirectory where plugin files are stored.
PLUGIN_ENABLE_DIRECT_URL_INSTALLSfalseEnable 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_DEPSfalseAllow plugins that declare python_dependencies to install them at install time. Off by default.
PLUGIN_PYPI_INDEX_URLhttps://pypi.org/simple/PyPI index used for hash-pinned dependency installs.
MARKETPLACE_REGISTRY_URLhttps://registry.freesdn.org/plugins.jsonRemote catalog URL used by POST /api/v1/marketplace/plugins/sync.
MARKETPLACE_PUBLISHER_PUBLIC_KEY(empty)Hex Ed25519 public key. When set, the catalog must carry a valid signature or sync is refused.
MARKETPLACE_ALLOW_UNSIGNEDfalseWhen 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.

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

MethodPathPurposeMinimum role
GET/api/v1/pluginsList installed plugins with org-effective statusorg_admin
GET/api/v1/plugins/{plugin_id}Plugin detail including cached manifestorg_admin
POST/api/v1/plugins/installInstall from uploaded ZIP (201)super_admin
POST/api/v1/plugins/install-urlInstall 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}/enableEnable globally (super_admin) or per-orgorg_admin
POST/api/v1/plugins/{plugin_id}/disableDisable globally or per-orgorg_admin
POST/api/v1/plugins/{plugin_id}/upgradeUpgrade via new ZIP; restarts everywheresuper_admin
GET/api/v1/plugins/{plugin_id}/settingsRead org-scoped settings maporg_admin
PUT/api/v1/plugins/{plugin_id}/settingsUpsert org-scoped settings (≤32 KiB)org_admin
GET/api/v1/plugins/{plugin_id}/healthRuntime health for caller’s orgorg_admin
GET/api/v1/plugins/{plugin_id}/public-authPublic-route HMAC auth statusorg_admin
POST/api/v1/plugins/{plugin_id}/public-auth/rotate-secretRotate org HMAC secret; returns plaintext onceorg_admin
MethodPathPurposeAuth
GET/api/v1/marketplace/pluginsBrowse published plugins (paginated)public
GET/api/v1/marketplace/plugins/featuredUp to 6 featured pluginspublic
GET/api/v1/marketplace/plugins/categoriesCategory list with countspublic
GET/api/v1/marketplace/plugins/{slug}Plugin detail by slugpublic
GET/api/v1/marketplace/plugins/{slug}/versionsVersion historypublic
POST/api/v1/marketplace/plugins/{slug}/installDownload, verify SHA-256, install (201)super_admin
GET/api/v1/marketplace/plugins/{slug}/reviewsPaginated reviewspublic
POST/api/v1/marketplace/plugins/{slug}/reviewsSubmit a review (one per user)active user
POST/api/v1/marketplace/plugins/syncSync catalog from registrysuper_admin

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.


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:

HeaderValue
X-FreeSDN-Plugin-OrgThe organisation UUID
X-FreeSDN-Plugin-TimestampUnix timestamp (seconds, string)
X-FreeSDN-Plugin-NonceRandom string, 16-128 characters
X-FreeSDN-Plugin-Signaturesha256=<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.


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.

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.


  • 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.