nemo_relay_plugin

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

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

Return the diagnostic with code, when present.

LlmSanitizeRequestContext

Structured per-call context provided to an LLM request sanitizer.

Methods

resolve_codec

1
def resolve_codec() -> 'WorkerRequestCodec | None'

Return the active request-codec proxy for this invocation.

LlmSanitizeResponseContext

Structured per-call context provided to an LLM response sanitizer.

Methods

resolve_codec

1
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

1
async def decode(request: LlmRequest) -> AnnotatedLlmRequest

Decode a wire request with the active host codec.

encode

1
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

1
async def decode(response: Json) -> Json

Decode a wire response with the active host codec.

DataSchema

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

Methods

to_json

1
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

1
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

1
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

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

Convert the schema tag to its canonical wire representation.

from_json

1
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

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

Convert the model identity to canonical JSON.

from_json

1
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

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

Convert the model transition to canonical JSON.

from_json

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

Decode a canonical model transition.

LlmOptimizationTokens

Retain token evidence independently from monetary pricing.

Methods

to_json

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

Convert populated token fields to canonical JSON.

from_json

1
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

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

Convert the token impact to canonical JSON.

from_json

1
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

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

Convert this contribution while flattening unknown top-level fields.

from_json

1
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

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

Convert this outcome to the canonical worker-envelope payload.

ToolExecutionResult

Canonical application-visible result of tool execution.

Methods

to_json

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

Convert this result to its canonical JSON representation.

from_json

1
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

1
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

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

Validate one component configuration before registration.

register

1
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

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

Create an instance.

runtime

1
def runtime() -> PluginRuntime

Return the host runtime handle for event and scope operations.

register_subscriber

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

Register a callback that receives Relay events.

register_conditional_middleware_guardrail

1
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

1
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

1
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

1
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

1
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

1
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

1
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

1
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

1
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

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

Register middleware around real tool execution.

register_llm_sanitize_request_guardrail

1
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

1
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

1
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

1
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

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

Register middleware around real LLM execution.

register_llm_stream_execution_intercept

1
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

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

Create an instance.

list_runtime_registrations

1
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

1
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

1
async def deregister_conditional_middleware_guardrail(handle: ConditionalMiddlewareGuardrailHandle) -> bool

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

emit_mark

1
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

1
async def runtime_diagnostics() -> RuntimeDiagnostics

Return a bounded snapshot of active host runtime diagnostics.

emit_metric

1
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

1
async def create_scope_stack() -> str

Create an isolated, host-owned scope stack.

drop_scope_stack

1
async def drop_scope_stack(scope_stack_id: str) -> None

Drop an isolated, host-owned scope stack.

push_scope

1
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) -> str

Start a scope on a Relay host-owned stack.

pop_scope

1
async def pop_scope(scope_handle_id: str, *, output: Json | None = None, metadata: Json | None = None) -> None

End a host scope by its handle identifier.

bind_scope_stack

1
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

1
def clear_scope_stack() -> Iterator[None]

Temporarily clear worker scope-stack correlation.

current_scope_stack_id

1
def current_scope_stack_id() -> str | None

Return the locally bound scope stack identifier.

current_parent_scope_id

1
def current_parent_scope_id() -> str | None

Return the locally bound parent scope identifier.

ToolNext

Continue the remaining tool execution chain.

Methods

ToolNext

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

Create an instance.

call

1
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

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

Create an instance.

call

1
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

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

Create an instance.

call

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

Call the remaining LLM stream chain with a replacement request.

Functions

serve_plugin

1
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
• 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