diff --git a/external/README.md b/external/README.md new file mode 100644 index 00000000..4732feda --- /dev/null +++ b/external/README.md @@ -0,0 +1,21 @@ + + +# 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. + +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/external/nat/README.md b/external/nat/README.md new file mode 100644 index 00000000..253dc50d --- /dev/null +++ b/external/nat/README.md @@ -0,0 +1,118 @@ + + +# 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. `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 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` | +| `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 `` | +| `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 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 +`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. NeMo Fabric MCP tool names remain bare +server-local names; NAT exposes a selected member as `__`. + +| 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`; 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. 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 +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 starts its source-only MCP server over stdio, so it does +not require a separately managed endpoint. Run the typed `FabricConfig` example: + +```bash +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 \ + "/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..4921f2e9 --- /dev/null +++ b/external/nat/examples/calculator.py @@ -0,0 +1,90 @@ +# 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 shlex +import sys +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 WorkflowConfig +from nemo_fabric import WorkflowEntrypointConfig + + +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", + ), + workflow=WorkflowConfig( + entrypoint=WorkflowEntrypointConfig( + kind="nat_workflow", + ref="react_agent", + ), + settings={"llm_name": "default"}, + ), + 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"), + ) + server = Path(__file__).with_name("calculator_mcp.py") + config.add_mcp_server( + "calculator", + transport="stdio", + url=shlex.join([sys.executable, str(server)]), + 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/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/external/nat/examples/email_phishing.py b/external/nat/examples/email_phishing.py new file mode 100644 index 00000000..e61748e4 --- /dev/null +++ b/external/nat/examples/email_phishing.py @@ -0,0 +1,95 @@ +# 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 +from nemo_fabric import WorkflowConfig +from nemo_fabric import WorkflowEntrypointConfig + + +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=WorkflowConfig( + entrypoint=WorkflowEntrypointConfig( + kind="nat_workflow", + ref="react_agent", + ), + settings={ + "llm_name": "default", + "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..691bc14f --- /dev/null +++ b/external/nat/fabric-adapter.json @@ -0,0 +1,101 @@ +{ + "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": { + "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." + } + }, + "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": {}, + "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..bc5aec51 --- /dev/null +++ b/external/nat/src/nemo_fabric_adapters/nat/adapter.py @@ -0,0 +1,915 @@ +#!/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" +WORKFLOW_KIND = "nat_workflow" +FUNCTION_GROUP_SEPARATOR = "__" +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", + "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_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): + 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, + ) + + functions = settings.get("functions", {}) + function_groups = settings.get("function_groups", {}) + return { + "workflow": _nat_workflow(payload), + "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 _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: + instruction = common_utils.system_instruction(payload) + if instruction is None: + return + + workflow = config["workflow"] + if not _is_react_agent(workflow): + 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 workflow.settings.additional_instructions", + fields=[ + "instructions.system", + "workflow.settings.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="workflow.settings.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: + 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) + 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 [] + 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") + if context is None: + context = {} + if not isinstance(context, dict): + raise ValueError("NAT invocation 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"NAT invocation 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: + 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) 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..5ca19a8d --- /dev/null +++ b/tests/adapters/test_external_nat_adapter.py @@ -0,0 +1,1052 @@ +# 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 runpy +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 +from nemo_fabric import Fabric + +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 + + +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.""" + + 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": { + "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: + 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, + } + + +@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") + ) + + 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", + ] + 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, + "streaming": False, + "updates": False, + } + + +def test_build_mapping_translates_components_models_and_instruction( + make_payload, + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setenv("NVIDIA_API_KEY", "test-key") + 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"]} + } + 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", + **workflow_settings, + "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"]["workflow"] == workflow + + +def test_system_instruction_rejects_duplicate_nat_instruction_source(make_payload): + payload = make_payload( + workflow=_fabric_workflow( + 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=_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_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, +): + 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( + "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", + 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( + 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) + + 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"), + [ + ( + {}, + { + "_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=_fabric_workflow( + 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=_fabric_workflow("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=_fabric_workflow( + 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"] + + +def test_blocking_last_group_member_and_function_removes_both_tool_refs( + make_payload, +): + payload = make_payload( + 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"]}, + ) + + result = adapter.build_nat_config_mapping(payload) + + assert result["functions"] == {} + assert result["function_groups"] == {} + assert result["workflow"]["tool_names"] == [] + + +@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=_fabric_workflow( + 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, + make_invocation_payload, + mock_nat, +): + runtime = adapter.NatRuntime() + await runtime.start(make_payload()) + + first = await runtime.invoke( + make_invocation_payload( + request_id="message-1", + context={"user_id": "user-1", "conversation_id": "conversation-1"}, + ) + ) + second = await runtime.invoke( + make_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_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, + 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 + + +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, + 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(make_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, + make_invocation_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(make_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 + + +@pytest.mark.parametrize("ref", ["custom_workflow", "custom/react_agent"]) +def test_system_instruction_rejects_unsupported_workflow(make_payload, ref): + payload = make_payload( + workflow=_fabric_workflow(ref), + 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", + } + + +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: + adapter.nat_mcp_server_config( + "docs", + {"transport": transport, "url": "https://mcp.test"}, + ) + + assert error.value.code == "nat_unsupported_mcp_transport"