feat: Add Deep Agents integration

pyproject.toml

@@ -84,6 +84,11 @@ langgraph = [
     "langgraph>=1.1.10,<2.0.0",
 ]
 
+deepagents = [
+    "nemo-flow[langgraph]",
+    "deepagents>=0.5.3,<0.6.0",
+]
+
 langchain-nvidia = [
     "nemo-flow[langchain]",
     "langchain-nvidia-ai-endpoints~=1.0",

python/nemo_flow/integrations/deepagents/README.md

@@ -0,0 +1,67 @@
+<!--
+SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+-->
+
+# NeMo Flow Deep Agents Integration
+
+This directory contains the `nemo_flow.integrations.deepagents` package, which
+adds Deep Agents-specific observability on top of the LangChain and LangGraph
+integrations.
+
+Deep Agents is built on LangChain `AgentMiddleware` and the LangGraph runtime.
+The integration therefore composes the existing NeMo Flow LangChain and
+LangGraph hooks, then emits stable mark events for Deep Agents concepts:
+subagents, async subagents, human-in-the-loop interrupts and resumes, skills,
+filesystem operations, and sandbox execution.
+
+## Setup
+
+```bash
+uv sync --all-groups --extra deepagents
+just build-python
+```
+
+## Usage Example
+
+```python
+import nemo_flow
+from deepagents import create_deep_agent
+from nemo_flow.integrations.deepagents import (
+    NemoFlowDeepAgentsCallbackHandler,
+    add_nemo_flow_integration,
+)
+
+agent = create_deep_agent(
+    **add_nemo_flow_integration(
+        model="nvidia:nvidia/nemotron-3-nano-30b-a3b",
+        tools=[],
+        skills=["/skills/research/"],
+        name="main-agent",
+    )
+)
+
+with nemo_flow.scope.scope("deepagents-request", nemo_flow.ScopeType.Agent):
+    result = agent.invoke(
+        {"messages": [{"role": "user", "content": "Research recent GPU news"}]},
+        config={"callbacks": [NemoFlowDeepAgentsCallbackHandler()]},
+    )
+```
+
+## What Is Captured
+
+- LangChain model and tool calls through NeMo Flow managed execution.
+- LangGraph run scopes through callbacks.
+- Human-in-the-loop interrupt and resume marks.
+- Deep Agents `task` delegation marks for synchronous subagents.
+- Deep Agents async subagent tool lifecycle marks.
+- Filesystem tool marks for `ls`, `read_file`, `write_file`, `edit_file`,
+  `glob`, and `grep`.
+- Sandbox/local shell `execute` marks.
+- Configured skills and subagent summaries at agent-run start.
+- Direct backend method calls when the backend is wrapped with
+  `observe_backend()` or via `add_nemo_flow_integration(..., backend=...)`.
+
+Remote async subagents still need NeMo Flow instrumentation in the remote graph
+or process to capture their internal model and tool calls. Supervisor-side
+observability captures the task lifecycle and returned status.

python/nemo_flow/integrations/deepagents/__init__.py

@@ -0,0 +1,127 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""NeMo Flow integrations for Deep Agents."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from nemo_flow.integrations.deepagents.backend import NemoFlowDeepAgentsBackend, observe_backend
+from nemo_flow.integrations.deepagents.callbacks import NemoFlowDeepAgentsCallbackHandler
+from nemo_flow.integrations.deepagents.middleware import NemoFlowDeepAgentsMiddleware
+
+
+def add_nemo_flow_integration(
+    kwargs: Mapping[str, Any] | None = None,
+    *,
+    instrument_subagents: bool = True,
+    instrument_backend: bool = True,
+    **overrides: Any,
+) -> dict[str, Any]:
+    """
+    Receives the keyword arguments for ``create_deep_agent`` and returns them with NeMo Flow observability attached.
+
+    Use this helper as ``create_deep_agent(**add_nemo_flow_integration(...))``.
+    It injects Deep Agents-aware middleware at the top level, adds the same
+    middleware to dictionary-style custom subagents that do not inherit parent
+    middleware, and optionally wraps the backend.
+    """
+    observed = dict(kwargs or {})
+    observed.update(overrides)
+
+    skills = _string_sequence(observed.get("skills"))
+    subagents = list(observed.get("subagents") or ())
+    subagent_summaries = [_subagent_summary(subagent) for subagent in subagents]
+    backend = observed.get("backend")
+
+    if isinstance(backend, NemoFlowDeepAgentsBackend):
+        observed["backend"] = backend
+        backend_name = type(backend.__wrapped__).__name__
+    elif instrument_backend and backend is not None:
+        observed["backend"] = observe_backend(backend)
+        backend_name = type(backend).__name__
+    elif backend is not None:
+        backend_name = type(backend).__name__
+    else:
+        backend_name = None
+
+    middleware = list(observed.get("middleware") or ())
+    _append_middleware(
+        middleware,
+        NemoFlowDeepAgentsMiddleware(
+            agent_name=observed.get("name"),
+            skills=skills,
+            subagents=subagent_summaries,
+            backend_name=backend_name,
+        ),
+    )
+    observed["middleware"] = middleware
+
+    if instrument_subagents and subagents:
+        observed["subagents"] = [_instrument_subagent(subagent) for subagent in subagents]
+
+    return observed
+
+
+def _append_middleware(middleware: list[Any], new_middleware: NemoFlowDeepAgentsMiddleware) -> None:
+    if any(isinstance(item, NemoFlowDeepAgentsMiddleware) for item in middleware):
+        return
+    middleware.append(new_middleware)
+
+
+def _instrument_subagent(subagent: Any) -> Any:
+    if not isinstance(subagent, dict):
+        return subagent
+
+    observed = dict(subagent)
+    middleware = list(observed.get("middleware") or ())
+    _append_middleware(
+        middleware,
+        NemoFlowDeepAgentsMiddleware(
+            agent_name=observed.get("name"),
+            skills=_string_sequence(observed.get("skills")),
+            subagents=None,
+        ),
+    )
+    observed["middleware"] = middleware
+    return observed
+
+
+def _subagent_summary(subagent: Any) -> Mapping[str, Any]:
+    if isinstance(subagent, Mapping):
+        summary: dict[str, Any] = {"type": "subagent"}
+        for key in ("name", "description", "model", "graph_id", "url"):
+            value = subagent.get(key)
+            if value is not None:
+                summary[key] = value
+        if "skills" in subagent:
+            summary["skills"] = _string_sequence(subagent.get("skills"))
+        return summary
+
+    summary = {"type": type(subagent).__name__}
+    for attr in ("name", "description", "graph_id", "url"):
+        value = getattr(subagent, attr, None)
+        if value is not None:
+            summary[attr] = value
+    return summary
+
+
+def _string_sequence(value: Any) -> Sequence[str] | None:
+    if value is None:
+        return None
+    if isinstance(value, str):
+        return [value]
+    if isinstance(value, Sequence):
+        return [str(item) for item in value]
+    return [str(value)]
+
+
+__all__ = [
+    "NemoFlowDeepAgentsBackend",
+    "NemoFlowDeepAgentsCallbackHandler",
+    "NemoFlowDeepAgentsMiddleware",
+    "add_nemo_flow_integration",
+    "observe_backend",
+]

python/nemo_flow/integrations/deepagents/_events.py

@@ -0,0 +1,188 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared Deep Agents observability helpers."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+import nemo_flow
+
+_logger = logging.getLogger(__name__)
+
+SYNC_SUBAGENT_TOOLS = frozenset({"task"})
+ASYNC_SUBAGENT_TOOLS = frozenset(
+    {
+        "start_async_task",
+        "check_async_task",
+        "update_async_task",
+        "cancel_async_task",
+        "list_async_tasks",
+    }
+)
+# Mirrors the Deep Agents built-in filesystem tools listed in the backend docs:
+# https://docs.langchain.com/oss/python/deepagents/backends
+FILESYSTEM_TOOLS = frozenset({"ls", "read_file", "write_file", "edit_file", "glob", "grep"})
+FILESYSTEM_BACKEND_METHODS = frozenset({"ls", "read", "write", "edit", "glob", "grep"})
+# Deep Agents sandbox backends expose execute()/aexecute(); the tool name is execute.
+SANDBOX_TOOLS = frozenset({"execute"})
+
+
+def tool_kind(tool_name: str) -> str | None:
+    """Return the Deep Agents semantic category for a known built-in tool."""
+    if tool_name in SYNC_SUBAGENT_TOOLS:
+        return "subagent"
+    if tool_name in ASYNC_SUBAGENT_TOOLS:
+        return "async_subagent"
+    if tool_name in SANDBOX_TOOLS:
+        return "sandbox"
+    if tool_name in FILESYSTEM_TOOLS:
+        return "filesystem"
+    return None
+
+
+def backend_kind(method_name: str) -> str:
+    """Return the Deep Agents semantic category for a backend method."""
+    normalized = method_name.removeprefix("a")
+    if normalized in SANDBOX_TOOLS:
+        return "sandbox"
+    if normalized in FILESYSTEM_TOOLS or normalized in FILESYSTEM_BACKEND_METHODS:
+        return "filesystem"
+    if normalized in {"upload_files", "download_file", "download_files"}:
+        return "filesystem"
+    return "backend"
+
+
+def mark_base_name(kind: str) -> str:
+    """Return a stable mark-event base name for a Deep Agents category."""
+    return {
+        "subagent": "DeepAgents Subagent",
+        "async_subagent": "DeepAgents Async Subagent",
+        "human_in_the_loop": "DeepAgents Human In The Loop",
+        "skill": "DeepAgents Skills",
+        "sandbox": "DeepAgents Sandbox",
+        "filesystem": "DeepAgents Filesystem",
+        "backend": "DeepAgents Backend",
+    }.get(kind, "DeepAgents")
+
+
+def json_safe(value: Any) -> nemo_flow.Json:
+    """Return a conservative JSON-compatible value."""
+    if value is None or isinstance(value, str | int | float | bool):
+        return value
+    if isinstance(value, Mapping):
+        return {str(key): json_safe(item) for key, item in value.items()}
+    if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray):
+        return [json_safe(item) for item in value]
+    if isinstance(value, bytes | bytearray):
+        return f"<{type(value).__name__}: {len(value)} bytes>"
+    return repr(value)
+
+
+def summarize_value(value: Any) -> nemo_flow.Json:
+    """Return a JSON-compatible representation for observability payloads."""
+    if value is None or isinstance(value, int | float | bool):
+        return value
+    if isinstance(value, str):
+        return value
+    if isinstance(value, bytes | bytearray):
+        return f"<{type(value).__name__}: {len(value)} bytes>"
+    if isinstance(value, Mapping):
+        return {str(key): summarize_value(item) for key, item in value.items()}
+    if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray):
+        return [summarize_value(item) for item in value]
+
+    content = getattr(value, "content", None)
+    if isinstance(content, str):
+        summary: dict[str, nemo_flow.Json] = {"content": content}
+        for attr in ("name", "id", "tool_call_id"):
+            attr_value = getattr(value, attr, None)
+            if attr_value is not None:
+                summary[attr] = summarize_value(attr_value)
+        return summary
+
+    return repr(value)
+
+
+def tool_event_data(
+    tool_name: str,
+    args: Mapping[str, Any],
+    *,
+    result: Any = None,
+    error: BaseException | None = None,
+) -> dict[str, nemo_flow.Json]:
+    """Build a stable Deep Agents tool event payload."""
+    data: dict[str, nemo_flow.Json] = {
+        "tool_name": tool_name,
+        "args": summarize_value(args),
+    }
+
+    for key in ("name", "agent_name", "task_id", "thread_id", "run_id", "graph_id", "status"):
+        value = args.get(key)
+        if value is not None:
+            data[key] = summarize_value(value)
+
+    for key in ("path", "file_path", "pattern", "glob", "command"):
+        value = args.get(key)
+        if value is not None:
+            data[key] = summarize_value(value)
+
+    if result is not None:
+        data["result"] = summarize_value(result)
+    if error is not None:
+        data["error"] = repr(error)
+
+    return data
+
+
+def backend_event_data(
+    backend_name: str,
+    method_name: str,
+    args: tuple[Any, ...],
+    kwargs: Mapping[str, Any],
+    *,
+    result: Any = None,
+    error: BaseException | None = None,
+) -> dict[str, nemo_flow.Json]:
+    """Build a stable Deep Agents backend event payload."""
+    data: dict[str, nemo_flow.Json] = {
+        "backend": backend_name,
+        "method": method_name,
+        "args": summarize_value(args),
+        "kwargs": summarize_value(kwargs),
+    }
+    if result is not None:
+        data["result"] = summarize_value(result)
+    if error is not None:
+        data["error"] = repr(error)
+    return data
+
+
+def emit_mark(
+    base_name: str,
+    kind: str,
+    phase: str,
+    data: Mapping[str, Any],
+    *,
+    metadata: Mapping[str, Any] | None = None,
+) -> None:
+    """Emit a Deep Agents mark event without changing framework behavior."""
+    event_metadata: dict[str, Any] = {
+        "integration": "deepagents",
+        "deepagents_kind": kind,
+        "phase": phase,
+    }
+    if metadata:
+        event_metadata.update(metadata)
+
+    try:
+        nemo_flow.scope.event(
+            f"{base_name} {phase.title()}",
+            data=json_safe(data),
+            metadata=json_safe(event_metadata),
+        )
+    except Exception:
+        _logger.debug("NeMo Flow: Deep Agents mark emission failed", exc_info=True)