neops-worker-sdk-py

Python SDK for building network automation function blocks for the neops 2.0 platform. Write small, typed Python units that the workflow engine orchestrates, schedules, and scales.

@register_function_block(Registration(name="show_version", run_on="device", ...))
class ShowVersion(FunctionBlock[ShowVersionParams, ShowVersionResult]):
    async def run(self, params, context):
        async with ConnectionProxy.connect(context.device) as conn:
            output = await conn.send_command("show version")
        return FunctionBlockResult(success=True, data=ShowVersionResult(output=output))

Prerequisites

Python 3.12+
uv (recommended) or pip
A running neops-workflow-engine instance (default: http://localhost:3030)

Quick Start

uv sync --extra test               # install all dependencies (dev + fb groups are defaults)
cp .env.example .env               # configure URL_BLACKBOARD, DIR_FUNCTION_BLOCKS
uv run neops_worker                # start worker (polls engine for jobs)
uv run pytest -q && make lint && make typeCheck-baseline  # verify

Architecture

Your Function Blocks (@register_function_block)
    |
    v
Registry (global singleton, discovers FBs from DIR_FUNCTION_BLOCKS dirs)
    |
    v
Worker Startup
  |- Register worker with engine (POST /workers/register -> UUID)
  |- Register function blocks (POST /function-blocks/register for each)
  |- Start heartbeat task (POST /workers/:uuid/ping every 20s)
  \- Start job polling task (POST /blackboard/job every 10s)
        |
        v
    ThreadPoolExecutor (max_workers=1)
        |
        v
    FunctionBlock.acquire() / run() / rollback()
        |
        v
    ConnectionProxy -> ConnectionPlugin -> BaseConnection
        |                                      |
        v                                      v
    WorkflowContext (snapshot + diff)    Device (netmiko/napalm/scrapli/ncclient)
        |
        v
    Push result to engine (POST /blackboard/job/result)

Development

Install

uv sync                              # default: dev tools + fb deps (default-groups in pyproject.toml)
uv sync --extra test                 # + test deps (pytest, remote-lab)
uv sync --no-dev                     # without dev tools (fb group stays — only dev is excluded)

Run

The worker connects to a workflow engine at URL_BLACKBOARD and polls it for jobs. This repo does not ship an engine or a CMS — there is no docker-compose.yml and no lab/ here any more (see Local environment). Point the worker at an engine you already run, or at the one in the lab stack.

uv run neops_worker                  # start worker process
python -m neops_worker_sdk.cli.neops_worker  # alternative

Code Quality

uv run ruff format --check neops_worker_sdk examples tests  # format check
uv run ruff check neops_worker_sdk examples tests            # lint
uv run ruff check --fix neops_worker_sdk examples tests      # lint + auto-fix
make lint                                                    # format + lint combined
make typeCheck                                               # pyrefly, full view incl. pre-existing debt
make typeCheck-baseline                                      # pyrefly, CI gate: fails only on NEW errors (pyrefly-baseline.json)

Testing

Tests are organized in tiers using pytest markers. By default, remote lab tests are excluded.

Command	What runs
`uv run pytest`	Unit + SDK tests (default)
`uv run pytest -m function_block`	Function block integration tests
`uv run pytest -m remote_lab`	Remote lab tests (needs `REMOTE_LAB_URL`)
`make test`	Unit tests (`uv run pytest -q`)
`make test-examples`	Example function block tests
`make test-function-blocks`	Function block + remote lab tests
`make test-all`	Everything

Remote lab tests require REMOTE_LAB_URL to be set:

export REMOTE_LAB_URL=http://<remote-lab-host>:8000
make test-all

Marker	Applied by	Purpose
`function_block`	`@fb_test_case`	Local function block lifecycle tests
`remote_lab`	`@fb_test_case_with_lab`	Tests requiring a provisioned lab topology
`examples`	Auto (conftest.py)	All tests collected from `examples/`
`sdk`	–	SDK internal tests

Local environment

The full local stack — CMS, workflow engine, web client, a worker and 15 containerlab devices (10 FRRouting + 5 Nokia SR Linux) — used to live in lab/ in this repo and no longer does. It was extracted into its own repo, zebbra/neops-lab, together with the root docker-compose.yml; the local-env-*, local-lab-* and apply-cms-config make targets went with it and do not exist here.

Get the lab next to this repo, then drive it from there:

git clone git@github.com:zebbra/neops-lab.git ../neops-lab   # skip if you have it

make build-docker                                  # HERE: -> neops-worker-sdk:latest
export NEOPS_WORKER_SDK_IMAGE=neops-worker-sdk:latest
make -C ../neops-lab local-env-init                # one-time; local-lab-up aborts without it
make -C ../neops-lab local-lab-up
make -C ../neops-lab local-lab-discover            # 15 devices + interfaces land in the CMS

Building the image here is required, not optional: the lab defaults to quay.io/zebbra/neops-worker-sdk:develop, which is built from origin/develop and carries no base function blocks, so discovery fails there with “Function block … not found”. The lab also needs a Linux host with sudo-less containerlab and several GB of RAM — read ../neops-lab/README.md § Prerequisites first. Deeper notes on this seam (including the exact function block version the lab pins) are in AGENTS.md § Local Lab.

Configuration

Environment variables loaded via python-dotenv from .env. See .env.example for all options.

Variable	Default	Purpose
`URL_BLACKBOARD`	(required)	Workflow engine base URL
`DIR_FUNCTION_BLOCKS`	(required)	Comma-separated dirs to scan for function blocks
`WORKER_NAME`	(none)	Human-readable worker name
`HEARTBEAT_INTERVAL`	`20`	Seconds between heartbeat pings
`POLL_INTERVAL`	`10`	Seconds between job poll requests
`SHUTDOWN_TIMEOUT`	`60`	Seconds to wait for running job on shutdown
`BLOCKING_DETECTION_THRESHOLD`	`0.5`	Seconds threshold for blocking warnings

Docker

make build-docker                                           # production image, exactly as CI builds it (-> neops-worker-sdk:latest)
docker run --env-file .env neops-worker-sdk:latest          # run worker (needs a reachable engine)
docker build --target linter -t neops-worker-sdk:lint .     # lint stage
docker build --target test -t neops-worker-sdk:test .       # test stage

Build stages: base (Python 3.12 + uv + deps), deps-dev (+ dev/test deps), linter (ruff + pyrefly), test (pytest), run-ci (combined results).

Project Structure

neops_worker_sdk/
  cli/                Worker entry point, job processing loop
  concurrency/        @run_in_thread, run_parallel, BlockingDetector
  connection/         3-tier device connection system
    capabilities/     Abstract capability interfaces
    plugins/          Platform-specific implementations (netmiko, napalm, scrapli, ncclient)
  function_block/     FunctionBlock ABC, result types
  logger/             Loguru-based structured logging
  registry/           FB discovery, registration decorator, global registry
  testing/            Test framework (@fb_test_case, context factories)
  workflow/           WorkflowContext, entity wrappers, DB update diffing
  worker/             Worker registration with engine

neops/fb/             Built-in function blocks (`fb.base.neops.io/*`), e.g.
                      base/global/discover_network.py. Shipped inside the image.
examples/             Example function blocks (getting-started, ping, use-cases)
docs/                 MkDocs documentation source
tests/                Test suites and topologies

Contributing

Default branch: develop. Branch from develop for all changes. Run verification before committing: uv run pytest -q && make lint && make typeCheck-baseline && make audit