Build PluginsLanguage Binding

Code Examples

View as Markdown

This page shows reusable Python and Node.js plugin patterns for request interception, event logging, multi-surface policies, and framework-neutral behavior. The same plugin contract applies to Rust. Refer to the Header Plugin Example for Rust-specific registration guidance.

Dynamic Header Injection

Use an LLM request intercept when a plugin needs to inject tenant or routing metadata into every provider request.

LLM request intercepts receive name, request, and annotated. Python intercepts receive an immutable request, so they return a new request with the required changes. Node.js intercepts receive a normal JavaScript request object; return a complete outcome containing the intended request rather than relying on in-place mutation. Rust intercepts receive a mutable request.

The following complete examples define, register, validate, and activate a plugin that adds a request header. Each example clears the active configuration during shutdown.

1import asyncio
2from typing import Any
3
4import nemo_relay
5
6class HeaderPlugin:
7    def validate(self, plugin_config: dict[str, Any]) -> list[dict[str, str]]:
8        diagnostics = []
9        for field in ("header_name", "value"):
10            if not isinstance(plugin_config.get(field), str):
11                diagnostics.append({
12                    "level": "error",
13                    "code": "header-plugin.invalid_config",
14                    "component": "header-plugin",
15                    "field": field,
16                    "message": f"{field} must be a string",
17                })
18        return diagnostics
19
20    def register(self, plugin_config: dict[str, Any], context: nemo_relay.plugin.PluginContext):
21        def add_header(
22            name: str,
23            request: nemo_relay.LLMRequest,
24            annotated: nemo_relay.AnnotatedLLMRequest | None
25        ) -> nemo_relay.LLMRequestInterceptOutcome:
26            # Return a new request with updated headers.
27            headers = request.headers.copy()
28            headers[plugin_config["header_name"]] = plugin_config["value"]
29            return nemo_relay.LLMRequestInterceptOutcome(
30                nemo_relay.LLMRequest(headers=headers, content=request.content),
31                annotated,
32            )
33
34        context.register_llm_request_intercept("inject-header", 100, False, add_header)
35
36async def main() -> None:
37    nemo_relay.plugin.register("header-plugin", HeaderPlugin())
38
39    config = nemo_relay.plugin.PluginConfig(
40        components=[
41            nemo_relay.plugin.ComponentSpec(
42                kind="header-plugin",
43                config={"header_name": "x-tenant", "value": "tenant-a"},
44            )
45        ]
46    )
47    report = nemo_relay.plugin.validate(config)
48    if any(diagnostic["level"] == "error" for diagnostic in report["diagnostics"]):
49        raise RuntimeError(report["diagnostics"])
50
51    try:
52        active_report = await nemo_relay.plugin.initialize(config)
53        print("Activation report:", active_report)
54        # Run instrumented application work here.
55    finally:
56        nemo_relay.plugin.clear()
57
58
59if __name__ == "__main__":
60    asyncio.run(main())

This pattern is useful for:

  • Tenant identity
  • Trace correlation
  • Region or deployment routing

Subscriber Logging Pattern

Use a subscriber-oriented plugin when the component should watch the full lifecycle rather than rewrite requests. The following examples log each event in Python and Node.js. They do not transform events into OpenInference spans. Refer to OpenInference for the built-in OpenInference exporter.

1import asyncio
2
3import nemo_relay
4
5class EventLoggingPlugin:
6    def register(self, plugin_config, context):
7        def on_event(event):
8            print("event", event)
9
10        context.register_subscriber("event-logging", on_event)
11
12async def main() -> None:
13    nemo_relay.plugin.register("event-logging", EventLoggingPlugin())
14    config = nemo_relay.plugin.PluginConfig(
15        components=[
16            nemo_relay.plugin.ComponentSpec(kind="event-logging", config={})
17        ]
18    )
19    report = nemo_relay.plugin.validate(config)
20    if any(diagnostic["level"] == "error" for diagnostic in report["diagnostics"]):
21        raise RuntimeError(report["diagnostics"])
22
23    try:
24        active_report = await nemo_relay.plugin.initialize(config)
25        print("Activation report:", active_report)
26        # Run managed tool or LLM work here to log lifecycle events.
27    finally:
28        nemo_relay.plugin.clear()
29
30
31if __name__ == "__main__":
32    asyncio.run(main())

This is the right pattern when the component:

  • Logs events across tools and LLMs
  • Adds application-specific logging or filtering
  • Should not change execution behavior

Multi-Surface Policy Bundle

A plugin can register more than one runtime surface when one configuration document controls a related behavior bundle.

The following Python example installs a subscriber and an LLM request intercept from one component configuration:

1import asyncio
2from typing import Any
3
4import nemo_relay
5
6class CorrelationPolicy:
7    def validate(self, plugin_config: dict[str, Any]) -> list[dict[str, str]]:
8        if isinstance(plugin_config.get("header_name"), str):
9            return []
10        return [{
11            "level": "error",
12            "code": "correlation-policy.invalid_config",
13            "component": "correlation-policy",
14            "field": "header_name",
15            "message": "header_name must be a string",
16        }]
17
18    def register(
19        self,
20        plugin_config: dict[str, Any],
21        context: nemo_relay.plugin.PluginContext,
22    ) -> None:
23        def on_event(event: nemo_relay.Event) -> None:
24            print("event", event)
25
26        def add_correlation_header(
27            name: str,
28            request: nemo_relay.LLMRequest,
29            annotated: nemo_relay.AnnotatedLLMRequest | None,
30        ) -> nemo_relay.LLMRequestInterceptOutcome:
31            headers = request.headers.copy()
32            headers[plugin_config["header_name"]] = "policy-bundle"
33            return nemo_relay.LLMRequestInterceptOutcome(
34                nemo_relay.LLMRequest(headers=headers, content=request.content),
35                annotated,
36            )
37
38        context.register_subscriber("event-logging", on_event)
39        context.register_llm_request_intercept(
40            "add-correlation-header", 100, False, add_correlation_header
41        )
42
43async def main() -> None:
44    nemo_relay.plugin.register("correlation-policy", CorrelationPolicy())
45    config = nemo_relay.plugin.PluginConfig(
46        components=[
47            nemo_relay.plugin.ComponentSpec(
48                kind="correlation-policy",
49                config={"header_name": "x-correlation-source"},
50            )
51        ]
52    )
53    report = nemo_relay.plugin.validate(config)
54    if any(diagnostic["level"] == "error" for diagnostic in report["diagnostics"]):
55        raise RuntimeError(report["diagnostics"])
56
57    try:
58        await nemo_relay.plugin.initialize(config)
59        # Run managed tool or LLM work here.
60    finally:
61        nemo_relay.plugin.clear()
62
63
64if __name__ == "__main__":
65    asyncio.run(main())

This bundle registers:

  • An event-logging subscriber
  • An LLM request intercept that adds correlation metadata

Use this pattern when one component makes the configured behavior easier to reason about than several unrelated plugin components. Keep each registered surface small. Make the component configuration explicit about which surfaces are enabled.

Framework-Neutral Plugin Design

Plugins can stay framework-agnostic if they operate on the normalized runtime data rather than framework-specific objects.

Good examples:

  • Rewrite provider headers
  • Emit tracing data
  • Attach scheduling hints
  • Apply cross-framework safety policies