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 --group dev --extra test   # install all dependencies
cp .env.example .env               # configure URL_BLACKBOARD, DIR_FUNCTION_BLOCKS
uv run neops_worker                # start worker (polls engine for jobs)
uv run pytest -x -q && uv run ruff format --check neops_worker_sdk && uv run ruff check neops_worker_sdk && uv run pyrefly check  # 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                              # production deps only
uv sync --group dev                  # + dev tools (ruff, pyrefly)
uv sync --group dev --extra test     # + test deps (pytest, remote-lab)

Run

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
uv run pyrefly check                                         # type check
make lint                                                    # format + lint combined

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

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

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

docker build -t neops-worker-sdk .                          # production image
docker run --env-file .env neops-worker-sdk                 # run worker
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

examples/             Example function blocks (getting-started, ping, use-cases)
tests/                Test suites and topologies

Contributing

Default branch: develop. Branch from develop for all changes. Run verification before committing: uv run pytest -x -q && uv run ruff format --check neops_worker_sdk && uv run ruff check neops_worker_sdk && uv run pyrefly check