> This page is for NeMo Relay, version v0.10.0.
> For other versions, use one of these documentation indexes:
> - Latest (v0.10.0) (default): https://docs.nvidia.com/nemo/relay/latest/llms.txt
> - dev: https://docs.nvidia.com/nemo/relay/dev/llms.txt
> - v0.10.0: https://docs.nvidia.com/nemo/relay/v0.10.0/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
> - v0.6.0: https://docs.nvidia.com/nemo/relay/v0.6.0/llms.txt
> - v0.5.0: https://docs.nvidia.com/nemo/relay/v0.5.0/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.

# nemo_relay_plugin

> Author out-of-process NeMo Relay worker plugins in Python.

Generated from `python/plugin/src/nemo_relay_plugin/_api.py`.

Module `nemo_relay_plugin`.

Author out-of-process NeMo Relay worker plugins in Python.

## Classes

### `EventSanitizeFields(TypedDict)`

Observability fields returned by event sanitizer callbacks.

### `LlmCodecIdentity`

Structured identity of the codec active for one worker invocation.

### `RuntimeDiagnostic`

One bounded runtime diagnostic reported by the Relay host.

### `RuntimeRegistrationKind(str, Enum)`

Global runtime registration surfaces that a worker can gate.

### `RuntimeRegistrationOwnerKind(str, Enum)`

Owner categories reported by host registration discovery.

### `RuntimeRegistrationOwner`

Owner metadata for a discovered runtime registration.

### `RuntimeRegistrationIdentity`

Structured identity for a global gateable runtime registration.

### `ConditionalMiddlewareGuardrailHandle`

Opaque activation-owned key for removing a worker-created gate.

### `RuntimeDiagnostics`

Bounded snapshot of active host runtime diagnostics.

#### Methods

##### `get`

```python
def get(code: str) -> RuntimeDiagnostic | None
```

Return the diagnostic with `code`, when present.

### `LlmSanitizeRequestContext`

Request codec context shared by sanitizer and execution callbacks.

#### Methods

##### `resolve_codec`

```python
def resolve_codec() -> 'WorkerRequestCodec | None'
```

Return the active request-codec proxy for this invocation.

### `LlmSanitizeResponseContext`

Response codec context shared by sanitizer and execution callbacks.

#### Methods

##### `resolve_codec`

```python
def resolve_codec() -> 'WorkerResponseCodec | None'
```

Return the active response-codec proxy for this invocation.

### `WorkerRequestCodec`

Invocation-scoped async proxy for an active request codec.

#### Methods

##### `decode`

```python
async def decode(request: LlmRequest) -> AnnotatedLlmRequest
```

Decode a wire request with the active host codec.

##### `encode`

```python
async def encode(annotated: AnnotatedLlmRequest, original: LlmRequest) -> LlmRequest
```

Encode an annotated request with the active host codec.

### `WorkerResponseCodec`

Invocation-scoped async proxy for an active response codec.

#### Methods

##### `decode`

```python
async def decode(response: Json) -> Json
```

Decode a wire response with the active host codec.

### `LlmExecutionContext`

Directional codec context for one LLM execution invocation.

### `DataSchema`

Identify the schema of a mark's opaque data payload.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, str]
```

Return the canonical JSON representation.

### `LogSeverity(str, Enum)`

Telemetry severity assigned to an exported mark log.

### `EventCategory(str, Enum)`

Well-known semantic categories for Relay events.

### `MetricKind(str, Enum)`

OpenTelemetry instrument kind recorded by a metric mark.

### `MetricValueType(str, Enum)`

Explicit numeric representation of a metric measurement value.

### `MetricMeasurement(TypedDict)`

One recording operation in a Relay metric-measurement mark.

### `WorkerSdkError(Exception)`

Report a worker SDK, host-call, or protocol error to plugin code.

### `DiagnosticLevel(str, Enum)`

Identify the severity of a plugin configuration diagnostic.

### `ConfigDiagnostic`

Describe one problem found while validating plugin configuration.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Any]
```

Convert the diagnostic to its Relay JSON representation.

### `PendingMarkSpec`

Describe a mark Relay emits under a managed lifecycle scope.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert this pending mark to its canonical JSON object.

### `LlmOptimizationDataSchema`

Identify the schema of an opaque optimization contribution payload.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert the schema tag to its canonical wire representation.

##### `from_json`

```python
def from_json(value: Mapping[str, Json]) -> LlmOptimizationDataSchema
```

Decode a canonical schema tag.

### `LlmOptimizationModel`

Identify one model for optimization accounting and repricing.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert the model identity to canonical JSON.

##### `from_json`

