Processor SDK quickstart

A processor is a program outside the platform that annotates like a person does. It pulls jobs over HTTPS, so it needs no inbound ports; it decrypts only the inputs it was granted, encrypts everything it writes, and posts its results as proposals for the project’s members to correct.

The Python SDK (yggdrasil-processor, Python 3.10 or later) does the protocol and the cryptography. You write one function that takes a Job.

1. Install

pip install "yggdrasil-processor[volumes]"   # the volumes extra adds numpy, nibabel and pydicom

To install from source instead:

git clone https://github.com/lorenzobrg/yggdrasil4
pip install "./yggdrasil4/sdk/python-processor[volumes]"

2. Write the function

threshold.py

from yggdrasil_processor import Job
from yggdrasil_processor.volumes import labelmap_bytes, load_series


def process(job: Job) -> None:
    for series in job.targets_as("series"):
        volume = load_series(series, job)  # an [i, j, k] numpy array on the canonical geometry
        mask = (volume.data > job.parameters["threshold"]).astype("uint8")
        job.log(f"{int(mask.sum())} voxels", progress=1.0)
        job.add_labelmap("mask", series, labelmap_bytes(volume.geometry, mask), {1: "foreground"})

• job.targets_as("series") lists the series of the job, with their decrypted files and metadata.
• load_series(series, job) reads a series’ volume. A series of several volumes (MR echoes, phases, diffusion directions, a 4D NIfTI file) gives its first by default; load_series(series, job, volume=1) reads another, and series_volumes(series) lists what tells them apart. Labelmaps and shapes lie on the series’ geometry, the first volume’s.
• job.parameters are validated against the descriptor’s parameter schema, with defaults filled in.
• job.log(...) sends an encrypted log line and the progress the project’s members see.
• job.add_labelmap, job.add_annotation and job.add_asset add outputs. The SDK checks each result against its type’s schema, encrypts and uploads the outputs, then posts the results. The server cannot read results, so an invalid one fails the job before anything is posted. job.add_frame_mask adds a mask of one video frame: a propagation writes one per frame.
• An exception fails the job with its message, encrypted for the project’s members. Long functions call job.check() to stop when the job is cancelled.

3. Describe it

A descriptor.json declares the processor’s inputs, outputs, parameters, triggers and access. It follows the processor descriptor schema:

descriptor.json

{
  "id": "ygg.threshold",
  "version": "1.0.0",
  "name": "Threshold segmentation",
  "description": "Labels every voxel above a threshold.",
  "inputs": { "level": "series", "modalities": ["ygg.volume@1"] },
  "outputs": [{ "id": "mask", "type": "ygg.segmentation.labelmap@1", "labels": ["foreground"] }],
  "parameters": {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "properties": { "threshold": { "type": "number", "default": 300 } },
    "additionalProperties": false
  },
  "triggers": ["manual", "on-upload"],
  "access": "per-job"
}

per-job access, the default, means the processor receives only the keys of each job’s inputs.

4. Register it

1. Generate the processor’s keys. The private keys are written with mode 0600; the public keys are printed.

   ygg-processor keygen --keys processor-keys.json

2. A project admin adds the processor to the project’s manifest (setup, Processors step) with the descriptor file, or picks it from the processors the installation offers (the catalog a superadmin keeps in the console; the descriptor’s bytes are the catalog’s, and the badge “Offered by …” names the installation’s institution, not a verification), publishes, then registers it under Processors with the descriptor and the public keys. The admin’s browser signs the registration; the registration ID, a token and the project root are shown. The token is shown once. A processor with full-project access receives the project key, so the admin confirms that before signing.

3. Check the registration and the project’s signed log without claiming a job:

   export YGG_PROCESSOR_URL=https://yggdrasil.example.org   # the instance's API
   export YGG_PROCESSOR_ID=<registration ID>
   export YGG_PROCESSOR_TOKEN=<token>
   export YGG_PROCESSOR_ROOT=<project root>                 # the trust anchor
   ygg-processor check

   The project root is the processor’s trust anchor: the processor refuses a project history with another root, so the server cannot show it a forged history. Get it from the admin through a channel you trust. Without a root the processor does not start. --tofu (YGG_PROCESSOR_TOFU=1) instead trusts the history it sees first, which a server could exploit; use it only for tests.

5. Run it

ygg-processor run threshold:process

The runner claims jobs until it is stopped; --once exits after one job. --jobs N runs up to N jobs at once in one process (1 by default), and --transfers N sets how many downloads or uploads one job runs at once (4 by default). A download cut off midway resumes where it stopped. The first SIGINT or SIGTERM lets the current jobs finish; a second one stops at once and the jobs return to the queue when their leases expire.

