Python Worker

The examples/python-grpc-worker-plugin package uses the 0.11.0 nemo-relay-plugin SDK and the shared documentation configuration. Its worker registers all 16 surfaces, accepts synchronous and asynchronous callback forms where the Python SDK permits them, and relies on the SDK for protobuf stubs, the authenticated server, and cooperative task cancellation.

Internal Diagnostic Observations

Use await runtime.diagnostics() to obtain diagnostics for the activation. The SDK negotiates delivery once and reuses the host’s selected context. It does not create a worker-local store or enable diagnostic collection.

This example matches the Rust native and Rust worker example. Register its component once during setup, then retain it in the callbacks that perform lookups. Python supports async def register, so that hook can await the helper using ctx.runtime. Keep both returned handles for the activation. It declares separate failure and success codes, records positive occurrence counts, and attaches a request ID. Call the failure helper only after observing an unavailable backend. Call the success helper only after a completed lookup.

Lookup diagnostics

from nemo_relay_plugin import (
    DiagnosticCodeDefinition,
    DiagnosticComponent,
    DiagnosticExtensionDefinition,
    DiagnosticObservation,
    PluginRuntime,
    WorkerDiagnostics,
)
from nemo_relay_plugin import (
    DiagnosticObservationOutcome as Outcome,
)
from nemo_relay_plugin import (
    DiagnosticObservationSeverity as Severity,
)
async def register_lookup_diagnostics(
    runtime: PluginRuntime,
) -> tuple[WorkerDiagnostics, DiagnosticComponent]:
    diagnostics = await runtime.diagnostics()
    component = await diagnostics.component(
        "example.lookup",
        [
            DiagnosticCodeDefinition(
                extension=DiagnosticExtensionDefinition(
                    name="lookup_failed",
                    description="The lookup backend was unavailable",
                    outcome=Outcome.DIAGNOSTIC_OBSERVATION_OUTCOME_FAILURE,
                    allowed_failure_classes=["unavailable"],
                )
            ),
            DiagnosticCodeDefinition(
                extension=DiagnosticExtensionDefinition(
                    name="lookup_completed",
                    description="A lookup completed",
                    outcome=Outcome.DIAGNOSTIC_OBSERVATION_OUTCOME_SUCCESS,
                    allowed_failure_classes=[],
                )
            ),
        ],
    )
    return diagnostics, component
def record_lookup_failure(component: DiagnosticComponent, request_id: str) -> None:
    component.try_record_lazy(
        lambda: DiagnosticObservation(
            code=1,
            count=1,
            outcome=Outcome.DIAGNOSTIC_OBSERVATION_OUTCOME_FAILURE,
            severity=Severity.DIAGNOSTIC_OBSERVATION_SEVERITY_ERROR,
            failure_class="unavailable",
            detail="The lookup backend was unavailable",
            request_id=request_id,
        )
    )
def record_lookup_success(component: DiagnosticComponent, request_id: str) -> None:
    component.try_record_lazy(
        lambda: DiagnosticObservation(
            code=2,
            count=1,
            outcome=Outcome.DIAGNOSTIC_OBSERVATION_OUTCOME_SUCCESS,
            severity=Severity.DIAGNOSTIC_OBSERVATION_SEVERITY_INFO,
            detail="A lookup completed",
            request_id=request_id,
        )
    )

Handle registration and recording failures separately from execution failures. Python raises DiagnosticDeliveryError, whose kind field identifies the error. The helpers expose those errors so the caller can apply its delivery policy. Do not propagate an optional diagnostic failure as the lookup’s result.

try_record_lazy returns None after local queue admission. It does not return a host acceptance receipt. An inactive component returns without recording. Disabled or unsupported collection skips the observation factory. A full queue also skips the factory. diagnostics.availability reports negotiated availability.

Check diagnostics.delivery_stats() for dropped, rejected, and uncertain work. Use await diagnostics.flush(timeout) to wait for earlier queue work. The timeout is in seconds. Flush does not return per-record receipts or acknowledge disk storage. Never replay uncertain observations into a new session.

Use await component.close() when the component’s work ends. SDK shutdown limits diagnostic drain time and closes delivery tasks. Explicit delivery shutdown uses await diagnostics.close(timeout). Canonical snapshot and memory reads are not available to workers, even when a native plugin has a read grant.

Refer to Plugin Diagnostics for the shared pattern, built-in code declarations, and native/worker API differences. The delivery protocol defines host limits, retries, and session authority.

Test the Package

Run the package’s atomic tests before creating a managed environment:

Enter the example directory and run its self-contained test project.
```
cd examples/python-grpc-worker-plugin
uv run --locked --group test pytest
```
Every test creates its own worker instance and mock host context, and any one test can be selected by node ID without running the rest of the suite. The tests separate configuration, JSON Schema, source digest, wheel packaging, registration metadata, sanitizers, policies, request outcomes, continuations, streams, and runtime cleanup. A digest mismatch is a packaging failure, not an activation warning.

Define the Installable Python Package

Relay creates the managed environment from the package root named by the manifest. A minimal pyproject.toml therefore needs a standard build backend, a package, and the 0.11 worker SDK dependency:

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "nemo-relay-python-grpc-worker-example"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = ["nemo-relay-plugin>=0.11.0"]
[dependency-groups]
test = ["pytest>=8", "pytest-asyncio>=0.26"]
[tool.setuptools.packages.find]
where = ["."]
include = ["nemo_relay_python_grpc_worker_example"]

