Skip to content

Anatomy of a Function Block

Every function block follows the same three-layer structure: typed parameters flow in, a run() method processes them, and typed results flow out. The platform handles discovery, scheduling, error handling, and persistence — you focus on the automation logic.

Lifecycle

A function block goes through six stages from deployment to database update:

flowchart LR
    A[Registration] --> B[Discovery]
    B --> C[Acquire]
    C --> D[Execute]
    D --> E[Result]
    E --> F[DB Updates]
  1. Registration – The @register_function_block decorator attaches metadata (name, version, schemas) to the class.
  2. Discovery – On startup, the worker recursively imports all Python modules in the configured directory and collects decorated classes.
  3. Acquire – The workflow engine dispatches an ACQUIRE job. The worker calls acquire() to declare resource needs; the engine locks entities and resolves context.
  4. Execute – The engine dispatches a separate EXECUTE job. The worker calls execute_function_block(), which validates parameters and invokes run() with a WorkflowContext holding entity state.
  5. Result – run() returns a typed FunctionBlockResult with a success flag, message, and structured data.
  6. DB Updates – The platform diffs entity state before and after execution, generating create/patch/delete operations automatically.

Parameters

FunctionBlockParams extends Pydantic’s BaseModel with extra="ignore". Unknown fields from the workflow engine are dropped (the SDK logs a warning listing them via parse_parameters_from_dto()), so schema extensions never break existing function blocks.

A minimal parameter model declares its fields with type annotations:

class EchoParameters(FunctionBlockParams):
    message: str

Pydantic validates automatically – if the engine sends {"text": 42}, Pydantic coerces it to "42". Send {} without a required field and validation fails with a clear error before run() is ever called.

Complex parameters: nested models, defaults, and validators

Real-world function blocks often need richer input structures:

class ThresholdConfig(FunctionBlockParams):
    """Nested model for configurable thresholds."""

    warning: float = 80.0
    critical: float = 95.0

    @field_validator("critical")
    @classmethod
    def critical_above_warning(cls, v: float, info: ValidationInfo) -> float:
        warning = info.data.get("warning", 80.0)
        if v <= warning:
            raise ValueError("critical threshold must exceed warning")
        return v


class ComplianceCheckParams(FunctionBlockParams):
    policy_name: str
    interfaces_to_check: list[str] | None = None
    cpu_thresholds: ThresholdConfig = ThresholdConfig()
    memory_thresholds: ThresholdConfig = ThresholdConfig()
    skip_unreachable: bool = True

Key patterns:

  • Nested models – compose FunctionBlockParams subclasses for structured input.
  • Optional fields – list[str] | None = None for parameters that may be omitted.
  • Defaults – assign values directly (skip_unreachable: bool = True).
  • Validators – @field_validator for cross-field or business-rule checks.

JSON schemas auto-generated from these models drive the parameter form in the neops UI.

Results

FunctionBlockResultData extends BaseModel with extra="forbid". Unlike parameters, results enforce strict output validation – any undeclared field raises a ValidationError. Downstream workflow steps and the UI can trust the schema completely.

The result goes to the engine as JSON, which has no NaN or infinity. A result whose data (or whose db_updates, such as a fact) holds one is not sent: the worker sends a failure without retry instead, naming the field, for example The result cannot be sent: data.ratio is NaN or infinite, which JSON cannot carry. The check reads the value before any JSON-only serializer runs, so a @field_serializer(..., when_used="json") that maps NaN away does not get it through: store None (or a finite number) in the field instead.

class EchoResult(FunctionBlockResultData):
    echoed_message: str

Wrap your result data in a FunctionBlockResult:

FunctionBlockResult(
    message="Version retrieved successfully.",
    success=True,
    data=ShowVersionResult(version_info=version_info),
)
Field Type Purpose
success bool Whether the block completed without errors
message str Human-readable summary for the UI and logs
data ResultDataT \| None Typed result payload, or None on failure
retryable bool (default True) Set False on a failure that would come out the same on every attempt, so the engine does not retry the job
side_effects bool \| None (default None) Set False when this attempt changed nothing, on success or failure; the engine then leaves the step out of the rollback. None means side effects are assumed. See Side effects of one attempt

side_effects belongs on run() results only. A keyword the result does not declare, such as a misspelt side_effect=False, is logged as a warning instead of being sent.

Registration

Every function block declares its identity through a Registration dataclass passed to the @register_function_block decorator:

