Code Examples

Use these examples when you need the direct runtime surfaces behind the application instrumentation guides.

Invocation API Selection

The following table shows which API to use based on your integration need:

Manual Tool Lifecycle

Use manual lifecycle calls only when the surrounding code owns the real tool invocation and only exposes reliable start and finish hooks. If you are replaying events or bridging a framework clock, pass an explicit timestamp to the manual start, end, or mark helpers. Python accepts timezone-aware datetime values, Node.js accepts Unix microseconds since epoch, and Rust accepts DateTime<Utc>. Every manual tool end helper accepts the canonical ToolExecutionResult; wrap the business payload and add an optional opaque annotation when needed.

import nemo_relay
handle = nemo_relay.tools.call("search", {"query": "weather"}, data={"attempt": 1})
try:
    result = {"hits": 2}
finally:
    nemo_relay.tools.call_end(handle, nemo_relay.ToolExecutionResult(result))

Managed LLM Execution

Use managed execution when NeMo Relay should run the full middleware pipeline around the provider call.

import nemo_relay
from nemo_relay import LLMRequest
request = LLMRequest({}, {"messages": [{"role": "user", "content": "hello"}]})
async def invoke(req: LLMRequest) -> dict[str, object]:
    return {"text": "hi", "request": req.content}
response = await nemo_relay.llm.execute(
    "demo-provider",
    request,
    invoke,
    model_name="demo-model",
)

Streaming LLM Execution

Use the streaming helper when a model returns its response in chunks. The collector gathers those chunks. The finalizer builds one response from the collected data for the LLM end event.

When the stream ends normally, Relay runs the finalizer automatically. If you stop reading early, close the stream to stop the producer and finalize the partial response. Relay records that early close as an interruption.

The following examples read one chunk and then close the stream:

from dataclasses import dataclass
from collections.abc import AsyncIterator
from nemo_relay import LLMRequest
from nemo_relay.typed import DataclassCodec, llm_stream_execute
@dataclass
class Chunk:
    delta: str
@dataclass
class FinalResponse:
    text: str
request = LLMRequest({}, {"messages": [{"role": "user", "content": "hello"}]})
collected: list[Chunk] = []
async def stream_impl(_request: LLMRequest) -> AsyncIterator[Chunk]:
    yield Chunk(delta="hi")
stream = await llm_stream_execute(
    "demo-provider",
    request,
    stream_impl,
    collector=collected.append,
    finalizer=lambda: FinalResponse(text="".join(chunk.delta for chunk in collected)),
    chunk_json_codec=DataclassCodec(Chunk),
    response_json_codec=DataclassCodec(FinalResponse),
)
try:
    first_chunk = await anext(stream)
    # Stop early after deciding no more output is needed.
finally:
    await stream.aclose()

Handle Stream Failures

Always close a stream after you finish reading it, including after a read error. Use await stream.aclose() in Python, await stream.close() in Node.js, or stream.close().await in Rust. Closing waits for producer cleanup and finalization. Repeated calls return the same close result.

In Node.js, both llmStreamCallExecute and typedLlmStreamExecute report failures at the following points:

Node.js collectors and finalizers can return a Promise. Relay waits for a collector before yielding its chunk. It waits for the finalizer before reporting the end of the stream.

If the finalizer fails, the combined response is null. Relay omits the response payload from the end event. It does not use the call’s data value as a fallback. A finalizer failure marks an otherwise completed LLM end event as ERROR.

In Node.js, stream.close() gives priority to a stream error that next() has not yet reported. Cleanup errors take priority over finalizer errors. If next() already reported the stream error, close does not report it again. It can still report a cleanup or finalizer error. Handle errors from both reading and closing. Keep the read error if cleanup also fails, and report the cleanup error separately. If a stream error has not yet been reported, close() reports that error and logs any accompanying producer cleanup failure.

Cancel the Producer

