nemo_flow.intercepts#

Global 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_flow

def add_header(name, request, annotated):
    request.headers["X-Trace"] = "demo"
    return request, annotated

nemo_flow.intercepts.register_llm_request("trace-header", 10, False, add_header)

Functions#

`register_tool_request`(name, priority, break_chain, fn)	Register an intercept that rewrites tool arguments before execution.
`deregister_tool_request`(name)	Remove a previously registered tool request intercept.
`register_tool_execution`(name, priority, fn)	Register middleware around tool execution.
`deregister_tool_execution`(name)	Remove a previously registered tool execution intercept.
`register_llm_request`(name, priority, break_chain, fn)	Register an intercept that rewrites an `LLMRequest` before execution.
`deregister_llm_request`(name)	Remove a previously registered LLM request intercept.
`register_llm_execution`(name, priority, fn)	Register middleware around non-streaming LLM execution.
`deregister_llm_execution`(name)	Remove a previously registered LLM execution intercept.
`register_llm_stream_execution`(name, priority, fn)	Register middleware around streaming LLM execution.
`deregister_llm_stream_execution`(name)	Remove a previously registered streaming LLM execution intercept.

Module Contents#

nemo_flow.intercepts.register_tool_request( name: str, priority: int, break_chain: bool, fn: nemo_flow.ToolRequestIntercept, ) → None#

Register an intercept that rewrites tool arguments before execution.

Parameters:

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:

This function returns after the intercept is registered.

Return type:

None

Notes

Request intercepts run after conditional-execution guardrails and before sanitize-request guardrails or execution intercepts.

Example:

import nemo_flow

def add_trace_id(tool_name, args):
    return {**args, "trace_id": "req-123"}

nemo_flow.intercepts.register_tool_request(
    "trace-id",
    10,
    False,
    add_trace_id,
)

nemo_flow.intercepts.deregister_tool_request(name: str) → bool#

Remove a previously registered tool request intercept.

Parameters:: name – Intercept name previously passed to register_tool_request().
Returns:: True if an intercept was removed, otherwise False.
Return type:: bool

Notes

Removal affects only future executions. In-flight calls continue using the intercept chain they already resolved.

nemo_flow.intercepts.register_tool_execution( name: str, priority: int, fn: nemo_flow.ToolExecutionIntercept, ) → None#

Register middleware around tool execution.

Parameters:

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.

Returns:

This function returns after the intercept is registered.

Return type:

None

Notes

Execution intercepts wrap the downstream tool callback. They are the right place for timing, retries, short-circuiting, or result shaping.

nemo_flow.intercepts.deregister_tool_execution(name: str) → bool#

Remove a previously registered tool execution intercept.

Parameters:: name – Intercept name previously passed to register_tool_execution().
Returns:: True if an intercept was removed, otherwise False.
Return type:: bool

Notes

Removal affects only future executions. In-flight calls continue using the execution chain they already resolved.

nemo_flow.intercepts.register_llm_request( name: str, priority: int, break_chain: bool, fn: nemo_flow.LlmRequestIntercept, ) → None#

Register an intercept that rewrites an LLMRequest before execution.

Parameters:

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 a tuple of (request, annotated) for the next intercept or the provider callback.

Returns:

This function returns after the intercept is registered.

Return type:

None

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_flow

def add_header(name, request, annotated):
    request.headers["X-Trace"] = "req-123"
    return request, annotated

nemo_flow.intercepts.register_llm_request(
    "trace-header",
    10,
    False,
    add_header,
)

nemo_flow.intercepts.deregister_llm_request(name: str) → bool#

Remove a previously registered LLM request intercept.

Parameters:: name – Intercept name previously passed to register_llm_request().
Returns:: True if an intercept was removed, otherwise False.
Return type:: bool

Notes

Removal affects only future executions. In-flight calls continue using the intercept chain they already resolved.

nemo_flow.intercepts.register_llm_execution( name: str, priority: int, fn: nemo_flow.LlmExecutionIntercept, ) → None#

Register middleware around non-streaming LLM execution.

Parameters:

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:

This function returns after the intercept is registered.

Return type:

None

Notes

Execution intercepts wrap only non-streaming LLM execution. Use register_llm_stream_execution() for streaming callbacks.

nemo_flow.intercepts.deregister_llm_execution(name: str) → bool#

Remove a previously registered LLM execution intercept.

Parameters:: name – Intercept name previously passed to register_llm_execution().
Returns:: True if an intercept was removed, otherwise False.
Return type:: bool

Notes

Removal affects only future executions. In-flight calls continue using the execution chain they already resolved.

nemo_flow.intercepts.register_llm_stream_execution( name: str, priority: int, fn: nemo_flow.LlmStreamExecutionIntercept, ) → None#

Register middleware around streaming LLM execution.

Parameters:

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:

This function returns after the intercept is registered.

Return type:

None

Notes

Streaming execution intercepts wrap chunk production only. They do not replace the separate collector or finalizer callbacks.

nemo_flow.intercepts.deregister_llm_stream_execution(name: str) → bool#

Remove a previously registered streaming LLM execution intercept.

Parameters:: name – Intercept name previously passed to register_llm_stream_execution().
Returns:: True if an intercept was removed, otherwise False.
Return type:: bool

Notes

Removal affects only future executions. In-flight streams continue using the execution chain they already resolved.