Automation & AI Bridges

Plugins can plug into two runtime buses: the automation engine (event-driven trigger/action chains) and the AI assistant (callable tools exposed to the agentic loop). Both are registered from inside on_start() using helper methods on FreeSDNPlugin. This page covers what you register, how the caps and naming rules work, and the security gates that apply to each bus.

---

What the bridges are

Two singleton objects, both initialised at startup, mediate between plugins and the platform:

Singleton	Module	Responsibility
automation_bridge	backend/app/plugins/bridges.py	Holds the registry of plugin triggers and action handlers; wires them into the automation engine’s _action_handlers dict
ai_bridge	backend/app/plugins/bridges.py	Holds the registry of plugin AI tools; exposes them to modules/ai/service.py for execution

Your plugin never imports these directly. You call the three methods on FreeSDNPlugin - register_automation_trigger, register_automation_action, and register_ai_tool - from on_start(), and the base class delegates to the singletons.

Call from on_start only

Register all triggers, actions, and AI tools inside on_start(). The bridges (automation_bridge, ai_bridge) are singletons shared across all organisations - registrations are keyed by plugin_id, not by (plugin_id, org_id). What is per-org is the plugin runtime: on_start() is called once per (plugin_id, org_id) pair, and that is the correct place to register. Do not call registration helpers from on_install() or module-level code.

---

Automation triggers

A trigger declares that your plugin can originate a named event. When your plugin later calls await self.ctx.events.emit(event_type, payload) on that event type, the automation engine treats it as a trigger firing, and any operator-authored Connection wired to it will execute.

Register a trigger

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

    self.register_automation_trigger(
        trigger_type="device_offline",
        description="Fired when a monitored device stops responding.",
        schema={
            "type": "object",
            "properties": {
                "device_id": {"type": "string"},
                "last_seen": {"type": "string", "format": "date-time"},
            },
        },
    )

Naming rules and caps

Constraint	Detail
trigger_type regex	^[a-z0-9][a-z0-9_-]{0,98}[a-z0-9]$ - lowercase alphanumeric, underscores, hyphens, 2-100 chars
Full registered name	plugin.{plugin_id}.{trigger_type} - always namespaced; you cannot claim a core event name
Cap	50 triggers per plugin (MAX_TRIGGERS_PER_PLUGIN)
Duplicates	Registering the same full type a second time is idempotently skipped - no error, no overwrite
Description	Truncated to 500 chars

How the event gets there

Emitting the event from your plugin code is the act that fires the trigger. The EventSDK.emit method automatically prefixes your plugin_id:

# Inside any async method with ctx available:
await self.ctx.events.emit("device_offline", {
    "device_id": str(device.id),
    "last_seen": last_seen.isoformat(),
})
# Published on the bus as: plugin.my-plugin.device_offline

Automation Connections authored by operators bind to the full namespaced name (plugin.my-plugin.device_offline). You declare the schema so the Connection builder can validate and hint the payload shape.

---

Automation actions

An action is a callable step in an automation chain. When an operator builds a Connection that includes your action as a step, the runtime calls your handler async function with the step parameters.

Register an action

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

    self.register_automation_action(
        action_type="send_summary",
        handler=self._handle_send_summary,
        description="POST a device-health summary to an external endpoint.",
        params_schema={
            "type": "object",
            "properties": {
                "webhook_url": {"type": "string", "format": "uri"},
                "include_offline": {"type": "boolean", "default": True},
            },
            "required": ["webhook_url"],
        },
    )

async def _handle_send_summary(self, params: dict) -> dict:
    url = params["webhook_url"]
    # Use self.ctx.http  -  SSRF-protected, not raw httpx/requests
    resp = await self.ctx.http.post(url, json={"status": "ok"})
    return {"http_status": resp.status_code}

The handler signature is async (params: dict) -> dict. Return a plain dict; any non-JSON-serialisable return causes an error response.

Naming rules and caps

Constraint	Detail
action_type regex	Same as trigger: ^[a-z0-9][a-z0-9_-]{0,98}[a-z0-9]$
Full registered name	plugin.{plugin_id}.{action_type}
Cap	50 actions per plugin (MAX_ACTIONS_PER_PLUGIN)
Clobbering	Registering the same action type a second time from the same plugin is idempotently skipped - no error, no overwrite. Because the full type is always namespaced as plugin.{plugin_id}.{action_type}, action types from different plugins can never collide.

Use the HTTP client, not raw networking

Inside automation action handlers you have self.ctx.http, which routes all outbound requests through safe_http_request - DNS-pinned, no redirects, 60-second timeout, 10 MB response cap. Do not import httpx, aiohttp, or requests directly; they are not in the blocked-module list, but they bypass the SSRF protections built into self.ctx.http (DNS-pinning, redirect blocking, response-size cap). Use self.ctx.http exclusively for outbound calls.

