Adds support for running Open Agent Spec agents as a new workflow type

docs/source/components/agents/agent-spec.md

@@ -0,0 +1,52 @@
+<!--
+    SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+    SPDX-License-Identifier: Apache-2.0
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+-->
+
+# Agent Spec Workflow
+
+This workflow allows running an [Agent Spec] configuration inside NeMo Agent Toolkit by converting it to a LangGraph component via the Agent Spec → LangGraph adapter.
+
+## Install
+
+- Install optional extra:
+
+```bash
+pip install 'nvidia-nat[agentspec]'
+```
+
+## Example configuration
+
+```yaml
+workflow:
+  _type: agent_spec
+  description: Agent Spec workflow
+  <!-- path-check-skip-next-line -->
+  agentspec_path: path/to/agent_spec.yaml  # or agentspec_yaml / agentspec_json
+  tool_names: [pretty_formatting]
+  max_history: 15
+  verbose: true
+```
+
+Exactly one of `agentspec_yaml`, `agentspec_json`, or `agentspec_path` must be provided.
+
+## Notes and limitations
+
+- Tools: NeMo Agent toolkit built-in tools provided in `tool_names` are exposed to the adapter `tool_registry` by name. If the Agent Spec also defines tools, the registries are merged; duplicate names are overwritten by built-in tools.
+- I/O: Inputs are standard `ChatRequest` messages; the workflow returns a `ChatResponse`.
+- Streaming: Non supported. 
+- Checkpointing: Not supported.
+
+[Agent Spec]: https://github.com/oracle/agent-spec

pyproject.toml

@@ -80,6 +80,7 @@ crewai = ["nvidia-nat-crewai"]
 data-flywheel = ["nvidia-nat-data-flywheel"]
 ingestion = ["nvidia-nat-ingestion"] # meta-package
 langchain = ["nvidia-nat-langchain"]
+agentspec = ["nvidia-nat-langchain", "pyagentspec>=0.1", "langgraph-agentspec-adapter>=0.1"]  # TODO: How do we actually reference the langgraph adapter as a dependency?
 llama-index = ["nvidia-nat-llama-index"]
 mcp = ["nvidia-nat-mcp"]
 mem0ai = ["nvidia-nat-mem0ai"]

scripts/agentspec_smoke.py

@@ -0,0 +1,73 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import asyncio
+import sys
+import types
+
+
+async def main():
+    # Ensure NAT src is importable if running from repo root
+    import os
+    repo_root = os.path.dirname(os.path.abspath(__file__))
+    src_dir = os.path.join(os.path.dirname(repo_root), "src")
+    if src_dir not in sys.path:
+        sys.path.insert(0, src_dir)
+
+    # Force registration imports
+    import nat.agent.register  # noqa: F401
+
+    # Create a fake adapter module that returns a stub component
+    class StubComponent:
+
+        async def ainvoke(self, value):
+            if isinstance(value, dict) and "messages" in value:
+                msgs = value["messages"]
+                last_user = next((m.get("content") for m in reversed(msgs) if m.get("role") == "user"), "")
+                return {"output": last_user}
+            return {"output": str(value)}
+
+    class StubLoader:
+
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def load_yaml(self, _):
+            return StubComponent()
+
+    fake_mod = types.ModuleType("langgraph_agentspec_adapter.agentspecloader")
+    fake_mod.AgentSpecLoader = StubLoader
+    sys.modules["langgraph_agentspec_adapter"] = types.ModuleType("langgraph_agentspec_adapter")
+    sys.modules["langgraph_agentspec_adapter.agentspecloader"] = fake_mod
+
+    # Import registers agent workflows (including Agent Spec)
+    import nat.agent.agentspec.register  # noqa: F401
+    from nat.agent.agentspec.config import AgentSpecWorkflowConfig
+    from nat.builder.workflow_builder import WorkflowBuilder
+
+    spec_yaml = """
+component_type: Agent
+name: echo-agent
+description: echo
+"""
+
+    cfg = AgentSpecWorkflowConfig(llm_name="dummy", agentspec_yaml=spec_yaml, tool_names=[])
+    async with WorkflowBuilder() as builder:
+        fn = await builder.set_workflow(cfg)
+        out = await fn.acall_invoke(input_message="hello agentspec")
+        print("OK:", out)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/nat/agent/agentspec/config.py

