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.

DynamicPluginValidationStatus(_DynamicPluginValidationStatusRequired)

DynamicPluginFailure(TypedDict)

Actionable reason a dynamic plugin could not be trusted or loaded.

DynamicPluginValidationReport(_DynamicPluginValidationReportRequired)

PluginHostReport(TypedDict)

Static and dynamic validation results owned by one plugin host.

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_conditional_middleware_guardrail

1
def register_conditional_middleware_guardrail(name: str, kinds: set[RuntimeRegistrationKind], registration_name: str, guardrail: ConditionalMiddlewareGuardrail) -> None

Register an activation-owned gate for a global runtime registration.

register_event_metadata_injector

1
def register_event_metadata_injector(name: str, priority: int, callback: EventMetadataInjectorCallback) -> None

Register an event metadata injector 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.

PluginHostActivation

Owned lifetime for one process-wide static and dynamic plugin host.

Methods

report

1
def report() -> PluginHostReport

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

initialize

1
async def initialize(config: PluginConfig | JsonObject, additional_plugins_toml: str | os.PathLike[str] | None = None) -> PluginHostActivation

Initialize the core-owned static and dynamic plugin host.

activate

1
def activate(config: PluginConfig | JsonObject, additional_plugins_toml: str | os.PathLike[str] | None = None) -> AbstractAsyncContextManager[PluginHostActivation]

Initialize and close the plugin host around an async context.

validate

1
def validate(config: PluginConfig | JsonObject, additional_plugins_toml: str | os.PathLike[str] | None = None) -> PluginHostReport

Validate the core-owned static and dynamic plugin host.

validate_exact

1
def validate_exact(config: PluginConfig | JsonObject) -> PluginHostReport

Validate only the supplied static plugin configuration.

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

• DynamicPluginCheckState
• DynamicPluginKind
• UnsupportedBehavior