@register_function_block(
    Registration(
        name="echo",
        description="Return the provided text without modification.",
        package="fb.examples.neops.io",
        version=(1, 0, 0),
        run_on="global",
        fb_type="execute",
        param_cls=EchoParameters,
        result_cls=EchoResult,
        is_pure=True,
        is_idempotent=True,
    )
)
Field Default Purpose
name – Unique name within the package
package – Reverse-DNS namespace (e.g. fb.acme.com)
version (1, 0, 0) Semantic version tuple (major, minor, patch)
description – Short description shown in the UI
run_on None Entity scope: "device", "interface", "group", or "global". Required for a plain FunctionBlock; taken from the class for a typed block (keyword-only)
fb_type – Category: "configure", "facts", "check", "execute", "none"
param_cls – The FunctionBlockParams subclass for this block
result_cls – The FunctionBlockResultData subclass for this block
is_pure False True if the block has no side effects; changes it declares to the CMS through the context do not count
is_idempotent False True if repeated execution produces the same result
reversible False True if the engine may undo the block through rollback()
rollback_data_cls None The FunctionBlockRollbackData subclass the block records in run() for its own rollback. Reversible blocks only; must match the class’s generic argument. See Rollback data
markdown_helptext "No helptext provided" Markdown documentation shown in the UI
deprecated False Marks the block as deprecated

The platform identifies each block as {package}/{name}:{major}.{minor}.{patch}. For the example above, that resolves to fb.examples.neops.io/echo:1.0.0.

JSON schemas for param_cls, result_cls and rollback_data_cls are auto-generated at registration time. The workflow engine uses the parameter and result schemas for input validation in both the UI and the API. The rollback data schema serves one purpose only: the engine validates each saved snapshot against it, exactly, and freezes it per exact block version. See Rollback data.

See Registration (Engine) for how the engine stores, resolves, and validates registrations.

Lifecycle Methods

The FunctionBlock base class defines three lifecycle methods:

Method Required Purpose
run(params, context) Yes Core execution logic
acquire(params) Yes Declare resource needs before execution (receives AcquireParams[YourParams])
rollback(params, context, forward) For reversible=True blocks Undo this block’s changes when the run fails

run() receives validated parameters and a WorkflowContext holding entity data. This is where your automation logic lives. Return a FunctionBlockResult with your typed output.

acquire() is dispatched as a separate job before run(). It declares which entities the block needs. The workflow engine locks resources and resolves additional context based on your response. For blocks that need no specific resources, return a simple success.

acquire() receives an AcquireParams[YourParams], not your parameter model directly. At ACQUIRE time the engine has not run earlier steps and has no entity context, so parameters fed by {{ <step>.result… }} or {{ context… }} are not yet knowable: they are listed in params.unresolved, and on params.partial a required field is then None while a field with a declared default keeps that default. Parameters fed by {{ parameters… }} or written literally in the workflow are resolved and can be trusted. The EXECUTE phase re-interpolates and strictly validates every parameter, so nothing you skip here is lost. See Advanced Patterns for how to branch on params.unresolved.

rollback() undoes this block’s changes when the workflow fails, at this step or a later one; the engine dispatches it as a ROLLBACK job with the same parameters and context the forward job had. Its third argument, forward, is a ForwardResult describing the earliest attempt of this step that may have changed something: its result data, the rollback data it saved, whether its outcome is known, and the step’s earlier rollbacks. The default implementation returns FunctionBlockRollbackResult(success=False, message="Rollback not implemented"). Override it, and register the block reversible=True, when your block modifies external systems that need cleanup. See Rollback.

