nat.middleware.dynamic.dynamic_function_middleware#

Classes#

DynamicFunctionMiddleware

Middleware extends FunctionMiddleware to provide dynamic discovery and

Module Contents#

class DynamicFunctionMiddleware(
config: nat.middleware.dynamic.dynamic_middleware_config.DynamicMiddlewareConfig,
builder: nat.builder.builder.Builder,
)#

Bases: nat.middleware.function_middleware.FunctionMiddleware

Middleware extends FunctionMiddleware to provide dynamic discovery and interception of all workflow components, including functions and components, without requiring explicit per-component configuration.

Initialize middleware and discover workflow functions.

Args:

config: Middleware configuration builder: Workflow builder

_config#
_builder#
_registered_callables: dict[str, nat.middleware.utils.workflow_inventory.RegisteredFunction | nat.middleware.utils.workflow_inventory.RegisteredComponentMethod]#
_builder_get_llm: collections.abc.Callable | None = None#
_builder_get_embedder: collections.abc.Callable | None = None#
_builder_get_retriever: collections.abc.Callable | None = None#
_builder_get_memory: collections.abc.Callable | None = None#
_builder_get_object_store: collections.abc.Callable | None = None#
_builder_get_auth_provider: collections.abc.Callable | None = None#
_builder_get_function: collections.abc.Callable | None = None#
_workflow_inventory: nat.middleware.utils.workflow_inventory.WorkflowInventory#
_component_allowlists: dict[nat.data_models.component.ComponentGroup, set[str]]#
property enabled: bool#

Whether this middleware should execute.

Returns config.enabled value. Framework checks this before invoking any middleware methods.

async pre_invoke(
context: nat.middleware.middleware.InvocationContext,
) → nat.middleware.middleware.InvocationContext | None#

Transform inputs before function execution.

Default implementation passes through unchanged. Override in subclass to add input transformation logic.

Args:
context: Invocation context (Pydantic model) containing:

  • function_context: Static function metadata (frozen)

  • original_args: What entered the middleware chain (frozen)

  • original_kwargs: What entered the middleware chain (frozen)

  • modified_args: Current args (mutable)

  • modified_kwargs: Current kwargs (mutable)

  • output: None (function not yet called)

Returns:

InvocationContext: Return the (modified) context to signal changes None: Pass through unchanged (framework uses current context state)

async post_invoke(
context: nat.middleware.middleware.InvocationContext,
) → nat.middleware.middleware.InvocationContext | None#

Transform output after function execution.

Default implementation passes through unchanged. Override in subclass to add output transformation logic.

Args:
context: Invocation context (Pydantic model) containing:

  • function_context: Static function metadata (frozen)

  • original_args: What entered the middleware chain (frozen)

  • original_kwargs: What entered the middleware chain (frozen)

  • modified_args: What the function received (mutable)

  • modified_kwargs: What the function received (mutable)

  • output: Current output value (mutable)

Returns:

InvocationContext: Return the (modified) context to signal changes None: Pass through unchanged (framework uses current context.output)

async _discover_and_register_llm(llm_name: str, wrapper_type: Any) → Any#

Intercept LLM creation and register allowlisted component functions with middleware.

Args:

llm_name: LLM component name wrapper_type: LLM framework wrapper type

Returns:

The LLM client instance

async _discover_and_register_embedder(
embedder_name: str,
wrapper_type: Any,
) → Any#

Intercept embedder creation and register allowlisted component functions with middleware.

Args:

embedder_name: Embedder component name wrapper_type: Embedder framework wrapper type

Returns:

The Embedder client instance

async _discover_and_register_retriever(
retriever_name: str,
wrapper_type: Any = None,
)#

Intercept retriever creation and register allowlisted component functions with middleware.

Args:

retriever_name: Retriever component name wrapper_type: Retriever framework wrapper type

Returns:

The retriever client instance

async _discover_and_register_memory(memory_name: str)#

Intercept memory creation and register allowlisted component functions with middleware.

Args:

memory_name: Memory component name

Returns:

The memory client instance

async _discover_and_register_object_store(object_store_name: str) → Any#

Intercept object store creation and register allowlisted component functions with middleware.

Args:

object_store_name: Object store component name

Returns:

The object store client instance

async _discover_and_register_auth_provider(auth_provider_name: str) → Any#

Intercept auth provider creation and register allowlisted component functions with middleware.

Args:

auth_provider_name: Auth provider component name

Returns:

The auth provider client instance

async _discover_and_register_function(
name: str | nat.data_models.component_ref.FunctionRef,
) → nat.builder.function.Function#

Intercept workflow function and register with middleware.

Args:

name: Function name or reference

Returns:

The function instance

_discover_workflow() → None#

Discover workflow functions and patch builder methods for runtime interception.

_discover_functions() → None#

Discover and register workflow functions already in the builder.

_should_intercept_llm(llm_name: str) → bool#

Check if LLM should be intercepted based on config.

Args:

llm_name: Name of the LLM to check

Returns:

True if should intercept, False otherwise

_should_intercept_embedder(embedder_name: str) → bool#

