Creating triggers

A trigger starts a workflow run in response to an external event. A plugin trigger is a polling trigger: the platform calls it on an interval, and it returns any new events plus the state it wants to see on the next poll. The platform owns everything hard about triggers — scheduling, state persistence, and routing events to workflow runs — exactly as it does for built-in triggers. Your trigger only answers one question: given my config and the state from last time, what’s new?

Trigger poll() runs inside the plugin’s sandbox worker, invoked over JSON-RPC (trigger.poll). Read How plugins run for the execution model — network posture, logging to stderr, and file callbacks all apply here too.

The interface

A trigger subclasses BasePollingTrigger[ConfigType] and implements poll.

from noxus_sdk.triggers import BasePollingTrigger, TriggerConfiguration
from noxus_sdk.ncl import Parameter, ConfigText
from noxus_sdk.plugins.context import RemoteExecutionContext


class TickConfig(TriggerConfiguration):
    label: str = Parameter(default="tick", display=ConfigText(label="Label"))


class TickTrigger(BasePollingTrigger[TickConfig]):
    trigger_name = "MyTick"           # unique id for this trigger type
    title = "Tick"
    description = "Emits a counter every interval"
    polling_interval = 60.0           # seconds between polls (default 300)
    outputs = {"message": "str", "tick": "number"}

    async def poll(
        self, ctx: RemoteExecutionContext, state: dict
    ) -> tuple[list[dict], dict]:
        tick = int(state.get("tick", 0)) + 1
        events = [{"message": f"{self.config.label}-{tick}", "tick": tick}]
        new_state = {"tick": tick}
        return events, new_state

Class attributes

Attribute	Meaning
`trigger_name`	Unique identifier for the trigger type (used for lookup and in the manifest).
`title` / `description`	Shown in the editor’s trigger picker.
`image`	Optional icon URL.
`polling_interval`	Seconds between polls. Defaults to `300.0`.
`outputs`	A `{field_name: type_label}` map. Each field becomes a workflow input the editor can wire from this trigger.
`integrations`	Credential types this trigger needs (see below).

ConfigType is your TriggerConfiguration subclass (a NodeConfiguration, so its fields are Parameter(...) with optional display= widgets, just like a node’s config). It reaches your instance as self.config.

`poll` contract

async def poll(self, ctx, state: dict) -> tuple[list[dict], dict]:

state is whatever your previous poll returned as its second element (an empty dict on the first poll). Use it as a cursor — a last-seen id, a timestamp, a page token.
Return (events, new_state). events is a list of JSON-serializable dicts; each dict’s fields become the trigger inputs for one workflow run. Return [] when nothing is new. new_state is persisted by the platform and handed back on the next poll.

Emit one event per thing that happened. The platform turns each event into a run; your outputs map tells the editor which fields the event carries so they can be bound to workflow inputs.

A dropped connection mid-poll is not retried (see at-most-once side effects). Advance your cursor in new_state only for events you actually returned, so a re-poll after a failure re-emits rather than skips.

Using credentials

If a trigger needs to authenticate to an external service, declare the credential type(s) it uses and read them from ctx:

class TickTrigger(BasePollingTrigger[TickConfig]):
    integrations = ["my_weather"]

    async def poll(self, ctx, state):
        creds = ctx.get_integration_credentials("my_weather")
        api_key = creds.get("api_key")
        ...

The credential type is defined by a BaseIntegration / BaseCredentials pair in the same plugin — see Creating integrations.

Registering the trigger

Return your trigger classes from the plugin’s triggers() method:

class MyPlugin(BasePlugin[MyPluginConfig]):
    ...
    def triggers(self) -> list[type[BasePollingTrigger]]:
        return [TickTrigger]

Each trigger is serialized into the plugin manifest (TriggerDefinition) at packaging time. Regenerate manifest.json whenever a trigger’s name, config, or outputs change.

A plugin must provide at least one node, integration, or data source — triggers alone don’t make a valid plugin. Ship a trigger alongside the node or integration it drives.

On the platform side, an installed plugin trigger is resolved on demand from the plugin components table (a read-through in the trigger registry), so it works without being eagerly registered at startup.

Testing

Locally, noxus plugin serve exercises your plugin’s surfaces, except trigger.poll — polling is driven only by the platform, so the local dev server does not expose it. Test poll directly instead:

import asyncio
from noxus_sdk.plugins.context import RemoteExecutionContext
from my_plugin import TickTrigger, TickConfig

trigger = TickTrigger(TickConfig(label="t"))
events, state = asyncio.run(trigger.poll(RemoteExecutionContext(), {}))
assert events[0]["tick"] == 1
assert state == {"tick": 1}
# feed the returned state back to prove the cursor advances
events2, state2 = asyncio.run(trigger.poll(RemoteExecutionContext(), state))
assert state2 == {"tick": 2}

End to end, install the plugin against a running stack and let the platform schedule the poll and create runs. The reference plugin in tests/plugins/ and its run_plugin_e2e.py driver exercise a trigger this way.

How plugins run

The sandbox execution model behind trigger.poll.

Creating integrations

Define the credentials a trigger reads from ctx.

Overview

Tutorial

Reference

GitOps

The interface

Class attributes

`poll` contract

Using credentials

Registering the trigger

Testing

How plugins run

Creating integrations

​The interface

​Class attributes

​poll contract

​Using credentials

​Registering the trigger

​Testing

​Related

How plugins run

Creating integrations

The interface

Class attributes

`poll` contract

Using credentials

Registering the trigger

Testing

Related