nemo_relay.plugin

Generated from python/nemo_relay/plugin.pyi.

Module nemo_relay.plugin.

Generic plugin configuration and registration helpers.

Classes

ConfigDiagnostic(_ConfigDiagnosticRequired)

One plugin validation diagnostic.

RuntimeDiagnostic(_RuntimeDiagnosticRequired)

One aggregated runtime failure from an active plugin.

ConfigReport(TypedDict)

Validation or activation report for a plugin config.

PluginContext(Protocol)

Component-scoped registration context passed to custom plugin handlers.

Methods

register_subscriber

1
def register_subscriber(name: str, callback: Callable[[Event], None]) -> None

Register an infallible event subscriber for this component.

register_mark_sanitize_guardrail

1
def register_mark_sanitize_guardrail(name: str, priority: int, callback: EventSanitizeGuardrail) -> None

Register a mark event sanitizer for this component.

register_scope_sanitize_start_guardrail

1
def register_scope_sanitize_start_guardrail(name: str, priority: int, callback: EventSanitizeGuardrail) -> None

Register a scope-start event sanitizer for this component.

register_scope_sanitize_end_guardrail

1
def register_scope_sanitize_end_guardrail(name: str, priority: int, callback: EventSanitizeGuardrail) -> None

Register a scope-end event sanitizer for this component.

register_tool_sanitize_request_guardrail

1
def register_tool_sanitize_request_guardrail(name: str, priority: int, callback: ToolSanitizeGuardrail) -> None

Register a tool sanitize-request guardrail for this component.

register_tool_sanitize_response_guardrail

1
def register_tool_sanitize_response_guardrail(name: str, priority: int, callback: ToolSanitizeGuardrail) -> None

Register a tool sanitize-response guardrail for this component.

register_tool_conditional_execution_guardrail

1
def register_tool_conditional_execution_guardrail(name: str, priority: int, callback: ToolConditionalExecutionGuardrail) -> None

Register a tool conditional-execution guardrail for this component.

register_llm_sanitize_request_guardrail

1
def register_llm_sanitize_request_guardrail(name: str, priority: int, callback: LlmSanitizeRequestGuardrail) -> None

Register an LLM sanitize-request guardrail for this component.

register_llm_sanitize_response_guardrail

1
def register_llm_sanitize_response_guardrail(name: str, priority: int, callback: LlmSanitizeResponseGuardrail) -> None

Register an LLM sanitize-response guardrail for this component.

register_llm_conditional_execution_guardrail

1
def register_llm_conditional_execution_guardrail(name: str, priority: int, callback: LlmConditionalExecutionGuardrail) -> None

Register an LLM conditional-execution guardrail for this component.

register_llm_request_intercept

1
def register_llm_request_intercept(name: str, priority: int, break_chain: bool, callback: LlmRequestIntercept) -> None

Register an LLM request intercept for this component.

register_llm_execution_intercept

1
def register_llm_execution_intercept(name: str, priority: int, callback: LlmExecutionIntercept) -> None

Register an LLM execution intercept for this component.

register_llm_stream_execution_intercept

1
def register_llm_stream_execution_intercept(name: str, priority: int, callback: LlmStreamExecutionIntercept) -> None

Register an LLM streaming execution intercept for this component.

register_tool_request_intercept

1
def register_tool_request_intercept(name: str, priority: int, break_chain: bool, callback: ToolRequestIntercept) -> None

Register a tool request intercept for this component.

register_tool_execution_intercept

1
def register_tool_execution_intercept(name: str, priority: int, callback: ToolExecutionIntercept) -> None

Register a tool execution intercept for this component.

Plugin(Protocol)

Custom plugin callback contract.

Methods

validate

1
def validate(plugin_config: JsonObject) -> list[ConfigDiagnostic] | None

Validate one component-local config object.

register

1
def register(plugin_config: JsonObject, context: PluginContext) -> None

Install middleware and subscribers for one component instance.

ConfigPolicy

Policy for unsupported plugin configuration.

Methods

ConfigPolicy

1
def ConfigPolicy(unknown_component: UnsupportedBehavior = 'warn', unknown_field: UnsupportedBehavior = 'warn', unsupported_value: UnsupportedBehavior = 'error') -> None

Create an instance.

to_dict

1
def to_dict() -> JsonObject

Serialize this policy to the canonical JSON object shape.

ComponentSpec

One top-level custom plugin component.

Methods

ComponentSpec

1
def ComponentSpec(kind: str, enabled: bool = True, config: JsonObject = ...) -> None

Create an instance.

to_dict

1
def to_dict() -> JsonObject

Serialize this component to the canonical JSON object shape.

PluginConfig

Canonical plugin configuration document.

Methods

PluginConfig

1
def PluginConfig(version: int = 1, components: list[object] = ..., policy: ConfigPolicy = ...) -> None

Create an instance.

to_dict

1
def to_dict() -> JsonObject

Serialize this config to the canonical JSON document shape.

DynamicPluginActivationSpec

One dynamic plugin component to load and activate.

Methods

DynamicPluginActivationSpec

1
def DynamicPluginActivationSpec(plugin_id: str, kind: DynamicPluginKind, manifest_ref: str, environment_ref: str | None = None, config: JsonObject = ...) -> None

Create an instance.

to_dict

1
def to_dict() -> JsonObject

Serialize this activation specification to its canonical JSON shape.

PluginHostActivation

Owned lifetime for one process-wide dynamic plugin host.

Methods

report

1
def report() -> ConfigReport

Return the validation report captured during activation.

is_active

1
def is_active() -> bool

Return whether this activation handle has not begun teardown.

close

1
async def close() -> None

Clear callbacks and unload plugins; repeated calls are safe.

Functions

load_dynamic_plugin_activation_specs

1
def load_dynamic_plugin_activation_specs(plugin_config_path: str | os.PathLike[str]) -> list[DynamicPluginActivationSpec]

Load dynamic activation specs from one standard plugins.toml.

validate

1
def validate(config: PluginConfig | JsonObject) -> ConfigReport

Validate a plugin configuration without changing runtime state.

initialize

1
async def initialize(config: PluginConfig | JsonObject) -> ConfigReport

Validate and activate a plugin configuration.

initialize_with_dynamic_plugins

1
async def initialize_with_dynamic_plugins(config: PluginConfig | JsonObject, dynamic_plugins: Sequence[DynamicPluginActivationSpec | JsonObject]) -> PluginHostActivation

Initialize registered components with dynamic plugins as one owned host.

clear

1
def clear() -> None

Clear the active plugin configuration.

clear_async

1
async def clear_async() -> None

Clear the active plugin configuration without blocking asyncio.

plugin

1
def plugin(config: PluginConfig | JsonObject) -> AsyncContextManager[ConfigReport]

Context manager for plugin initialization and cleanup.

report

1
def report() -> ConfigReport | None

Return the active plugin report or a failed-teardown diagnostic report.

list_kinds

1
def list_kinds() -> list[str]

List registered custom plugin kinds.

register

1
def register(plugin_kind: str, plugin: Plugin) -> None

Register a custom plugin implementation.

deregister

1
def deregister(plugin_kind: str) -> bool

Deregister a custom plugin kind.

Type Aliases And Constants

• DynamicPluginKind
• UnsupportedBehavior