6. Advanced Techniques

This is part 6 of the Your First Plugin tutorial. Make sure you’ve completed 5. Working with Files first.

Node configuration with UI controls

Add configuration fields that appear in the node’s settings panel in the editor:

from noxus_sdk.ncl import (
    ConfigSelect,
    ConfigToggle,
    ConfigNumberSlider,
    Parameter,
)


class GetWeatherConfig(NodeConfiguration):
    units: str = Parameter(
        default="metric",
        description="Choose temperature units",
        display=ConfigSelect(label="Temperature Units", values=["metric", "imperial", "kelvin"]),
    )

    include_humidity: bool = Parameter(
        default=False,
        description="Also return humidity data",
        display=ConfigToggle(label="Include Humidity"),
    )

Every display widget requires a label — it has no default. The description on Parameter is optional helper text; the label on the widget is what renders as the field name.

Access config values in your node’s call():

async def call(self, ctx: RemoteExecutionContext) -> dict:
    # Both config fields and bindable inputs are read from self.config.
    units = self.config.units  # "metric", "imperial", or "kelvin"
    include_humidity = self.config.include_humidity
    # ...

Available config display types

Every widget takes a label; the argument column shows what else each accepts.

Type	Description
`ConfigText(label=...)`	Single-line text input
`ConfigBigText(label=...)`	Multi-line textarea
`ConfigPassword(label=...)`	Masked secret input
`ConfigSelect(label=..., values=[...])`	Dropdown select
`ConfigMultiSelect(label=..., values=[...])`	Multi-select dropdown
`ConfigToggle(label=...)`	Boolean switch
`ConfigNumber(label=...)`	Number input
`ConfigNumberSlider(label=..., min=, max=, step=)`	Slider
`ConfigFile(label=...)` / `ConfigFileArray(label=...)`	File / file-list input
`ConfigRichTextVariables(label=...)`	Rich text with variable insertion

Dynamic configuration

Override get_config() to compute config server-side at edit time — for example, fetching a select’s options from an upstream API. It runs in the plugin’s worker; mutate the ConfigResponse and return it. Write dynamic options into config_response.config_values keyed by the field name:

from noxus_sdk.nodes.schemas import ConfigResponse
from noxus_sdk.plugins.context import RemoteExecutionContext


class GetWeatherNode(BaseNodeV2[GetWeatherConfig, GetWeatherOutputs]):
    # ...

    @classmethod
    async def get_config(
        cls,
        ctx: RemoteExecutionContext,
        config_response: ConfigResponse,
        *,
        skip_cache: bool = False,
    ) -> ConfigResponse:
        # e.g. fetch the available options from an API using the connected
        # credentials, then populate the "region" dropdown dynamically.
        config_response.config_values["region"] = ["eu-west-1", "us-east-1"]
        return config_response

The platform only wakes the worker for get_config when a ConfigSelect has no static values. Nodes with fully static config never pay the round-trip, so add get_config only when options genuinely depend on context.

Polling triggers

A plugin can also emit events on a schedule with BasePollingTrigger. The platform owns scheduling, state persistence, and turning events into runs — your trigger only implements poll(ctx, state), returning (events, new_state). Cursor state round-trips between polls so you emit each event once:

from noxus_sdk.triggers import BasePollingTrigger, TriggerConfiguration
from noxus_sdk.plugins.context import RemoteExecutionContext


class TickConfig(TriggerConfiguration):
    pass


class TickTrigger(BasePollingTrigger[TickConfig]):
    trigger_name = "weather_tick"
    title = "Weather Tick"
    description = "Emits an incrementing tick on an interval"
    polling_interval = 60.0  # seconds
    # event field name -> human-readable type label
    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"tick-{tick}", "tick": tick}]
        return events, {"tick": tick}

class WeatherPlugin(BasePlugin[WeatherPluginConfig]):
    # ...
    def triggers(self):
        return [TickTrigger]

Each event dict’s fields become the workflow’s trigger inputs (matching the outputs you declared). A trigger can also declare integrations = ["weather_api"] to require a connected credential.

List handling

When a list output connects to a non-list input, the platform runs the node once per item automatically. To process the whole list in a single run, declare the input as a list instead. On a V2 node that’s a bindable list[...] config field:

class SummariseCitiesConfig(NodeConfiguration):
    cities: list[str] = Parameter(
        default_factory=list, bindable=True, display=ConfigMultiSelect(label="Cities", values=[]),
    )


class SummariseCitiesOutputs(NodeOutputs):
    results: list[str]


class SummariseCitiesNode(BaseNodeV2[SummariseCitiesConfig, SummariseCitiesOutputs]):
    node_name = "SummariseCitiesNode"
    title = "Summarise Cities"
    description = "Fetches weather for every city in one run"

    async def call(self, ctx: RemoteExecutionContext) -> dict:
        # Process all cities at once — self.config.cities is the full list.
        results = [await fetch_weather(city) for city in self.config.cities]
        return {"results": results}