@@ -0,0 +1,64 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from pathlib import Path
+
+from pydantic import Field
+from pydantic import model_validator
+
+from nat.data_models.agent import AgentBaseConfig
+from nat.data_models.component_ref import FunctionGroupRef
+from nat.data_models.component_ref import FunctionRef
+
+
+class AgentSpecWorkflowConfig(AgentBaseConfig, name="agent_spec"):
+    """
+    NAT function that executes an Agent Spec configuration via the LangGraph adapter.
+
+    Provide exactly one of agentspec_yaml, agentspec_json, or agentspec_path.
+    Optionally supply tool_names to make NAT/LC tools available to the Agent Spec runtime.
+    """
+
+    description: str = Field(default="Agent Spec Workflow", description="Description of this workflow.")
+
+    agentspec_yaml: str | None = Field(default=None, description="Inline Agent Spec YAML content")
+    agentspec_json: str | None = Field(default=None, description="Inline Agent Spec JSON content")
+    agentspec_path: str | None = Field(default=None, description="Path to an Agent Spec YAML/JSON file")
+
+    tool_names: list[FunctionRef | FunctionGroupRef] = Field(
+        default_factory=list, description="Optional list of tool names/groups to expose to the Agent Spec runtime.")
+
+    max_history: int = Field(default=15, description="Maximum number of messages to keep in conversation history.")
+
+    @model_validator(mode="after")
+    def _validate_sources(self):
+        provided = [self.agentspec_yaml, self.agentspec_json, self.agentspec_path]
+        cnt = sum(1 for v in provided if v)
+        if cnt != 1:
+            raise ValueError("Exactly one of agentspec_yaml, agentspec_json, or agentspec_path must be provided")
+        return self
+
+
+def read_agentspec_payload(config: AgentSpecWorkflowConfig) -> tuple[str, str]:
+    """Return (format, payload_str) where format is 'yaml' or 'json'."""
+    if config.agentspec_yaml:
+        return ("yaml", config.agentspec_yaml)
+    if config.agentspec_json:
+        return ("json", config.agentspec_json)
+    assert config.agentspec_path
+    path = Path(config.agentspec_path)
+    text = path.read_text(encoding="utf-8")
+    ext = path.suffix.lower()
+    fmt = "json" if ext == ".json" else "yaml"
+    return (fmt, text)

src/nat/agent/agentspec/register.py

