Skip to content

Python API

cloudg is a library first. The CLI is a thin layer over CloudGEngine, so anything a command does you can do from Python, one phase at a time if you like.

Package cloudg · Python 3.11 or newer · Install pip install cloudg

Start here

Most programs need three names, all importable from the package root: CloudGConfig to say what to collect, CloudGEngine to run it, and the result object that comes back.

No credentials and no scanner binaries: point the engine at scanner output produced elsewhere.

ingest_only.py
from cloudg import CloudGConfig, CloudGEngine

engine = CloudGEngine(CloudGConfig())
result = engine.run_from_reports_sync(
    {"trivy": ["./trivy.json"], "checkov": ["./results_json.json"]},
    output_dir="./reports",
)

print(result.total_findings, result.severity_breakdown)
for f in result.findings:
    print(f.severity.value, f.source_tool, f.title)

Read-only credentials and the aws extra. No scanner runs.

map_account.py
from cloudg import CloudGConfig, CloudGEngine

config = CloudGConfig(providers=["aws"])
config.aws.profile = "audit"
config.aws.regions = ["ALL"]

inventory = CloudGEngine(config).map_inventory_sync(output_dir="./reports")
print(inventory.summary["total_assets"], "assets")
print(inventory.summary["internet_exposed"], "internet-exposed")

What cloudg run does: collect, scan, analyse and report.

pipeline.py
import asyncio

from cloudg import CloudGConfig, CloudGEngine


async def main() -> None:
    config = CloudGConfig(providers=["aws"])
    config.scanners.enabled = ["prowler"]
    result = await CloudGEngine(config).run_pipeline(output_dir="./reports")
    print(result.to_summary())


asyncio.run(main())

How the pieces fit

flowchart LR
  CFG["CloudGConfig"] --> ENG["CloudGEngine"]
  ENG -->|collect| ASSETS["CloudAsset, NetworkEdge"]
  ENG -->|scan or ingest| FIND["Finding"]
  ASSETS --> GB["GraphBuilder"]
  GB --> RA["ReachabilityAnalyzer"]
  ASSETS --> ONT["CloudOntology, RAGExporter"]
  FIND --> NORM["FindingsNormaliser"]
  RA --> NORM
  NORM --> RES["PipelineResult"]
  ENG -->|map_inventory| IM["InventoryMapper"]
  IM --> IR["InventoryResult"]
  IR --> DG["DependencyGraph"]

Each box is a page below. The engine wires them together, but every one of them also works on its own: build a graph from assets you loaded yourself, normalise findings from a CMDB export, or link assets that came from your own collector.

Reference

Engine

  • CloudGEngine

    The engine behind cloudg run, cloudg ingest and cloudg map: the whole pipeline in one call, or each phase yourself.

  • CloudGConfig

    The pydantic model behind config.yaml. Build it in code, load it from a file, or both.

  • Result objects

    PipelineResult, CollectionResult and AnalysisResult: what the engine hands back.

  • Event hooks

    Six optional callbacks that report phases, findings, results and failures during a run.

  • Exceptions

    The CloudGError hierarchy, strict mode for the scanner wrappers and raise_for_errors().

Inventory

  • InventoryMapper

    Maps what is deployed and how it connects, then overlays scanner findings on the map.

  • InventoryResult

    The map as one object: assets, edges, summary, dependency analysis, save and load.

  • RelationshipLinker

    Turns a flat list of assets from any source into a connected map.

  • DependencyGraph

    What an asset needs, and what breaks if it goes away.

Graph

  • GraphBuilder

    Assets and edges as a NetworkX DiGraph: attack paths, centrality, exports.

  • ReachabilityAnalyzer

    What the internet can reach, along the edges traffic can actually take.

  • CloudOntology

    An rdflib graph with 64 typed relations, exported as RDF and queried with SPARQL.

  • RAGExporter

    Text chunks with metadata, ready to embed for retrieval-augmented generation.

Findings

  • FindingsNormaliser

    Findings from any number of scanners, merged, rescored, sorted and mapped to controls.

  • parse_report

    Native Prowler, ScoutSuite, Checkov and Trivy output as Finding objects.

  • Data models

    The pydantic models every part of cloudg exchanges.

Extend

  • Entry points

    Ship a collector or a scanner as its own package; PluginRegistry finds it.

  • cloudg.resilience

    Your own cloud API calls through cloudg's rate limits, circuit breakers and retry budget.

  • CloudGMCPLayer

    The MCP catalog as a Python object: call it in-process, mount it, or export it.

Conventions

  • Signatures, parameter tables and the "source" links on these pages are generated from the code at build time, so they always match the release the site was built from. The prose and the examples above each reference are written by hand.
  • Methods that end in _sync wrap the async method of the same name in asyncio.run(). Use them from scripts; inside a running event loop, await the async method.
  • Everything cloudg raises on purpose derives from CloudGError. See Exceptions.