Before decrypting or sealing anything, the SDK replays the project’s signed log, checks that its root is the configured one, pins it with the furthest head seen, and checks that the registration names these keys and the exact descriptor. Keys granted to the processor are bound to the project root, so they open only under the real history. Every output the SDK writes (masks, results, log lines, failure details) is signed with the processor’s key; members’ browsers check that signature against the registration before showing or accepting the output, and grant a job only inputs they verified to lie inside its target. Keep the trust state (--state) on persistent storage, such as a Docker volume. Decrypted inputs are written under a working directory that is deleted when the job ends, so run processors on machines trusted with the project’s data.

New series

A processor can also write a new series next to the one it ran on, of the same modality or another: a panoramic radiograph projected from a CBCT, a resampled volume, a picture of a slide. Its descriptor declares a series output with the modality plugin it produces, and the project’s manifest binds that output to one of its modality entries:

{ "id": "projection", "kind": "series", "modality": "ygg.image2d@1", "formats": ["png"] }

def process(job: Job) -> None:
    (series,) = job.targets_as("series")
    derived = job.add_series(series, "projection", {"description": "Maximum-intensity projection"})
    derived.add_asset(png_bytes, {"format": "png", "mediaType": "image/png", "image": {"width": w, "height": h}})

The original is never changed. The new series is encrypted and signed like every output, and its signed metadata names its source, the processor and the job, which members’ browsers check. Members see it once the job succeeded, in the worklist with a “Derived” mark and in the workspace with a link to its source; it opens in its modality’s viewer, gets that modality’s tasks and is exported like any series. It is plain data, not a proposal: whoever may delete series deletes it, and deleting its source deletes it too. The processors that run on upload also run on new series (chaining), started by the browser of a member allowed to run processors while the workspace is open; each runs once per series, never on a series derived from its own output. examples/projection projects volumes to pictures this way.

Stacking processors

A processor can read what another one wrote. A multi-class classifier that needs the segmentation of a CBCT, for example, runs after a tooth segmenter: its descriptor names the input with an ID, the type it reads and, optionally, the statuses that count (submitted annotations by default; a descriptor’s accepted means submitted since review gave way to correction):

"inputs": {
  "level": "series",
  "modalities": ["ygg.volume@1"],
  "annotations": [
    { "id": "segmentation", "type": "ygg.segmentation.labelmap@1", "description": "The series' segmentation" }
  ]
}

The project’s manifest binds that input to a task: the one the segmenter’s output writes. The setup wizard binds it to the only task of its type (or makes one) and shows it as “Needs segmentation from task …”, which the admin can change:

"processors": [
  { "id": "segmenter", "trigger": "on-upload", "access": "per-job", "outputs": [{ "output": "mask", "task": "teeth" }] },
  {
    "id": "classifier", "trigger": "on-upload", "access": "per-job",
    "inputs": [{ "input": "segmentation", "task": "teeth" }],
    "outputs": [{ "output": "burden", "task": "caries" }]
  }
]

The function reads the input by its ID and opens the labelmap on its series:

def process(job: Job) -> None:
    for series in job.targets_as("series"):
        total_ml = 0.0
        for annotation in job.inputs("segmentation", series):  # the bound task's, submitted
            total_ml += sum(job.read_labelmap(annotation).volumes_ml().values())
        job.add_annotation("burden", series, {"choice": "large" if total_ml >= 1 else "small"})

Both run on upload, and the classifier waits: a new CBCT starts the segmenter, a member corrects or confirms its segmentation (or the task takes its results as they are), and then the browser of a member allowed to run processors starts the classifier on that series, once. The upload page and the run dialog say what an automatic processor waits for, and the Processors page shows each chain in one line, “CBCT → Tooth segmentation (Tooth segmenter) → Caries (Caries classifier)”. A later change to the segmentation does not run the classifier again: run it by hand. examples/classify-from-segmentation is this classifier.

More

• The SDK README lists every setting, exit code and check.
• docker/Dockerfile packages a processor as an image.
• examples/totalsegmentator is the reference processor: TotalSegmentator, with a descriptor for whole-body CT segmentation (total) and one for craniofacial structures on CT and CBCT.
• examples/sam2-video propagates a reader’s masks of video frames through the video with SAM 2.1, writing a mask proposal on each annotated frame, which the reader takes over on every frame at once.
• Interactive processors answer a member while they work, with ygg-processor serve module:handler: examples/region-grow segments from clicks, and examples/scripted-captions shows live dictation with transcription_handler (a transcription processor receives the reader’s voice and no project data; it must keep no audio).
• The processor HTTP API is the contract, for processors written in other languages.