# How to Invoke SkillSpector's LangGraph Workflow Programmatically Using the Python API

> Learn to programmatically invoke SkillSpector's LangGraph workflow using the Python API. Import the graph object, create a state dictionary, and execute synchronously or asynchronously.

- Repository: [NVIDIA Corporation/SkillSpector](https://github.com/NVIDIA/SkillSpector)
- Tags: how-to-guide
- Published: 2026-07-08

---

**To invoke SkillSpector's LangGraph workflow programmatically, import the compiled `graph` object from `skillspector.graph`, construct a state dictionary matching the `SkillspectorState` schema, and call `graph.invoke(state)` for synchronous execution or `await graph.ainvoke(state)` for asynchronous execution.**

The NVIDIA/SkillSpector repository packages its core security analysis engine as a **LangGraph** state machine. While the tool ships with a command-line interface, you can embed the scanner directly into Python applications, CI pipelines, or notebooks by interacting with the underlying graph API. This approach gives you fine-grained control over execution, output formats, and cleanup.

## Importing and Configuring the Graph

### Locating the Compiled Graph

The entry point for programmatic access is the module-level variable `graph` defined in [`src/skillspector/graph.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/graph.py). This object is created by the `create_graph()` function, which constructs a `StateGraph[SkillspectorState]`, wires the analyzer nodes, and compiles the workflow. Import it directly without needing to recompile:

```python
from skillspector.graph import graph

```

The graph is already compiled and ready for invocation, so you avoid the overhead of rebuilding the state machine on every run.

### Constructing the Initial State

Before invocation, you must populate a dictionary that conforms to the `SkillspectorState` schema defined in [`src/skillspector/state.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/state.py). The most critical keys include:

- **`skill_path`** – Absolute path to the skill bundle (directory, zip file, or URL).
- **`output_format`** – Choose `"terminal"`, `"json"`, `"markdown"`, or `"sarif"` (defaults to `"terminal"`).
- **`use_llm`** – Boolean flag; set `True` to enable LLM-backed analyzers or `False` for static-only scans.
- **Optional settings** – `yara_rules_dir`, `baseline`, `show_suppressed`, and other configuration parameters.

The state dictionary flows through the graph's nodes—including `resolve_input`, `build_context`, the analyzer fan-out, `meta_analyzer`, and finally the `report` node—accumulating findings and metadata at each step.

## Synchronous and Asynchronous Invocation

### Blocking Execution with graph.invoke

For straightforward scripts or single-threaded applications, use `graph.invoke()` to run the entire workflow blocking:

```python
from pathlib import Path
from skillspector.graph import graph

state = {
    "skill_path": str(Path("/path/to/your/skill_bundle")),
    "output_format": "json",
    "use_llm": False,
}

result = graph.invoke(state)

print("Risk score:", result["risk_score"])
print("Total findings:", len(result["findings"]))
print("Report:", result["report_body"])

```

The method returns a plain Python dictionary containing all final state fields populated by the graph nodes.

### Async Execution with graph.ainvoke

For applications requiring non-blocking execution, such as web servers or async data pipelines, use `graph.ainvoke()`:

```python
import asyncio
from pathlib import Path
from skillspector.graph import graph

async def scan_skill():
    state = {
        "skill_path": "./my_skill.zip",
        "output_format": "terminal",
        "use_llm": True,
    }
    
    result = await graph.ainvoke(state)
    
    print(result["report_body"])
    
    for entry in result["llm_call_log"]:
        print(entry)

asyncio.run(scan_skill())

```

This coroutine-based approach prevents I/O blocking when the graph downloads remote skill bundles or performs LLM inference. Both invocation methods return an identical result structure.

## Understanding the Result Dictionary

### Core Output Fields

After `graph.invoke()` or `await graph.ainvoke()` completes, the returned dictionary contains the complete `SkillspectorState` plus generated reports. Key fields include:

- **`findings`** – List of `Finding` objects (see [`src/skillspector/models.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/models.py)) containing vulnerability details.
- **`filtered_findings`** – Findings remaining after baseline suppression and filtering.
- **`report_body`** – Human-readable report formatted according to your `output_format` selection.
- **`sarif_report`** – Full SARIF JSON payload for integration with security platforms.
- **`risk_score`**, **`risk_severity`**, **`risk_recommendation`** – Aggregated risk assessment computed by the `meta_analyzer` node.
- **`llm_call_log`** – Telemetry data for each LLM invocation, including latency and token usage, present when `use_llm=True`.

The graph's reducer (using `operator.add`) automatically merges list-type fields like `findings` and `llm_call_log` as nodes execute in parallel.

### Handling Temporary Directories

When the input path is a URL or zip file, the `resolve_input` node creates a temporary directory for extraction. The result dictionary includes a **`temp_dir_for_cleanup`** key pointing to this directory. You must delete it manually or use the helper function from [`src/skillspector/cleanup.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/cleanup.py):

```python
from skillspector.cleanup import cleanup_result
from skillspector.graph import graph

result = graph.invoke(state)

# Automatic cleanup

cleanup_result(result)

```

Failing to clean up leaves temporary clones or extracted archives in your filesystem.

## Complete Code Examples

### Static Analysis Only (Synchronous)

```python
from pathlib import Path
from skillspector.graph import graph

state = {
    "skill_path": str(Path("/path/to/your/skill_bundle")),
    "output_format": "json",
    "use_llm": False,
}

result = graph.invoke(state)

print("Risk score:", result["risk_score"])
print("Findings:", len(result["findings"]))
print("Report (JSON):", result["report_body"])

```

### LLM-Enabled Async Scan

```python
import asyncio
from pathlib import Path
from skillspector.graph import graph

async def async_scan():
    state = {
        "skill_path": str(Path("./my_skill")),
        "output_format": "terminal",
        "use_llm": True,
    }

    result = await graph.ainvoke(state)

    print(result["report_body"])

    for entry in result["llm_call_log"]:
        print(entry)

asyncio.run(async_scan())

```

### Remote URL with Cleanup

```python
import shutil
from pathlib import Path
from skillspector.graph import graph

state = {
    "skill_path": "https://github.com/example/skill.git",
    "output_format": "markdown",
    "use_llm": True,
}

result = graph.invoke(state)

temp_dir = result.get("temp_dir_for_cleanup")
if temp_dir:
    shutil.rmtree(temp_dir)

Path("skill_report.md").write_text(result["report_body"], encoding="utf-8")
print("Report saved to skill_report.md")

```

## Summary

- **Import the compiled graph** from [`src/skillspector/graph.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/graph.py) to access the `graph` variable.
- **Construct a valid state dictionary** using keys defined in [`src/skillspector/state.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/state.py), specifying `skill_path`, `output_format`, and `use_llm`.
- **Invoke synchronously** with `graph.invoke(state)` or **asynchronously** with `await graph.ainvoke(state)`.
- **Process the result** to extract `findings`, `report_body`, `sarif_report`, and risk metrics.
- **Clean up temporary directories** referenced by `temp_dir_for_cleanup` in the result, or use `skillspector.cleanup.cleanup_result`.

## Frequently Asked Questions

### What is the difference between graph.invoke and graph.ainvoke?

`graph.invoke()` blocks the current thread until the LangGraph workflow completes, making it suitable for scripts and batch processing. `graph.ainvoke()` returns a coroutine that you must await, allowing concurrent execution in async applications like FastAPI servers or Jupyter notebooks. Both methods accept the same state dictionary and return identical result structures.

### Can I modify the graph before invocation?

Yes. Because `graph` is a standard LangGraph `CompiledStateGraph` object, you can access its underlying structure or create a custom graph using `create_graph()` from [`src/skillspector/graph.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/graph.py). This allows you to replace specific nodes, add tracing hooks, or insert middleware between the analyzer fan-out and the meta-analyzer step.

### What output formats does the programmatic API support?

The `output_format` state key accepts `"terminal"`, `"json"`, `"markdown"`, or `"sarif"`. The `report_body` field in the result dictionary contains the human-readable formatted report, while `sarif_report` contains the structured SARIF JSON payload regardless of the chosen format.

### How do I handle authentication for private skill repositories?

When specifying a URL in `skill_path`, include authentication tokens in the URL string (e.g., `https://token@github.com/user/repo.git`) or configure Git credentials in the environment before invocation. The `resolve_input` node in `src/skillspector/nodes/` handles the cloning process using standard Git commands, so it respects system-level credential helpers.