```python
def from_json(value: Mapping[str, Json]) -> LlmOptimizationModel
```

Decode a canonical model identity.

### `LlmOptimizationModelTransition`

Describe the counterfactual and effective models for one optimization.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert the model transition to canonical JSON.

##### `from_json`

```python
def from_json(value: Mapping[str, Json]) -> LlmOptimizationModelTransition
```

Decode a canonical model transition.

### `LlmOptimizationTokens`

Retain token evidence independently from monetary pricing.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert populated token fields to canonical JSON.

##### `from_json`

```python
def from_json(value: Mapping[str, Json]) -> LlmOptimizationTokens
```

Decode canonical token evidence.

### `LlmOptimizationEvidenceQuality(str, Enum)`

Classify token evidence as directly observed or estimated.

### `LlmOptimizationTokenImpact`

Describe baseline, effective, and explicitly saved token evidence.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert the token impact to canonical JSON.

##### `from_json`

```python
def from_json(value: Mapping[str, Json]) -> LlmOptimizationTokenImpact
```

Decode a canonical token impact.

### `LlmOptimizationContribution`

One plugin's lossless, forward-compatible optimization evidence.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert this contribution while flattening unknown top-level fields.

##### `from_json`

```python
def from_json(value: Mapping[str, Json]) -> LlmOptimizationContribution
```

Decode all known fields and retain unknown top-level fields in `extra`.

### `LlmRequestInterceptOutcome`

Canonical result returned by a Python worker LLM request intercept.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert this outcome to the canonical worker-envelope payload.

### `ToolExecutionContext`

Per-call context provided to a tool execution intercept.

### `ToolExecutionResult`

Canonical application-visible result of tool execution.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert this result to its canonical JSON representation.

##### `from_json`

```python
def from_json(value: Json) -> 'ToolExecutionResult'
```

Decode and validate a canonical JSON representation.

### `ToolExecutionInterceptOutcome`

Canonical result returned by a Python worker tool execution intercept.

#### Methods

##### `to_json`

```python
def to_json() -> dict[str, Json]
```

Convert this outcome to its canonical JSON representation.

### `ScopeType(str, Enum)`

Identify the semantic category of a Relay execution scope.

### `WorkerPlugin`

Define the validation and registration contract for a worker plugin.

#### Methods

##### `validate`

```python
def validate(config: Json) -> list[ConfigDiagnostic | dict[str, Any]] | Awaitable[list[ConfigDiagnostic | dict[str, Any]]]
```

Validate one component configuration before registration.

##### `register`

```python
def register(ctx: PluginContext, config: Json) -> None | Awaitable[None]
```

Register plugin callbacks for one component configuration.

### `PluginContext`

Register component-scoped callbacks and access the Relay host runtime.

#### Methods

##### `PluginContext`

```python
def PluginContext(runtime: PluginRuntime | None = None) -> None
```

Create an instance.

##### `runtime`

```python
def runtime() -> PluginRuntime
```

Return the host runtime handle for event and scope operations.

##### `register_subscriber`

```python
def register_subscriber(name: str, callback: SubscriberCallback) -> None
```

Register a callback that receives Relay events.

##### `register_conditional_middleware_guardrail`

```python
def register_conditional_middleware_guardrail(name: str, kinds: set[RuntimeRegistrationKind], registration_name: str, callback: ConditionalMiddlewareCallback) -> None
```

Declare a callback-based gate installed with this activation.

##### `register_event_metadata_injector`

```python
def register_event_metadata_injector(name: str, callback: EventMetadataInjectorCallback, *, priority: int = 0) -> None
```

Register a callback that proposes additions to Event metadata.

##### `register_mark_sanitize_guardrail`

```python
def register_mark_sanitize_guardrail(name: str, callback: EventSanitizeCallback, *, priority: int = 0) -> None
```

Register a sanitizer for mark event observability fields.

##### `register_scope_sanitize_start_guardrail`

```python
def register_scope_sanitize_start_guardrail(name: str, callback: EventSanitizeCallback, *, priority: int = 0) -> None
```

Register a sanitizer for scope start event observability fields.

##### `register_scope_sanitize_end_guardrail`

```python
def register_scope_sanitize_end_guardrail(name: str, callback: EventSanitizeCallback, *, priority: int = 0) -> None
```

Register a sanitizer for scope end event observability fields.

##### `register_tool_sanitize_request_guardrail`

```python
def register_tool_sanitize_request_guardrail(name: str, callback: ToolSanitizeCallback, *, priority: int = 0) -> None
```

Register a guardrail that sanitizes tool input for observability.

##### `register_tool_sanitize_response_guardrail`

