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/fabric/environment.py

@@ -0,0 +1,19 @@
+# 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
+
+    workspace = Path(agent_config.environment.workspace)
+    if not workspace.is_absolute():
+        workspace = base_dir / workspace
+    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)

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

@@ -20,12 +20,29 @@
 from typing import Any
 
 # CI type-checks this plugin via ty extra-paths without installing nemo-agents deps.
-from nemo_fabric import Fabric, FabricConfig, FabricError, RunRequest, RunResult  # ty: ignore[unresolved-import]
+from nemo_fabric import (  # ty: ignore[unresolved-import]
+    Fabric,
+    FabricConfig,
+    FabricError,
+    RunRequest,
+    RunResult,
+    Runtime,
+)
 
 
 @dataclass(frozen=True, slots=True)
-class FabricRuntimeRequest:
-    """Platform-owned request for one Fabric runtime invocation.
+class FabricInvocationRequest:
+    """Platform-owned request for one invocation on an active Fabric runtime."""
+
+    input: Any = ""
+    request_id: str | None = None
+    caller_context: dict[str, Any] = field(default_factory=dict)
+    timeout_seconds: float | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class FabricOneShotRequest:
+    """Platform-owned request for one ephemeral Fabric runtime invocation.
 
     This is an internal bridge type. The fields are intentionally close to
     Fabric's ``RunRequest`` while preserving Platform-owned lifecycle inputs
@@ -70,8 +87,30 @@ class FabricRuntimeTimeoutError(FabricRuntimeExecutionError):
     """Raised when a Fabric runtime invocation exceeds the Platform timeout."""
 
 
+async def invoke_fabric_runtime(
+    runtime: Runtime,
+    request: FabricInvocationRequest,
+) -> FabricRuntimeResult:
+    """Invoke an active Fabric runtime without changing its lifecycle."""
+    try:
+        result = await asyncio.wait_for(
+            runtime.invoke(request=_with_platform_invocation_context(request)),
+            timeout=request.timeout_seconds,
+        )
+    except TimeoutError as error:
+        raise FabricRuntimeTimeoutError(
+            f"Fabric runtime invocation timed out after {request.timeout_seconds:g}s.",
+        ) from error
+    except FabricError as error:
+        raise FabricRuntimeExecutionError(
+            f"Fabric runtime invocation failed: {error}",
+        ) from error
+
+    return _normalize_fabric_run_result(result)
+
+
 async def run_fabric_agent_once(
-    request: FabricRuntimeRequest,
+    request: FabricOneShotRequest,
     *,
     fabric: Any | None = None,
 ) -> FabricRuntimeResult:
@@ -96,7 +135,7 @@ async def run_fabric_agent_once(
 
 
 async def _invoke_fabric_agent_once(
-    request: FabricRuntimeRequest,
+    request: FabricOneShotRequest,
     *,
     fabric: Any,
 ) -> RunResult:
@@ -108,7 +147,7 @@ async def _invoke_fabric_agent_once(
         return await runtime.invoke(request=_with_platform_invocation_context(request))
 
 
-def _with_platform_invocation_context(request: FabricRuntimeRequest) -> RunRequest:
+def _with_platform_invocation_context(request: FabricInvocationRequest | FabricOneShotRequest) -> RunRequest:
     """Preserve Platform invocation metadata when calling Fabric."""
     request_kwargs: dict[str, Any] = {
         "context": request.caller_context,