The module entrypoint is an async function, not a server factory or a prebound port. Relay supplies the authenticated endpoint and activation credentials when it starts the managed command.

Configure the Snapshot Location

Relay copies the worker runtime closure into an isolated activation snapshot before starting the worker. Set NEMO_RELAY_PLUGIN_SNAPSHOT_DIR in the environment that starts Relay to use a directory of your choice; Relay creates it when it does not exist. Choose a location outside every directory copied into a snapshot, including active plugin packages and external entrypoint directories:

$env:NEMO_RELAY_PLUGIN_SNAPSHOT_DIR = "C:\\nrs"

export NEMO_RELAY_PLUGIN_SNAPSHOT_DIR=/var/tmp/nrs

from nemo_relay_plugin import ConfigDiagnostic, DiagnosticLevel, Json, PluginContext, WorkerPlugin, serve_plugin
class ExamplePythonWorker(WorkerPlugin):
    plugin_id = "examples.python_grpc_worker"
    allows_multiple_components = False
    def validate(self, config: Json) -> list[ConfigDiagnostic]:
        return validate_config(config)
    def register(self, ctx: PluginContext, config: Json) -> None:
        diagnostics = validate_config(config)
        errors = [
            item for item in diagnostics
            if item.level == DiagnosticLevel.ERROR
        ]
        if errors:
            raise ValueError(errors[0].message)
        settings = normalized_config(config)
        if settings["observe"]["enabled"]:
            self._register_observation(
                ctx, settings["tag"], settings["observe"]
            )
        if settings["requests"]["enabled"]:
            self._register_requests(
                ctx,
                settings["tag"],
                settings["requests"],
                settings["execution"],
            )
        self._register_runtime(ctx, settings["tag"], settings["runtime"])
        if settings["execution"]["enabled"]:
            self._register_execution(ctx, settings["tag"], settings["execution"])
async def main() -> None:
    await serve_plugin(ExamplePythonWorker())

Relay calls validation before registration, but register still rejects invalid direct use instead of assuming every possible host followed the expected sequence.

The corresponding manifest tells Relay to install this directory and import main from the installed module. Calculate the <worker-source-sha256> placeholder from the current worker.py; changing that file requires a new digest.

manifest_version = 1
[plugin]
id = "examples.python_grpc_worker"
kind = "worker"
[compat]
relay = ">=0.10.0,<1.0"
worker_protocol = "grpc-v1"
[capabilities]
items = ["plugin_worker", "config_schema"]
[config_schema]
path = "config.schema.json"
[source]
manifest_root = "."
artifact = "nemo_relay_python_grpc_worker_example/worker.py"
[integrity]
sha256 = "sha256:<worker-source-sha256>"
[load]
runtime = "python"
entrypoint = "nemo_relay_python_grpc_worker_example.worker:main"

The manifest requires Relay 0.10 because this worker uses the context-aware tool execution contract and the LlmExecutionContext callback shape. The protocol identifier remains grpc-v1. ToolNext.call() returns ToolExecutionResult, whose result contains the application payload and whose optional annotation remains adjacent opaque metadata. See the protocol reference for the complete boundary.

Install and Activate the Package

Use the following procedure to install the package into Relay’s managed environment and start the worker:

From examples/python-grpc-worker-plugin, create a clean temporary Relay state and add the manifest.
```
relay_tmp="$(mktemp -d)"
relay_config="$relay_tmp/gateway.toml"
: > "$relay_config"
nemo-relay --config "$relay_config" plugins add ./relay-plugin.toml
```
plugins add creates an isolated managed environment and installs source.manifest_root with pip. Set NEMO_RELAY_PYTHON only for this add operation when Relay must use a non-default base interpreter. Standard pip index, proxy, certificate, and wheelhouse variables control dependency resolution.
Enable the component and start Relay.
```
nemo-relay --config "$relay_config" plugins enable examples.python_grpc_worker
nemo-relay --config "$relay_config" --bind 127.0.0.1:4040
```
The activation report should identify examples.python_grpc_worker, and the worker handshake should advertise all 16 supported surfaces.

Verify Behavior and Clean Up

Use the following procedure to verify each callback family and clean up the managed environment:

Exercise one allowed and one blocked tool, one allowed and one blocked model, a unary LLM continuation, and a multi-chunk stream. Confirm configured headers, sanitized event fields, preserved annotations, pending marks, and lazy chunk transformation.
Cancel a long-running async callback and abandon a worker stream. Confirm the SDK task receives cancellation and the worker’s finally cleanup runs. A synchronous callback cannot be preempted, so the example keeps synchronous work bounded.
Emit a mark, use a nested scope, create and bind an isolated stack, then force a failure. Confirm the prior scope context is restored and the stack is dropped.
Stop Relay with Ctrl+C, then remove the plugin and delete the temporary state from the shell where relay_tmp and relay_config remain defined.
```
nemo-relay --config "$relay_config" plugins remove examples.python_grpc_worker
rm -rf -- "$relay_tmp"
```

plugins remove deletes the Relay-managed environment. Copying a Python worker manifest into plugins.toml is not an equivalent installation path because no attested environment would exist. Success means the managed environment is created and later removed, every feature group has an observable call-path result, cancellation cleanup runs, and no worker process remains after shutdown.