feat(nemo-agents): support for nemo agents run and nemo agents deploy for Fabric backed agents

plugins/nemo-agents/src/nemo_agents_plugin/agent_config_formats.py

@@ -0,0 +1,112 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared validation and deployment resolution for agent config formats."""
+
+from __future__ import annotations
+
+from typing import Any, Protocol
+
+from nemo_agents_plugin.agent_config import AgentConfig
+from nemo_agents_plugin.entities import NAT_WORKFLOW_CONFIG_FORMAT, NEMO_AGENTS_SPEC_CONFIG_FORMAT
+from nemo_agents_plugin.utils import inject_default_model, inject_gateway_url, inject_nemo_trace_fields
+from pydantic import ValidationError
+
+
+class AgentConfigFormatError(ValueError):
+    """Base error for unsupported or invalid agent config formats."""
+
+
+class UnsupportedAgentConfigFormatError(AgentConfigFormatError):
+    """Raised when no handler exists for an agent config format."""
+
+
+class InvalidAgentConfigError(AgentConfigFormatError):
+    """Raised when an agent config does not satisfy its format contract."""
+
+
+class AgentConfigFormatHandler(Protocol):
+    """Validate and resolve one persisted agent config format."""
+
+    def validate(self, config: dict[str, Any]) -> dict[str, Any]: ...
+
+    def resolve_for_deployment(
+        self,
+        config: dict[str, Any],
+        *,
+        workspace: str,
+        agent_name: str,
+    ) -> dict[str, Any]: ...
+
+
+class _NatWorkflowConfigHandler:
+    def validate(self, config: dict[str, Any]) -> dict[str, Any]:
+        return config
+
+    def resolve_for_deployment(
+        self,
+        config: dict[str, Any],
+        *,
+        workspace: str,
+        agent_name: str,
+    ) -> dict[str, Any]:
+        resolved = inject_gateway_url(config, workspace)
+        resolved = inject_default_model(resolved)
+        inject_nemo_trace_fields(resolved, workspace=workspace, agent_name=agent_name)
+        return resolved
+
+
+class _NemoAgentsSpecConfigHandler:
+    def validate(self, config: dict[str, Any]) -> dict[str, Any]:
+        return self._normalize(config)
+
+    def resolve_for_deployment(
+        self,
+        config: dict[str, Any],
+        *,
+        workspace: str,
+        agent_name: str,
+    ) -> dict[str, Any]:
+        del workspace, agent_name
+        return self._normalize(config)
+
+    @staticmethod
+    def _normalize(config: dict[str, Any]) -> dict[str, Any]:
+        try:
+            return AgentConfig.model_validate(config).model_dump(exclude_none=True)
+        except ValidationError as error:
+            raise InvalidAgentConfigError(f"Invalid agent config: {error}") from error
+
+
+_AGENT_CONFIG_FORMAT_HANDLERS: dict[str, AgentConfigFormatHandler] = {
+    NAT_WORKFLOW_CONFIG_FORMAT: _NatWorkflowConfigHandler(),
+    NEMO_AGENTS_SPEC_CONFIG_FORMAT: _NemoAgentsSpecConfigHandler(),
+}
+
+
+def get_agent_config_format_handler(config_format: str) -> AgentConfigFormatHandler:
+    """Return the handler registered for an agent config format."""
+    try:
+        return _AGENT_CONFIG_FORMAT_HANDLERS[config_format]
+    except KeyError as error:
+        raise UnsupportedAgentConfigFormatError(f"Unsupported config_format {config_format!r}.") from error
+
+
+def validate_agent_config(config_format: str, config: dict[str, Any]) -> dict[str, Any]:
+    """Validate and normalize an agent config before persistence."""
+    return get_agent_config_format_handler(config_format).validate(config)
+
+
+def resolve_agent_config_for_deployment(
+    config_format: str,
+    config: dict[str, Any],
+    *,
+    workspace: str,
+    agent_name: str,
+) -> dict[str, Any]:
+    """Resolve a persisted agent config for deployment."""
+    return get_agent_config_format_handler(config_format).resolve_for_deployment(
+        config,
+        workspace=workspace,
+        agent_name=agent_name,
+    )

plugins/nemo-agents/src/nemo_agents_plugin/api/v2/agents.py

@@ -14,16 +14,11 @@
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, Query
-from nemo_agents_plugin.agent_config import AgentConfig
+from nemo_agents_plugin.agent_config_formats import AgentConfigFormatError, validate_agent_config
 from nemo_agents_plugin.api.v2._perms import AgentPerms
 from nemo_agents_plugin.api.v2.dependencies import get_entity_client
 from nemo_agents_plugin.authz import scope
-from nemo_agents_plugin.entities import (
-    NAT_WORKFLOW_CONFIG_FORMAT,
-    NEMO_AGENTS_SPEC_CONFIG_FORMAT,
-    Agent,
-    AgentDeployment,
-)
+from nemo_agents_plugin.entities import Agent, AgentDeployment
 from nemo_agents_plugin.schema import (
     AgentFilter,
     AgentPage,
@@ -33,7 +28,6 @@
 from nemo_platform_plugin.authz import CallerKind, path_rule
 from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityConflictError, NemoEntityNotFoundError
 from nemo_platform_plugin.schema import PaginationData
-from pydantic import ValidationError
 
 # Deployment statuses that block agent deletion.
 # "failed" and "deleting" are excluded — they are terminal/in-cleanup and
@@ -192,13 +186,7 @@ async def delete_agent(
 
 
 def _validate_agent_config_for_create(body: CreateAgentRequest) -> dict[str, Any]:
-    if body.config_format == NAT_WORKFLOW_CONFIG_FORMAT:
-        return body.config
-
-    if body.config_format == NEMO_AGENTS_SPEC_CONFIG_FORMAT:
-        try:
-            return AgentConfig.model_validate(body.config).model_dump(exclude_none=True)
-        except ValidationError as exc:
-            raise HTTPException(status_code=400, detail=f"Invalid agent config: {exc}") from exc
-
-    raise HTTPException(status_code=400, detail=f"Unsupported config_format {body.config_format!r}.")
+    try:
+        return validate_agent_config(body.config_format, body.config)
+    except AgentConfigFormatError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc

plugins/nemo-agents/src/nemo_agents_plugin/api/v2/deployments.py

@@ -21,13 +21,11 @@
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, Query
-from nemo_agents_plugin.agent_config import AgentConfig
+from nemo_agents_plugin.agent_config_formats import AgentConfigFormatError, resolve_agent_config_for_deployment
 from nemo_agents_plugin.api.v2._perms import DeploymentPerms
 from nemo_agents_plugin.api.v2.dependencies import get_entity_client
 from nemo_agents_plugin.authz import scope
 from nemo_agents_plugin.entities import (
-    NAT_WORKFLOW_CONFIG_FORMAT,
-    NEMO_AGENTS_SPEC_CONFIG_FORMAT,
     Agent,
     AgentDeployment,
     is_container_deployment_mode,
@@ -37,12 +35,10 @@
     DeploymentFilter,
     DeploymentPage,
 )
-from nemo_agents_plugin.utils import inject_default_model, inject_gateway_url, inject_nemo_trace_fields
 from nemo_platform_plugin.api.filters import make_filter_obj_dep
 from nemo_platform_plugin.authz import CallerKind, path_rule
 from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityConflictError, NemoEntityNotFoundError
 from nemo_platform_plugin.schema import PaginationData
-from pydantic import ValidationError
 
 logger = logging.getLogger(__name__)
 
@@ -112,19 +108,15 @@ async def create_deployment(
 
 
 def _resolve_deployment_config(agent: Agent, *, workspace: str) -> dict[str, Any]:
-    if agent.config_format == NAT_WORKFLOW_CONFIG_FORMAT:
-        resolved_config = inject_gateway_url(agent.config, workspace)
-        resolved_config = inject_default_model(resolved_config)
-        inject_nemo_trace_fields(resolved_config, workspace=workspace, agent_name=agent.name)
-        return resolved_config
-
-    if agent.config_format == NEMO_AGENTS_SPEC_CONFIG_FORMAT:
-        try:
-            return AgentConfig.model_validate(agent.config).model_dump(exclude_none=True)
-        except ValidationError as exc:
-            raise HTTPException(status_code=400, detail=f"Invalid agent config: {exc}") from exc
-
-    raise HTTPException(status_code=400, detail=f"Unsupported config_format {agent.config_format!r}.")
+    try:
+        return resolve_agent_config_for_deployment(
+            agent.config_format,
+            agent.config,
+            workspace=workspace,
+            agent_name=agent.name,
+        )
+    except AgentConfigFormatError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
 
 
 @router.get("/deployments", response_model=DeploymentPage, tags=["Agent Deployments"])

plugins/nemo-agents/src/nemo_agents_plugin/cli.py

@@ -12,7 +12,7 @@
 instance.
 
 - ``invoke``   — single invocation
-- ``run``      — start a persistent local FastAPI server for NAT configs
+- ``run``      — start a persistent local FastAPI server
 
 The ``evaluate`` and ``optimize`` commands are auto-generated from the
 ``EvaluateAgentJob`` and ``OptimizeAgentJob`` registered under the
@@ -38,6 +38,7 @@
 import logging
 import os
 import re
+import sys
 import time
 from dataclasses import asdict
 from datetime import datetime
@@ -205,26 +206,52 @@ def run(
             ...,
             "--agent-config",
             "-c",
-            help="Path to a NAT workflow YAML config file.",
+            help="Path to an agent YAML config file.",
             exists=True,
             file_okay=True,
             dir_okay=False,
         ),
         host: str = typer.Option("0.0.0.0", "--host"),
         port: int = typer.Option(8080, "--port", "-p"),
     ) -> None:
-        """Run an agent locally as a persistent FastAPI server (wraps ``nat start fastapi``)."""
+        """Run an agent locally as a persistent FastAPI server."""
         import subprocess
 
-        cmd = ["nat", "start", "fastapi", "--config_file", agent_config.name, "--host", host, "--port", str(port)]
+        config = _load_yaml(agent_config)
+        if not isinstance(config, dict):
+            typer.echo(f"Error: agent config {agent_config} root must be a YAML mapping.", err=True)
+            raise typer.Exit(code=1)
+
+        config_format = config.get("config_format", NAT_WORKFLOW_CONFIG_FORMAT)
+        if config_format == NEMO_AGENTS_SPEC_CONFIG_FORMAT:
+            cmd = [
+                sys.executable,
+                "-m",
+                "nemo_agents_plugin.fabric.server",
+                "--agent-config",
+                agent_config.name,
+                "--host",
+                host,
+                "--port",
+                str(port),
+            ]
+        elif config_format == NAT_WORKFLOW_CONFIG_FORMAT:
+            cmd = ["nat", "start", "fastapi", "--config_file", agent_config.name, "--host", host, "--port", str(port)]
+        else:
+            typer.echo(f"Error: unsupported config_format {config_format!r}", err=True)
+            raise typer.Exit(code=1)
+
         typer.echo(f"Starting agent server: {' '.join(cmd)}")
         try:
             subprocess.run(cmd, check=True, cwd=agent_config.parent)
         except subprocess.CalledProcessError as exc:
             typer.echo(f"Agent server exited with code {exc.returncode}.", err=True)
             raise typer.Exit(code=exc.returncode)
         except FileNotFoundError:
-            typer.echo("Error: 'nat' command not found.  Install nvidia-nat-core.", err=True)
+            if config_format == NAT_WORKFLOW_CONFIG_FORMAT:
+                typer.echo("Error: 'nat' command not found.  Install nvidia-nat-core.", err=True)
+            else:
+                typer.echo(f"Error: server command {cmd[0]!r} was not found.", err=True)
             raise typer.Exit(code=1)
 
 

plugins/nemo-agents/src/nemo_agents_plugin/fabric/environment.py

@@ -0,0 +1,25 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Prepare Platform-owned environment paths before Fabric runtime startup."""
+
+from pathlib import Path
+
+from nemo_agents_plugin.agent_config import AgentConfig
+
+
+def ensure_local_workspace_dir(agent_config: AgentConfig, base_dir: Path) -> None:
+    """Create the configured local workspace relative to the agent base directory."""
+    if agent_config.environment.provider != "local":
+        return
+
+    configured_workspace = Path(agent_config.environment.workspace)
+    if configured_workspace.is_absolute():
+        raise ValueError("Local workspace path must be relative to the agent base directory.")
+
+    resolved_base_dir = base_dir.resolve()
+    workspace = (resolved_base_dir / configured_workspace).resolve()
+    if not workspace.is_relative_to(resolved_base_dir):
+        raise ValueError("Local workspace path must remain within the agent base directory.")
+
+    workspace.mkdir(parents=True, exist_ok=True)

