From 3ea79cdde93dd955e803d23165d7549571d70bd4 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 4 Aug 2026 21:49:41 -0700 Subject: [PATCH 1/5] feat: add external NAT reference adapter Signed-off-by: Ajay Thorve --- external/README.md | 19 + external/nat/README.md | 115 +++ external/nat/examples/calculator.py | 86 ++ external/nat/examples/email_phishing.py | 89 ++ external/nat/fabric-adapter.json | 73 ++ .../src/nemo_fabric_adapters/nat/__init__.py | 4 + .../src/nemo_fabric_adapters/nat/adapter.py | 858 ++++++++++++++++++ tests/adapters/test_external_nat_adapter.py | 708 +++++++++++++++ 8 files changed, 1952 insertions(+) create mode 100644 external/README.md create mode 100644 external/nat/README.md create mode 100644 external/nat/examples/calculator.py create mode 100644 external/nat/examples/email_phishing.py create mode 100644 external/nat/fabric-adapter.json create mode 100644 external/nat/src/nemo_fabric_adapters/nat/__init__.py create mode 100644 external/nat/src/nemo_fabric_adapters/nat/adapter.py create mode 100644 tests/adapters/test_external_nat_adapter.py diff --git a/external/README.md b/external/README.md new file mode 100644 index 00000000..e9a1719e --- /dev/null +++ b/external/README.md @@ -0,0 +1,19 @@ + + +# External Adapter References + +This directory contains source-only reference adapters that exercise the public +NVIDIA NeMo Fabric adapter contract without becoming bundled NeMo Fabric +adapters. Each reference owns its descriptor, implementation, examples, and +documentation. + +These adapters are not published as wheels and are not wired into bundled or +installed-adapter discovery. Packaging and discovery are separate concerns from +the adapter contract demonstrated here. + +| Harness | Adapter ID | Reference | +| --- | --- | --- | +| NVIDIA NeMo Agent Toolkit | `nvidia.fabric.nat` | [NAT adapter](nat/README.md) | diff --git a/external/nat/README.md b/external/nat/README.md new file mode 100644 index 00000000..493173fe --- /dev/null +++ b/external/nat/README.md @@ -0,0 +1,115 @@ + + +# NVIDIA NeMo Fabric NAT Reference Adapter + +This source-only adapter runs an NVIDIA NeMo Agent Toolkit (NAT) workflow behind +the NeMo Fabric lifecycle contract. It is a third-party adapter reference, not a +bundled NeMo Fabric adapter or a published package. + +The implementation is generic. It constructs NAT configuration in memory from +`FabricConfig`; it does not read a NAT YAML file and does not hardcode the +calculator or email-phishing components. + +## Configuration Boundary + +NeMo Fabric owns portable configuration. `harness.settings` contains only +NAT-native component configuration that has no portable NeMo Fabric equivalent. + +| NeMo Fabric input | NAT configuration | +| --- | --- | +| `models.` | `llms.`; every Fabric model-role name is preserved | +| `instructions.system` | Built-in `react_agent` workflow `additional_instructions`; other workflow types reject this field in the initial adapter | +| `harness.settings.workflow` | `workflow` | +| `harness.settings.functions` | `functions` | +| `harness.settings.function_groups` | `function_groups` | +| Harness-native `mcp.servers.` | Generated `mcp_client` function group named `` | +| `tools.enabled`, `tools.blocked` | Effective NAT-native workflow tool selection | + +The adapter loads installed `nat.components` entry points before NAT validates +the generated configuration. A custom function, function group, or workflow is +therefore supplied as an installed NAT component package and selected by its +registered `_type`. No Python import path or callable crosses `FabricConfig`. + +At runtime, `start` loads components, enters one `WorkflowBuilder`, creates a +`SessionManager` with that shared builder, and retains both resources. Each +`invoke` opens a session from the retained manager, enters `session.run(...)`, +and awaits `runner.result()`. `stop` shuts down the session manager and exits +the builder context. This first reference does not claim cancellation, service, +streaming, or live-update support. + +## MCP Tool Filters + +The adapter consumes the routed `capability_plan.native.mcp_servers` entries, +including the normalized per-server filters. Fabric MCP tool names remain bare +server-local names; NAT exposes a selected member as `__`. + +| Fabric server policy | Generated NAT function group | +| --- | --- | +| `allowed_tools` omitted and `blocked_tools=[]` | No `include` or `exclude`; expose all discovered tools | +| Nonempty `allowed_tools` only | `include=allowed_tools` | +| `allowed_tools` omitted and nonempty `blocked_tools` | `exclude=blocked_tools` | +| Both lists configured | `include=allowed_tools`; Fabric requires `blocked_tools` to be disjoint, so those names are already outside the allowlist | +| `allowed_tools=[]` | Omit the generated group; expose no tools from that server | + +NAT rejects a function group that sets both `include` and `exclude`, so the +adapter emits only `include` whenever an allowlist is present. Fabric rejects +blank names and an allow/block overlap before adapter startup. A nonempty +generated MCP group is added to workflows that expose `tool_names`; callers do +not repeat portable MCP servers in `harness.settings`. A workflow implementation +that requires at least one tool can still reject a configuration whose effective +tool set is empty. + +Per-server MCP filters and root `tools.enabled` or `tools.blocked` solve +different problems. MCP filters select members within one server. Root tool +policy selects across the effective NAT-native tool surface. + +## Development Bootstrap + +This directory intentionally has no package metadata or discovery wiring. Until +source resolution and third-party descriptor discovery are available, use one +Python environment for NeMo Fabric, the common adapter host, NAT, and every NAT +component referenced by the config: + +```bash +uv pip install \ + nemo-fabric-adapters-common \ + nvidia-nat-core \ + nvidia-nat-langchain \ + nvidia-nat-mcp +export PYTHONPATH="$PWD/external/nat/src${PYTHONPATH:+:$PYTHONPATH}" +``` + +`PYTHONPATH` is a development bootstrap limitation, not the target installation +contract. Stage the descriptor in the current agent-local discovery location: + +```bash +mkdir -p .tmp/nat-reference/adapters/nat +cp external/nat/fabric-adapter.json \ + .tmp/nat-reference/adapters/nat/fabric-adapter.json +``` + +The calculator example also requires a streamable-HTTP calculator MCP server. +Set its endpoint and run the typed `FabricConfig` example: + +```bash +export CALCULATOR_MCP_URL=http://127.0.0.1:9901/mcp +uv run python external/nat/examples/calculator.py \ + --base-dir "$PWD/.tmp/nat-reference" +``` + +The email-phishing example uses the NAT example component registered by +`nat_email_phishing_analyzer`. Install that component from a NAT checkout, then +run the example: + +```bash +uv pip install -e \ + /path/to/NeMo-Agent-Toolkit/examples/evaluation_and_profiling/email_phishing_analyzer +uv run python external/nat/examples/email_phishing.py \ + --base-dir "$PWD/.tmp/nat-reference" +``` + +Both examples accept `--plan` to inspect the resolved plan without starting the +runtime. They use Python `FabricConfig` objects only; no YAML is involved. diff --git a/external/nat/examples/calculator.py b/external/nat/examples/calculator.py new file mode 100644 index 00000000..8570f58d --- /dev/null +++ b/external/nat/examples/calculator.py @@ -0,0 +1,86 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Run a NAT ReAct workflow with a portable calculator MCP server.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +from pathlib import Path + +from nemo_fabric import Fabric +from nemo_fabric import FabricConfig +from nemo_fabric import HarnessConfig +from nemo_fabric import InstructionConfig +from nemo_fabric import InstructionsConfig +from nemo_fabric import MetadataConfig +from nemo_fabric import ModelConfig +from nemo_fabric import RuntimeConfig + + +def build_config() -> FabricConfig: + """Build the portable calculator configuration.""" + + config = FabricConfig( + metadata=MetadataConfig( + name="nat-calculator", + description="Uses calculator tools exposed by an MCP server.", + ), + harness=HarnessConfig( + adapter_id="nvidia.fabric.nat", + resolution="preinstalled", + settings={ + "workflow": { + "_type": "react_agent", + "llm_name": "default", + "tool_names": [], + } + }, + ), + models={ + "default": ModelConfig( + provider="nvidia", + model="nvidia/nemotron-3-nano-30b-a3b", + api_key_env="NVIDIA_API_KEY", + temperature=0.0, + ) + }, + instructions=InstructionsConfig( + system=InstructionConfig( + content="Use the calculator tools for arithmetic. Return a concise answer." + ) + ), + runtime=RuntimeConfig(input_schema="text", output_schema="message"), + ) + config.add_mcp_server( + "calculator", + transport="streamable-http", + url=os.environ.get("CALCULATOR_MCP_URL", "http://127.0.0.1:9901/mcp"), + exposure="harness_native", + blocked_tools=["divide"], + ) + return config + + +async def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base-dir", type=Path, default=Path.cwd()) + parser.add_argument("--input", default="What is 21 multiplied by 2?") + parser.add_argument("--plan", action="store_true") + args = parser.parse_args() + + fabric = Fabric() + config = build_config() + output = ( + fabric.plan(config, base_dir=args.base_dir) + if args.plan + else await fabric.run(config, base_dir=args.base_dir, input=args.input) + ) + print(json.dumps(output.to_mapping(), indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/external/nat/examples/email_phishing.py b/external/nat/examples/email_phishing.py new file mode 100644 index 00000000..49297b45 --- /dev/null +++ b/external/nat/examples/email_phishing.py @@ -0,0 +1,89 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Run a NAT workflow with the installed email-phishing analyzer function.""" + +from __future__ import annotations + +import argparse +import asyncio +import json +from pathlib import Path + +from nemo_fabric import Fabric +from nemo_fabric import FabricConfig +from nemo_fabric import HarnessConfig +from nemo_fabric import InstructionConfig +from nemo_fabric import InstructionsConfig +from nemo_fabric import MetadataConfig +from nemo_fabric import ModelConfig +from nemo_fabric import RuntimeConfig +from nemo_fabric import ToolsConfig + + +def build_config() -> FabricConfig: + """Build the NAT-native email-phishing configuration.""" + + return FabricConfig( + metadata=MetadataConfig( + name="nat-email-phishing-analyzer", + description="Classifies an email with an installed NAT function.", + ), + harness=HarnessConfig( + adapter_id="nvidia.fabric.nat", + resolution="preinstalled", + settings={ + "functions": { + "email_phishing_analyzer": { + "_type": "email_phishing_analyzer", + "llm": "default", + } + }, + "workflow": { + "_type": "react_agent", + "llm_name": "default", + "tool_names": [], + "use_native_tool_calling": True, + }, + }, + ), + models={ + "default": ModelConfig( + provider="nvidia", + model="nvidia/nemotron-3-nano-30b-a3b", + api_key_env="NVIDIA_API_KEY", + temperature=0.0, + ) + }, + instructions=InstructionsConfig( + system=InstructionConfig( + content='State whether the email is "phishing" or "benign" and explain why.' + ) + ), + tools=ToolsConfig(enabled=["email_phishing_analyzer"]), + runtime=RuntimeConfig(input_schema="text", output_schema="message"), + ) + + +async def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base-dir", type=Path, default=Path.cwd()) + parser.add_argument( + "--input", + default="Urgent: confirm your password at http://example.invalid today.", + ) + parser.add_argument("--plan", action="store_true") + args = parser.parse_args() + + fabric = Fabric() + config = build_config() + output = ( + fabric.plan(config, base_dir=args.base_dir) + if args.plan + else await fabric.run(config, base_dir=args.base_dir, input=args.input) + ) + print(json.dumps(output.to_mapping(), indent=2)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/external/nat/fabric-adapter.json b/external/nat/fabric-adapter.json new file mode 100644 index 00000000..554567c8 --- /dev/null +++ b/external/nat/fabric-adapter.json @@ -0,0 +1,73 @@ +{ + "contract_version": "fabric.adapter/v1alpha1", + "adapter_id": "nvidia.fabric.nat", + "harness": "nat", + "adapter_kind": "python", + "runner": { + "module": "nemo_fabric_adapters.nat.adapter" + }, + "settings_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$defs": { + "component": { + "type": "object", + "properties": { + "_type": { + "type": "string", + "minLength": 1, + "description": "NAT component type registered through an installed nat.components entry point." + } + }, + "required": ["_type"], + "additionalProperties": true + }, + "components": { + "type": "object", + "propertyNames": { + "minLength": 1 + }, + "additionalProperties": { + "$ref": "#/$defs/component" + } + } + }, + "type": "object", + "properties": { + "workflow": { + "$ref": "#/$defs/component", + "description": "NAT workflow component configuration." + }, + "functions": { + "$ref": "#/$defs/components", + "default": {}, + "description": "NAT function component instances keyed by their configured names." + }, + "function_groups": { + "$ref": "#/$defs/components", + "default": {}, + "description": "NAT function-group component instances keyed by their configured names." + } + }, + "required": ["workflow"], + "additionalProperties": false + }, + "requirements": {}, + "config": { + "accepts": [ + "models", + "models.base_url", + "models.temperature", + "instructions.system", + "tools.enabled", + "tools.blocked", + "mcp", + "mcp.tool_filters" + ] + }, + "capabilities": { + "cancellation": false, + "service": false, + "streaming": false, + "updates": false + } +} diff --git a/external/nat/src/nemo_fabric_adapters/nat/__init__.py b/external/nat/src/nemo_fabric_adapters/nat/__init__.py new file mode 100644 index 00000000..91312c7e --- /dev/null +++ b/external/nat/src/nemo_fabric_adapters/nat/__init__.py @@ -0,0 +1,4 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""NeMo Agent Toolkit reference adapter for NeMo Fabric.""" diff --git a/external/nat/src/nemo_fabric_adapters/nat/adapter.py b/external/nat/src/nemo_fabric_adapters/nat/adapter.py new file mode 100644 index 00000000..9cf5308d --- /dev/null +++ b/external/nat/src/nemo_fabric_adapters/nat/adapter.py @@ -0,0 +1,858 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""NeMo Agent Toolkit adapter for NeMo Fabric. + +The adapter builds one in-memory NAT configuration from Fabric's normalized +configuration and adapter-owned NAT component settings. One persistent adapter +host owns the resulting workflow for the complete Fabric runtime lifecycle. +""" + +from __future__ import annotations + +import asyncio +import copy +import logging +import os +import shlex +from contextlib import AsyncExitStack +from typing import Any + +from nemo_fabric_adapters.common import lifecycle +import nemo_fabric_adapters.common.utils as common_utils + +HARNESS = "nat" +MODE = "nat_workflow" +FUNCTION_GROUP_SEPARATOR = "__" +NAT_SETTINGS_FIELDS = frozenset({"workflow", "functions", "function_groups"}) +RESERVED_MODEL_SETTINGS = frozenset( + { + "_type", + "type", + "provider", + "model", + "model_name", + "api_key", + "api_key_env", + "base_url", + "temperature", + } +) + +LOGGER = logging.getLogger(__name__) + + +def main() -> None: + """Serve the persistent local-host lifecycle protocol.""" + + lifecycle.serve(NatRuntime) + + +def _config_error(code: str, message: str, **metadata: Any) -> lifecycle.LifecycleError: + return lifecycle.LifecycleError(code, message, metadata=metadata or None) + + +def _runtime_id(payload: dict[str, Any]) -> str: + try: + return common_utils.runtime_id(payload) + except ValueError as error: + raise _config_error( + "nat_invalid_runtime_context", + "NAT lifecycle payload is missing a runtime ID", + ) from error + + +def _mapping(value: Any, field: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise _config_error( + "nat_invalid_harness_settings", + f"{field} must be a mapping", + field=field, + ) + return copy.deepcopy(value) + + +def _nat_component_settings(payload: dict[str, Any]) -> dict[str, Any]: + settings = common_utils.settings_payload(payload) + if not isinstance(settings, dict): + raise _config_error( + "nat_invalid_harness_settings", + "harness.settings must be a mapping", + field="harness.settings", + ) + + unknown = sorted(set(settings).difference(NAT_SETTINGS_FIELDS)) + if unknown: + raise _config_error( + "nat_invalid_harness_settings", + "NAT harness settings contain unsupported top-level fields", + fields=unknown, + ) + + workflow = _mapping(settings.get("workflow"), "harness.settings.workflow") + workflow_type = workflow.get("_type") + if not isinstance(workflow_type, str) or not workflow_type.strip(): + raise _config_error( + "nat_invalid_harness_settings", + "harness.settings.workflow._type must be a non-empty string", + field="harness.settings.workflow._type", + ) + if _workflow_type(workflow) == "react_agent": + workflow.setdefault("tool_names", []) + + functions = settings.get("functions", {}) + function_groups = settings.get("function_groups", {}) + return { + "workflow": workflow, + "functions": _mapping(functions, "harness.settings.functions"), + "function_groups": _mapping( + function_groups, + "harness.settings.function_groups", + ), + } + + +def _nat_llm_type(provider: str) -> str: + # NAT calls NVIDIA's OpenAI-compatible provider ``nim``. Other provider + # identifiers are already NAT component type names and are validated after + # installed NAT plugins register their config objects. + return "nim" if provider == "nvidia" else provider + + +def _nat_llms(payload: dict[str, Any]) -> dict[str, dict[str, Any]]: + models = common_utils.models_payload(payload) + if not isinstance(models, dict): + raise _config_error("nat_invalid_models", "Fabric models must be a mapping") + + llms: dict[str, dict[str, Any]] = {} + for role, raw_model in models.items(): + if not isinstance(role, str) or not role: + raise _config_error( + "nat_invalid_models", + "Fabric model role names must be non-empty strings", + ) + if not isinstance(raw_model, dict): + raise _config_error( + "nat_invalid_models", + f"Fabric model {role!r} must be a mapping", + role=role, + ) + + provider = raw_model.get("provider") + model_name = raw_model.get("model") + if not isinstance(provider, str) or not provider: + raise _config_error( + "nat_invalid_models", + f"Fabric model {role!r} requires a non-empty provider", + role=role, + ) + if not isinstance(model_name, str) or not model_name: + raise _config_error( + "nat_invalid_models", + f"Fabric model {role!r} requires a non-empty model", + role=role, + ) + + settings = raw_model.get("settings") or {} + if not isinstance(settings, dict): + raise _config_error( + "nat_invalid_models", + f"Fabric model {role!r} settings must be a mapping", + role=role, + ) + reserved = sorted(RESERVED_MODEL_SETTINGS.intersection(settings)) + if reserved: + raise _config_error( + "nat_model_settings_reserved", + f"Fabric model {role!r} settings cannot replace normalized model fields", + role=role, + fields=reserved, + ) + + llm = copy.deepcopy(settings) + llm.update( + { + "_type": _nat_llm_type(provider), + "model_name": model_name, + } + ) + if raw_model.get("base_url") is not None: + llm["base_url"] = raw_model["base_url"] + if raw_model.get("temperature") is not None: + llm["temperature"] = raw_model["temperature"] + + api_key_env = raw_model.get("api_key_env") + if api_key_env is not None: + if not isinstance(api_key_env, str) or not api_key_env: + raise _config_error( + "nat_invalid_models", + f"Fabric model {role!r} api_key_env must be a non-empty string", + role=role, + ) + api_key = os.environ.get(api_key_env) + if not api_key: + raise _config_error( + "nat_model_api_key_missing", + f"Fabric model {role!r} requires environment variable {api_key_env!r}", + role=role, + api_key_env=api_key_env, + ) + llm["api_key"] = api_key + + llms[role] = llm + return llms + + +def _workflow_type(workflow: dict[str, Any]) -> str: + value = workflow.get("_type") + return value.rsplit("/", 1)[-1] if isinstance(value, str) else "" + + +def _apply_system_instruction(config: dict[str, Any], payload: dict[str, Any]) -> None: + instruction = common_utils.system_instruction(payload) + if instruction is None: + return + + workflow = config["workflow"] + if _workflow_type(workflow) != "react_agent": + raise _config_error( + "nat_system_instruction_unsupported", + "instructions.system is supported only for a NAT react_agent workflow", + workflow_type=workflow.get("_type"), + ) + if "additional_instructions" in workflow: + raise _config_error( + "nat_system_instruction_conflict", + "instructions.system conflicts with harness.settings.workflow.additional_instructions", + fields=[ + "instructions.system", + "harness.settings.workflow.additional_instructions", + ], + ) + workflow["additional_instructions"] = instruction + + +def _string_list(value: Any, field: str, *, optional: bool = False) -> list[str] | None: + if value is None and optional: + return None + if value is None: + return [] + if not isinstance(value, list) or any( + not isinstance(item, str) or not item.strip() or item != item.strip() + for item in value + ): + raise _config_error( + "nat_invalid_tool_policy", + f"{field} must be a list of non-empty strings", + field=field, + ) + return list(dict.fromkeys(value)) + + +def nat_mcp_server_config(name: str, server: Any) -> dict[str, Any]: + """Translate one Fabric MCP server plan into a NAT server mapping.""" + + if not isinstance(server, dict): + raise _config_error( + "nat_invalid_mcp_server", + f"NAT MCP server {name!r} must be a mapping", + server=name, + ) + + transport = str(server.get("transport") or "").strip().lower().replace("_", "-") + target = os.path.expandvars(str(server.get("url") or "")).strip() + if not target: + raise _config_error( + "nat_invalid_mcp_server", + f"NAT MCP server {name!r} requires a non-empty url", + server=name, + ) + + if transport == "stdio": + try: + command = shlex.split(target) + except ValueError as error: + raise _config_error( + "nat_invalid_mcp_server", + f"NAT MCP server {name!r} has an invalid stdio command", + server=name, + ) from error + if not command: + raise _config_error( + "nat_invalid_mcp_server", + f"NAT MCP server {name!r} has an empty stdio command", + server=name, + ) + result: dict[str, Any] = { + "transport": "stdio", + "command": command[0], + } + if command[1:]: + result["args"] = command[1:] + return result + + if transport in {"http", "streamablehttp"}: + transport = "streamable-http" + if transport not in {"sse", "streamable-http"}: + raise _config_error( + "nat_unsupported_mcp_transport", + f"NAT MCP server {name!r} has unsupported transport {transport!r}", + server=name, + transport=transport, + ) + return {"transport": transport, "url": target} + + +def _workflow_tool_names(config: dict[str, Any], reason: str) -> list[str]: + tool_names = config["workflow"].get("tool_names") + if not isinstance(tool_names, list) or any( + not isinstance(name, str) or not name.strip() or name != name.strip() + for name in tool_names + ): + raise _config_error( + "nat_workflow_tools_unsupported", + f"{reason} requires a NAT workflow with a string-list tool_names field", + field="harness.settings.workflow.tool_names", + ) + return tool_names + + +def _native_mcp_servers(payload: dict[str, Any]) -> dict[str, Any]: + plan = common_utils.capability_plan(payload) + if not isinstance(plan, dict): + raise _config_error( + "nat_invalid_capability_plan", + "NAT capability plan must be a mapping", + ) + native = plan.get("native", {}) + if not isinstance(native, dict): + raise _config_error( + "nat_invalid_capability_plan", + "NAT native capability plan must be a mapping", + ) + servers = native.get("mcp_servers", {}) + if not isinstance(servers, dict): + raise _config_error( + "nat_invalid_mcp_config", + "NAT native MCP capability plan must be a mapping", + ) + return servers + + +def _mcp_group(name: str, server: dict[str, Any]) -> dict[str, Any] | None: + allowed = _string_list( + server.get("allowed_tools"), + f"capability_plan.native.mcp_servers.{name}.allowed_tools", + optional=True, + ) + blocked = _string_list( + server.get("blocked_tools"), + f"capability_plan.native.mcp_servers.{name}.blocked_tools", + ) + assert blocked is not None + + if allowed is not None: + overlap = sorted(set(allowed).intersection(blocked)) + if overlap: + raise _config_error( + "nat_invalid_mcp_tool_policy", + f"NAT MCP server {name!r} cannot both allow and block a tool", + server=name, + tool=overlap[0], + ) + if not allowed: + return None + + group: dict[str, Any] = { + "_type": "mcp_client", + "server": nat_mcp_server_config(name, server), + } + # NAT forbids include and exclude together. A non-empty Fabric allowlist is + # already the effective exposed set after the disjoint blocklist is applied. + if allowed is not None: + group["include"] = allowed + elif blocked: + group["exclude"] = blocked + return group + + +def _apply_mcp_servers(config: dict[str, Any], payload: dict[str, Any]) -> set[str]: + servers = _native_mcp_servers(payload) + if not servers: + return set() + + functions = config["functions"] + function_groups = config["function_groups"] + suppressed: set[str] = set() + tool_names: list[str] | None = None + + for name, raw_server in sorted(servers.items()): + if not isinstance(name, str) or not name.strip() or name != name.strip(): + raise _config_error( + "nat_invalid_mcp_server", + "NAT MCP server names must be non-empty strings", + ) + if not isinstance(raw_server, dict): + raise _config_error( + "nat_invalid_mcp_server", + f"NAT MCP server {name!r} must be a mapping", + server=name, + ) + + if name in functions or name in function_groups: + raise _config_error( + "nat_mcp_name_conflict", + f"NAT MCP server {name!r} conflicts with an existing function or function group", + server=name, + ) + + group = _mcp_group(name, raw_server) + if group is None: + # An explicit empty allowlist means the server exposes no tools. It + # must not remain reachable through a pre-existing workflow ref. + existing_names = config["workflow"].get("tool_names") + if isinstance(existing_names, list): + existing_names[:] = [ + tool_name for tool_name in existing_names if tool_name != name + ] + suppressed.add(name) + continue + + if tool_names is None: + tool_names = _workflow_tool_names(config, "Normalized MCP configuration") + function_groups[name] = group + if name not in tool_names: + tool_names.append(name) + + return suppressed + + +def _group_member_identity(name: str) -> tuple[str, str] | None: + if FUNCTION_GROUP_SEPARATOR not in name: + return None + group, member = name.split(FUNCTION_GROUP_SEPARATOR, 1) + if not group or not member: + return None + return group, member + + +def _group_mapping(groups: dict[str, Any], name: str) -> dict[str, Any] | None: + value = groups.get(name) + if value is None: + return None + if not isinstance(value, dict): + raise _config_error( + "nat_invalid_harness_settings", + f"NAT function group {name!r} must be a mapping", + group=name, + ) + return value + + +def _unknown_tool_selector(selector: str) -> lifecycle.LifecycleError: + return _config_error( + "nat_unknown_tool_selector", + f"Fabric tool selector {selector!r} does not match a configured NAT function or function group", + selector=selector, + ) + + +def _validate_tool_selectors( + config: dict[str, Any], + selectors: list[str], + suppressed: set[str], + *, + enabling: bool, +) -> None: + functions = config["functions"] + groups = config["function_groups"] + for selector in selectors: + if selector in suppressed: + raise _unknown_tool_selector(selector) + if selector in functions or selector in groups: + continue + member = _group_member_identity(selector) + if member is None: + raise _unknown_tool_selector(selector) + group = _group_mapping(groups, member[0]) + if group is None: + raise _unknown_tool_selector(selector) + if not enabling: + continue + + include = _string_list(group.get("include"), "function group include") or [] + exclude = _string_list(group.get("exclude"), "function group exclude") or [] + if (include and member[1] not in include) or member[1] in exclude: + raise _unknown_tool_selector(selector) + + +def _select_group_members(requested: list[str]) -> list[str]: + # Selector compatibility with existing include/exclude policy is validated + # before mutation. Preserve caller order while removing duplicates. + return list(dict.fromkeys(requested)) + + +def _apply_enabled_tools( + config: dict[str, Any], enabled: list[str], suppressed: set[str] +) -> None: + functions = config["functions"] + groups = config["function_groups"] + selected_functions: dict[str, Any] = {} + selected_groups: dict[str, Any] = {} + selected_refs: list[str] = [] + + requested_members: dict[str, list[str]] = {} + for identity in enabled: + if identity in functions or identity in groups: + continue + member = _group_member_identity(identity) + if member is not None: + requested_members.setdefault(member[0], []).append(member[1]) + + for identity in enabled: + if identity in suppressed: + continue + if identity in functions: + selected_functions[identity] = functions[identity] + if identity not in selected_refs: + selected_refs.append(identity) + continue + + if identity in groups: + if identity in suppressed: + continue + selected_groups[identity] = groups[identity] + if identity not in selected_refs: + selected_refs.append(identity) + continue + + member = _group_member_identity(identity) + if member is None: + continue + group_name, _ = member + if group_name in suppressed or group_name in selected_groups: + continue + group = _group_mapping(groups, group_name) + if group is None: + continue + selected = _select_group_members(requested_members[group_name]) + if not selected: + continue + selected_group = copy.deepcopy(group) + selected_group["include"] = selected + selected_group.pop("exclude", None) + selected_groups[group_name] = selected_group + if group_name not in selected_refs: + selected_refs.append(group_name) + + functions.clear() + functions.update(selected_functions) + groups.clear() + groups.update(selected_groups) + config["workflow"]["tool_names"] = selected_refs + + +def _block_group_member( + config: dict[str, Any], group_name: str, member_name: str +) -> None: + groups = config["function_groups"] + group = _group_mapping(groups, group_name) + if group is None: + return + + include = _string_list(group.get("include"), "function group include") or [] + if include: + remaining = [name for name in include if name != member_name] + if remaining: + group["include"] = remaining + return + groups.pop(group_name, None) + config["workflow"]["tool_names"] = [ + name for name in config["workflow"]["tool_names"] if name != group_name + ] + return + + exclude = _string_list(group.get("exclude"), "function group exclude") or [] + if member_name not in exclude: + group["exclude"] = [*exclude, member_name] + + +def _apply_blocked_tools(config: dict[str, Any], blocked: list[str]) -> None: + functions = config["functions"] + groups = config["function_groups"] + tool_names = config["workflow"]["tool_names"] + + for identity in blocked: + if identity in functions: + functions.pop(identity, None) + tool_names[:] = [name for name in tool_names if name != identity] + continue + if identity in groups: + groups.pop(identity, None) + tool_names[:] = [name for name in tool_names if name != identity] + continue + member = _group_member_identity(identity) + if member is not None: + _block_group_member(config, *member) + + +def _apply_tool_policy( + config: dict[str, Any], payload: dict[str, Any], suppressed: set[str] +) -> None: + tools = common_utils.tools_config(payload) + enabled = _string_list(tools.get("enabled"), "config.tools.enabled", optional=True) + blocked = _string_list(tools.get("blocked"), "config.tools.blocked") + assert blocked is not None + if enabled is None and not blocked: + return + + _workflow_tool_names(config, "Normalized tool policy") + if enabled is not None: + _validate_tool_selectors( + config, + enabled, + suppressed, + enabling=True, + ) + _validate_tool_selectors( + config, + blocked, + suppressed, + enabling=False, + ) + if enabled is not None: + _apply_enabled_tools(config, enabled, suppressed) + else: + config["workflow"]["tool_names"] = [ + name for name in config["workflow"]["tool_names"] if name not in suppressed + ] + if blocked: + _apply_blocked_tools(config, blocked) + + +def apply_nat_capabilities(config: dict[str, Any], payload: dict[str, Any]) -> None: + """Compile Fabric MCP routing and tool policy into a raw NAT config.""" + + suppressed = _apply_mcp_servers(config, payload) + _apply_tool_policy(config, payload, suppressed) + + +def build_nat_config_mapping(payload: dict[str, Any]) -> dict[str, Any]: + """Build the raw in-memory NAT mapping from one Fabric start payload.""" + + config = _nat_component_settings(payload) + llms = _nat_llms(payload) + if llms: + config["llms"] = llms + _apply_system_instruction(config, payload) + apply_nat_capabilities(config, payload) + return config + + +def build_nat_config(payload: dict[str, Any]) -> Any: + """Build and validate a typed NAT Config without a workflow YAML file.""" + + raw_config = build_nat_config_mapping(payload) + + try: + from nat.runtime.loader import PluginTypes + from nat.runtime.loader import discover_and_register_plugins + + discover_and_register_plugins(PluginTypes.CONFIG_OBJECT) + + from nat.data_models.config import Config + + return Config.model_validate(raw_config) + except lifecycle.LifecycleError: + raise + except Exception as error: + raise _config_error( + "nat_config_translation_failed", + "Fabric config could not be translated into a valid NAT config", + ) from error + + +def _session_kwargs(request: dict[str, Any]) -> dict[str, str]: + context = request.get("context") or {} + if not isinstance(context, dict): + raise ValueError("request.context must be a mapping") + + values = { + "user_id": context.get("user_id"), + "conversation_id": context.get("conversation_id"), + "user_message_id": context.get("user_message_id") or request.get("request_id"), + } + result: dict[str, str] = {} + for name, value in values.items(): + if value is None: + continue + if not isinstance(value, str) or not value: + raise ValueError(f"request context {name} must be a non-empty string") + result[name] = value + return result + + +def _success_output(response: Any) -> dict[str, Any]: + return { + "harness": HARNESS, + "adapter": "python", + "mode": MODE, + "response": response, + "completed": True, + "failed": False, + "error": None, + } + + +def _failure_output(code: str, message: str) -> dict[str, Any]: + return { + "harness": HARNESS, + "adapter": "python", + "mode": MODE, + "response": None, + "completed": False, + "failed": True, + "error": { + "code": code, + "message": message, + "retryable": False, + }, + } + + +async def _close_after_failed_start(stack: AsyncExitStack) -> None: + try: + await stack.aclose() + except asyncio.CancelledError: + raise + except Exception as error: + LOGGER.error( + "NAT workflow cleanup failed after start error (error_type=%s)", + type(error).__name__, + ) + + +class NatRuntime: + """One NAT WorkflowBuilder and SessionManager owned by a Fabric runtime.""" + + def __init__(self) -> None: + self._runtime_id: str | None = None + self._sessions: Any = None + self._exit_stack: AsyncExitStack | None = None + + async def start(self, payload: dict[str, Any]) -> None: + if self._exit_stack is not None: + raise lifecycle.LifecycleError( + "nat_runtime_already_started", + "NAT runtime is already started", + ) + + runtime_id = _runtime_id(payload) + stack = AsyncExitStack() + try: + from nat.builder.workflow_builder import WorkflowBuilder + from nat.runtime.session import SessionManager + + config = build_nat_config(payload) + builder = await stack.enter_async_context( + WorkflowBuilder.from_config(config=config) + ) + sessions = await SessionManager.create( + config=config, + shared_builder=builder, + ) + stack.push_async_callback(sessions.shutdown) + except asyncio.CancelledError: + await _close_after_failed_start(stack) + raise + except lifecycle.LifecycleError: + await _close_after_failed_start(stack) + raise + except Exception as error: + await _close_after_failed_start(stack) + raise lifecycle.LifecycleError( + "nat_workflow_start_failed", + "NAT workflow failed to load; inspect adapter stderr for details", + ) from error + + self._runtime_id = runtime_id + self._sessions = sessions + self._exit_stack = stack + + async def invoke(self, payload: dict[str, Any]) -> dict[str, Any]: + if self._sessions is None or self._runtime_id is None: + raise lifecycle.LifecycleError( + "nat_runtime_not_started", + "NAT runtime is not started", + ) + if _runtime_id(payload) != self._runtime_id: + raise lifecycle.LifecycleError( + "nat_runtime_mismatch", + "NAT invocation does not match the active runtime", + ) + + request = common_utils.request_payload(payload) + if not isinstance(request, dict): + return _failure_output( + "nat_invalid_request", + "NAT invocation request must be a mapping", + ) + + try: + from nat.data_models.runtime_enum import RuntimeTypeEnum + + async with self._sessions.session(**_session_kwargs(request)) as session: + async with session.run( + request.get("input", ""), + runtime_type=RuntimeTypeEnum.RUN_OR_SERVE, + ) as runner: + result = await runner.result() + except asyncio.CancelledError: + raise + except Exception as error: + LOGGER.error( + "NAT workflow invocation failed (error_type=%s)", + type(error).__name__, + ) + return _failure_output( + "nat_workflow_invoke_failed", + "NAT workflow invocation failed; inspect adapter stderr for details", + ) + + try: + from pydantic_core import to_jsonable_python + + response = to_jsonable_python(result, serialize_unknown=False) + except (TypeError, ValueError) as error: + LOGGER.error( + "NAT workflow returned a non-JSON result (error_type=%s)", + type(error).__name__, + ) + return _failure_output( + "nat_result_not_json_serializable", + "NAT workflow returned a result that cannot be represented as JSON", + ) + return _success_output(response) + + async def stop(self) -> None: + stack = self._exit_stack + self._runtime_id = None + self._sessions = None + self._exit_stack = None + + if stack is None: + return + try: + await stack.aclose() + except asyncio.CancelledError: + raise + except Exception as error: + raise lifecycle.LifecycleError( + "nat_runtime_stop_failed", + "NAT runtime failed to stop cleanly", + ) from error + + +if __name__ == "__main__": + main() diff --git a/tests/adapters/test_external_nat_adapter.py b/tests/adapters/test_external_nat_adapter.py new file mode 100644 index 00000000..9743ffd6 --- /dev/null +++ b/tests/adapters/test_external_nat_adapter.py @@ -0,0 +1,708 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Focused tests for the source-only NeMo Agent Toolkit reference adapter.""" + +from __future__ import annotations + +import json +import os +import sys +import types +from copy import deepcopy +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock +from unittest.mock import MagicMock +from unittest.mock import call + +import pytest + +ROOT = Path(__file__).parents[2] +NAT_ADAPTER_SOURCE = ROOT / "external" / "nat" / "src" +sys.path.insert(0, str(NAT_ADAPTER_SOURCE)) + +from nemo_fabric_adapters.nat import adapter # noqa: E402 + + +@pytest.fixture(name="make_payload") +def make_payload_fixture(tmp_path: Path): + """Return a factory for canonical Fabric lifecycle payloads.""" + + def make( + *, + workflow: dict[str, Any] | None = None, + functions: dict[str, Any] | None = None, + function_groups: dict[str, Any] | None = None, + models: dict[str, Any] | None = None, + instruction: str | None = None, + tools: dict[str, Any] | None = None, + mcp_servers: dict[str, Any] | None = None, + ) -> dict[str, Any]: + config: dict[str, Any] = { + "harness": { + "settings": { + "workflow": deepcopy( + workflow + or { + "_type": "react_agent", + "llm_name": "default", + "tool_names": [], + } + ), + "functions": deepcopy(functions or {}), + "function_groups": deepcopy(function_groups or {}), + } + }, + "models": deepcopy(models or {}), + } + if instruction is not None: + config["instructions"] = { + "system": {"content": instruction, "mode": "replace"} + } + if tools is not None: + config["tools"] = deepcopy(tools) + + return { + "base_dir": str(tmp_path), + "config": config, + "runtime_context": { + "runtime_id": "runtime-1", + "environment": {"workspace": str(tmp_path)}, + }, + "capability_plan": { + "native": {"mcp_servers": deepcopy(mcp_servers or {})} + }, + } + + return make + + +@pytest.fixture(name="mock_nat") +def mock_nat_fixture(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: + """Install mocked NAT modules and return lifecycle call recorders.""" + + mock_typed_config = MagicMock(name="typed_nat_config") + mock_config_type = MagicMock(name="Config") + mock_config_type.model_validate = MagicMock(return_value=mock_typed_config) + mock_discover = MagicMock(name="discover_and_register_plugins") + mock_plugin_types = MagicMock(name="PluginTypes") + mock_plugin_types.CONFIG_OBJECT = "config-object" + + mock_builder = MagicMock(name="builder") + mock_builder_context = MagicMock(name="builder_context") + mock_builder_context.__aenter__ = AsyncMock(return_value=mock_builder) + mock_builder_context.__aexit__ = AsyncMock(return_value=False) + mock_workflow_builder = MagicMock(name="WorkflowBuilder") + mock_workflow_builder.from_config = MagicMock(return_value=mock_builder_context) + + mock_runner = MagicMock(name="runner") + mock_runner.result = AsyncMock(side_effect=[{"answer": 42}, {"answer": 84}]) + mock_run_context = MagicMock(name="run_context") + mock_run_context.__aenter__ = AsyncMock(return_value=mock_runner) + mock_run_context.__aexit__ = AsyncMock(return_value=False) + mock_session = MagicMock(name="session") + mock_session.run = MagicMock(return_value=mock_run_context) + mock_session_context = MagicMock(name="session_context") + mock_session_context.__aenter__ = AsyncMock(return_value=mock_session) + mock_session_context.__aexit__ = AsyncMock(return_value=False) + mock_sessions = MagicMock(name="sessions") + mock_sessions.session = MagicMock(return_value=mock_session_context) + mock_sessions.shutdown = AsyncMock() + mock_session_manager = MagicMock(name="SessionManager") + mock_session_manager.create = AsyncMock(return_value=mock_sessions) + + mock_runtime_type = MagicMock(name="RuntimeTypeEnum") + mock_runtime_type.RUN_OR_SERVE = "run-or-serve" + mock_to_jsonable = MagicMock( + name="to_jsonable_python", + side_effect=lambda value, **_kwargs: value, + ) + + modules: dict[str, types.ModuleType] = {} + for name in ( + "nat", + "nat.builder", + "nat.builder.workflow_builder", + "nat.runtime", + "nat.runtime.loader", + "nat.runtime.session", + "nat.data_models", + "nat.data_models.config", + "nat.data_models.runtime_enum", + "pydantic_core", + ): + module = types.ModuleType(name) + if name in {"nat", "nat.builder", "nat.runtime", "nat.data_models"}: + module.__path__ = [] # type: ignore[attr-defined] + modules[name] = module + monkeypatch.setitem(sys.modules, name, module) + + modules["nat.runtime.loader"].PluginTypes = mock_plugin_types + modules["nat.runtime.loader"].discover_and_register_plugins = mock_discover + modules["nat.data_models.config"].Config = mock_config_type + modules["nat.builder.workflow_builder"].WorkflowBuilder = mock_workflow_builder + modules["nat.runtime.session"].SessionManager = mock_session_manager + modules["nat.data_models.runtime_enum"].RuntimeTypeEnum = mock_runtime_type + modules["pydantic_core"].to_jsonable_python = mock_to_jsonable + + return { + "typed_config": mock_typed_config, + "config_type": mock_config_type, + "discover": mock_discover, + "plugin_types": mock_plugin_types, + "workflow_builder": mock_workflow_builder, + "builder_context": mock_builder_context, + "builder": mock_builder, + "session_manager": mock_session_manager, + "sessions": mock_sessions, + "session": mock_session, + "runner": mock_runner, + "run_context": mock_run_context, + "session_context": mock_session_context, + "runtime_type": mock_runtime_type, + "to_jsonable": mock_to_jsonable, + } + + +def invocation_payload( + *, + input_value: Any = "hello", + request_id: str = "request-1", + context: dict[str, Any] | None = None, +) -> dict[str, Any]: + request: dict[str, Any] = {"input": input_value, "request_id": request_id} + if context is not None: + request["context"] = context + return { + "runtime_context": {"runtime_id": "runtime-1"}, + "request": request, + } + + +def test_descriptor_declares_exact_source_reference_contract(): + descriptor = json.loads( + (ROOT / "external" / "nat" / "fabric-adapter.json").read_text( + encoding="utf-8" + ) + ) + + assert descriptor["adapter_id"] == "nvidia.fabric.nat" + assert descriptor["harness"] == "nat" + assert descriptor["adapter_kind"] == "python" + assert descriptor["runner"] == { + "module": "nemo_fabric_adapters.nat.adapter" + } + assert descriptor["requirements"] == {} + assert descriptor["config"]["accepts"] == [ + "models", + "models.base_url", + "models.temperature", + "instructions.system", + "tools.enabled", + "tools.blocked", + "mcp", + "mcp.tool_filters", + ] + assert descriptor["settings_schema"]["required"] == ["workflow"] + assert descriptor["settings_schema"]["additionalProperties"] is False + assert descriptor["capabilities"] == { + "cancellation": False, + "service": False, + "streaming": False, + "updates": False, + } + + +def test_build_mapping_translates_components_models_and_instruction( + make_payload, +): + os.environ["NVIDIA_API_KEY"] = "test-key" + workflow = { + "_type": "react_agent", + "llm_name": "default", + "tool_names": ["clock", "calculator"], + } + functions = {"clock": {"_type": "current_datetime"}} + function_groups = { + "calculator": {"_type": "calculator", "include": ["add", "subtract"]} + } + payload = make_payload( + workflow=workflow, + functions=functions, + function_groups=function_groups, + models={ + "default": { + "provider": "nvidia", + "model": "nvidia/test-model", + "api_key_env": "NVIDIA_API_KEY", + "base_url": "https://integrate.api.nvidia.com/v1", + "temperature": 0.2, + "settings": {"max_tokens": 512}, + }, + "reviewer": { + "provider": "openai", + "model": "gpt-test", + }, + }, + instruction="Use portable instructions.", + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result == { + "workflow": { + "_type": "react_agent", + "llm_name": "default", + "tool_names": ["clock", "calculator"], + "additional_instructions": "Use portable instructions.", + }, + "functions": functions, + "function_groups": function_groups, + "llms": { + "default": { + "_type": "nim", + "model_name": "nvidia/test-model", + "api_key": "test-key", + "base_url": "https://integrate.api.nvidia.com/v1", + "temperature": 0.2, + "max_tokens": 512, + }, + "reviewer": { + "_type": "openai", + "model_name": "gpt-test", + }, + }, + } + assert payload["config"]["harness"]["settings"]["workflow"] == workflow + + +def test_system_instruction_rejects_duplicate_nat_instruction_source(make_payload): + payload = make_payload( + workflow={ + "_type": "react_agent", + "llm_name": "default", + "additional_instructions": "adapter-local value", + }, + instruction="Portable instruction", + ) + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.build_nat_config_mapping(payload) + + assert error.value.code == "nat_system_instruction_conflict" + + +def test_react_agent_without_tool_names_defaults_to_empty_list(make_payload): + payload = make_payload( + workflow={"_type": "react_agent", "llm_name": "default"} + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result["workflow"]["tool_names"] == [] + + +def test_build_typed_config_discovers_components_before_validation( + make_payload, + mock_nat, +): + events: list[str] = [] + mock_nat["discover"].side_effect = lambda _plugin_type: events.append("discover") + mock_nat["config_type"].model_validate.side_effect = ( + lambda _mapping: events.append("validate") or mock_nat["typed_config"] + ) + + result = adapter.build_nat_config(make_payload()) + + assert result is mock_nat["typed_config"] + assert events == ["discover", "validate"] + mock_nat["discover"].assert_called_once_with( + mock_nat["plugin_types"].CONFIG_OBJECT + ) + + +@pytest.mark.parametrize( + ("server_policy", "expected_group"), + [ + ( + {}, + { + "_type": "mcp_client", + "server": {"transport": "sse", "url": "https://mcp.test/sse"}, + }, + ), + ( + {"blocked_tools": ["delete"]}, + { + "_type": "mcp_client", + "server": {"transport": "sse", "url": "https://mcp.test/sse"}, + "exclude": ["delete"], + }, + ), + ( + {"allowed_tools": ["read", "list"], "blocked_tools": ["delete"]}, + { + "_type": "mcp_client", + "server": {"transport": "sse", "url": "https://mcp.test/sse"}, + "include": ["read", "list"], + }, + ), + ], +) +def test_mcp_filter_states_generate_one_nat_group_policy( + make_payload, + server_policy: dict[str, Any], + expected_group: dict[str, Any], +): + server = { + "transport": "sse", + "url": "https://mcp.test/sse", + **server_policy, + } + payload = make_payload(mcp_servers={"docs": server}) + + result = adapter.build_nat_config_mapping(payload) + + assert result["function_groups"]["docs"] == expected_group + assert result["workflow"]["tool_names"] == ["docs"] + + +def test_mcp_policy_deduplicates_valid_fabric_names(make_payload): + payload = make_payload( + mcp_servers={ + "docs": { + "transport": "sse", + "url": "https://mcp.test/sse", + "allowed_tools": ["read", "read"], + "blocked_tools": ["delete", "delete"], + } + }, + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result["function_groups"] == { + "docs": { + "_type": "mcp_client", + "server": {"transport": "sse", "url": "https://mcp.test/sse"}, + "include": ["read"], + }, + } + assert result["workflow"]["tool_names"] == ["docs"] + + +def test_root_tool_policy_deduplicates_valid_fabric_names(make_payload): + payload = make_payload( + function_groups={"calculator": {"_type": "calculator"}}, + tools={"enabled": ["calculator", "calculator"]}, + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result["function_groups"] == { + "calculator": {"_type": "calculator"}, + } + assert result["workflow"]["tool_names"] == ["calculator"] + + +def test_empty_mcp_allowlist_suppresses_server_and_existing_workflow_ref( + make_payload, +): + payload = make_payload( + workflow={ + "_type": "react_agent", + "llm_name": "default", + "tool_names": ["clock", "docs"], + }, + functions={"clock": {"_type": "current_datetime"}}, + mcp_servers={ + "docs": { + "transport": "streamable-http", + "url": "https://mcp.test/mcp", + "allowed_tools": [], + } + }, + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result["workflow"]["tool_names"] == ["clock"] + assert "docs" not in result["function_groups"] + + +@pytest.mark.parametrize("component_field", ["functions", "function_groups"]) +def test_empty_mcp_allowlist_rejects_same_name_nat_component( + make_payload, + component_field: str, +): + components = {"docs": {"_type": "native_docs"}} + payload = make_payload( + **{component_field: components}, + mcp_servers={ + "docs": { + "transport": "sse", + "url": "https://mcp.test/sse", + "allowed_tools": [], + } + }, + ) + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.build_nat_config_mapping(payload) + + assert error.value.code == "nat_mcp_name_conflict" + + +def test_all_suppressed_mcp_does_not_require_custom_workflow_tool_names( + make_payload, +): + payload = make_payload( + workflow={"_type": "custom_workflow"}, + mcp_servers={ + "docs": { + "transport": "sse", + "url": "https://mcp.test/sse", + "allowed_tools": [], + } + }, + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result["workflow"] == {"_type": "custom_workflow"} + assert result["function_groups"] == {} + + +def test_root_tool_policy_selects_and_blocks_exact_group_members(make_payload): + payload = make_payload( + workflow={ + "_type": "react_agent", + "llm_name": "default", + "tool_names": ["clock", "unused", "calculator", "search"], + }, + functions={ + "clock": {"_type": "current_datetime"}, + "unused": {"_type": "unused_function"}, + }, + function_groups={ + "calculator": { + "_type": "calculator", + "include": ["add", "subtract", "multiply"], + }, + "search": {"_type": "search", "exclude": ["delete"]}, + }, + tools={ + "enabled": [ + "clock", + "calculator__add", + "calculator__subtract", + "search__find", + ], + "blocked": ["calculator__subtract", "search__secret"], + }, + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result["functions"] == {"clock": {"_type": "current_datetime"}} + assert result["function_groups"] == { + "calculator": {"_type": "calculator", "include": ["add"]}, + "search": {"_type": "search", "include": ["find"]}, + } + assert result["workflow"]["tool_names"] == ["clock", "calculator", "search"] + + +@pytest.mark.parametrize( + "tools", + [ + {"enabled": ["missing"]}, + {"blocked": ["missing"]}, + {"enabled": ["calculator__missing"]}, + ], +) +def test_root_tool_policy_rejects_unknown_exact_selectors(make_payload, tools): + payload = make_payload( + workflow={ + "_type": "react_agent", + "llm_name": "default", + "tool_names": ["calculator"], + }, + function_groups={ + "calculator": {"_type": "calculator", "include": ["add"]} + }, + tools=tools, + ) + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.build_nat_config_mapping(payload) + + assert error.value.code == "nat_unknown_tool_selector" + + +async def test_runtime_reuses_one_builder_across_invocations_and_cleans_up( + make_payload, + mock_nat, +): + runtime = adapter.NatRuntime() + await runtime.start(make_payload()) + + first = await runtime.invoke( + invocation_payload( + request_id="message-1", + context={"user_id": "user-1", "conversation_id": "conversation-1"}, + ) + ) + second = await runtime.invoke( + invocation_payload( + input_value="again", + request_id="message-2", + context={"user_id": "user-1", "conversation_id": "conversation-1"}, + ) + ) + await runtime.stop() + await runtime.stop() + + assert first == { + "harness": "nat", + "adapter": "python", + "mode": "nat_workflow", + "response": {"answer": 42}, + "completed": True, + "failed": False, + "error": None, + } + assert second["response"] == {"answer": 84} + mock_nat["workflow_builder"].from_config.assert_called_once_with( + config=mock_nat["typed_config"] + ) + mock_nat["session_manager"].create.assert_awaited_once_with( + config=mock_nat["typed_config"], + shared_builder=mock_nat["builder"], + ) + assert mock_nat["sessions"].session.call_args_list == [ + call( + user_id="user-1", + conversation_id="conversation-1", + user_message_id="message-1", + ), + call( + user_id="user-1", + conversation_id="conversation-1", + user_message_id="message-2", + ), + ] + assert mock_nat["session"].run.call_args_list == [ + call("hello", runtime_type="run-or-serve"), + call("again", runtime_type="run-or-serve"), + ] + assert mock_nat["runner"].result.await_count == 2 + mock_nat["sessions"].shutdown.assert_awaited_once_with() + assert mock_nat["builder_context"].__aexit__.await_count == 1 + + +async def test_start_failure_cleans_builder_and_redacts_cause( + make_payload, + mock_nat, + caplog, +): + mock_nat["session_manager"].create.side_effect = RuntimeError( + "api-key=super-secret" + ) + runtime = adapter.NatRuntime() + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + await runtime.start(make_payload()) + + assert error.value.code == "nat_workflow_start_failed" + assert "super-secret" not in error.value.message + assert "super-secret" not in caplog.text + assert mock_nat["builder_context"].__aexit__.await_count == 1 + await runtime.stop() + assert mock_nat["builder_context"].__aexit__.await_count == 1 + + +async def test_invoke_failure_is_normalized_and_redacts_cause( + make_payload, + mock_nat, + caplog, +): + mock_nat["runner"].result = AsyncMock( + side_effect=RuntimeError("api-key=super-secret") + ) + runtime = adapter.NatRuntime() + await runtime.start(make_payload()) + try: + result = await runtime.invoke(invocation_payload()) + finally: + await runtime.stop() + + assert result["failed"] is True + assert result["completed"] is False + assert result["response"] is None + assert result["error"] == { + "code": "nat_workflow_invoke_failed", + "message": "NAT workflow invocation failed; inspect adapter stderr for details", + "retryable": False, + } + assert "super-secret" not in json.dumps(result) + assert "super-secret" not in caplog.text + + +async def test_non_json_result_is_normalized_without_value_leak( + make_payload, + mock_nat, + caplog, +): + mock_nat["runner"].result = AsyncMock(return_value=object()) + mock_nat["to_jsonable"].side_effect = TypeError("secret-object-repr") + runtime = adapter.NatRuntime() + await runtime.start(make_payload()) + try: + result = await runtime.invoke(invocation_payload()) + finally: + await runtime.stop() + + assert result["error"] == { + "code": "nat_result_not_json_serializable", + "message": "NAT workflow returned a result that cannot be represented as JSON", + "retryable": False, + } + assert "secret-object-repr" not in json.dumps(result) + assert "secret-object-repr" not in caplog.text + + +def test_system_instruction_rejects_unsupported_workflow(make_payload): + payload = make_payload( + workflow={"_type": "custom_workflow"}, + instruction="Portable instruction", + ) + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.build_nat_config_mapping(payload) + + assert error.value.code == "nat_system_instruction_unsupported" + + +@pytest.mark.parametrize("transport", ["http", "streamable_http", "streamablehttp"]) +def test_mcp_server_normalizes_streamable_http_aliases(transport: str): + result = adapter.nat_mcp_server_config( + "docs", + {"transport": transport, "url": "https://mcp.test"}, + ) + + assert result == { + "transport": "streamable-http", + "url": "https://mcp.test", + } + + +@pytest.mark.parametrize("transport", ["websocket", ""]) +def test_mcp_server_rejects_unsupported_transport(transport: str): + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.nat_mcp_server_config( + "docs", + {"transport": transport, "url": "https://mcp.test"}, + ) + + assert error.value.code == "nat_unsupported_mcp_transport" From c6d8df0228c7e2545e3b48919539b6e69ee6ea29 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Tue, 4 Aug 2026 22:13:12 -0700 Subject: [PATCH 2/5] fix: address NAT adapter review feedback Signed-off-by: Ajay Thorve --- external/nat/README.md | 2 +- .../src/nemo_fabric_adapters/nat/adapter.py | 23 +- tests/adapters/test_external_nat_adapter.py | 262 +++++++++++++++--- 3 files changed, 239 insertions(+), 48 deletions(-) diff --git a/external/nat/README.md b/external/nat/README.md index 493173fe..a24b7913 100644 --- a/external/nat/README.md +++ b/external/nat/README.md @@ -106,7 +106,7 @@ run the example: ```bash uv pip install -e \ - /path/to/NeMo-Agent-Toolkit/examples/evaluation_and_profiling/email_phishing_analyzer + "/examples/evaluation_and_profiling/email_phishing_analyzer" uv run python external/nat/examples/email_phishing.py \ --base-dir "$PWD/.tmp/nat-reference" ``` diff --git a/external/nat/src/nemo_fabric_adapters/nat/adapter.py b/external/nat/src/nemo_fabric_adapters/nat/adapter.py index 9cf5308d..a9164a53 100644 --- a/external/nat/src/nemo_fabric_adapters/nat/adapter.py +++ b/external/nat/src/nemo_fabric_adapters/nat/adapter.py @@ -520,8 +520,6 @@ def _apply_enabled_tools( continue if identity in groups: - if identity in suppressed: - continue selected_groups[identity] = groups[identity] if identity not in selected_refs: selected_refs.append(identity) @@ -568,9 +566,8 @@ def _block_group_member( group["include"] = remaining return groups.pop(group_name, None) - config["workflow"]["tool_names"] = [ - name for name in config["workflow"]["tool_names"] if name != group_name - ] + tool_names = config["workflow"]["tool_names"] + tool_names[:] = [name for name in tool_names if name != group_name] return exclude = _string_list(group.get("exclude"), "function group exclude") or [] @@ -674,9 +671,11 @@ def build_nat_config(payload: dict[str, Any]) -> Any: def _session_kwargs(request: dict[str, Any]) -> dict[str, str]: - context = request.get("context") or {} + context = request.get("context") + if context is None: + context = {} if not isinstance(context, dict): - raise ValueError("request.context must be a mapping") + raise ValueError("NAT invocation request.context must be a mapping") values = { "user_id": context.get("user_id"), @@ -688,7 +687,9 @@ def _session_kwargs(request: dict[str, Any]) -> dict[str, str]: if value is None: continue if not isinstance(value, str) or not value: - raise ValueError(f"request context {name} must be a non-empty string") + raise ValueError( + f"NAT invocation request context {name} must be a non-empty string" + ) result[name] = value return result @@ -798,11 +799,15 @@ async def invoke(self, payload: dict[str, Any]) -> dict[str, Any]: "nat_invalid_request", "NAT invocation request must be a mapping", ) + try: + session_kwargs = _session_kwargs(request) + except ValueError as error: + return _failure_output("nat_invalid_request", str(error)) try: from nat.data_models.runtime_enum import RuntimeTypeEnum - async with self._sessions.session(**_session_kwargs(request)) as session: + async with self._sessions.session(**session_kwargs) as session: async with session.run( request.get("input", ""), runtime_type=RuntimeTypeEnum.RUN_OR_SERVE, diff --git a/tests/adapters/test_external_nat_adapter.py b/tests/adapters/test_external_nat_adapter.py index 9743ffd6..d6d386af 100644 --- a/tests/adapters/test_external_nat_adapter.py +++ b/tests/adapters/test_external_nat_adapter.py @@ -6,7 +6,6 @@ from __future__ import annotations import json -import os import sys import types from copy import deepcopy @@ -70,9 +69,7 @@ def make( "runtime_id": "runtime-1", "environment": {"workspace": str(tmp_path)}, }, - "capability_plan": { - "native": {"mcp_servers": deepcopy(mcp_servers or {})} - }, + "capability_plan": {"native": {"mcp_servers": deepcopy(mcp_servers or {})}}, } return make @@ -165,34 +162,40 @@ def mock_nat_fixture(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: } -def invocation_payload( - *, - input_value: Any = "hello", - request_id: str = "request-1", - context: dict[str, Any] | None = None, -) -> dict[str, Any]: - request: dict[str, Any] = {"input": input_value, "request_id": request_id} - if context is not None: - request["context"] = context - return { - "runtime_context": {"runtime_id": "runtime-1"}, - "request": request, - } +@pytest.fixture(name="make_invocation_payload") +def make_invocation_payload_fixture(): + """Return a factory for canonical Fabric invocation payloads.""" + + def make( + *, + input_value: Any = "hello", + request_id: str = "request-1", + context: Any = None, + raw_request: Any = None, + runtime_id: str = "runtime-1", + ) -> dict[str, Any]: + request = raw_request + if request is None: + request = {"input": input_value, "request_id": request_id} + if context is not None: + request["context"] = context + return { + "runtime_context": {"runtime_id": runtime_id}, + "request": request, + } + + return make def test_descriptor_declares_exact_source_reference_contract(): descriptor = json.loads( - (ROOT / "external" / "nat" / "fabric-adapter.json").read_text( - encoding="utf-8" - ) + (ROOT / "external" / "nat" / "fabric-adapter.json").read_text(encoding="utf-8") ) assert descriptor["adapter_id"] == "nvidia.fabric.nat" assert descriptor["harness"] == "nat" assert descriptor["adapter_kind"] == "python" - assert descriptor["runner"] == { - "module": "nemo_fabric_adapters.nat.adapter" - } + assert descriptor["runner"] == {"module": "nemo_fabric_adapters.nat.adapter"} assert descriptor["requirements"] == {} assert descriptor["config"]["accepts"] == [ "models", @@ -216,8 +219,9 @@ def test_descriptor_declares_exact_source_reference_contract(): def test_build_mapping_translates_components_models_and_instruction( make_payload, + monkeypatch: pytest.MonkeyPatch, ): - os.environ["NVIDIA_API_KEY"] = "test-key" + monkeypatch.setenv("NVIDIA_API_KEY", "test-key") workflow = { "_type": "react_agent", "llm_name": "default", @@ -294,9 +298,7 @@ def test_system_instruction_rejects_duplicate_nat_instruction_source(make_payloa def test_react_agent_without_tool_names_defaults_to_empty_list(make_payload): - payload = make_payload( - workflow={"_type": "react_agent", "llm_name": "default"} - ) + payload = make_payload(workflow={"_type": "react_agent", "llm_name": "default"}) result = adapter.build_nat_config_mapping(payload) @@ -309,18 +311,46 @@ def test_build_typed_config_discovers_components_before_validation( ): events: list[str] = [] mock_nat["discover"].side_effect = lambda _plugin_type: events.append("discover") - mock_nat["config_type"].model_validate.side_effect = ( - lambda _mapping: events.append("validate") or mock_nat["typed_config"] + mock_nat["config_type"].model_validate.side_effect = lambda _mapping: ( + events.append("validate") or mock_nat["typed_config"] ) result = adapter.build_nat_config(make_payload()) assert result is mock_nat["typed_config"] assert events == ["discover", "validate"] - mock_nat["discover"].assert_called_once_with( - mock_nat["plugin_types"].CONFIG_OBJECT + mock_nat["discover"].assert_called_once_with(mock_nat["plugin_types"].CONFIG_OBJECT) + + +def test_build_typed_config_contract_with_installed_nat( + make_payload, + monkeypatch: pytest.MonkeyPatch, +): + config_module = pytest.importorskip( + "nat.data_models.config", + reason="NAT is not installed in the base Fabric test environment", + ) + pytest.importorskip( + "nat.plugins.langchain.agent.react_agent", + reason="The NAT LangChain extra is not installed", + ) + monkeypatch.setenv("NVIDIA_API_KEY", "test-key") + payload = make_payload( + models={ + "default": { + "provider": "nvidia", + "model": "nvidia/test-model", + "api_key_env": "NVIDIA_API_KEY", + } + } ) + result = adapter.build_nat_config(payload) + + assert isinstance(result, config_module.Config) + assert result.workflow.type == "react_agent" + assert result.llms["default"].model_name == "nvidia/test-model" + @pytest.mark.parametrize( ("server_policy", "expected_group"), @@ -513,6 +543,27 @@ def test_root_tool_policy_selects_and_blocks_exact_group_members(make_payload): assert result["workflow"]["tool_names"] == ["clock", "calculator", "search"] +def test_blocking_last_group_member_and_function_removes_both_tool_refs( + make_payload, +): + payload = make_payload( + workflow={ + "_type": "react_agent", + "llm_name": "default", + "tool_names": ["calculator", "clock"], + }, + functions={"clock": {"_type": "current_datetime"}}, + function_groups={"calculator": {"_type": "calculator", "include": ["add"]}}, + tools={"blocked": ["calculator__add", "clock"]}, + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result["functions"] == {} + assert result["function_groups"] == {} + assert result["workflow"]["tool_names"] == [] + + @pytest.mark.parametrize( "tools", [ @@ -528,9 +579,7 @@ def test_root_tool_policy_rejects_unknown_exact_selectors(make_payload, tools): "llm_name": "default", "tool_names": ["calculator"], }, - function_groups={ - "calculator": {"_type": "calculator", "include": ["add"]} - }, + function_groups={"calculator": {"_type": "calculator", "include": ["add"]}}, tools=tools, ) @@ -542,19 +591,20 @@ def test_root_tool_policy_rejects_unknown_exact_selectors(make_payload, tools): async def test_runtime_reuses_one_builder_across_invocations_and_cleans_up( make_payload, + make_invocation_payload, mock_nat, ): runtime = adapter.NatRuntime() await runtime.start(make_payload()) first = await runtime.invoke( - invocation_payload( + make_invocation_payload( request_id="message-1", context={"user_id": "user-1", "conversation_id": "conversation-1"}, ) ) second = await runtime.invoke( - invocation_payload( + make_invocation_payload( input_value="again", request_id="message-2", context={"user_id": "user-1", "conversation_id": "conversation-1"}, @@ -601,6 +651,94 @@ async def test_runtime_reuses_one_builder_across_invocations_and_cleans_up( assert mock_nat["builder_context"].__aexit__.await_count == 1 +async def test_start_rejects_an_already_started_runtime(make_payload, mock_nat): + runtime = adapter.NatRuntime() + await runtime.start(make_payload()) + try: + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + await runtime.start(make_payload()) + finally: + await runtime.stop() + + assert error.value.code == "nat_runtime_already_started" + assert error.value.message == "NAT runtime is already started" + mock_nat["workflow_builder"].from_config.assert_called_once() + + +async def test_invoke_rejects_a_runtime_that_has_not_started( + make_invocation_payload, +): + runtime = adapter.NatRuntime() + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + await runtime.invoke(make_invocation_payload()) + + assert error.value.code == "nat_runtime_not_started" + assert error.value.message == "NAT runtime is not started" + + +async def test_invoke_rejects_a_different_runtime_id( + make_payload, + make_invocation_payload, + mock_nat, +): + runtime = adapter.NatRuntime() + await runtime.start(make_payload()) + try: + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + await runtime.invoke(make_invocation_payload(runtime_id="runtime-2")) + finally: + await runtime.stop() + + assert error.value.code == "nat_runtime_mismatch" + assert error.value.message == "NAT invocation does not match the active runtime" + mock_nat["sessions"].session.assert_not_called() + + +async def test_invoke_normalizes_a_non_mapping_request( + make_payload, + make_invocation_payload, + mock_nat, +): + runtime = adapter.NatRuntime() + await runtime.start(make_payload()) + try: + result = await runtime.invoke( + make_invocation_payload(raw_request=["not", "a", "mapping"]) + ) + finally: + await runtime.stop() + + assert result["error"] == { + "code": "nat_invalid_request", + "message": "NAT invocation request must be a mapping", + "retryable": False, + } + mock_nat["sessions"].session.assert_not_called() + + +async def test_invoke_normalizes_a_non_mapping_request_context( + make_payload, + make_invocation_payload, + mock_nat, +): + runtime = adapter.NatRuntime() + await runtime.start(make_payload()) + try: + result = await runtime.invoke( + make_invocation_payload(context=["not", "a", "mapping"]) + ) + finally: + await runtime.stop() + + assert result["error"] == { + "code": "nat_invalid_request", + "message": "NAT invocation request.context must be a mapping", + "retryable": False, + } + mock_nat["sessions"].session.assert_not_called() + + async def test_start_failure_cleans_builder_and_redacts_cause( make_payload, mock_nat, @@ -624,6 +762,7 @@ async def test_start_failure_cleans_builder_and_redacts_cause( async def test_invoke_failure_is_normalized_and_redacts_cause( make_payload, + make_invocation_payload, mock_nat, caplog, ): @@ -633,7 +772,7 @@ async def test_invoke_failure_is_normalized_and_redacts_cause( runtime = adapter.NatRuntime() await runtime.start(make_payload()) try: - result = await runtime.invoke(invocation_payload()) + result = await runtime.invoke(make_invocation_payload()) finally: await runtime.stop() @@ -651,6 +790,7 @@ async def test_invoke_failure_is_normalized_and_redacts_cause( async def test_non_json_result_is_normalized_without_value_leak( make_payload, + make_invocation_payload, mock_nat, caplog, ): @@ -659,7 +799,7 @@ async def test_non_json_result_is_normalized_without_value_leak( runtime = adapter.NatRuntime() await runtime.start(make_payload()) try: - result = await runtime.invoke(invocation_payload()) + result = await runtime.invoke(make_invocation_payload()) finally: await runtime.stop() @@ -697,6 +837,52 @@ def test_mcp_server_normalizes_streamable_http_aliases(transport: str): } +def test_mcp_stdio_expands_environment_and_parses_quoted_arguments( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv("NAT_TEST_MCP_COMMAND", "/opt/nat/bin/mcp-server") + + result = adapter.nat_mcp_server_config( + "calculator", + { + "transport": "stdio", + "url": "$NAT_TEST_MCP_COMMAND --label 'safe mode' --port 9000", + }, + ) + + assert result == { + "transport": "stdio", + "command": "/opt/nat/bin/mcp-server", + "args": ["--label", "safe mode", "--port", "9000"], + } + + +def test_mcp_stdio_rejects_unbalanced_quotes(): + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.nat_mcp_server_config( + "calculator", + {"transport": "stdio", "url": "mcp-server --label 'unterminated"}, + ) + + assert error.value.code == "nat_invalid_mcp_server" + assert error.value.message == ( + "NAT MCP server 'calculator' has an invalid stdio command" + ) + + +def test_mcp_stdio_rejects_a_whitespace_only_command(): + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.nat_mcp_server_config( + "calculator", + {"transport": "stdio", "url": " \t\n "}, + ) + + assert error.value.code == "nat_invalid_mcp_server" + assert error.value.message == ( + "NAT MCP server 'calculator' requires a non-empty url" + ) + + @pytest.mark.parametrize("transport", ["websocket", ""]) def test_mcp_server_rejects_unsupported_transport(transport: str): with pytest.raises(adapter.lifecycle.LifecycleError) as error: From 9a271c22798568cc8030740fcede94a50f4c7861 Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Wed, 5 Aug 2026 15:01:32 -0700 Subject: [PATCH 3/5] refactor: adopt typed workflow configuration Signed-off-by: Ajay Thorve --- external/nat/README.md | 12 +- external/nat/examples/calculator.py | 16 +- external/nat/examples/email_phishing.py | 20 +- external/nat/fabric-adapter.json | 38 +++- .../src/nemo_fabric_adapters/nat/adapter.py | 92 ++++++-- tests/adapters/test_external_nat_adapter.py | 196 ++++++++++++++---- 6 files changed, 286 insertions(+), 88 deletions(-) diff --git a/external/nat/README.md b/external/nat/README.md index a24b7913..68b607d3 100644 --- a/external/nat/README.md +++ b/external/nat/README.md @@ -15,14 +15,17 @@ calculator or email-phishing components. ## Configuration Boundary -NeMo Fabric owns portable configuration. `harness.settings` contains only -NAT-native component configuration that has no portable NeMo Fabric equivalent. +NeMo Fabric owns portable configuration. `workflow` selects and configures the +NAT executable, while `harness.settings` contains NAT-native function components +that have no portable NeMo Fabric equivalent. | NeMo Fabric input | NAT configuration | | --- | --- | | `models.` | `llms.`; every Fabric model-role name is preserved | | `instructions.system` | Built-in `react_agent` workflow `additional_instructions`; other workflow types reject this field in the initial adapter | -| `harness.settings.workflow` | `workflow` | +| `workflow.entrypoint.kind=nat_workflow` | Resolve a registered NAT workflow component | +| `workflow.entrypoint.ref` | `workflow._type` | +| `workflow.settings` | Remaining `workflow` component fields | | `harness.settings.functions` | `functions` | | `harness.settings.function_groups` | `function_groups` | | Harness-native `mcp.servers.` | Generated `mcp_client` function group named `` | @@ -31,7 +34,8 @@ NAT-native component configuration that has no portable NeMo Fabric equivalent. The adapter loads installed `nat.components` entry points before NAT validates the generated configuration. A custom function, function group, or workflow is therefore supplied as an installed NAT component package and selected by its -registered `_type`. No Python import path or callable crosses `FabricConfig`. +registered type in `workflow.entrypoint.ref` or the component `_type` in +`harness.settings`. No Python import path or callable crosses `FabricConfig`. At runtime, `start` loads components, enters one `WorkflowBuilder`, creates a `SessionManager` with that shared builder, and retains both resources. Each diff --git a/external/nat/examples/calculator.py b/external/nat/examples/calculator.py index 8570f58d..69135b9a 100644 --- a/external/nat/examples/calculator.py +++ b/external/nat/examples/calculator.py @@ -19,6 +19,8 @@ from nemo_fabric import MetadataConfig from nemo_fabric import ModelConfig from nemo_fabric import RuntimeConfig +from nemo_fabric import WorkflowConfig +from nemo_fabric import WorkflowEntrypointConfig def build_config() -> FabricConfig: @@ -32,13 +34,13 @@ def build_config() -> FabricConfig: harness=HarnessConfig( adapter_id="nvidia.fabric.nat", resolution="preinstalled", - settings={ - "workflow": { - "_type": "react_agent", - "llm_name": "default", - "tool_names": [], - } - }, + ), + workflow=WorkflowConfig( + entrypoint=WorkflowEntrypointConfig( + kind="nat_workflow", + ref="react_agent", + ), + settings={"llm_name": "default"}, ), models={ "default": ModelConfig( diff --git a/external/nat/examples/email_phishing.py b/external/nat/examples/email_phishing.py index 49297b45..e61748e4 100644 --- a/external/nat/examples/email_phishing.py +++ b/external/nat/examples/email_phishing.py @@ -19,6 +19,8 @@ from nemo_fabric import ModelConfig from nemo_fabric import RuntimeConfig from nemo_fabric import ToolsConfig +from nemo_fabric import WorkflowConfig +from nemo_fabric import WorkflowEntrypointConfig def build_config() -> FabricConfig: @@ -38,13 +40,17 @@ def build_config() -> FabricConfig: "_type": "email_phishing_analyzer", "llm": "default", } - }, - "workflow": { - "_type": "react_agent", - "llm_name": "default", - "tool_names": [], - "use_native_tool_calling": True, - }, + } + }, + ), + workflow=WorkflowConfig( + entrypoint=WorkflowEntrypointConfig( + kind="nat_workflow", + ref="react_agent", + ), + settings={ + "llm_name": "default", + "use_native_tool_calling": True, }, ), models={ diff --git a/external/nat/fabric-adapter.json b/external/nat/fabric-adapter.json index 554567c8..691bc14f 100644 --- a/external/nat/fabric-adapter.json +++ b/external/nat/fabric-adapter.json @@ -33,10 +33,6 @@ }, "type": "object", "properties": { - "workflow": { - "$ref": "#/$defs/component", - "description": "NAT workflow component configuration." - }, "functions": { "$ref": "#/$defs/components", "default": {}, @@ -48,7 +44,39 @@ "description": "NAT function-group component instances keyed by their configured names." } }, - "required": ["workflow"], + "additionalProperties": false + }, + "workflow_schema": { + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "properties": { + "entrypoint": { + "type": "object", + "properties": { + "kind": { + "const": "nat_workflow", + "description": "Resolve the reference through NAT's registered workflow components." + }, + "ref": { + "type": "string", + "minLength": 1, + "pattern": "^\\S+$", + "description": "Registered NAT workflow component type, copied to NAT _type." + } + }, + "required": ["kind", "ref"], + "additionalProperties": false + }, + "settings": { + "type": "object", + "properties": { + "_type": false + }, + "additionalProperties": true, + "description": "NAT workflow component settings; _type is derived from entrypoint.ref." + } + }, + "required": ["entrypoint"], "additionalProperties": false }, "requirements": {}, diff --git a/external/nat/src/nemo_fabric_adapters/nat/adapter.py b/external/nat/src/nemo_fabric_adapters/nat/adapter.py index a9164a53..bc5aec51 100644 --- a/external/nat/src/nemo_fabric_adapters/nat/adapter.py +++ b/external/nat/src/nemo_fabric_adapters/nat/adapter.py @@ -24,8 +24,15 @@ HARNESS = "nat" MODE = "nat_workflow" +WORKFLOW_KIND = "nat_workflow" FUNCTION_GROUP_SEPARATOR = "__" -NAT_SETTINGS_FIELDS = frozenset({"workflow", "functions", "function_groups"}) +NAT_SETTINGS_FIELDS = frozenset({"functions", "function_groups"}) +REACT_AGENT_REFS = frozenset( + { + "react_agent", + "nat.plugins.langchain.agent.react_agent/react_agent", + } +) RESERVED_MODEL_SETTINGS = frozenset( { "_type", @@ -73,6 +80,63 @@ def _mapping(value: Any, field: str) -> dict[str, Any]: return copy.deepcopy(value) +def _nat_workflow(payload: dict[str, Any]) -> dict[str, Any]: + fabric_config = common_utils.fabric_config(payload) + workflow_config = fabric_config.get("workflow") + if not isinstance(workflow_config, dict): + raise _config_error( + "nat_invalid_workflow", + "workflow must be a mapping", + field="workflow", + ) + + entrypoint = workflow_config.get("entrypoint") + if not isinstance(entrypoint, dict): + raise _config_error( + "nat_invalid_workflow", + "workflow.entrypoint must be a mapping", + field="workflow.entrypoint", + ) + kind = entrypoint.get("kind") + if kind != WORKFLOW_KIND: + raise _config_error( + "nat_invalid_workflow", + f"workflow.entrypoint.kind must equal {WORKFLOW_KIND!r}", + field="workflow.entrypoint.kind", + ) + workflow_ref = entrypoint.get("ref") + if ( + not isinstance(workflow_ref, str) + or not workflow_ref + or any(character.isspace() for character in workflow_ref) + ): + raise _config_error( + "nat_invalid_workflow", + "workflow.entrypoint.ref must be a non-empty string without whitespace", + field="workflow.entrypoint.ref", + ) + + settings = workflow_config.get("settings", {}) + if not isinstance(settings, dict): + raise _config_error( + "nat_invalid_workflow", + "workflow.settings must be a mapping", + field="workflow.settings", + ) + if "_type" in settings: + raise _config_error( + "nat_invalid_workflow", + "workflow.settings._type is reserved; use workflow.entrypoint.ref", + field="workflow.settings._type", + ) + + workflow = copy.deepcopy(settings) + workflow["_type"] = workflow_ref + if _is_react_agent(workflow): + workflow.setdefault("tool_names", []) + return workflow + + def _nat_component_settings(payload: dict[str, Any]) -> dict[str, Any]: settings = common_utils.settings_payload(payload) if not isinstance(settings, dict): @@ -90,21 +154,10 @@ def _nat_component_settings(payload: dict[str, Any]) -> dict[str, Any]: fields=unknown, ) - workflow = _mapping(settings.get("workflow"), "harness.settings.workflow") - workflow_type = workflow.get("_type") - if not isinstance(workflow_type, str) or not workflow_type.strip(): - raise _config_error( - "nat_invalid_harness_settings", - "harness.settings.workflow._type must be a non-empty string", - field="harness.settings.workflow._type", - ) - if _workflow_type(workflow) == "react_agent": - workflow.setdefault("tool_names", []) - functions = settings.get("functions", {}) function_groups = settings.get("function_groups", {}) return { - "workflow": workflow, + "workflow": _nat_workflow(payload), "functions": _mapping(functions, "harness.settings.functions"), "function_groups": _mapping( function_groups, @@ -204,9 +257,8 @@ def _nat_llms(payload: dict[str, Any]) -> dict[str, dict[str, Any]]: return llms -def _workflow_type(workflow: dict[str, Any]) -> str: - value = workflow.get("_type") - return value.rsplit("/", 1)[-1] if isinstance(value, str) else "" +def _is_react_agent(workflow: dict[str, Any]) -> bool: + return workflow.get("_type") in REACT_AGENT_REFS def _apply_system_instruction(config: dict[str, Any], payload: dict[str, Any]) -> None: @@ -215,7 +267,7 @@ def _apply_system_instruction(config: dict[str, Any], payload: dict[str, Any]) - return workflow = config["workflow"] - if _workflow_type(workflow) != "react_agent": + if not _is_react_agent(workflow): raise _config_error( "nat_system_instruction_unsupported", "instructions.system is supported only for a NAT react_agent workflow", @@ -224,10 +276,10 @@ def _apply_system_instruction(config: dict[str, Any], payload: dict[str, Any]) - if "additional_instructions" in workflow: raise _config_error( "nat_system_instruction_conflict", - "instructions.system conflicts with harness.settings.workflow.additional_instructions", + "instructions.system conflicts with workflow.settings.additional_instructions", fields=[ "instructions.system", - "harness.settings.workflow.additional_instructions", + "workflow.settings.additional_instructions", ], ) workflow["additional_instructions"] = instruction @@ -313,7 +365,7 @@ def _workflow_tool_names(config: dict[str, Any], reason: str) -> list[str]: raise _config_error( "nat_workflow_tools_unsupported", f"{reason} requires a NAT workflow with a string-list tool_names field", - field="harness.settings.workflow.tool_names", + field="workflow.settings.tool_names", ) return tool_names diff --git a/tests/adapters/test_external_nat_adapter.py b/tests/adapters/test_external_nat_adapter.py index d6d386af..96142e82 100644 --- a/tests/adapters/test_external_nat_adapter.py +++ b/tests/adapters/test_external_nat_adapter.py @@ -6,6 +6,7 @@ from __future__ import annotations import json +import runpy import sys import types from copy import deepcopy @@ -16,6 +17,7 @@ from unittest.mock import call import pytest +from nemo_fabric import Fabric ROOT = Path(__file__).parents[2] NAT_ADAPTER_SOURCE = ROOT / "external" / "nat" / "src" @@ -24,6 +26,18 @@ from nemo_fabric_adapters.nat import adapter # noqa: E402 +def _fabric_workflow( + ref: str = "react_agent", + *, + kind: str = "nat_workflow", + **settings: Any, +) -> dict[str, Any]: + return { + "entrypoint": {"kind": kind, "ref": ref}, + "settings": settings, + } + + @pytest.fixture(name="make_payload") def make_payload_fixture(tmp_path: Path): """Return a factory for canonical Fabric lifecycle payloads.""" @@ -41,18 +55,13 @@ def make( config: dict[str, Any] = { "harness": { "settings": { - "workflow": deepcopy( - workflow - or { - "_type": "react_agent", - "llm_name": "default", - "tool_names": [], - } - ), "functions": deepcopy(functions or {}), "function_groups": deepcopy(function_groups or {}), } }, + "workflow": deepcopy( + _fabric_workflow(llm_name="default") if workflow is None else workflow + ), "models": deepcopy(models or {}), } if instruction is not None: @@ -207,8 +216,19 @@ def test_descriptor_declares_exact_source_reference_contract(): "mcp", "mcp.tool_filters", ] - assert descriptor["settings_schema"]["required"] == ["workflow"] - assert descriptor["settings_schema"]["additionalProperties"] is False + settings_schema = descriptor["settings_schema"] + assert set(settings_schema["properties"]) == {"functions", "function_groups"} + assert "required" not in settings_schema + assert settings_schema["additionalProperties"] is False + workflow_schema = descriptor["workflow_schema"] + assert workflow_schema["required"] == ["entrypoint"] + assert workflow_schema["additionalProperties"] is False + entrypoint_schema = workflow_schema["properties"]["entrypoint"] + assert entrypoint_schema["properties"]["kind"]["const"] == "nat_workflow" + assert entrypoint_schema["properties"]["ref"]["pattern"] == r"^\S+$" + assert entrypoint_schema["required"] == ["kind", "ref"] + assert entrypoint_schema["additionalProperties"] is False + assert workflow_schema["properties"]["settings"]["properties"]["_type"] is False assert descriptor["capabilities"] == { "cancellation": False, "service": False, @@ -222,11 +242,11 @@ def test_build_mapping_translates_components_models_and_instruction( monkeypatch: pytest.MonkeyPatch, ): monkeypatch.setenv("NVIDIA_API_KEY", "test-key") - workflow = { - "_type": "react_agent", + workflow_settings = { "llm_name": "default", "tool_names": ["clock", "calculator"], } + workflow = _fabric_workflow(**workflow_settings) functions = {"clock": {"_type": "current_datetime"}} function_groups = { "calculator": {"_type": "calculator", "include": ["add", "subtract"]} @@ -257,8 +277,7 @@ def test_build_mapping_translates_components_models_and_instruction( assert result == { "workflow": { "_type": "react_agent", - "llm_name": "default", - "tool_names": ["clock", "calculator"], + **workflow_settings, "additional_instructions": "Use portable instructions.", }, "functions": functions, @@ -278,16 +297,15 @@ def test_build_mapping_translates_components_models_and_instruction( }, }, } - assert payload["config"]["harness"]["settings"]["workflow"] == workflow + assert payload["config"]["workflow"] == workflow def test_system_instruction_rejects_duplicate_nat_instruction_source(make_payload): payload = make_payload( - workflow={ - "_type": "react_agent", - "llm_name": "default", - "additional_instructions": "adapter-local value", - }, + workflow=_fabric_workflow( + llm_name="default", + additional_instructions="adapter-local value", + ), instruction="Portable instruction", ) @@ -298,13 +316,95 @@ def test_system_instruction_rejects_duplicate_nat_instruction_source(make_payloa def test_react_agent_without_tool_names_defaults_to_empty_list(make_payload): - payload = make_payload(workflow={"_type": "react_agent", "llm_name": "default"}) + payload = make_payload(workflow=_fabric_workflow(llm_name="default")) result = adapter.build_nat_config_mapping(payload) assert result["workflow"]["tool_names"] == [] +def test_namespaced_workflow_ref_is_preserved(make_payload): + payload = make_payload( + workflow=_fabric_workflow( + "nat.plugins.langchain.agent.react_agent/react_agent", + llm_name="default", + ) + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result["workflow"] == { + "_type": "nat.plugins.langchain.agent.react_agent/react_agent", + "llm_name": "default", + "tool_names": [], + } + + +def test_custom_workflow_named_react_agent_does_not_receive_built_in_defaults( + make_payload, +): + payload = make_payload(workflow=_fabric_workflow("custom/react_agent")) + + result = adapter.build_nat_config_mapping(payload) + + assert result["workflow"] == {"_type": "custom/react_agent"} + + +def test_missing_root_workflow_is_rejected(make_payload): + payload = make_payload() + payload["config"].pop("workflow") + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.build_nat_config_mapping(payload) + + assert error.value.code == "nat_invalid_workflow" + assert error.value.metadata["field"] == "workflow" + + +@pytest.mark.parametrize( + ("workflow", "field"), + [ + ({}, "workflow.entrypoint"), + ({"entrypoint": []}, "workflow.entrypoint"), + (_fabric_workflow(kind="python_callable"), "workflow.entrypoint.kind"), + (_fabric_workflow("bad ref"), "workflow.entrypoint.ref"), + ( + { + "entrypoint": {"kind": "nat_workflow", "ref": "react_agent"}, + "settings": [], + }, + "workflow.settings", + ), + (_fabric_workflow(_type="react_agent"), "workflow.settings._type"), + ], +) +def test_invalid_root_workflow_is_rejected(make_payload, workflow, field): + payload = make_payload(workflow=workflow) + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.build_nat_config_mapping(payload) + + assert error.value.code == "nat_invalid_workflow" + assert error.value.metadata["field"] == field + + +@pytest.mark.parametrize("example", ["calculator.py", "email_phishing.py"]) +def test_typed_examples_plan_with_root_workflow(tmp_path: Path, example: str): + descriptor = ROOT / "external" / "nat" / "fabric-adapter.json" + staged_descriptor = tmp_path / "adapters" / "nat" / "fabric-adapter.json" + staged_descriptor.parent.mkdir(parents=True) + staged_descriptor.write_text( + descriptor.read_text(encoding="utf-8"), encoding="utf-8" + ) + namespace = runpy.run_path(str(ROOT / "external" / "nat" / "examples" / example)) + + plan = Fabric().plan(namespace["build_config"](), base_dir=tmp_path) + + assert plan.config.workflow.entrypoint.kind == "nat_workflow" + assert plan.config.workflow.entrypoint.ref == "react_agent" + assert "workflow" not in plan.config.harness.settings + + def test_build_typed_config_discovers_components_before_validation( make_payload, mock_nat, @@ -322,9 +422,17 @@ def test_build_typed_config_discovers_components_before_validation( mock_nat["discover"].assert_called_once_with(mock_nat["plugin_types"].CONFIG_OBJECT) +@pytest.mark.parametrize( + "workflow_ref", + [ + "react_agent", + "nat.plugins.langchain.agent.react_agent/react_agent", + ], +) def test_build_typed_config_contract_with_installed_nat( make_payload, monkeypatch: pytest.MonkeyPatch, + workflow_ref: str, ): config_module = pytest.importorskip( "nat.data_models.config", @@ -336,13 +444,14 @@ def test_build_typed_config_contract_with_installed_nat( ) monkeypatch.setenv("NVIDIA_API_KEY", "test-key") payload = make_payload( + workflow=_fabric_workflow(workflow_ref, llm_name="default"), models={ "default": { "provider": "nvidia", "model": "nvidia/test-model", "api_key_env": "NVIDIA_API_KEY", } - } + }, ) result = adapter.build_nat_config(payload) @@ -440,11 +549,10 @@ def test_empty_mcp_allowlist_suppresses_server_and_existing_workflow_ref( make_payload, ): payload = make_payload( - workflow={ - "_type": "react_agent", - "llm_name": "default", - "tool_names": ["clock", "docs"], - }, + workflow=_fabric_workflow( + llm_name="default", + tool_names=["clock", "docs"], + ), functions={"clock": {"_type": "current_datetime"}}, mcp_servers={ "docs": { @@ -488,7 +596,7 @@ def test_all_suppressed_mcp_does_not_require_custom_workflow_tool_names( make_payload, ): payload = make_payload( - workflow={"_type": "custom_workflow"}, + workflow=_fabric_workflow("custom_workflow"), mcp_servers={ "docs": { "transport": "sse", @@ -506,11 +614,10 @@ def test_all_suppressed_mcp_does_not_require_custom_workflow_tool_names( def test_root_tool_policy_selects_and_blocks_exact_group_members(make_payload): payload = make_payload( - workflow={ - "_type": "react_agent", - "llm_name": "default", - "tool_names": ["clock", "unused", "calculator", "search"], - }, + workflow=_fabric_workflow( + llm_name="default", + tool_names=["clock", "unused", "calculator", "search"], + ), functions={ "clock": {"_type": "current_datetime"}, "unused": {"_type": "unused_function"}, @@ -547,11 +654,10 @@ def test_blocking_last_group_member_and_function_removes_both_tool_refs( make_payload, ): payload = make_payload( - workflow={ - "_type": "react_agent", - "llm_name": "default", - "tool_names": ["calculator", "clock"], - }, + workflow=_fabric_workflow( + llm_name="default", + tool_names=["calculator", "clock"], + ), functions={"clock": {"_type": "current_datetime"}}, function_groups={"calculator": {"_type": "calculator", "include": ["add"]}}, tools={"blocked": ["calculator__add", "clock"]}, @@ -574,11 +680,10 @@ def test_blocking_last_group_member_and_function_removes_both_tool_refs( ) def test_root_tool_policy_rejects_unknown_exact_selectors(make_payload, tools): payload = make_payload( - workflow={ - "_type": "react_agent", - "llm_name": "default", - "tool_names": ["calculator"], - }, + workflow=_fabric_workflow( + llm_name="default", + tool_names=["calculator"], + ), function_groups={"calculator": {"_type": "calculator", "include": ["add"]}}, tools=tools, ) @@ -812,9 +917,10 @@ async def test_non_json_result_is_normalized_without_value_leak( assert "secret-object-repr" not in caplog.text -def test_system_instruction_rejects_unsupported_workflow(make_payload): +@pytest.mark.parametrize("ref", ["custom_workflow", "custom/react_agent"]) +def test_system_instruction_rejects_unsupported_workflow(make_payload, ref): payload = make_payload( - workflow={"_type": "custom_workflow"}, + workflow=_fabric_workflow(ref), instruction="Portable instruction", ) From 16c360c3be2887330607684d50eb23a766fdb31f Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Wed, 5 Aug 2026 16:13:08 -0700 Subject: [PATCH 4/5] refactor: make NAT calculator example self-contained Signed-off-by: Ajay Thorve --- external/nat/README.md | 15 ++++--- external/nat/examples/calculator.py | 8 ++-- external/nat/examples/calculator_mcp.py | 44 +++++++++++++++++++++ tests/adapters/test_external_nat_adapter.py | 13 ++++++ 4 files changed, 69 insertions(+), 11 deletions(-) create mode 100644 external/nat/examples/calculator_mcp.py diff --git a/external/nat/README.md b/external/nat/README.md index 68b607d3..253dc50d 100644 --- a/external/nat/README.md +++ b/external/nat/README.md @@ -21,7 +21,7 @@ that have no portable NeMo Fabric equivalent. | NeMo Fabric input | NAT configuration | | --- | --- | -| `models.` | `llms.`; every Fabric model-role name is preserved | +| `models.` | `llms.`; every NeMo Fabric model-role name is preserved | | `instructions.system` | Built-in `react_agent` workflow `additional_instructions`; other workflow types reject this field in the initial adapter | | `workflow.entrypoint.kind=nat_workflow` | Resolve a registered NAT workflow component | | `workflow.entrypoint.ref` | `workflow._type` | @@ -47,19 +47,19 @@ streaming, or live-update support. ## MCP Tool Filters The adapter consumes the routed `capability_plan.native.mcp_servers` entries, -including the normalized per-server filters. Fabric MCP tool names remain bare +including the normalized per-server filters. NeMo Fabric MCP tool names remain bare server-local names; NAT exposes a selected member as `__`. -| Fabric server policy | Generated NAT function group | +| NeMo Fabric server policy | Generated NAT function group | | --- | --- | | `allowed_tools` omitted and `blocked_tools=[]` | No `include` or `exclude`; expose all discovered tools | | Nonempty `allowed_tools` only | `include=allowed_tools` | | `allowed_tools` omitted and nonempty `blocked_tools` | `exclude=blocked_tools` | -| Both lists configured | `include=allowed_tools`; Fabric requires `blocked_tools` to be disjoint, so those names are already outside the allowlist | +| Both lists configured | `include=allowed_tools`; NeMo Fabric requires `blocked_tools` to be disjoint, so those names are already outside the allowlist | | `allowed_tools=[]` | Omit the generated group; expose no tools from that server | NAT rejects a function group that sets both `include` and `exclude`, so the -adapter emits only `include` whenever an allowlist is present. Fabric rejects +adapter emits only `include` whenever an allowlist is present. NeMo Fabric rejects blank names and an allow/block overlap before adapter startup. A nonempty generated MCP group is added to workflows that expose `tool_names`; callers do not repeat portable MCP servers in `harness.settings`. A workflow implementation @@ -95,11 +95,10 @@ cp external/nat/fabric-adapter.json \ .tmp/nat-reference/adapters/nat/fabric-adapter.json ``` -The calculator example also requires a streamable-HTTP calculator MCP server. -Set its endpoint and run the typed `FabricConfig` example: +The calculator example starts its source-only MCP server over stdio, so it does +not require a separately managed endpoint. Run the typed `FabricConfig` example: ```bash -export CALCULATOR_MCP_URL=http://127.0.0.1:9901/mcp uv run python external/nat/examples/calculator.py \ --base-dir "$PWD/.tmp/nat-reference" ``` diff --git a/external/nat/examples/calculator.py b/external/nat/examples/calculator.py index 69135b9a..4921f2e9 100644 --- a/external/nat/examples/calculator.py +++ b/external/nat/examples/calculator.py @@ -8,7 +8,8 @@ import argparse import asyncio import json -import os +import shlex +import sys from pathlib import Path from nemo_fabric import Fabric @@ -57,10 +58,11 @@ def build_config() -> FabricConfig: ), runtime=RuntimeConfig(input_schema="text", output_schema="message"), ) + server = Path(__file__).with_name("calculator_mcp.py") config.add_mcp_server( "calculator", - transport="streamable-http", - url=os.environ.get("CALCULATOR_MCP_URL", "http://127.0.0.1:9901/mcp"), + transport="stdio", + url=shlex.join([sys.executable, str(server)]), exposure="harness_native", blocked_tools=["divide"], ) diff --git a/external/nat/examples/calculator_mcp.py b/external/nat/examples/calculator_mcp.py new file mode 100644 index 00000000..212638e7 --- /dev/null +++ b/external/nat/examples/calculator_mcp.py @@ -0,0 +1,44 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Self-contained calculator MCP server for the NAT reference example.""" + +from __future__ import annotations + +from mcp.server.fastmcp import FastMCP + +server = FastMCP("calculator") + + +@server.tool() +def add(left: float, right: float) -> float: + """Add two numbers.""" + + return left + right + + +@server.tool() +def subtract(left: float, right: float) -> float: + """Subtract the right value from the left value.""" + + return left - right + + +@server.tool() +def multiply(left: float, right: float) -> float: + """Multiply two numbers.""" + + return left * right + + +@server.tool() +def divide(left: float, right: float) -> float: + """Divide the left value by the right value.""" + + if right == 0: + raise ValueError("cannot divide by zero") + return left / right + + +if __name__ == "__main__": + server.run(transport="stdio") diff --git a/tests/adapters/test_external_nat_adapter.py b/tests/adapters/test_external_nat_adapter.py index 96142e82..0cc7fa07 100644 --- a/tests/adapters/test_external_nat_adapter.py +++ b/tests/adapters/test_external_nat_adapter.py @@ -405,6 +405,19 @@ def test_typed_examples_plan_with_root_workflow(tmp_path: Path, example: str): assert "workflow" not in plan.config.harness.settings +def test_calculator_example_uses_the_source_stdio_server(): + namespace = runpy.run_path( + str(ROOT / "external" / "nat" / "examples" / "calculator.py") + ) + + config = namespace["build_config"]() + calculator = config.mcp.servers["calculator"] + + assert calculator.transport == "stdio" + assert "calculator_mcp.py" in calculator.url + assert calculator.blocked_tools == ["divide"] + + def test_build_typed_config_discovers_components_before_validation( make_payload, mock_nat, From bce23f0deabc2b563597783d5b2bf4722a4d639c Mon Sep 17 00:00:00 2001 From: Ajay Thorve Date: Wed, 5 Aug 2026 16:19:49 -0700 Subject: [PATCH 5/5] test: cover NAT adapter cleanup failures Signed-off-by: Ajay Thorve --- external/README.md | 2 ++ tests/adapters/test_external_nat_adapter.py | 39 +++++++++++++++++++++ 2 files changed, 41 insertions(+) diff --git a/external/README.md b/external/README.md index e9a1719e..4732feda 100644 --- a/external/README.md +++ b/external/README.md @@ -14,6 +14,8 @@ These adapters are not published as wheels and are not wired into bundled or installed-adapter discovery. Packaging and discovery are separate concerns from the adapter contract demonstrated here. +The following source-only reference adapter is available: + | Harness | Adapter ID | Reference | | --- | --- | --- | | NVIDIA NeMo Agent Toolkit | `nvidia.fabric.nat` | [NAT adapter](nat/README.md) | diff --git a/tests/adapters/test_external_nat_adapter.py b/tests/adapters/test_external_nat_adapter.py index 0cc7fa07..5ca19a8d 100644 --- a/tests/adapters/test_external_nat_adapter.py +++ b/tests/adapters/test_external_nat_adapter.py @@ -878,6 +878,45 @@ async def test_start_failure_cleans_builder_and_redacts_cause( assert mock_nat["builder_context"].__aexit__.await_count == 1 +def test_config_translation_failure_is_normalized_and_redacts_cause( + make_payload, + mock_nat, +): + mock_nat["config_type"].model_validate.side_effect = RuntimeError( + "api-key=super-secret" + ) + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + adapter.build_nat_config(make_payload()) + + assert error.value.code == "nat_config_translation_failed" + assert error.value.message == ( + "Fabric config could not be translated into a valid NAT config" + ) + assert "super-secret" not in str(error.value) + + +async def test_stop_failure_clears_runtime_state_and_redacts_cause( + make_payload, + mock_nat, +): + mock_nat["sessions"].shutdown.side_effect = RuntimeError("api-key=super-secret") + runtime = adapter.NatRuntime() + await runtime.start(make_payload()) + + with pytest.raises(adapter.lifecycle.LifecycleError) as error: + await runtime.stop() + + assert error.value.code == "nat_runtime_stop_failed" + assert error.value.message == "NAT runtime failed to stop cleanly" + assert "super-secret" not in str(error.value) + mock_nat["sessions"].shutdown.assert_awaited_once_with() + assert mock_nat["builder_context"].__aexit__.await_count == 1 + + await runtime.stop() + mock_nat["sessions"].shutdown.assert_awaited_once_with() + + async def test_invoke_failure_is_normalized_and_redacts_cause( make_payload, make_invocation_payload,