> This page is for NeMo Relay, version v0.11.0. > For other versions, use one of these documentation indexes: > - Latest (v0.10.4) (default): https://docs.nvidia.com/nemo/relay/latest/llms.txt > - dev: https://docs.nvidia.com/nemo/relay/dev/llms.txt > - v0.11.0: https://docs.nvidia.com/nemo/relay/v0.11.0/llms.txt > - v0.10.4: https://docs.nvidia.com/nemo/relay/v0.10.4/llms.txt > - v0.9.4: https://docs.nvidia.com/nemo/relay/v0.9.4/llms.txt > - v0.8.4: https://docs.nvidia.com/nemo/relay/v0.8.4/llms.txt > - v0.7.3: https://docs.nvidia.com/nemo/relay/v0.7.3/llms.txt > For clean Markdown of any page, append .md to the page URL. > For a complete documentation index, see https://docs.nvidia.com/nemo/relay/llms.txt. > For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/relay/_mcp/server. # Python Worker > Build, package, activate, verify, and remove the checked 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](/nemo/relay/v0.11.0/build-plugins/workers/middleware-and-continuations), 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. ```python title={"Lookup diagnostics"} showLineNumbers={false} 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](/nemo/relay/v0.11.0/build-plugins/diagnostics) for the shared pattern, built-in code declarations, and native/worker API differences. The [delivery protocol](/nemo/relay/v0.11.0/build-plugins/workers/grpc-v1-protocol#diagnostic-delivery) defines host limits, retries, and session authority. ## Test the Package Run the package's atomic tests before creating a managed environment: 1. Enter the example directory and run its self-contained test project. ```bash 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](/nemo/relay/v0.11.0/build-plugins/package-discoverable-plugins). A minimal `pyproject.toml` therefore needs a standard build backend, a package, and the 0.11 worker SDK dependency: ```toml [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: ```powershell $env:NEMO_RELAY_PLUGIN_SNAPSHOT_DIR = "C:\\nrs" ``` ```bash export NEMO_RELAY_PLUGIN_SNAPSHOT_DIR=/var/tmp/nrs ``` ```python 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 `` placeholder from the current `worker.py`; changing that file requires a new digest. ```toml 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:" [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](/nemo/relay/v0.11.0/build-plugins/workers/grpc-v1-protocol) 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: 1. From `examples/python-grpc-worker-plugin`, create a clean temporary Relay state and add the manifest. ```bash 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. 2. Enable the component and start Relay. ```bash 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: 1. 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. 2. 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. 3. 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. 4. 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. ```bash 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.