Skip to content

neops-worker-sdk-py

Python SDK for building typed, testable function blocks that execute within the neops 2.0 workflow system. Provides the FunctionBlock ABC, a decorator-based registration system, pluggable device connections, automatic context change tracking, and a CLI worker process that polls the workflow engine’s blackboard API.

Tech: Python 3.12, Pydantic, httpx, asyncio, netmiko, napalm, ncclient, scrapli, loguru, pytest, uv, ruff, pyrefly

Development

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 for jobs)
uv run pytest                             # unit tests (excludes remote_lab by default)
uv run pytest -m function_block           # function block integration tests
uv run pytest -m remote_lab               # remote lab tests (needs REMOTE_LAB_URL)
uv run ruff check neops_worker_sdk        # lint
uv run ruff format --check neops_worker_sdk  # format check
uv run pyrefly check                      # type check
make lint                                 # `ruff format --check` + `ruff check` (read-only verification, does not auto-fix)
make test                                 # same as uv run pytest -q

Conventions

  • Default branch is develop. Branch from develop for all changes.
  • Function block parameters: Pydantic BaseModel with extra="ignore"
  • Function block results: Pydantic BaseModel with extra="forbid"
  • Use @run_in_thread for blocking I/O in async contexts
  • Example function block package name: fb.examples.neops.io (consistent across all neops repos)
  • Logging: loguru-based (neops_worker_sdk/logger/), use logger.with_context() for structured fields

Gotchas & Boundaries

  • NEVER: modify excluded fields on entities (including but not limited to: id, created_at, updated_at, permission (many fields total – see workflow/db_updates.py)) – they are silently ignored by the diff engine
  • NEVER: rely on nested object diffing – compute_db_updates() compares top-level fields only; nested changes produce full field replacement
  • ALWAYS: run uv run pytest && make lint && uv run pyrefly check before pushing
  • ALWAYS: set URL_BLACKBOARD and DIR_FUNCTION_BLOCKS env vars before starting the worker
  • Config lives in examples/config.py, not the SDK – registry.py imports from examples.config, so examples/ must be on the Python path
  • Registry is a module-level singleton – @register_function_block adds to a global Registry() instance; import order matters for discovery
  • Function blocks run in a single thread (ThreadPoolExecutor(max_workers=1)) – jobs execute sequentially, not in parallel
  • Each job gets its own asyncio.run() event loop inside the thread pool worker
  • Base FunctionBlock.rollback() returns success=False, message="Rollback not implemented" by default
  • Connection plugin precedence: user plugins silently override SDK defaults for default_for_platform; two user plugins conflicting raises PluginRegistrationError
  • Version compatibility check (client_version_compatible_with_backend()) always returns True – it is not implemented
  • neops_workflow_engine_client is exact-pinned to 0.42.0-beta.26 – SDK updates required for engine API changes
  • Worker-engine communication has no authentication

Ecosystem Context

This SDK is the worker-side component of neops 2.0. The workflow engine (neops-workflow-engine, NestJS) orchestrates execution and provides the blackboard REST API that workers poll. The CMS (neops-core, Django/GraphQL) provides device/interface/group entity data. The SDK uses an OpenAPI-generated client (neops_workflow_engine_client) built from the engine’s Swagger spec.

Called by: user function block implementations. Calls: neops-workflow-engine (blackboard API via OpenAPI client). This SDK consumes neops-remote-lab as a pytest plugin for integration testing function blocks against real virtual devices via @fb_test_case_with_lab and remote_lab_fixture().

Key Configuration

Loaded in examples/config.py via python-dotenv. Only non-obvious vars listed:

Variable Default Purpose
URL_BLACKBOARD (required) Workflow engine base URL (e.g., http://localhost:3030)
DIR_FUNCTION_BLOCKS (required) Comma-separated dirs to scan for function blocks
HEARTBEAT_INTERVAL 20 Seconds between pings – engine expects pings within 120s
POLL_INTERVAL 10 Seconds between job poll requests
SHUTDOWN_TIMEOUT 60 Seconds to wait for running job on graceful shutdown
BLOCKING_DETECTION_THRESHOLD 0.5 Seconds threshold for blocking warnings (0 to disable)

Architecture

Three-layer connection system: CapabilityInterface (abstract contracts) -> ConnectionPlugin (platform/library-specific) -> ConnectionProxy (metaclass-based proxy resolving plugins by platform/type/library). Function blocks discovered via @register_function_block decorator -> global Registry -> registered with engine at startup. Worker runs two concurrent async tasks: heartbeat pings + job polling with ThreadPoolExecutor(max_workers=1) dispatch.

Verification

uv run pytest -x -q && uv run ruff format --check neops_worker_sdk && uv run ruff check neops_worker_sdk && uv run pyrefly check