Rmj`+ vdZddlmZmZmZmZmZddlmZ ddlm Z ddlm Z ddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZd ededededdf dZd edefdZ d edededdfdZ!d edefdZ"d ededededdf dZ#d edefdZ$d edededdfdZ%d edefdZ&d edededdfdZ'd edefdZ(gdZ)dS)aRGlobal middleware intercept registration for tools and LLMs. Request intercepts transform inputs before execution. Execution intercepts wrap the downstream callable and can observe, modify, or replace the result. Example:: import nemo_relay def add_header( name: str, request: nemo_relay.LLMRequest, annotated: nemo_relay.AnnotatedLLMRequest | None ) -> nemo_relay.LLMRequestInterceptOutcome: # The request object is immutable, however we can return a new instance with updated headers. headers = request.headers.copy() headers["X-Trace"] = "demo" return nemo_relay.LLMRequestInterceptOutcome( nemo_relay.LLMRequest(headers=headers, content=request.content), annotated ) nemo_relay.intercepts.register_llm_request("trace-header", 10, False, add_header) )LlmExecutionInterceptLlmRequestInterceptLlmStreamExecutionInterceptToolExecutionInterceptToolRequestIntercept)"deregister_llm_execution_intercept) deregister_llm_request_intercept))deregister_llm_stream_execution_intercept)#deregister_tool_execution_intercept)!deregister_tool_request_intercept) register_llm_execution_intercept)register_llm_request_intercept)'register_llm_stream_execution_intercept)!register_tool_execution_intercept)register_tool_request_interceptnamepriority break_chainfnreturnNc&t||||S)aRegister an intercept that rewrites tool arguments before execution. Args: name: Unique intercept name used for later replacement or removal. priority: Execution order for the intercept. Lower values run first. break_chain: Whether to stop applying lower-priority request intercepts after this intercept runs. fn: Callable invoked as ``fn(tool_name, args)`` that returns the rewritten tool arguments. Returns: None: This function returns after the intercept is registered. Notes: Request intercepts run after conditional-execution guardrails and before sanitize-request guardrails or execution intercepts. Example:: import nemo_relay def add_trace_id(tool_name, args): return {**args, "trace_id": "req-123"} nemo_relay.intercepts.register_tool_request( "trace-id", 10, False, add_trace_id, ) )_native_register_tool_requestrrrrs ]/home/thesage/.hermes/hermes-agent/venv/lib/python3.11/site-packages/nemo_relay/intercepts.pyregister_tool_requestrGs@ )xb I IIc t|S)azRemove a previously registered tool request intercept. Args: name: Intercept name previously passed to ``register_tool_request()``. Returns: bool: ``True`` if an intercept was removed, otherwise ``False``. Notes: Removal affects only future executions. In-flight calls continue using the intercept chain they already resolved. )_native_deregister_tool_requestrs rderegister_tool_requestr js +4 0 00rc$t|||S)aRegister middleware around tool execution. Args: name: Unique intercept name used for later replacement or removal. priority: Execution order for the intercept. Lower values run first. fn: Callable invoked as ``fn(tool_name, args, next_call)``. The callback may await or call ``next_call(args)`` to continue the chain, modify the result, or bypass downstream execution entirely. It must return ``ToolExecutionInterceptOutcome``. Returns: None: This function returns after the intercept is registered. Notes: Execution intercepts wrap the downstream tool callback. They are the right place for timing, retries, short-circuiting, or result shaping. )_native_register_tool_executionrrrs rregister_tool_executionr$zs$ +42 > >>rc t|S)aRemove a previously registered tool execution intercept. Args: name: Intercept name previously passed to ``register_tool_execution()``. Returns: bool: ``True`` if an intercept was removed, otherwise ``False``. Notes: Removal affects only future executions. In-flight calls continue using the execution chain they already resolved. )!_native_deregister_tool_executionrs rderegister_tool_executionr's -T 2 22rc&t||||S)aRegister an intercept that rewrites an ``LLMRequest`` before execution. Args: name: Unique intercept name used for later replacement or removal. priority: Execution order for the intercept. Lower values run first. break_chain: Whether to stop applying lower-priority request intercepts after this intercept runs. fn: Callable invoked as ``fn(name, request, annotated)`` that returns an ``nemo_relay.LLMRequestInterceptOutcome`` for the next intercept or the provider callback. Returns: None: This function returns after the intercept is registered. Notes: ``annotated`` is ``None`` unless a request codec was supplied to the managed LLM call. Intercepts should preserve both values when they do not need to mutate them. Example:: import nemo_relay def add_header( name: str, request: nemo_relay.LLMRequest, annotated: nemo_relay.AnnotatedLLMRequest | None ) -> nemo_relay.LLMRequestInterceptOutcome: headers = request.headers.copy() headers["X-Trace"] = "req-123" return nemo_relay.LLMRequestInterceptOutcome( nemo_relay.LLMRequest(headers=headers, content=request.content), annotated ) nemo_relay.intercepts.register_llm_request( "trace-header", 10, False, add_header, ) )_native_register_llm_requestrs rregister_llm_requestr*sR (h R H HHrc t|S)axRemove a previously registered LLM request intercept. Args: name: Intercept name previously passed to ``register_llm_request()``. Returns: bool: ``True`` if an intercept was removed, otherwise ``False``. Notes: Removal affects only future executions. In-flight calls continue using the intercept chain they already resolved. )_native_deregister_llm_requestrs rderegister_llm_requestr-s *$ / //rc$t|||S)aRegister middleware around non-streaming LLM execution. Args: name: Unique intercept name used for later replacement or removal. priority: Execution order for the intercept. Lower values run first. fn: Callable invoked as ``fn(name, request, next_call)``. The callback may call ``next_call(request)`` to continue execution, modify the result, or short-circuit the provider call. Returns: None: This function returns after the intercept is registered. Notes: Execution intercepts wrap only non-streaming LLM execution. Use ``register_llm_stream_execution()`` for streaming callbacks. )_native_register_llm_executionr#s rregister_llm_executionr0s" *$" = ==rc t|S)aRemove a previously registered LLM execution intercept. Args: name: Intercept name previously passed to ``register_llm_execution()``. Returns: bool: ``True`` if an intercept was removed, otherwise ``False``. Notes: Removal affects only future executions. In-flight calls continue using the execution chain they already resolved. ) _native_deregister_llm_executionrs rderegister_llm_executionr3s ,D 1 11rc$t|||S)aRegister middleware around streaming LLM execution. Args: name: Unique intercept name used for later replacement or removal. priority: Execution order for the intercept. Lower values run first. fn: Callable invoked as ``fn(request, next_call)`` that returns an async iterator of JSON chunks, either by delegating to ``next_call(request)`` or by replacing the stream entirely. Returns: None: This function returns after the intercept is registered. Notes: Streaming execution intercepts wrap chunk production only. They do not replace the separate collector or finalizer callbacks. )%_native_register_llm_stream_executionr#s rregister_llm_stream_executionr6s* 1x D DDrc t|S)aRemove a previously registered streaming LLM execution intercept. Args: name: Intercept name previously passed to ``register_llm_stream_execution()``. Returns: bool: ``True`` if an intercept was removed, otherwise ``False``. Notes: Removal affects only future executions. In-flight streams continue using the execution chain they already resolved. )'_native_deregister_llm_stream_executionrs rderegister_llm_stream_executionr9s 34 8 88r)rrrrrrr r$r'r*r-r0r3r6r9)*__doc__ nemo_relayrrrrrnemo_relay._nativerr2r r,r r8r r&r rr r/rr)rr5rr"rrstrintboolrr r$r'r*r-r0r3r6r9__all__rrrBsj0 J Js J JK_ Jdh J J J JF 1# 1$ 1 1 1 1 ?#??:P?UY????*3C3D3333,)Is)Ic)I)IJ])Ibf)I)I)I)IX 0 0 0 0 0 0 >>>9N>SW>>>>(23242222"E EE $E EEEE09#9$9999"   r