```python
def register_tool_sanitize_response_guardrail(name: str, callback: ToolSanitizeCallback, *, priority: int = 0) -> None
```

Register a guardrail that sanitizes tool output for observability.

##### `register_tool_conditional_execution_guardrail`

```python
def register_tool_conditional_execution_guardrail(name: str, callback: ToolConditionalCallback, *, priority: int = 0) -> None
```

Register a guardrail that can block tool execution.

##### `register_tool_request_intercept`

```python
def register_tool_request_intercept(name: str, callback: ToolRequestCallback, *, priority: int = 0, break_chain: bool = False) -> None
```

Register an intercept that rewrites tool arguments.

##### `register_tool_execution_intercept`

```python
def register_tool_execution_intercept(name: str, callback: ToolExecutionCallback, *, priority: int = 0) -> None
```

Register middleware around real tool execution.

##### `register_llm_sanitize_request_guardrail`

```python
def register_llm_sanitize_request_guardrail(name: str, callback: LlmSanitizeRequestCallback, *, priority: int = 0) -> None
```

Register a guardrail that sanitizes an LLM request for observability.

##### `register_llm_sanitize_response_guardrail`

```python
def register_llm_sanitize_response_guardrail(name: str, callback: LlmSanitizeResponseCallback, *, priority: int = 0) -> None
```

Register a guardrail that sanitizes an LLM response for observability.

##### `register_llm_conditional_execution_guardrail`

```python
def register_llm_conditional_execution_guardrail(name: str, callback: LlmConditionalCallback, *, priority: int = 0) -> None
```

Register a guardrail that can block LLM execution.

##### `register_llm_request_intercept`

```python
def register_llm_request_intercept(name: str, callback: LlmRequestCallback, *, priority: int = 0, break_chain: bool = False) -> None
```

Register an intercept that rewrites an LLM request.

##### `register_llm_execution_intercept`

```python
def register_llm_execution_intercept(name: str, callback: LlmExecutionCallback, *, priority: int = 0) -> None
```

Register middleware around real LLM execution.

##### `register_llm_stream_execution_intercept`

```python
def register_llm_stream_execution_intercept(name: str, callback: LlmStreamExecutionCallback, *, priority: int = 0) -> None
```

Register middleware around real streaming LLM execution.

### `PluginRuntime`

Call event, scope, and continuation operations on the Relay host.

#### Methods

##### `PluginRuntime`

```python
def PluginRuntime(*, activation_id: str, auth_token: str, host_stub: Any) -> None
```

Create an instance.

##### `log`

```python
async def log(level: str, message: str, *, target: str = '', fields: Mapping[str, Json] | None = None) -> None
```

Emit one operational log record through Relay's configured sinks.

##### `trace`

```python
async def trace(message: str, *, target: str = '', fields: Mapping[str, Json] | None = None) -> None
```

Emit a trace-level operational log record through Relay.

##### `debug`

```python
async def debug(message: str, *, target: str = '', fields: Mapping[str, Json] | None = None) -> None
```

Emit a debug-level operational log record through Relay.

##### `info`

```python
async def info(message: str, *, target: str = '', fields: Mapping[str, Json] | None = None) -> None
```

Emit an info-level operational log record through Relay.

##### `warn`

```python
async def warn(message: str, *, target: str = '', fields: Mapping[str, Json] | None = None) -> None
```

Emit a warn-level operational log record through Relay.

##### `error`

```python
async def error(message: str, *, target: str = '', fields: Mapping[str, Json] | None = None) -> None
```

Emit an error-level operational log record through Relay.

##### `list_runtime_registrations`

```python
async def list_runtime_registrations(kinds: Iterable[RuntimeRegistrationKind] | None = None) -> list[RuntimeRegistrationIdentity]
```

List global gateable registrations known to the Relay host.

##### `register_conditional_middleware_guardrail`

```python
async def register_conditional_middleware_guardrail(name: str, kinds: set[RuntimeRegistrationKind], registration_name: str, callback: ConditionalMiddlewareCallback) -> ConditionalMiddlewareGuardrailHandle
```

Register a callback-based gate owned by this worker activation.

##### `deregister_conditional_middleware_guardrail`

```python
async def deregister_conditional_middleware_guardrail(handle: ConditionalMiddlewareGuardrailHandle) -> bool
```

Remove a callback-based gate owned by this worker activation.

##### `emit_mark`

```python
async def emit_mark(name: str, data: Json | None = None, metadata: Json | None = None, *, data_schema: DataSchema | Mapping[str, Json] | None = None, severity: LogSeverity | str | None = None, category: EventCategory | str | None = None, scope_stack_id: str | None = None, parent_scope_id: str | None = None) -> None
```

