Your First Function Block

In this guide you will build a function block from scratch — a small, self-contained unit of automation logic that the neops platform can schedule and execute. By the end you will understand parameters, results, registration, and the run() method.

Python concepts you'll see in this guide

This guide uses a few Python features that may be new to you. Here is a quick orientation:

Type hints — annotations like text: str or count: int that tell Python (and your editor) what type each variable should be. They enable autocomplete, catch bugs early, and are required by the framework.

Pydantic — a popular Python library for data validation. You define a class with typed fields, and Pydantic ensures the data always matches. If someone passes a number where a string is expected, Pydantic raises a clear error automatically.

async def — the framework uses Python's async features internally. For now, treat async def as boilerplate: your code inside run() works just like regular Python. You can use normal for loops, if statements, and library calls.

Generics (FunctionBlock[X, Y]) — the square brackets tell the framework which input and output types this block uses. Think of it as a template: FunctionBlock[InputType, OutputType]. Your editor uses this for autocomplete.

What you'll build

An Echo function block that receives a piece of text and returns it unchanged. It is deliberately simple so you can focus on the structure rather than the logic. Every function block you write later — configuring interfaces, collecting facts, pushing templates — follows the exact same pattern.

Step 1: Imports

Every function block needs a few imports from the SDK:

from neops_worker_sdk.function_block.function_block import (
    FunctionBlock,
    FunctionBlockAcquireResult,
    FunctionBlockParams,
    FunctionBlockResult,
    FunctionBlockResultData,
)
from neops_worker_sdk.registry.decorator import register_function_block
from neops_worker_sdk.registry.registration import Registration
from neops_worker_sdk.workflow.workflow_context import WorkflowContext

These give you the base classes for parameters, results, the function block itself, and the registration decorator.

Step 2: Define parameters and results

Function blocks communicate through typed data classes. Parameters are the inputs your block receives when the workflow engine schedules it; results are the structured outputs it returns.

Parameters

class EchoParameters(FunctionBlockParams):
    message: str

FunctionBlockParams is a Pydantic model. By writing text: str, you are saying: "this block requires a string input called text." Pydantic validates this automatically — if someone passes a number instead of a string, you get a clear error.

Results

class EchoResult(FunctionBlockResultData):
    echoed_message: str

FunctionBlockResultData works the same way. Whatever you define here is persisted on the blackboard and available to downstream steps in the workflow.

Step 3: Register the function block

Registration tells the platform what your block is called, which entities it operates on, and how it behaves. You do this with a decorator above the class:

@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,
    )
)

Key fields:

Field	Purpose
`name`	Unique identifier for this function block
`package`	Reverse-DNS namespace to avoid collisions across teams
`version`	Semantic version tuple `(major, minor, patch)`
`run_on`	Entity scope — `"device"`, `"interface"`, `"group"`, or `"global"`
`fb_type`	Operation category — `"execute"`, `"facts"`, `"check"`, `"configure"`, `"none"`
`param_cls` / `result_cls`	Links to the parameter and result models defined above
`is_pure`	`True` if the block has no side effects (e.g. reading config, not writing it)
`is_idempotent`	`True` if running it twice produces the same outcome

Note

is_pure and is_idempotent help the workflow engine decide when it is safe to retry or cache. You will learn more about these in the Pure and Idempotent guide.

Step 4: Implement the logic

With the metadata in place, write the class that does the actual work:

class Echo(FunctionBlock[EchoParameters, EchoResult]):
    async def run(self, params: EchoParameters, context: WorkflowContext) -> FunctionBlockResult[EchoResult]:
        del context
        return FunctionBlockResult(
            message="Echo executed successfully.",
            success=True,
            data=EchoResult(echoed_message=params.message),
        )

    async def acquire(self, params: EchoParameters) -> FunctionBlockAcquireResult:
        del params
        return FunctionBlockAcquireResult(message="No resources required.", success=True, acquires=None)

Breaking this down:

FunctionBlock[EchoParameters, EchoResult] — the square brackets tell the framework (and your editor) which parameter and result types this block uses. This enables full autocomplete on params.message and data.echoed_message.
async def run(...) — the entry point the worker calls. The async keyword is required by the framework; your code inside works like regular Python. It receives validated params and a WorkflowContext that carries entity data.
FunctionBlockResult — wrap your response with a success flag, a human-readable message, and your typed data.
del context / del params — a Python idiom that tells linters and readers "this argument is intentionally unused." The statement discards the local reference but has no effect on the actual data. An alternative is to prefix the argument with an underscore (_context).
acquire() — called before run() to request additional data from the CMS or lock shared resources. For blocks that don't need this (like our Echo), return a simple success. The method is required by the framework — every function block must implement it.

No custom __init__

The FunctionBlock.__init__ method is marked @final — you cannot override it. This ensures consistent logger setup and lifecycle management. If your block needs initialization state, use the acquire() phase or set it up in run().

Keep run() focused

A function block should do one thing well. If you find yourself writing hundreds of lines in run(), consider splitting the work into multiple blocks that compose inside a workflow.

What happens next

When you deploy a worker, it automatically discovers every decorated function block, registers them with the platform, and polls for jobs. But you do not need a full neops instance to verify your code works — the next section shows you how to test it locally with pytest.

Full example

Here is the complete Echo function block in one file:

from neops_worker_sdk.function_block.function_block import (
    FunctionBlock,
    FunctionBlockAcquireResult,
    FunctionBlockParams,
    FunctionBlockResult,
    FunctionBlockResultData,
)
from neops_worker_sdk.registry.decorator import register_function_block
from neops_worker_sdk.registry.registration import Registration
from neops_worker_sdk.workflow.workflow_context import WorkflowContext



class EchoParameters(FunctionBlockParams):
    message: str




class EchoResult(FunctionBlockResultData):
    echoed_message: str




@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,
    )
)
class Echo(FunctionBlock[EchoParameters, EchoResult]):
    async def run(self, params: EchoParameters, context: WorkflowContext) -> FunctionBlockResult[EchoResult]:
        del context
        return FunctionBlockResult(
            message="Echo executed successfully.",
            success=True,
            data=EchoResult(echoed_message=params.message),
        )

    async def acquire(self, params: EchoParameters) -> FunctionBlockAcquireResult:
        del params
        return FunctionBlockAcquireResult(message="No resources required.", success=True, acquires=None)

See it in action

The echo function block you just built is used in the Workflow Engine's Your First Workflow tutorial. Deploy the hello-world workflow, start your worker, and watch the engine orchestrate the function block you just wrote.

Next: Connecting to Devices to open your first device connection, or jump to Test your function block to verify it works locally with pytest.