LangGraph in SkillSpector: Orchestrating AI Security Scanning Workflows

LangGraph serves as the core execution engine in SkillSpector, orchestrating a state-driven workflow that chains together input resolution, static analysis, meta-analysis, and reporting into a single, deterministic pipeline.

NVIDIA's SkillSpector is an open-source security scanner for AI agent skills that leverages LangGraph to transform independent analysis modules into a cohesive, end-to-end pipeline. By using LangGraph's StateGraph abstraction, the architecture separates concerns between individual analyzers and workflow orchestration, making the system both testable and extensible. This article examines how LangGraph functions as the backbone of SkillSpector's architecture based on the actual implementation in the NVIDIA/SkillSpector repository.

State-Centric Workflow Design

SkillSpector's workflow revolves around a strongly-typed shared state that persists across all execution nodes.

The SkillspectorState Schema

In src/skillspector/state.py, the system defines a SkillspectorState typed dictionary that carries data from the initial input resolution through to the final report generation. This state object includes fields for the skill path, discovered components, accumulated findings, SARIF reports, risk scores, and formatted output.

Each node in the graph receives the current state as input and returns updates to that state. This state-centric approach ensures that nodes remain pure functions without side effects—each analyzer reads the context built by previous steps and appends its own findings without mutating external globals.

Modular Node Architecture

The pipeline is decomposed into discrete, single-responsibility nodes that conform to LangGraph's node signature requirements.

Core Node Functions

In src/skillspector/graph.py, the workflow registers nodes such as resolve_input, build_context, meta_analyzer, and report. Each node is implemented as an independent function in the src/skillspector/nodes/ directory:

• resolve_input – Normalizes input paths and validates skill artifacts
• build_context – Identifies components within the skill package
• meta_analyzer – Aggregates and filters findings from all analyzers
• report – Formats final output as JSON, Markdown, or terminal text

The graph definition explicitly wires these nodes together using workflow.add_node() and workflow.add_edge():

workflow.add_node("resolve_input", resolve_input)
workflow.add_node("build_context", build_context)
workflow.add_edge(START, "resolve_input")
workflow.add_edge("resolve_input", "build_context")

Parallel Analyzer Execution

After build_context completes, the graph fans out to multiple static-pattern analyzers. The ANALYZER_NODE_IDS list in src/skillspector/graph.py (lines 42-44) defines which analyzers participate in the workflow. Each analyzer node runs independently, reading from the shared state and returning findings to the collective pool before the graph converges at the meta_analyzer node.

Compiling and Invoking the Graph

The workflow is compiled into a runnable object that consumers can invoke with a simple dictionary interface.

Graph Compilation

The create_graph() function in src/skillspector/graph.py (line 55) compiles the StateGraph into an executable object:

graph = create_graph()

This compilation step validates the node connectivity and prepares the state machine for execution.

Workflow Invocation

Consumers trigger the full security scan by calling invoke() with an input dictionary, as demonstrated in tests/integration/test_graph.py (lines 21-33):

from skillspector.graph import graph

result = graph.invoke(
    {
        "skill_path": "/path/to/your/skill",   # directory or zip file

        "output_format": "json",                # or "markdown", "terminal"

    }
)

# Access accumulated results

findings = result["findings"]
sarif = result["sarif_report"]
risk = result["risk_score"]
report = result["report_body"]

LangGraph handles the routing, state propagation, and termination automatically, returning a unified dictionary containing all accumulated analysis results.

Extending the Pipeline with Custom Analyzers

One advantage of LangGraph's architecture is the simplicity of adding new analysis capabilities.

Registering New Nodes

To add a custom analyzer, implement a node function that accepts SkillspectorState and returns an AnalyzerNodeResponse, then register it in src/skillspector/nodes/analyzers/__init__.py:

# 1. Implement the node function

def my_custom_analyzer(state: SkillspectorState) -> AnalyzerNodeResponse:
    # Perform custom analysis

    return {"findings": [...]}

# 2. Register in the analyzers module

ANALYZER_NODES["my_custom_analyzer"] = my_custom_analyzer
ANALYZER_NODE_IDS.append("my_custom_analyzer")

Because src/skillspector/graph.py iterates over ANALYZER_NODE_IDS when wiring edges, the new node automatically connects between build_context and meta_analyzer without modifying the core graph logic.

Summary

LangGraph's role in SkillSpector's architecture can be summarized as follows:

• State Management: Provides the StateGraph infrastructure that carries SkillspectorState across all analysis nodes in src/skillspector/state.py
• Execution Orchestration: Compiles a deterministic workflow graph in src/skillspector/graph.py that routes data from input resolution through parallel analyzers to final reporting
• Modularity: Enforces a node-based architecture where each analyzer in src/skillspector/nodes/ operates as a pure function with explicit inputs and outputs
• Extensibility: Allows new analyzers to be added by appending to ANALYZER_NODE_IDS, with LangGraph automatically handling the edge connections
• Invocation Interface: Exposes a simple graph.invoke() method that triggers the entire pipeline and returns comprehensive results

Frequently Asked Questions

What is LangGraph's specific role in SkillSpector?

LangGraph acts as the execution engine and orchestrator that transforms SkillSpector's collection of independent security analyzers into a coherent, state-driven pipeline. It manages the flow of data between input resolution, context building, parallel static analysis, meta-analysis, and report generation according to the graph definition in src/skillspector/graph.py.

How does SkillSpector manage state between analysis steps?

SkillSpector uses a strongly-typed SkillspectorState dictionary defined in src/skillspector/state.py that persists across the entire LangGraph workflow. Each node receives the current state as input and returns updates to specific fields (such as findings or risk_score), allowing subsequent nodes to access previous results without side effects or global state pollution.

Can I add custom analyzers to the SkillSpector pipeline?

Yes. You can add custom analyzers by implementing a node function that conforms to the SkillspectorState interface and registering it in src/skillspector/nodes/analyzers/__init__.py. By appending the analyzer name to ANALYZER_NODE_IDS, LangGraph automatically wires the new node between build_context and meta_analyzer in the compiled graph.

How do I invoke the SkillSpector scanning workflow?

Import the compiled graph from skillspector.graph and call graph.invoke() with a dictionary containing at minimum the skill_path (pointing to a directory or zip file) and optionally an output_format (such as "json", "markdown", or "terminal"). LangGraph executes the entire workflow and returns a dictionary containing findings, SARIF reports, risk scores, and formatted output.

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