Emit a mark event through the Relay host runtime.

##### `runtime_diagnostics`

```python
async def runtime_diagnostics() -> RuntimeDiagnostics
```

Return a bounded snapshot of active host runtime diagnostics.

##### `emit_metric`

```python
async def emit_metric(name: str, measurements: Iterable[MetricMeasurement | Mapping[str, Json]], metadata: Json | None = None, *, category: EventCategory | str | None = None, scope_stack_id: str | None = None, parent_scope_id: str | None = None) -> None
```

Emit a Relay metric-measurement mark through the host runtime.

##### `create_scope_stack`

```python
async def create_scope_stack() -> str
```

Create an isolated, host-owned scope stack.

##### `drop_scope_stack`

```python
async def drop_scope_stack(scope_stack_id: str) -> None
```

Drop an isolated, host-owned scope stack.

##### `push_scope`

```python
async def push_scope(name: str, *, scope_type: ScopeType = ScopeType.CUSTOM, data: Json | None = None, metadata: Json | None = None, input: Json | None = None, scope_stack_id: str | None = None, parent_scope_id: str | None = None, timestamp: datetime | None = None) -> str
```

Start a scope on a Relay host-owned stack.

##### `pop_scope`

```python
async def pop_scope(scope_handle_id: str, *, output: Json | None = None, metadata: Json | None = None, timestamp: datetime | None = None) -> None
```

End a host scope by its handle identifier.

##### `bind_scope_stack`

```python
def bind_scope_stack(scope_stack_id: str | None, *, parent_scope_id: str | None = None) -> Iterator[None]
```

Temporarily bind host calls to a worker-selected scope stack.

##### `clear_scope_stack`

```python
def clear_scope_stack() -> Iterator[None]
```

Temporarily clear worker scope-stack correlation.

##### `current_scope_stack_id`

```python
def current_scope_stack_id() -> str | None
```

Return the locally bound scope stack identifier.

##### `current_parent_scope_id`

```python
def current_parent_scope_id() -> str | None
```

Return the locally bound parent scope identifier.

### `ToolNext`

Continue the remaining tool execution chain.

#### Methods

##### `ToolNext`

```python
def ToolNext(runtime: PluginRuntime, continuation_id: str) -> None
```

Create an instance.

##### `call`

```python
async def call(value: Json) -> ToolExecutionResult
```

Call the remaining tool execution chain with replacement arguments.

### `LlmNext`

Continue the remaining unary LLM execution chain.

#### Methods

##### `LlmNext`

```python
def LlmNext(runtime: PluginRuntime, continuation_id: str) -> None
```

Create an instance.

##### `call`

```python
async def call(request: LlmRequest) -> Json
```

Call the remaining LLM execution chain with a replacement request.

### `LlmStreamNext`

Continue the remaining streaming LLM execution chain.

#### Methods

##### `LlmStreamNext`

```python
def LlmStreamNext(runtime: PluginRuntime, continuation_id: str) -> None
```

Create an instance.

##### `call`

```python
def call(request: LlmRequest) -> AsyncIterator[Json]
```

Call the remaining LLM stream chain with a replacement request.

## Functions

### `serve_plugin`

```python
async def serve_plugin(plugin: _SupportsWorkerPlugin) -> None
```

Run a local `grpc-v1` worker until the Relay host shuts it down.

## Type Aliases And Constants

* `ANNOTATED_LLM_REQUEST_SCHEMA`
* `AnnotatedLlmRequest`
* `ConditionalMiddlewareCallback`
* `DATA_SCHEMA_SCHEMA`
* `EVENT_SCHEMA`
* `Event`
* `EventMetadataInjectorCallback`
* `EventSanitizeCallback`
* `JSON_SCHEMA`
* `Json`
* `LLM_REQUEST_INTERCEPT_OUTCOME_SCHEMA`
* `LLM_REQUEST_SCHEMA`
* `LlmConditionalCallback`
* `LlmExecutionCallback`
* `LlmRequest`
* `LlmRequestCallback`
* `LlmRequestContext`
* `LlmResponseContext`
* `LlmSanitizeRequestCallback`
* `LlmSanitizeResponseCallback`
* `LlmStreamExecutionCallback`
* `METRIC_DATA_SCHEMA_NAME`
* `METRIC_DATA_SCHEMA_VERSION`
* `PLUGIN_DIAGNOSTICS_SCHEMA`
* `SubscriberCallback`
* `ToolConditionalCallback`
* `ToolExecutionCallback`
* `ToolRequestCallback`
* `ToolSanitizeCallback`
* `WORKER_PROTOCOL`
* `grpc`
* `pb`
* `pb_grpc`