Plugin-level configuration

Use PluginConfiguration for settings that apply to the entire plugin (not per-node). These are set in Settings → Plugins → Configure:

from noxus_sdk.plugins import PluginConfiguration
from noxus_sdk.schemas import ValidationResult


class WeatherPluginConfig(PluginConfiguration):
    default_units: str = Parameter(
        default="metric",
        display=ConfigSelect(label="Default Units", values=["metric", "imperial"]),
    )

    cache_ttl: int = Parameter(
        default=300,
        display=ConfigNumberSlider(label="Cache TTL (seconds)", min=60, max=3600, step=60),
    )

    def validate_config(self) -> ValidationResult:
        # Gate the plugin into RUNNING only when config is coherent.
        if self.cache_ttl < 60:
            return ValidationResult(valid=False, errors=["Cache TTL must be >= 60s"])
        return ValidationResult(valid=True, errors=[])

Access plugin config in any node — ctx.plugin_config is a plain dict:

async def call(self, ctx: RemoteExecutionContext) -> dict:
    default_units = ctx.plugin_config.get("default_units", "metric")
    # ...

Error handling

Raise the SDK’s typed errors from noxus_sdk.errors. The message crosses the JSON-RPC sandbox boundary and becomes the node’s user-visible failure reason — the exception type itself does not survive, so make every message actionable:

from noxus_sdk.errors import IntegrationFailedError, UnexpectedError


async def call(self, ctx: RemoteExecutionContext) -> dict:
    city = self.config.city  # bindable input
    if not city or not city.strip():
        raise IntegrationFailedError("City name cannot be empty")

    import httpx

    try:
        result = await fetch_weather(city)
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            raise IntegrationFailedError(f"City '{city}' not found") from e
        raise UnexpectedError(f"Weather API error: {e.response.status_code}") from e

    return {"temperature": result["temp"], "description": result["desc"]}

Best practices:

Use IntegrationFailedError for credential/API/user-fixable problems.
Use UnexpectedError for failures you did not anticipate.
Degrade, don’t raise, for expected external flakiness. A raised exception fails the whole run. For flaky network or parsing, catch it and return an empty typed result; reserve exceptions for genuine misconfiguration (e.g. no credential connected). The Proxy API scrapers do this — a failed scrape yields an empty CSV rather than crashing the flow.
Chain with from e and include actionable information in the message.

Runtime & sandbox gotchas

Your code runs as a warm worker inside a sandbox VM, reached over JSON-RPC. That environment imposes a few rules the local dev server does not:

stdout is the protocol. The worker speaks JSON-RPC on stdout; anything you (or a dependency) print() there corrupts the channel and breaks the call. Redirect noisy code to stderr:
import contextlib import sys with contextlib.redirect_stdout(sys.stderr): run_chatty_library()
Import heavy deps lazily, inside call/helpers — not at module top. Manifest generation and noxus plugin validate import your module in a lightweight environment; a top-level import pandas / playwright / curl_cffi makes get_manifest() fail with ModuleNotFoundError even though the sandbox has the dep. Put the import in the function that uses it.
Degrade, don’t raise, for expected external failures (see above).
Don’t shadow a package with a same-named module. A weather.py next to a weather/ package makes import your_pkg.weather resolve to the package — the module becomes unreachable. Name adapters distinctly (weather_adapter.py).
Zip only source. Exclude .venv, __pycache__, .git, node_modules. An accidental .venv can balloon the archive from ~300 KB to tens of MB.
Browser / JIT plugins can’t run under the default syd sandbox. syd’s seccomp filter SIGSYS-kills V8’s JIT, so Playwright / Chromium / any Node or JIT-heavy workload dies on startup — even when you only drive a remote browser, the local Playwright Node driver still runs. Those plugins need the gvisor sandbox provider (Linux/EKS) or none (local dev — gVisor can’t pivot_root inside Docker Desktop on macOS). This is a deployment setting (SANDBOX_PROVIDER), but know it before shipping a browser plugin.
Keep noxus-sdk>= aligned with the platform. The sandbox resolves noxus-sdk alongside your pyproject.toml deps; pinning an older SDK can win and the worker fails to start. Depend on noxus-sdk at or above the platform’s version, not an old floor.

Deploy your plugin

Option 1: From a Git repository

Push your plugin to a Git repository, then install from the Noxus UI:

Go to Settings → Plugins → Install Plugin
Choose Git source
Enter your repository URL, branch, and path (if the plugin is in a subdirectory)
For private repos, provide an access token

Option 2: Upload directly

Package and upload:

noxus plugin package --path ./weather-plugin --output weather-plugin.tar.gz

Then upload the .tar.gz file through the Noxus UI.

Option 3: Marketplace

Publish to the Noxus plugins marketplace for public distribution.