plugins/nemo-agents/src/nemo_agents_plugin/fabric/invocation.py

@@ -11,7 +11,8 @@
 from typing import Any
 
 from nemo_agents_plugin.agent_config import AgentConfig
-from nemo_agents_plugin.fabric.runtime import FabricRuntimeRequest, FabricRuntimeResult, run_fabric_agent_once
+from nemo_agents_plugin.fabric.environment import ensure_local_workspace_dir
+from nemo_agents_plugin.fabric.runtime import FabricOneShotRequest, FabricRuntimeResult, run_fabric_agent_once
 from nemo_agents_plugin.fabric.translator import translate_agent_config
 
 
@@ -23,27 +24,17 @@ async def invoke_agent_config_once(
 ) -> list[FabricRuntimeResult]:
     """Translate a Platform agent config and run each input through Fabric once."""
     fabric_config = translate_agent_config(agent_config)
-    await asyncio.to_thread(_ensure_local_workspace_dir, agent_config, base_dir)
+    await asyncio.to_thread(ensure_local_workspace_dir, agent_config, base_dir)
 
     results: list[FabricRuntimeResult] = []
     for item in inputs:
         results.append(
             await run_fabric_agent_once(
-                FabricRuntimeRequest(
+                FabricOneShotRequest(
                     fabric_config=fabric_config,
                     base_dir=base_dir,
                     input=item,
                 )
             )
         )
     return results
-
-
-def _ensure_local_workspace_dir(agent_config: AgentConfig, base_dir: Path) -> None:
-    if agent_config.environment.provider != "local":
-        return
-
-    workspace = Path(agent_config.environment.workspace)
-    if not workspace.is_absolute():
-        workspace = base_dir / workspace
-    workspace.mkdir(parents=True, exist_ok=True)