---

AI tools

An AI tool is a callable function exposed to the FreeSDN AI assistant’s agentic loop. When a user asks the assistant a question that warrants your tool, the assistant calls it with validated parameters and incorporates the result.

The assistant runs a bounded agentic loop (maximum 5 iterations). Your tool is one of the callable steps in that loop, alongside the 11 built-in tools.

Register an AI tool

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

    self.register_ai_tool(
        name="get_device_summary",
        description="Return a structured health summary for a specific device by ID.",
        parameters={
            "type": "object",
            "properties": {
                "device_id": {
                    "type": "string",
                    "description": "UUID of the device to summarise.",
                },
            },
            "required": ["device_id"],
        },
        handler=self._ai_get_device_summary,
        permission="device:read",  # REQUIRED  -  core permission string, e.g. "device:read", "alert:read"
    )

async def _ai_get_device_summary(self, user, db, device_id: str) -> dict:
    device = await self.ctx.devices.get(device_id)
    if device is None:
        return {"error": "Device not found"}
    return {"id": str(device["id"]), "status": device["status"], "name": device["name"]}

Handler signature

AI tool handlers receive (user, db, **kwargs) where kwargs are the validated call arguments from the assistant. They do NOT receive self.ctx automatically - the handler runs with the calling user’s raw session, not the org-scoped plugin context.

This is intentional: the AI tool runs on behalf of the authenticated user who asked the question, so access control must be the caller’s access, not the plugin’s ambient authority.

Always declare a permission - the fail-closed sentinel

This is the most important rule for AI tools.

If permission is falsy (empty string, None, or omitted), the bridge coerces it to the sentinel value plugin:undeclared_ai_tool. No role’s explicit permission set contains that string, but has_permission() returns True for super_admin unconditionally via the role’s implicit universal grant (see dependencies.py:456-458). Only super_admin passes the has_permission() check for that sentinel - via the role’s implicit universal grant - so the tool is effectively super_admin-only until the author declares a real permission. Because the bridge always writes the sentinel into the registered AITool object (bridges.py:238), the live enforcement path is service.py:502: if tool.permission and not user.has_permission(tool.permission) - which allows only super_admin (via implicit grant) and denies everyone else. The service.py:499 check (not tool.permission) is defense-in-depth for hypothetical out-of-band registry writes and does not fire for any tool registered through the standard bridge API.

A plugin AI tool with no declared permission is not “open to everyone” - it is blocked for everyone. Always set permission to a real core permission string (e.g. "device:read", "alert:read").

Naming rules, caps, and result limits

Constraint	Detail
Registered name	Auto-prefixed: plugin_{plugin_id}_{name} - you cannot overwrite a built-in tool
Cap	20 AI tools per plugin (MAX_TOOLS_PER_PLUGIN)
Overwrite guard	If a tool with the same prefixed name already exists (built-in or another plugin), the registration is skipped silently
Result size	Success-path dict is serialised and checked; results over 256 KB are replaced with {"error": "Plugin tool result too large", "truncated": True}
Error sanitisation	Exceptions from your handler are caught and returned as a generic error message - stack traces are not forwarded to the assistant

Permission codes recognised by the confused-deputy guard

The CAN-015 guard intersects the plugin’s declared permissions with the calling user’s actual permissions when the tool is invoked through an SDK method call (e.g. self.ctx.devices.list(), self.ctx.alerts.create()). At that point _require_permission checks two things: (a) the SDK capability (devices.read, alerts.write, etc.) is listed in plugin.yaml permissions[].code, and (b) the authenticated caller holds the mapped core permission. The mapping is:

Permission code	Core permission required
devices.read	device:read
devices.write	device:write
alerts.read	alert:read
alerts.write	alert:write

For AI tool invocations (not a REST endpoint hit), execution succeeds if the caller (the authenticated user running the assistant session) holds the core permission declared in the permission= argument of register_ai_tool(). There is no separate check against plugin.yaml at AI tool execution time - the permission is embedded in the registered AITool object (modules/ai/tools/__init__.py:18-26). The plugin.yaml declaration governs which SDK capabilities the handler may use inside its body, not whether the tool fires at all.

---

Caps summary

Resource	Cap	Constant
Automation triggers per plugin	50	MAX_TRIGGERS_PER_PLUGIN
Automation actions per plugin	50	MAX_ACTIONS_PER_PLUGIN
AI tools per plugin	20	MAX_TOOLS_PER_PLUGIN
AI tool result size	256 KB	_MAX_TOOL_RESULT_BYTES
HTTP response (via ctx.http)	10 MB	MAX_RESPONSE_SIZE
HTTP timeout	60 s	MAX_TIMEOUT