Verify installation

Once installed, you should see:

Your plugin listed with status Running in Settings → Plugins
Your nodes available in the flow editor palette
Your integration available in workspace control for credential configuration

Complete example

Here’s the full weather_plugin/__init__.py putting everything together:

from typing import ClassVar

from noxus_sdk.errors import IntegrationFailedError
from noxus_sdk.integrations.base import BaseIntegration, BaseCredentials
from noxus_sdk.ncl import ConfigPassword, ConfigSelect, ConfigText, ConfigToggle, Parameter
from noxus_sdk.nodes.base import BaseNodeV2, NodeConfiguration, NodeOutputs
from noxus_sdk.nodes.types import NodeCategory
from noxus_sdk.plugins import BasePlugin, PluginConfiguration
from noxus_sdk.plugins.context import RemoteExecutionContext
from noxus_sdk.plugins.types import PluginCategory


# ── Integration ──────────────────────────────────────────────────────

class WeatherAPICredentials(BaseCredentials):
    type: ClassVar[str] = "weather_api"
    api_key: str = Parameter(
        default="",
        description="OpenWeatherMap API key",
        display=ConfigPassword(label="API Key"),
    )

    def is_ready(self) -> bool:
        return bool(self.api_key)


class WeatherAPIIntegration(BaseIntegration[WeatherAPICredentials]):
    display_name = "Weather API"
    image = "https://cdn-icons-png.flaticon.com/512/1779/1779940.png"


# ── Node schemas (V2) ────────────────────────────────────────────────

class GetWeatherConfig(NodeConfiguration):
    # bindable=True → a node input; read from self.config in call().
    city: str = Parameter(
        default="", bindable=True,
        display=ConfigText(label="City", placeholder="e.g. London"),
    )
    units: str = Parameter(
        default="metric",
        display=ConfigSelect(label="Units", values=["metric", "imperial"]),
    )
    include_humidity: bool = Parameter(
        default=False,
        display=ConfigToggle(label="Include Humidity"),
    )


class GetWeatherOutputs(NodeOutputs):
    temperature: str
    description: str


# ── Node ─────────────────────────────────────────────────────────────

class GetWeatherNode(BaseNodeV2[GetWeatherConfig, GetWeatherOutputs]):
    node_name = "GetWeatherNode"
    title = "Get Weather"
    description = "Fetches current weather for a city"
    category = NodeCategory.DATA
    color = "#4A90E2"
    integrations = {"weather_api": ["api_key"]}

    async def call(self, ctx: RemoteExecutionContext) -> dict:
        import httpx

        creds = ctx.get_integration_credentials("weather_api") or {}
        api_key = creds.get("api_key", "")
        if not api_key:
            raise IntegrationFailedError("Weather API key not configured")

        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.openweathermap.org/data/2.5/weather",
                params={"q": self.config.city, "appid": api_key, "units": self.config.units},
            )
            response.raise_for_status()
            data = response.json()

        result = {
            "temperature": f"{data['main']['temp']}°{'C' if self.config.units == 'metric' else 'F'}",
            "description": data["weather"][0]["description"].capitalize(),
        }

        if self.config.include_humidity:
            result["description"] += f" (Humidity: {data['main']['humidity']}%)"

        return result


# ── Plugin ───────────────────────────────────────────────────────────

class WeatherPluginConfig(PluginConfiguration):
    pass


class WeatherPlugin(BasePlugin[WeatherPluginConfig]):
    name = "weather-plugin"
    display_name = "Weather Plugin"
    version = "0.1.0"
    description = "Weather data nodes with OpenWeatherMap integration"
    category = PluginCategory.GENERAL
    author = "Your Name"

    def nodes(self):
        return [GetWeatherNode]

    def integrations(self):
        return [WeatherAPIIntegration]

Examples

Browse more code examples

Architecture

Understand how plugins run under the hood

Overview

Tutorial

Reference

GitOps

Node configuration with UI controls

Available config display types

Dynamic configuration

Polling triggers

List handling

Plugin-level configuration

Error handling

Runtime & sandbox gotchas

Deploy your plugin

Option 1: From a Git repository

Option 2: Upload directly

Option 3: Marketplace

Verify installation

Complete example

Examples

Architecture

​Node configuration with UI controls

​Available config display types

​Dynamic configuration

​Polling triggers

​List handling

​Plugin-level configuration

​Error handling

​Runtime & sandbox gotchas

​Deploy your plugin

​Option 1: From a Git repository

​Option 2: Upload directly

​Option 3: Marketplace

​Verify installation

​Complete example

Examples

Architecture

Node configuration with UI controls

Available config display types

Dynamic configuration

Polling triggers

List handling

Plugin-level configuration

Error handling

Runtime & sandbox gotchas

Deploy your plugin

Option 1: From a Git repository

Option 2: Upload directly

Option 3: Marketplace

Verify installation

Complete example