feat(nemo-agents): one shot local invocation support for fabric backed agents

e2e/test_nemo_agents.py

@@ -16,6 +16,7 @@
 pytestmark = [pytest.mark.e2e_config("e2e/configs/local-subprocess.yaml")]
 
 _TEST_AGENT_RESPONSE = "The answer to your question is 42."
+_NEMO_AGENTS_SPEC_CONFIG_FORMAT = "nemo-agents-spec-v1"
 
 
 def _unique_name(prefix: str) -> str:
@@ -67,6 +68,20 @@ def _agent_config(label: str) -> dict[str, Any]:
     }
 
 
+def _platform_agent_config(label: str) -> dict[str, Any]:
+    """Return a minimal Platform-owned agent config for API persistence tests."""
+    return {
+        "config_format": _NEMO_AGENTS_SPEC_CONFIG_FORMAT,
+        "name": label,
+        "default_harness": "hermes",
+        "harnesses": {
+            "hermes": {
+                "kind": "hermes",
+            }
+        },
+    }
+
+
 def _page_data(page: Any) -> list[dict[str, Any]]:
     if isinstance(page, dict):
         data = page.get("data", [])
@@ -257,8 +272,8 @@ def test_agent_list_pagination_sorting_and_filtering(sdk: NeMoPlatform, workspac
         sdk.agents.create(
             workspace=workspace,
             name=alternate_name,
-            config=_agent_config(alternate_name),
-            config_format="e2e-other-format",
+            config=_platform_agent_config(alternate_name),
+            config_format=_NEMO_AGENTS_SPEC_CONFIG_FORMAT,
         )
 
         first_page = _get_agents_page(sdk, workspace, params={"page": 1, "page_size": 2, "sort": "name"})
@@ -274,7 +289,7 @@ def test_agent_list_pagination_sorting_and_filtering(sdk: NeMoPlatform, workspac
         filtered_page = _get_agents_page(
             sdk,
             workspace,
-            params={"page_size": 100, "filter[config_format]": "e2e-other-format"},
+            params={"page_size": 100, "filter[config_format]": _NEMO_AGENTS_SPEC_CONFIG_FORMAT},
         )
         filtered_names = {agent["name"] for agent in _page_data(filtered_page)}
         assert alternate_name in filtered_names

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

@@ -11,12 +11,19 @@
 from __future__ import annotations
 
 import logging
+from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+from nemo_agents_plugin.agent_config import AgentConfig
 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 Agent, AgentDeployment
+from nemo_agents_plugin.entities import (
+    NAT_WORKFLOW_CONFIG_FORMAT,
+    NEMO_AGENTS_SPEC_CONFIG_FORMAT,
+    Agent,
+    AgentDeployment,
+)
 from nemo_agents_plugin.schema import (
     AgentFilter,
     AgentPage,
@@ -26,6 +33,7 @@
 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
@@ -50,12 +58,14 @@ async def create_agent(
     body: CreateAgentRequest,
     entity_client: NemoEntitiesClient = Depends(get_entity_client),
 ) -> Agent:
-    """Create a new agent from a NAT workflow config."""
+    """Create a new agent from an agent config."""
+    config = _validate_agent_config_for_create(body)
+
     agent = Agent(
         name=body.name,
         workspace=workspace,
         description=body.description,
-        config=body.config,
+        config=config,
         config_format=body.config_format,
     )
     try:
@@ -179,3 +189,16 @@ async def delete_agent(
     except Exception as exc:
         logger.exception("Failed to delete agent '%s'", name)
         raise HTTPException(status_code=500, detail="Failed to delete agent.") from exc
+
+
+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}.")

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

@@ -18,12 +18,20 @@
 
 import logging
 import secrets
+from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+from nemo_agents_plugin.agent_config import AgentConfig
 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 Agent, AgentDeployment, is_container_deployment_mode
+from nemo_agents_plugin.entities import (
+    NAT_WORKFLOW_CONFIG_FORMAT,
+    NEMO_AGENTS_SPEC_CONFIG_FORMAT,
+    Agent,
+    AgentDeployment,
+    is_container_deployment_mode,
+)
 from nemo_agents_plugin.schema import (
     CreateDeploymentRequest,
     DeploymentFilter,
@@ -34,6 +42,7 @@
 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__)
 
@@ -73,10 +82,9 @@ async def create_deployment(
     # 2. Build deployment name (auto-generate if not provided)
     deployment_name = body.name or f"{body.agent}-{secrets.token_hex(4)}"
 
-    # 3. Deep-copy config and inject IGW URL, telemetry fields, and default model.
-    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=body.agent)
+    # 3. Resolve deployment-time config. NAT workflows need legacy injection;
+    # Platform-owned agent specs stay strict and are translated by the runner.
+    resolved_config = _resolve_deployment_config(agent, workspace=workspace)
 
     # 4. Create the entity with status "pending"
     deployment = AgentDeployment(
@@ -103,6 +111,22 @@ async def create_deployment(
     return saved
 
 
+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}.")
+
+
 @router.get("/deployments", response_model=DeploymentPage, tags=["Agent Deployments"])
 @scope.read
 @path_rule(