These limits are also exported by the published SDK as PLUGIN_LIMITS from freesdn_sdk.types so you can reference them in tests.

---

Declaring event subscriptions

If your automation triggers depend on receiving core events - for example, you fire your trigger only in response to a vpn.connection_down event - you must declare those subscriptions in plugin.yaml:

event_subscriptions:
  - alert.created
  - alert.resolved
  - vpn.connection_down
  - vpn.connection_restored

The EventSDK.subscribe validation rejects patterns not listed here, and it also blocks bare wildcards (*, #). Your on_event handler fires for each matching event, already filtered to your organisation.

The notify-hub example plugin (freesdn-sdk/examples/plugins/notify-hub/) demonstrates this pattern: it subscribes to six core events and uses them to dispatch external notifications.

---

Fabric catalog projection

All registered triggers, actions, and AI tools appear in the Fabric universal-interconnect catalog at GET /api/v1/fabric/catalog with tier: "plugin". Operators use the catalog to build Connections.

Two important constraints apply to the plugin tier in the Fabric:

1. A plugin operation with no declared permission is refused at the catalog level.
2. Plugin operations are not invocable through the generic Fabric invoke path - they run only through their own (plugin, org) runtime.

You do not need to do anything special to appear in the catalog. Registration via the bridge methods is sufficient.

---

Cleanup on uninstall or disable

When the last organisation stops using a plugin (or the plugin is uninstalled), the loader automatically calls:

• automation_bridge.unregister_plugin(plugin_id) - removes all triggers and action handlers for the plugin
• ai_bridge.unregister_plugin_tools(plugin_id) - removes all AI tools for the plugin

You do not need to manually deregister in on_uninstall or on_stop. The bridges handle it.

---

Introspection endpoints

You can inspect what is currently registered without touching the bridge singletons directly. From the REST API:

Method	Path	Purpose
GET	/api/v1/plugins/{plugin_id}/health	Runtime health including active state per org; 410 if not active
GET	/api/v1/fabric/catalog	Full tier-tagged catalog of all operations, events, and AI-tool projections

From within tests, use the bridge accessors:

from backend.app.plugins.bridges import automation_bridge, ai_bridge

triggers = automation_bridge.get_plugin_triggers("my-plugin")
actions  = automation_bridge.get_plugin_actions("my-plugin")   # includes 'handler' key; pass plugin_id=None to get all plugins with handlers stripped
tools    = ai_bridge.get_plugin_tools("my-plugin")

---

Common mistakes

Registering after on_start returns. Background tasks or deferred coroutines that call registration helpers after on_start completes are unreliable - the loader binds event subscriptions immediately after on_start. Do all registration synchronously within on_start.

Forgetting await super().on_start(...) is still a risk for forward-compatibility and lifecycle correctness, but it does not leave self.ctx as None. The loader pre-initialises self.ctx via _init_context before invoking your on_start (see loader.py start_for_org), so SDK access such as self.ctx.devices, self.ctx.alerts, and self.ctx.events works even without the super() call. Always call await super().on_start(organization_id, db) first anyway - the base class owns shared lifecycle steps, and omitting it can break those.

Using raw HTTP in action or tool handlers. Modules like socket, http, and urllib are in the load-time blocked-module list. Third-party HTTP clients (httpx, aiohttp, requests) are not in the blocked-module list - but they bypass the SSRF protections (DNS-pinning, redirect blocking, 10 MB response cap) built into self.ctx.http. Use self.ctx.http exclusively for outbound calls.

Returning non-serialisable objects from an AI tool handler. The result must be a plain dict with JSON-serialisable values. Returning a SQLAlchemy model, a UUID object, or a datetime without converting it to a string will trigger the generic error path and the assistant will receive an error, not your data.

The security model is cooperative, not sandboxed

Plugins run inside the same Python process as the backend. The import hygiene layer (sandbox.py) blocks blocked modules at load time but does not prevent a determined author from accessing process memory or the interpreter object graph. Treat plugin install the same way you treat adding a Python package to a production application - only install plugins from authors you trust, and review the source. Real subprocess isolation is not yet available.

---

Next steps

• Manifest reference - complete plugin.yaml field list including event_subscriptions, permissions, and python_dependencies pinning rules
• Getting started - scaffold, validate, and package your first plugin
• Plugins overview - the two-tier trust model and what SDK plugins can and cannot do
• Fabric catalog - how Connections, Operations, and Events fit together at the platform level