Closing requests cancellation, but the producer must be able to stop its work. In Node.js, the provider callback receives (request, signal). Pass this AbortSignal to provider operations that support it. Stop pending work when the signal aborts. Release resources in the iterator’s cleanup path.

Close has no cleanup timeout. It can remain pending if the producer does not stop. An asynchronous collector must also finish after the producer stops. Set time limits on provider operations and make cleanup finish promptly.

Wait for Subscriber Delivery

Stream completion and subscriber delivery are separate steps. Reading the last chunk or closing the stream does not wait for queued response sanitizers and subscribers. Close the stream first. Then flush subscribers before reading captured events. Before checking exporter output or shutting down, also complete the exporter’s flush, export, or shutdown operation. Refer to Waiting for Delivery for the flush API and exporter requirements for each binding.

Temporarily Exclude a Global Registration

The following examples discover a global subscriber, register a state-driven gate, and remove the gate in a guaranteed cleanup path. Use the same registration kind during discovery and gate registration.

from nemo_relay.runtime_registrations import (
    RuntimeRegistrationKind,
    deregister_conditional_middleware_guardrail,
    list_runtime_registrations,
    register_conditional_middleware_guardrail,
)
target = next(
    (
        registration
        for registration in list_runtime_registrations(
            {RuntimeRegistrationKind.SUBSCRIBER}
        )
        if registration.local_name == "opentelemetry"
        and registration.owner.plugin_kind == "observability"
    ),
    None,
)
if target is None:
    raise RuntimeError("OpenTelemetry subscriber was not found")
paused = True
register_conditional_middleware_guardrail(
    "pause-opentelemetry",
    {RuntimeRegistrationKind.SUBSCRIBER},
    target.effective_name,
    lambda _kinds, _name: "export paused" if paused else None,
)
try:
    await run_application()
finally:
    deregister_conditional_middleware_guardrail("pause-opentelemetry")

Python and Node.js use try and finally for cleanup. Rust must preserve both the application result and deregistration result, as shown, or use an owning guard in the host application. A callback failure keeps the target eligible and adds a runtime diagnostic.

Partial Middleware Calls

These helpers are useful when framework code cannot use managed execution but still wants a request rewrite or block decision.

import nemo_relay
from nemo_relay import LLMRequest
tool_args = nemo_relay.tools.request_intercepts("search", {"query": "weather"})
nemo_relay.tools.conditional_execution("search", tool_args)
llm_request = LLMRequest({}, {"messages": [{"role": "user", "content": "hello"}]})
outcome = nemo_relay.llm.request_intercepts("demo-provider", llm_request)
nemo_relay.llm.conditional_execution(outcome.request)

Scope and Context Helpers

Use normal scope helpers first. Reach for explicit stack helpers only when work crosses thread, task, worker, or request boundaries.

from concurrent.futures import ThreadPoolExecutor
import nemo_relay
with nemo_relay.scope.scope("request", nemo_relay.ScopeType.Agent):
    nemo_relay.scope.event("started", data={"ok": True})
    shared = nemo_relay.propagate_scope_to_thread()
    def worker() -> None:
        nemo_relay.set_thread_scope_stack(shared)
        nemo_relay.scope.event("worker-ran")
    with ThreadPoolExecutor() as pool:
        pool.submit(worker).result()

Middleware Registration Families

The runtime exposes the same registration families for tool and LLM calls:

• Sanitize-request guardrails change emitted start-event payloads only
• Sanitize-response guardrails change emitted end-event payloads only
• Conditional-execution guardrails return an allow-or-block decision
• Request intercepts change the real request before execution
• Execution intercepts wrap the callback and may post-process or short-circuit
• LLM stream execution intercepts wrap streaming provider callbacks

Every family also has a scope-local surface:

• Python: nemo_relay.scope_local.register_*
• Node.js: scopeRegister*
• Rust: middleware scope_register_* functions under nemo_relay::api::registry; subscriber scope registration under nemo_relay::api::subscriber

Use Add Middleware for an end-to-end policy example and API Reference for symbol-level details.