@@ -0,0 +1,154 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import io
+import logging
+from typing import Any
+
+from nat.builder.framework_enum import LLMFrameworkEnum
+from nat.builder.function_info import FunctionInfo
+from nat.cli.register_workflow import register_function
+from nat.data_models.api_server import ChatRequest
+from nat.data_models.api_server import ChatRequestOrMessage
+from nat.data_models.api_server import ChatResponse
+from nat.data_models.api_server import Usage
+from nat.utils.type_converter import GlobalTypeConverter
+
+from .config import AgentSpecWorkflowConfig
+from .config import read_agentspec_payload
+
+logger = logging.getLogger(__name__)
+
+
+def _to_plain_messages(messages: list[Any]) -> list[dict[str, Any]]:
+    plain: list[dict[str, Any]] = []
+    for m in messages:
+        # Accept either NAT Message models or LangChain BaseMessage dicts
+        role = None
+        content = None
+        if isinstance(m, dict):
+            role = m.get("role")
+            content = m.get("content")
+        else:
+            # Try NAT Message model
+            if hasattr(m, "role"):
+                role = getattr(m.role, "value", None) or str(getattr(m, "role"))
+            # Various content shapes
+            if hasattr(m, "content"):
+                c = getattr(m, "content")
+                if isinstance(c, str):
+                    content = c
+                else:
+                    try:
+                        buf = io.StringIO()
+                        for part in c:
+                            if hasattr(part, "text"):
+                                buf.write(str(getattr(part, "text")))
+                            else:
+                                buf.write(str(part))
+                        content = buf.getvalue()
+                    except Exception:
+                        content = str(c)
+            # Fallback: LangChain BaseMessage has .type
+            if role is None and hasattr(m, "type"):
+                role = str(getattr(m, "type"))
+            if content is None and hasattr(m, "content"):
+                content = str(getattr(m, "content"))
+        plain.append({"role": role or "user", "content": content or ""})
+    return plain
+
+
+@register_function(config_type=AgentSpecWorkflowConfig, framework_wrappers=[LLMFrameworkEnum.LANGCHAIN])
+async def agent_spec_workflow(config: AgentSpecWorkflowConfig, builder):
+    # Lazy import to make the dependency optional unless this workflow is used
+    try:
+        from langgraph_agentspec_adapter.agentspecloader import AgentSpecLoader  # type: ignore
+    except Exception as e:  # pragma: no cover - import error path
+        raise ImportError("Agent Spec adapter not installed. Install with: pip install 'nvidia-nat[agentspec]'") from e
+
+    # Build tool registry from NAT tool names if provided
+    tools = await builder.get_tools(tool_names=config.tool_names, wrapper_type=LLMFrameworkEnum.LANGCHAIN)
+    tool_registry = {getattr(t, "name", f"tool_{i}"): t for i, t in enumerate(tools)} if tools else {}
+
+    fmt, payload = read_agentspec_payload(config)
+    loader = AgentSpecLoader(tool_registry=tool_registry, checkpointer=None, config=None)
+
+    # Compile Agent Spec to a LangGraph component
+    if fmt == "yaml":
+        component = loader.load_yaml(payload)
+    else:
+        component = loader.load_json(payload)
+
+    async def _response_fn(chat_request_or_message: ChatRequestOrMessage) -> ChatResponse | str:
+        from langchain_core.messages import trim_messages  # lazy import with LANGCHAIN wrapper
+
+        from nat.agent.base import AGENT_LOG_PREFIX
+
+        try:
+            message = GlobalTypeConverter.get().convert(chat_request_or_message, to_type=ChatRequest)
+
+            # Trim message history
+            trimmed = trim_messages(messages=[m.model_dump() for m in message.messages],
+                                    max_tokens=config.max_history,
+                                    strategy="last",
+                                    token_counter=len,
+                                    start_on="human",
+                                    include_system=True)
+
+            # Best-effort: pass messages in a generic shape expected by adapter graphs
+            input_state: dict[str, Any] = {"messages": _to_plain_messages(trimmed)}
+
+            result: Any
+            result = await component.ainvoke(input_state)
+
+            # Heuristic extraction of assistant content
+            content: str | None = None
+            if isinstance(result, dict):
+                msgs = result.get("messages")
+                if isinstance(msgs, list) and msgs:
+                    for entry in reversed(msgs):
+                        # LangChain BaseMessage objects have `.type` (e.g., 'ai', 'human') and `.content`
+                        if hasattr(entry, "type") and hasattr(entry, "content"):
+                            role = getattr(entry, "type", None)
+                            if role in ("ai", "assistant", "system"):
+                                content = str(getattr(entry, "content", ""))
+                                break
+                        # Dict-shaped message
+                        if isinstance(entry, dict):
+                            role = entry.get("role")
+                            if role in ("assistant", "system", "ai"):
+                                content = str(entry.get("content", ""))
+                                break
+                if content is None and "output" in result:
+                    content = str(result.get("output"))
+            if content is None and isinstance(result, str):
+                content = result
+            if content is None:
+                content = str(result)
+
+            prompt_tokens = sum(len(str(msg.content).split()) for msg in message.messages)
+            completion_tokens = len(content.split()) if content else 0
+            usage = Usage(prompt_tokens=prompt_tokens,
+                          completion_tokens=completion_tokens,
+                          total_tokens=prompt_tokens + completion_tokens)
+            response = ChatResponse.from_string(content, usage=usage)
+            if chat_request_or_message.is_string:
+                return GlobalTypeConverter.get().convert(response, to_type=str)
+            return response
+        except Exception as ex:  # pragma: no cover - surface original exception
+            logger.error("%s Agent Spec workflow failed: %s", AGENT_LOG_PREFIX, str(ex))
+            raise
+
+    yield FunctionInfo.from_fn(_response_fn, description=config.description)

src/nat/agent/register.py

@@ -23,3 +23,4 @@
 from .responses_api_agent import register as responses_api_agent
 from .rewoo_agent import register as rewoo_agent
 from .tool_calling_agent import register as tool_calling_agent
+from .agentspec import register as agentspec

tests/conftest.py

@@ -53,7 +53,9 @@
 PROJECT_DIR = os.path.dirname(TESTS_DIR)
 SRC_DIR = os.path.join(PROJECT_DIR, "src")
 EXAMPLES_DIR = os.path.join(PROJECT_DIR, "examples")
-sys.path.append(SRC_DIR)
+# Prepend local src so tests run against workspace code rather than any installed package
+if SRC_DIR not in sys.path:
+    sys.path.insert(0, SRC_DIR)
 
 os.environ.setdefault("DASK_DISTRIBUTED__WORKER__PYTHON", sys.executable)