Check if embedder should be intercepted based on config.

Args:

embedder_name: Name of the embedder to check

Returns:

True if should intercept, False otherwise

_should_intercept_retriever(retriever_name: str) → bool#

Check if retriever should be intercepted based on config.

Args:

retriever_name: Name of the retriever to check

Returns:

True if should intercept, False otherwise

_should_intercept_memory(memory_name: str) → bool#

Check if memory provider should be intercepted based on config.

Args:

memory_name: Name of the memory provider to check

Returns:

True if should intercept, False otherwise

_should_intercept_object_store(store_name: str) → bool#

Check if object store should be intercepted based on config.

Args:

store_name: Name of the object store to check

Returns:

True if should intercept, False otherwise

_should_intercept_auth_provider(auth_name: str) → bool#

Check if auth provider should be intercepted based on config.

Args:

auth_name: Name of the auth provider to check

Returns:

True if should intercept, False otherwise

_register_function(
discovered: nat.middleware.utils.workflow_inventory.DiscoveredFunction,
) → None#

Register a discovered workflow function with this middleware.

Args:

discovered: A DiscoveredFunction from the workflow inventory

_register_component_function(
discovered: nat.middleware.utils.workflow_inventory.DiscoveredComponent,
function_name: str,
) → None#

Register a specific component function from a discovered component.

Args:

discovered: A DiscoveredComponent from the workflow inventory function_name: Name of the component function to register

get_registered(
key: str,
) → nat.middleware.utils.workflow_inventory.RegisteredFunction | nat.middleware.utils.workflow_inventory.RegisteredComponentMethod | None#

Get a registered callable by its key.

Args:

key: The registration key (for example, “my_llm.invoke” or “calculator__add”)

Returns:

The RegisteredFunction or RegisteredComponentMethod if found, None otherwise

get_registered_keys() → list[str]#

Get all registered callable keys.

Returns:

List of all registration keys currently tracked by this middleware

unregister(
registered: nat.middleware.utils.workflow_inventory.RegisteredFunction | nat.middleware.utils.workflow_inventory.RegisteredComponentMethod,
) → None#

Unregister a callable from middleware interception.

Args:

registered: The registered function or component method to unregister

Raises:

ValueError: If not currently registered

_configure_component_function_middleware(
discovered: nat.middleware.utils.workflow_inventory.DiscoveredComponent,
function_name: str,
) → Any#

Wrap a component function with middleware interception.

Args:

discovered: The DiscoveredComponent from the workflow inventory function_name: Name of the component function to wrap

Returns:

Wrapped component function

_build_component_allowlists() → dict[nat.data_models.component.ComponentGroup, set[str]]#

Build component allowlists from config (merged with defaults).

Returns:

Dict mapping ComponentGroup enums to sets of allowed function names

_extract_component_attributes(
discovered: nat.middleware.utils.workflow_inventory.DiscoveredComponent,
attr_name: str,
) → Any#

Safely extract an attribute from a discovered component’s instance.

Args:

discovered: DiscoveredComponent containing the component instance attr_name: Name of the attribute to extract from the component instance

Returns:

Attribute value or None if it cannot be safely extracted

_get_callable_functions(
instance: Any,
component_type: str | None = None,
) → set[str]#

Get all callable functions from component instance that can be safely wrapped.

This discovers ALL potentially wrappable component functions without allowlist filtering. Safety checks ensure only valid, callable, bound functions are included.

Args:

instance: The component instance to introspect component_type: Type of component (for logging/metadata, not filtering)

Returns:

Set of all valid component function names that could be wrapped

_is_valid_wrappable_function(
instance: Any,
function_name: str,
) → bool#

Check if a component function passes all safety checks for wrapping.

This is the gatekeeper for what CAN be wrapped (not what SHOULD be).

Args:

instance: The component instance function_name: Name of the component function to check

Returns:

True if component function is safe to wrap, False otherwise

_patch_components()#

Patch builder getter methods to enable runtime discovery and registration.

_patch_get_llm()#

Patch builder.get_llm() for runtime LLM interception.

_patch_get_embedder()#

Patch builder.get_embedder() for runtime embedder interception.

_patch_get_retriever()#

Patch builder.get_retriever() for runtime retriever interception.

_patch_get_memory()#

Patch builder.get_memory_client() for runtime memory provider interception.

_patch_get_object_store()#

Patch builder.get_object_store_client() for runtime object store interception.

_patch_get_auth_provider()#

Patch builder.get_auth_provider() for runtime auth provider interception.

_patch_get_function()#

Patch builder.get_function() for runtime function interception.

_get_builder_get_llm()#

Return original builder.get_llm method.

_get_builder_get_embedder()#

Return original builder.get_embedder method.

_get_builder_get_retriever()#

Return original builder.get_retriever method.

_get_builder_get_memory_client()#

Return original builder.get_memory_client method.

_get_builder_get_object_store()#

Return original builder.get_object_store_client method.

_get_builder_get_auth_provider()#

Return original builder.get_auth_provider method.

_get_builder_get_function()#

Return original builder.get_function method.