How to Integrate Custom Analyzer Nodes into NVIDIA SkillSpector’s LangGraph Workflow

You can integrate a custom analyzer into SkillSpector’s LangGraph workflow by creating a node function that accepts and mutates the shared SkillspectorState dictionary, then registering it in the analyzer registry at src/skillspector/nodes/analyzers/__init__.py, where it is automatically discovered and wired into the execution graph.

NVIDIA’s SkillSpector implements its security-scanning pipeline as a LangGraph state machine defined in src/skillspector/graph.py. To extend this framework with custom static analysis or LLM-based security checks, developers must implement nodes that conform to the state schema and expose them through the central analyzer registry. This guide demonstrates the exact steps to integrate custom analyzer nodes into the LangGraph workflow using source-accurate examples from the repository.

Understanding SkillSpector’s LangGraph Architecture

The workflow is built around three core components that govern how custom analyzers integrate.

SkillspectorState defines the state schema used by all nodes in the graph. Located in src/skillspector/state.py, it structures fields such as inputs, context, and analysis_results that every analyzer reads and writes.

src/skillspector/graph.py constructs the StateGraph and handles the automatic wiring of analyzer nodes. The construction logic iterates over every analyzer ID registered in ANALYZER_NODE_IDS, adding each to the workflow:

for analyzer_id in ANALYZER_NODE_IDS:
    workflow.add_node(analyzer_id, ANALYZER_NODES[analyzer_id])
    workflow.add_edge("build_context", analyzer_id)
    workflow.add_edge(analyzer_id, "meta_analyzer")

The Analyzer Registry in src/skillspector/nodes/analyzers/__init__.py exports two critical objects: ANALYZER_NODES (a mapping of node IDs to callables) and ANALYZER_NODE_IDS (a flat list of those IDs). Any callable added to this dictionary is automatically connected to the pipeline when the graph initializes.

Implementing a Custom Analyzer Node

Creating the Node Function

Create a new Python file in src/skillspector/nodes/analyzers/, for example my_custom_analyzer.py. Your function must accept the current state as a dictionary and return the mutated state:

# src/skillspector/nodes/analyzers/my_custom_analyzer.py

from typing import Dict, Any

def my_custom_analyzer(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Custom analyzer that detects unsafe eval() usage.
    """
    results = []
    
    # Access files prepared by the build_context node

    for path, content in state["context"]["files"].items():
        if "eval(" in content:
            results.append({
                "file": path,
                "message": "Use of eval() can lead to code injection.",
                "severity": "high",
            })
    
    # Store findings under a unique key in analysis_results

    state["analysis_results"]["my_custom_analyzer"] = results
    return state

Conforming to the State Contract

To ensure compatibility with LangGraph and the downstream meta_analyzer, your node must follow three rules:

• Return the same mutable state dictionary you received. LangGraph passes state by reference; nodes must mutate and return this object so subsequent nodes see the changes.
• Use a unique key in analysis_results. The meta_analyzer aggregates findings by iterating over this dictionary, so namespace collisions will overwrite data.
• Remain pure. Avoid side effects such as file I/O or external network calls outside of the state mutation. This guarantees deterministic execution and simplifies testing.

Registering the Analyzer in the Registry

Edit src/skillspector/nodes/analyzers/__init__.py to import your function and add it to the ANALYZER_NODES mapping:

# src/skillspector/nodes/analyzers/__init__.py

from .static_yara import static_yara
from .my_custom_analyzer import my_custom_analyzer

ANALYZER_NODES = {
    "static_yara": static_yara,
    "my_custom_analyzer": my_custom_analyzer,  # Registration

}

# Automatically generates the list used by graph.py

ANALYZER_NODE_IDS = list(ANALYZER_NODES.keys())

No changes to src/skillspector/graph.py are required. When the graph module imports ANALYZER_NODE_IDS, it will include "my_custom_analyzer" and automatically invoke workflow.add_node() and workflow.add_edge() to connect your analyzer between build_context and meta_analyzer.

Optional: Declaring Provider Requirements

If your analyzer requires an external API key, expose a requires_api_key attribute on the function:

my_custom_analyzer.requires_api_key = True

While the current graph implementation treats this as a stub for future use, declaring this attribute follows the public contract and ensures forward compatibility when the framework adds key-validation logic.

Testing and Execution

Validate your implementation with a unit test that mocks the state structure:

# tests/nodes/test_my_custom_analyzer.py

from skillspector.nodes.analyzers.my_custom_analyzer import my_custom_analyzer

def test_detects_eval_usage():
    state = {
        "context": {"files": {"vulnerable.py": "print(eval('1 + 1'))"}},
        "analysis_results": {}
    }
    
    updated = my_custom_analyzer(state)
    
    assert "my_custom_analyzer" in updated["analysis_results"]
    assert len(updated["analysis_results"]["my_custom_analyzer"]) == 1
    assert updated["analysis_results"]["my_custom_analyzer"][0]["severity"] == "high"

Run the full pipeline via the CLI to see your analyzer integrated into the final SARIF report:

python -m skillspector.cli scan path/to/skill_dir

The graph execution will automatically invoke your analyzer after build_context and feed its output into meta_analyzer, including your findings in the generated security report.

Summary

• Implement a node function in src/skillspector/nodes/analyzers/ that accepts Dict[str, Any], mutates state["analysis_results"], and returns the state.
• Register the callable in src/skillspector/nodes/analyzers/__init__.py by adding it to the ANALYZER_NODES dictionary.
• Leverage automatic wiring via ANALYZER_NODE_IDS—graph.py handles all LangGraph node and edge creation without further edits.
• Use unique keys in analysis_results to prevent collisions with built-in analyzers like static_yara.
• Test locally with pytest, then execute through the standard CLI to generate reports.

Frequently Asked Questions

Do I need to modify graph.py to add my custom analyzer?

No. The registry pattern in src/skillspector/nodes/analyzers/__init__.py decouples analyzer implementation from graph construction. As long as your node is listed in ANALYZER_NODES, the loop in src/skillspector/graph.py automatically adds it to the StateGraph with the correct edges.

What data is available in the state dictionary passed to analyzer nodes?

According to src/skillspector/state.py, the SkillspectorState dictionary contains inputs (the scan targets), context (including file contents prepared by build_context), and analysis_results (a dictionary where you store your findings). Analyzers typically read from state["context"]["files"] and write to state["analysis_results"][your_analyzer_id].

Can I register multiple custom analyzers in the same workflow?

Yes. Simply import each node function into src/skillspector/nodes/analyzers/__init__.py and add multiple entries to the ANALYZER_NODES dictionary. Each unique key becomes a distinct node ID in the LangGraph workflow, allowing you to run static analysis, YARA rules, and LLM-based checks in parallel.

How does the meta_analyzer aggregate results from my custom node?

The meta_analyzer node (defined in src/skillspector/nodes/meta_analyzer.py) iterates over the analysis_results dictionary after all analyzer nodes complete. It processes every key—including your custom analyzer’s unique ID—to compile the final security report, requiring no special configuration to recognize new analyzers.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}