Full FunctionBlock base class 
class FunctionBlock(Generic[ParamsT, ResultDataT, ContextT, RollbackDataT], ABC):
    run_on: ClassVar[str | None] = None
    """The entity type the block runs on, fixed by the typed flavours (`DeviceFunctionBlock`, …).
    ``None`` on a plain ``FunctionBlock``: its ``Registration.run_on`` decides."""

    context_cls: ClassVar[type[WorkflowContext]] = WorkflowContext
    """The context type ``run`` and ``rollback`` receive; the typed flavours narrow it."""

    rollback_data_cls: ClassVar[type[FunctionBlockRollbackData] | None] = None
    """The class ``save_rollback_data`` takes, resolved from the ``RollbackDataT`` argument.
    ``None`` when the block declares none."""

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        cls.rollback_data_cls = resolve_rollback_data_cls(_bound_argument(cls, FunctionBlock, RollbackDataT))

    @final
    def __init__(self, logger: BaseLogger | None = None) -> None:
        self.logger: BaseLogger = Logger.or_default(logger).child(self)
        self._run_entered = False
        self._rolling_back = False
        self._rollback_data = RollbackDataRecorder(type(self).__name__, type(self).rollback_data_cls)

    @property
    def run_entered(self) -> bool:
        """Whether ``execute_function_block`` got as far as calling ``run``: before that, nothing was changed."""
        return self._run_entered

    @property
    def recorded_rollback_data(self) -> dict[str, Any] | None:
        """The JSON value of the last successful ``save_rollback_data``, or ``None``."""
        return self._rollback_data.value

    @final
    def save_rollback_data(self, data: RollbackDataT, *, replace: bool = False) -> None:
        """
        Record what ``rollback`` needs to undo this attempt; call it before ``run`` changes anything.

        The model is serialized at once, so a later change to it is not sent: save a changed
        model again with ``replace=True``, from the thread that made the first save. A save
        that fails raises `RollbackDataError` and keeps the earlier snapshot. The value is sent
        with the result whatever ``run`` returns or raises.
        """
        if self._rolling_back:
            raise RollbackDataError(f"{type(self).__name__}: save_rollback_data() belongs in run(), not in rollback().")
        self._rollback_data.save(data, replace=replace)

    @abstractmethod
    async def run(self, params: ParamsT, context: ContextT) -> FunctionBlockResult[ResultDataT]:
        """
        Run the function block with the given context and return a result.
        """

    @abstractmethod
    async def acquire(self, params: "AcquireParams[ParamsT]") -> FunctionBlockAcquireResult:
        """
        Acquire resources or perform setup before running the function block.
        This method should be implemented by subclasses to define how to acquire necessary resources.

        ``params`` is a best-effort view, not the final parameters. No earlier
        step has run and there is no entity context yet, so anything fed by
        ``{{ <step>.result… }}`` or ``{{ context… }}`` is listed in
        ``params.unresolved``; ``{{ parameters… }}`` and literals are resolved
        and can be trusted. Branch on ``params.unresolved`` rather than checking
        for ``None`` — a required field's static type is non-optional, so a type
        checker cannot see that ``None`` is possible.
        """

    async def rollback(
        self,
        params: ParamsT,
        context: ContextT,
        forward: ForwardResult[ResultDataT, RollbackDataT],
    ) -> FunctionBlockRollbackResult:
        """
        Undo what ``run`` did, from the same parameters and entity context ``run`` received.

        The engine dispatches a rollback job when a later step fails; this step itself may
        have succeeded. ``forward`` is the earliest attempt that may have changed something,
        with its snapshot in ``forward.rollback_data``. What to do without a snapshot, with an
        unknown outcome or on a second run (``forward.previous_rollbacks``) is the block's call;
        a refusal ends the entry ROLLBACK_FAILED. Override it in every block registered
        ``reversible=True``; the default refuses.
        """
        return FunctionBlockRollbackResult(success=False, message="Rollback not implemented")

    @final
    async def rollback_function_block(
        self,
        params: ParamsT,
        context: WorkflowContext,
        forward: ForwardResult[ResultDataT, RollbackDataT],
    ) -> FunctionBlockRollbackResult:
        """Runs ``rollback`` behind the same error boundary ``execute_function_block`` gives ``run``."""
        self._rolling_back = True
        try:
            return await self.rollback(params, self._narrow_context(context), forward)
        except Exception as e:
            return FunctionBlockRollbackResult(success=False, message=f"Error rolling back function block: {e!s}")
        finally:
            self._rolling_back = False

    @final
    async def execute_function_block(
        self,
        params: ParamsT,
        context: WorkflowContext,
        *,
        propagate_exceptions: bool = False,
    ) -> FunctionBlockResult[ResultDataT]:
        """
        Execute the function block and return its result.

        If ``propagate_exceptions`` is True any exception raised by ``run`` will be
        re-raised so that tooling (e.g. pytest) can show a full stack-trace that
        points to the actual error location.
        This is essential for debugging and testing.
        In production code the default (``False``) keeps the previous behaviour and converts exceptions into a
        failed ``FunctionBlockResult``.
        """
        if propagate_exceptions:
            # Let exceptions bubble up for better debuggability
            return await self._enter_run(params, context)

        try:
            return await self._enter_run(params, context)
        except Exception as e:
            return FunctionBlockResult(
                message=f"Error executing function block: {e!s}",
                success=False,
                data=None,
            )

    @final
    async def _enter_run(self, params: ParamsT, context: WorkflowContext) -> FunctionBlockResult[ResultDataT]:
        narrowed = self._narrow_context(context)
        self._run_entered = True
        return await self.run(params, narrowed)

    @final
    def _narrow_context(self, context: WorkflowContext) -> ContextT:
        """Hand ``run``/``rollback`` the context type the block declared.

        A plain block gets ``context`` itself. A typed flavour gets a view on the same
        state that guarantees its entity, or a `MissingRunOnEntityError` when the job's
        context does not carry it.
        """
        # `context_cls` is the runtime twin of `ContextT`; the type system cannot tie a
        # class variable to a type parameter, hence the cast.
        return cast("ContextT", self.context_cls.narrow(context))

    @final
    def parse_parameters_from_dto(self, dto: dict[str, Any], param_cls: type[ParamsT]) -> ParamsT:
        """
        Parse and validate parameters from a DTO dictionary.

        Uses Pydantic's model_validate for automatic type coercion and validation.
        Logs a warning if extra properties are present in the input.
        """
        # pyrefly doesn't understand Pydantic's model_fields descriptor
        expected_fields = set(param_cls.model_fields.keys())  # type: ignore[no-matching-overload]
        provided_fields = set(dto.keys())
        extra_fields = provided_fields - expected_fields
        if extra_fields:
            expected_fields_sorted = sorted(field for field in expected_fields if field is not None)
            self.logger.warning(
                f"Extra properties in parameters for {param_cls.__name__}: {sorted(extra_fields)}. "
                f"Expected: {expected_fields_sorted}"
            )
        return param_cls.model_validate(dto)

    @final
    def parse_forward_result_from_dto(
        self, dto: Mapping[str, Any], result_cls: type[ResultDataT]
    ) -> ForwardResult[ResultDataT, RollbackDataT]:
        """
        Parse a rollback job's forward result with this block's result and rollback data classes.

        Raises pydantic's ``ValidationError`` when it does not fit them. Result data is accepted
        by field name too, because the worker pushes it that way.
        """
        return cast(
            "ForwardResult[ResultDataT, RollbackDataT]",
            self.forward_result_cls(result_cls).model_validate(dto, by_alias=True, by_name=True),
        )

    @final
    def forward_result_cls(self, result_cls: type[ResultDataT]) -> type[ForwardResult[Any, Any]]:
        """The ForwardResult model this block's rollback jobs are parsed with."""
        rollback_data_cls = type(self).rollback_data_cls or NoRollbackData
        return ForwardResult[result_cls, rollback_data_cls]  # type: ignore[valid-type]

    @final
    def parse_acquire_parameters_from_dto(
        self, dto: dict[str, Any], registration: "Registration"
    ) -> "AcquireParams[ParamsT]":
        """Parse an ACQUIRE job's parameters leniently.

        The engine cannot resolve `{{ <step>.result… }}` or `{{ context… }}`
        references before any step has run, so it omits or nulls those fields.
        Validating that payload strictly would fail the job and, with it, the
        whole workflow — so ACQUIRE gets the lenient twin of the params model
        while EXECUTE keeps `parse_parameters_from_dto`'s full strictness.
        """
        # Imported here, not at module scope: acquire_params and partial_params
        # both import FunctionBlockParams from this module.
        from neops_worker_sdk.function_block.acquire_params import AcquireParams
        from neops_worker_sdk.function_block.partial_params import parse_leniently

        parsed, unresolved = parse_leniently(registration.param_cls, dto)

        if unresolved:
            self.logger.warning(
                f"Partial parameters at ACQUIRE for {registration.identifier}: "
                f"unresolved={sorted(unresolved)} reasons={unresolved}"
            )

        # `Registration` is not generic over its params class, so `param_cls` is
        # statically `type[FunctionBlockParams]` and its link to `ParamsT` is
        # invisible here. The worker only ever hands a block its OWN
        # registration, so the two agree at runtime.
        return AcquireParams(partial=cast("ParamsT", parsed), unresolved=frozenset(unresolved), raw=dto)

execute_function_block()

You never call run() directly. The @final method execute_function_block() wraps every execution:

  • Production (default) – catches all exceptions from run() and converts them into a failed FunctionBlockResult. A single failing block never crashes the worker process.
  • Testing (propagate_exceptions=True) – lets exceptions propagate with full stack traces so pytest points you to the exact error location.
result = await block.execute_function_block(
    params=MyParams(name="test"),
    context=context,
    propagate_exceptions=True,
)

Tip

Always use execute_function_block() in tests rather than calling run() directly. With propagate_exceptions=True you get the same execution boundary as production but with full traceability.