Python API¶

Use the public Python API when another tool needs typed agent-bom results without parsing terminal output, or when services and notebooks need a small control-plane client for stable REST endpoints.

Install¶

pip install agent-bom

Use the API extra when the same environment also starts a local control plane:

pip install 'agent-bom[api]'

Local Scan Helpers¶

from agent_bom import check, diff, scan
from agent_bom.sdk import inventory

report = scan(project=".", offline=True)
for finding in report.to_findings():
    print(finding.id, finding.severity)

package = check("requests@2.31.0", ecosystem="pypi", offline=True)
print(package.status, package.vulnerabilities)

fleet = inventory("fleet-inventory.json")
print(fleet.agent_count, fleet.package_count)

delta = diff("baseline.json", "current.json")
print(delta.summary)

The API delegates to the same scanner, inventory, and history-diff primitives used elsewhere in the product. It is not a parallel scan engine.

Control-Plane Client¶

from agent_bom import AgentBomClient

with AgentBomClient(
    base_url="https://agent-bom.internal",
    api_key="agent-bom-api-key",
    tenant_id="tenant-a",
) as client:
    print(client.health()["status"])
    manifest = client.agent_manifest()
    runtime = client.runtime_production_index()
    intel = client.intel_sources()
    decision = client.should_i_deploy("flask@2.0.0", block_risk=80)

print(manifest["schema_version"], runtime["schema_version"], len(intel.get("sources", [])), decision["decision"])

Run the packaged smoke example against a live API:

AGENT_BOM_BASE_URL=http://127.0.0.1:8422 \
AGENT_BOM_API_KEY=dev-key \
python examples/python_sdk/control_plane_smoke.py

The client accepts either api_key or bearer_token. tenant_id is sent as X-Agent-Bom-Tenant-ID and used as the default tenant scope for tenant-aware methods.

Payload-first methods accept the obvious positional form:

client.ingest_findings(
    [{"id": "finding-1", "severity": "high", "title": "External scanner finding"}],
    source="external-scanner",
)
client.register_dataset_version("training-set", version_id="2026-05-24")
client.should_i_deploy("flask@2.0.0", block_risk=80)

agent_bom.sdk ¶

Public Python API for embedding agent-bom in other tools.

This module is intentionally thin: it exposes stable Python functions while delegating to the same scanner, inventory, and history primitives used by the CLI and MCP surfaces. It is not a second scan implementation.

Asset `dataclass` ¶

What is affected by this finding.

Source code in src/agent_bom/finding.py

@dataclass
class Asset:
    """What is affected by this finding."""

    name: str  # human-readable name (server name, package name, cloud resource ID)
    asset_type: str  # "mcp_server" | "package" | "container" | "cloud_resource" | "agent"
    identifier: Optional[str] = None  # purl, ARN, image digest, etc.
    location: Optional[str] = None  # file path, URL, cloud region

    # Explicit scope — where this asset lives. Optional/nullable so non-cloud
    # assets (packages, files) serialize unchanged. ``account_ref`` is a single
    # normalized string (e.g. ``aws:123456789012``) built by finding_scope.
    provider: Optional[str] = None  # aws | azure | gcp | snowflake | ...
    account_ref: Optional[str] = None  # normalized ``<provider>:<account>``
    region: Optional[str] = None
    environment: Optional[str] = None  # prod | staging | dev | ...

    @property
    def stable_id(self) -> str:
        """Deterministic UUID derived from asset content.

        Same asset type + identifier always produces the same ID across scans.
        This enables tracking: first seen, last seen, resolved, re-emerged.
        """
        identifier = self.identifier or f"{self.name}:{self.location or ''}"
        return _stable_id(self.asset_type, identifier)

    @property
    def canonical_id(self) -> str:
        """Canonical alias for stable_id used by reports and graph joins."""
        return self.stable_id

    @property
    def source_ids(self) -> dict[str, str]:
        """Original source identifiers retained as provenance."""
        return source_ids(identifier=self.identifier, location=self.location)

stable_id `property` ¶

stable_id: str

Deterministic UUID derived from asset content.

Same asset type + identifier always produces the same ID across scans. This enables tracking: first seen, last seen, resolved, re-emerged.

canonical_id `property` ¶

canonical_id: str

Canonical alias for stable_id used by reports and graph joins.

source_ids `property` ¶

source_ids: dict[str, str]

Original source identifiers retained as provenance.

Finding `dataclass` ¶

Unified finding — one model for all issue types across all sources.

Phase 1 covers CVE findings (migrated from BlastRadius). Phase 2 will add cloud CIS, proxy, SAST, skill findings.

Source code in src/agent_bom/finding.py

@dataclass
class Finding:
    """Unified finding — one model for all issue types across all sources.

    Phase 1 covers CVE findings (migrated from BlastRadius).
    Phase 2 will add cloud CIS, proxy, SAST, skill findings.
    """

    # Core identity
    finding_type: FindingType
    source: FindingSource
    asset: Asset
    severity: str  # mirrors Severity enum value; str for forward-compat

    # Explicit scope (issue #3946) — carried on the finding for query/filter
    # convenience and mirrored onto the asset at ingest. All optional/nullable
    # so existing findings serialize unchanged.
    provider: Optional[str] = None  # aws | azure | gcp | snowflake | ...
    account_ref: Optional[str] = None  # normalized ``<provider>:<account>``
    region: Optional[str] = None
    environment: Optional[str] = None  # prod | staging | dev | ...

    # Vendor severity (from source scanner) vs normalised CVSS severity
    vendor_severity: Optional[str] = None  # severity as reported by vendor/scanner
    cvss_severity: Optional[str] = None  # normalised from CVSS base score

    # Finding content
    title: str = ""
    description: str = ""
    cve_id: Optional[str] = None  # e.g. "CVE-2024-1234"
    cwe_ids: list[str] = field(default_factory=list)  # e.g. ["CWE-79"]
    cvss_score: Optional[float] = None
    cvss_vector: Optional[str] = None
    attack_vector: Optional[str] = None
    attack_complexity: Optional[str] = None
    privileges_required: Optional[str] = None
    user_interaction: Optional[str] = None
    network_exploitable: bool = False
    epss_score: Optional[float] = None
    is_kev: bool = False  # CISA Known Exploited Vulnerability
    is_malicious: bool = False  # Known-malicious package (MAL- IDs, typosquat, etc.)
    malicious_reason: Optional[str] = None

    # Remediation
    fixed_version: Optional[str] = None
    remediation_guidance: Optional[str] = None
    # Structured advisory remediation (fix + least-privilege-to-apply + optional
    # artifact). Additive + optional: findings without it serialize unchanged.
    # Read-only forever — agent-bom recommends, the user applies.
    remediation: Optional["Remediation"] = None

    # Compliance mappings (same tags as BlastRadius for parity)
    compliance_tags: list[str] = field(default_factory=list)  # all framework tags combined
    # Framework slugs that govern this finding (set by compliance_hub.apply_hub_classification).
    # Distinct from the per-framework `*_tags` fields below, which hold control codes.
    applicable_frameworks: list[str] = field(default_factory=list)
    controls: list[ControlTag] = field(default_factory=list)
    owasp_tags: list[str] = field(default_factory=list)
    atlas_tags: list[str] = field(default_factory=list)
    attack_tags: list[str] = field(default_factory=list)
    nist_ai_rmf_tags: list[str] = field(default_factory=list)
    owasp_mcp_tags: list[str] = field(default_factory=list)
    owasp_agentic_tags: list[str] = field(default_factory=list)
    eu_ai_act_tags: list[str] = field(default_factory=list)
    nist_csf_tags: list[str] = field(default_factory=list)
    iso_27001_tags: list[str] = field(default_factory=list)
    soc2_tags: list[str] = field(default_factory=list)
    cis_tags: list[str] = field(default_factory=list)
    cmmc_tags: list[str] = field(default_factory=list)
    nist_800_53_tags: list[str] = field(default_factory=list)
    fedramp_tags: list[str] = field(default_factory=list)
    pci_dss_tags: list[str] = field(default_factory=list)

    # Graph / correlation
    related_findings: list[str] = field(default_factory=list)  # IDs of related findings
    evidence: dict = field(default_factory=dict)  # raw evidence payload
    # First-class graph FKs (optional, additive). ``node_id`` is the estate /
    # asset UnifiedNode this finding attaches to; ``finding_node_id`` is the
    # vulnerability/misconfiguration node (e.g. ``vuln:CVE-…``) when materialised.
    # ``entity_type`` mirrors EntityType.value for the asset node when known.
    node_id: Optional[str] = None
    finding_node_id: Optional[str] = None
    entity_type: Optional[str] = None

    # Risk
    risk_score: float = 0.0  # 0-10 unified risk score
    reachability: Optional[str] = None
    is_actionable: Optional[bool] = None
    impact_category: Optional[str] = None

    # Suppression state (mirrors BlastRadius; preserved through the unified stream
    # so a suppressed finding never appears unsuppressed downstream)
    suppressed: bool = False
    suppression_id: Optional[str] = None
    suppression_state: Optional[str] = None
    suppression_reason: Optional[str] = None
    unsuppressed_risk_score: Optional[float] = None

    # AI-native risk context (mirrors BlastRadius)
    ai_risk_context: Optional[str] = None
    ai_summary: Optional[str] = None
    attack_vector_summary: Optional[str] = None

    # Reach / blast-radius lists — kept structured rather than collapsed to counts
    affected_servers: list[str] = field(default_factory=list)  # MCP server names on the impacted path
    affected_agents: list[str] = field(default_factory=list)  # agent names reachable along the path
    exposed_credentials: list[str] = field(default_factory=list)  # credential env-var names at risk
    exposed_tools: list[str] = field(default_factory=list)  # tool names accessible through the path

    # CWPP runtime/EDR workload evidence (optional, additive). Never implies the
    # workload is clean — summaries carry clean_workload_assertion=False.
    workload_runtime_evidence: Optional[dict] = None

    # Unique ID — deterministic UUID v5 based on content (computed in __post_init__)
    # Pass an explicit id= to override (e.g. when ingesting from external scanner)
    id: str = field(default="")

    def __post_init__(self) -> None:
        """Compute stable ID from finding content if not explicitly set."""
        from agent_bom.graph.severity import normalize_severity

        self.severity = normalize_severity(self.severity)
        # Keep finding scope and asset scope consistent: mirror finding-level
        # scope down to the asset when the asset does not already carry it (and
        # lift asset scope up when only the asset was populated). Non-cloud
        # findings leave every field None, so this is a no-op for them.
        for _scope_field in ("provider", "account_ref", "region", "environment"):
            finding_val = getattr(self, _scope_field)
            asset_val = getattr(self.asset, _scope_field, None)
            if finding_val is not None and asset_val is None:
                setattr(self.asset, _scope_field, finding_val)
            elif finding_val is None and asset_val is not None:
                setattr(self, _scope_field, asset_val)
        if self.vendor_severity is not None:
            self.vendor_severity = normalize_severity(self.vendor_severity)
        if self.cvss_severity is not None:
            self.cvss_severity = normalize_severity(self.cvss_severity)
        self.controls = _dedupe_control_tags(
            [
                *(tag if isinstance(tag, ControlTag) else ControlTag.from_dict(tag) for tag in self.controls),
                *self._legacy_control_tags(),
            ]
        )
        # Derive entity_type from asset_type aliases without mutating asset_type
        # (asset_type feeds stable_id / finding id — must stay byte-stable).
        if not self.entity_type:
            try:
                from agent_bom.graph.asset_entity import entity_type_for_asset_type

                mapped = entity_type_for_asset_type(self.asset.asset_type)
                if mapped is not None:
                    self.entity_type = mapped.value
            except Exception:  # noqa: BLE001 — finding construction must stay resilient
                pass
        if not self.id:
            # Deterministic ID: same CVE on same asset always same ID
            cve_part = self.vulnerability_id or self.title
            pkg_name = ""
            pkg_version = ""
            if self.asset.asset_type == "package" and self.asset.identifier:
                # purl like "pkg:pypi/torch@2.3.0" — extract name/version
                purl = self.asset.identifier
                pkg_part = purl.split("/")[-1] if "/" in purl else purl
                if "@" in pkg_part:
                    pkg_name, pkg_version = pkg_part.rsplit("@", 1)
            elif isinstance(self.evidence, dict):
                # Asset is a server/container/etc — the affected package lives in
                # evidence. Fold it into the discriminator so two CVEs on distinct
                # packages under one asset don't collide on the same id.
                pkg_name = str(self.evidence.get("package_name") or "")
                pkg_version = str(self.evidence.get("package_version") or "")
            self.id = canonical_finding_id(
                self.asset.stable_id,
                cve_part,
                pkg_name,
                pkg_version,
            )

    @property
    def canonical_id(self) -> str:
        """Canonical alias for id used by report and graph consumers."""
        return self.id

    @property
    def vulnerability_id(self) -> Optional[str]:
        """Canonical advisory identity, regardless of CVE/GHSA/OSV namespace.

        ``cve_id`` remains the wire-compatible legacy field. New producers may
        populate ``evidence.vulnerability_id`` and consumers should prefer this
        namespace-neutral alias when joining findings to advisories.
        """
        if self.cve_id:
            return self.cve_id
        raw = self.evidence.get("vulnerability_id") if isinstance(self.evidence, dict) else None
        return str(raw).strip() or None if raw is not None else None

    @property
    def advisory_ids(self) -> list[str]:
        """Return deterministic, de-duplicated CVE/GHSA/OSV advisory aliases."""
        raw: list[object] = [self.vulnerability_id]
        if isinstance(self.evidence, dict):
            raw.extend(self.evidence.get("cve_ids") or [])
            raw.extend(self.evidence.get("advisory_aliases") or [])
            raw.extend(self.evidence.get("advisory_ids") or [])
        seen: set[str] = set()
        result: list[str] = []
        for value in raw:
            item = str(value or "").strip()
            if item and item not in seen:
                seen.add(item)
                result.append(item)
        return result

    @property
    def finding_category(self) -> str:
        """Stable category for consumers while legacy finding types remain intact."""
        if self.finding_type is FindingType.CVE:
            return "vulnerability"
        if self.finding_type in {FindingType.CIS_FAIL, FindingType.CIS_ERROR}:
            return "compliance"
        return self.finding_type.value.lower()

    def _legacy_control_tags(self) -> list[ControlTag]:
        """Return normalized controls derived from legacy tag arrays."""
        tags: list[ControlTag] = []
        for field_name, framework in LEGACY_CONTROL_FIELDS:
            values = getattr(self, field_name)
            for value in values:
                if value:
                    tags.append(
                        ControlTag(
                            framework=framework,
                            control=str(value),
                            version=_LEGACY_CONTROL_VERSION_BY_FRAMEWORK.get(framework, "legacy"),
                            confidence=0.75,
                            source=f"legacy:{field_name}",
                            via=field_name,
                        )
                    )
        return tags

    def normalized_controls(self) -> list[ControlTag]:
        """Return deduplicated structured controls for this finding."""
        return _dedupe_control_tags([*self.controls, *self._legacy_control_tags()])

    def all_compliance_tags(self) -> list[str]:
        """Return deduplicated union of all compliance tag lists."""
        seen: set[str] = set()
        result: list[str] = []
        for tag in (
            self.compliance_tags
            + self.owasp_tags
            + self.atlas_tags
            + self.attack_tags
            + self.nist_ai_rmf_tags
            + self.owasp_mcp_tags
            + self.owasp_agentic_tags
            + self.eu_ai_act_tags
            + self.nist_csf_tags
            + self.iso_27001_tags
            + self.soc2_tags
            + self.cis_tags
            + self.cmmc_tags
            + self.nist_800_53_tags
            + self.fedramp_tags
            + self.pci_dss_tags
            + [tag.control for tag in self.normalized_controls()]
        ):
            if tag not in seen:
                seen.add(tag)
                result.append(tag)
        return result

    def effective_severity(self) -> str:
        """Return the best severity value: vendor > cvss > base severity."""
        return self.vendor_severity or self.cvss_severity or self.severity

    @property
    def security_domain(self) -> str:
        """Derived posture lane: one of cspm/vuln/aspm/dspm/aispm.

        A pure function of source + finding type (+ evidence for the cloud
        data-vs-config split), so the overview and findings surfaces route each
        finding to exactly one coverage lane without double counting.
        """
        from agent_bom.finding_scope import security_domain_for

        return security_domain_for(self.source, self.finding_type, self.evidence)

    def to_dict(self) -> dict:
        """Return a JSON-serializable finding payload."""
        return {
            "schema_version": FINDING_SCHEMA_VERSION,
            "id": self.id,
            "canonical_id": self.canonical_id,
            "finding_type": self.finding_type.value,
            "finding_category": self.finding_category,
            "source": self.source.value,
            "asset": {
                "name": self.asset.name,
                "asset_type": self.asset.asset_type,
                "identifier": self.asset.identifier,
                "location": self.asset.location,
                "stable_id": self.asset.stable_id,
                "canonical_id": self.asset.canonical_id,
                "source_ids": self.asset.source_ids,
                "provider": self.asset.provider,
                "account_ref": self.asset.account_ref,
                "region": self.asset.region,
                "environment": self.asset.environment,
            },
            # First-class scope + taxonomy (issue #3946)
            "provider": self.provider,
            "account_ref": self.account_ref,
            "region": self.region,
            "environment": self.environment,
            "security_domain": self.security_domain,
            "severity": self.severity,
            "effective_severity": self.effective_severity(),
            "vendor_severity": self.vendor_severity,
            "cvss_severity": self.cvss_severity,
            "title": self.title,
            "description": self.description,
            "cve_id": self.cve_id,
            "vulnerability_id": self.vulnerability_id,
            "advisory_ids": self.advisory_ids,
            "cve_ids": self.evidence.get("cve_ids") or ([self.cve_id] if self.cve_id else []),
            "match_confidence_tier": self.evidence.get("match_confidence_tier"),
            "advisory_aliases": self.evidence.get("advisory_aliases") or [],
            "cwe_ids": self.cwe_ids,
            "cvss_score": self.cvss_score,
            "cvss_vector": self.cvss_vector,
            "attack_vector": self.attack_vector,
            "attack_complexity": self.attack_complexity,
            "privileges_required": self.privileges_required,
            "user_interaction": self.user_interaction,
            "network_exploitable": self.network_exploitable,
            "epss_score": self.epss_score,
            "is_kev": self.is_kev,
            "is_malicious": self.is_malicious,
            "malicious_reason": self.malicious_reason,
            "fixed_version": self.fixed_version,
            "remediation_guidance": self.remediation_guidance,
            # Structured advisory remediation — emitted only when populated so
            # findings without it keep their existing serialization shape.
            **({"remediation": self.remediation.to_dict()} if self.remediation is not None else {}),
            "compliance_tags": self.all_compliance_tags(),
            "applicable_frameworks": list(self.applicable_frameworks),
            "controls": [tag.to_dict() for tag in self.normalized_controls()],
            "owasp_tags": self.owasp_tags,
            "atlas_tags": self.atlas_tags,
            "attack_tags": self.attack_tags,
            "nist_ai_rmf_tags": self.nist_ai_rmf_tags,
            "owasp_mcp_tags": self.owasp_mcp_tags,
            "owasp_agentic_tags": self.owasp_agentic_tags,
            "eu_ai_act_tags": self.eu_ai_act_tags,
            "nist_csf_tags": self.nist_csf_tags,
            "iso_27001_tags": self.iso_27001_tags,
            "soc2_tags": self.soc2_tags,
            "cis_tags": self.cis_tags,
            "cmmc_tags": self.cmmc_tags,
            "nist_800_53_tags": self.nist_800_53_tags,
            "fedramp_tags": self.fedramp_tags,
            "pci_dss_tags": self.pci_dss_tags,
            "related_findings": self.related_findings,
            "evidence": self.evidence,
            "node_id": self.node_id,
            "finding_node_id": self.finding_node_id,
            "entity_type": self.entity_type,
            "risk_score": self.risk_score,
            "reachability": self.reachability,
            "is_actionable": self.is_actionable,
            "impact_category": self.impact_category,
            # Suppression state — a suppressed finding must never surface as
            # unsuppressed downstream (mirrors BlastRadius / SARIF suppressions[]).
            "suppressed": self.suppressed,
            "suppression_id": self.suppression_id,
            "suppression_state": self.suppression_state,
            "suppression_reason": self.suppression_reason,
            "unsuppressed_risk_score": self.unsuppressed_risk_score,
            # AI-native risk context
            "ai_risk_context": self.ai_risk_context,
            "ai_summary": self.ai_summary,
            "attack_vector_summary": self.attack_vector_summary,
            # Structured reach / blast-radius lists (not collapsed to counts)
            "affected_servers": list(self.affected_servers),
            "affected_agents": list(self.affected_agents),
            "exposed_credentials": list(self.exposed_credentials),
            "exposed_tools": list(self.exposed_tools),
            # CWPP runtime/EDR — omit when unset so plain findings stay unchanged
            **(
                {"workload_runtime_evidence": dict(self.workload_runtime_evidence)}
                if isinstance(self.workload_runtime_evidence, dict)
                else {}
            ),
        }

canonical_id `property` ¶

canonical_id: str

Canonical alias for id used by report and graph consumers.

vulnerability_id `property` ¶

vulnerability_id: Optional[str]

Canonical advisory identity, regardless of CVE/GHSA/OSV namespace.

cve_id remains the wire-compatible legacy field. New producers may populate evidence.vulnerability_id and consumers should prefer this namespace-neutral alias when joining findings to advisories.

advisory_ids `property` ¶

advisory_ids: list[str]

Return deterministic, de-duplicated CVE/GHSA/OSV advisory aliases.

finding_category `property` ¶

finding_category: str

Stable category for consumers while legacy finding types remain intact.

security_domain `property` ¶

security_domain: str

Derived posture lane: one of cspm/vuln/aspm/dspm/aispm.

A pure function of source + finding type (+ evidence for the cloud data-vs-config split), so the overview and findings surfaces route each finding to exactly one coverage lane without double counting.

__post_init__ ¶

__post_init__() -> None

Compute stable ID from finding content if not explicitly set.

Source code in src/agent_bom/finding.py

def __post_init__(self) -> None:
    """Compute stable ID from finding content if not explicitly set."""
    from agent_bom.graph.severity import normalize_severity

    self.severity = normalize_severity(self.severity)
    # Keep finding scope and asset scope consistent: mirror finding-level
    # scope down to the asset when the asset does not already carry it (and
    # lift asset scope up when only the asset was populated). Non-cloud
    # findings leave every field None, so this is a no-op for them.
    for _scope_field in ("provider", "account_ref", "region", "environment"):
        finding_val = getattr(self, _scope_field)
        asset_val = getattr(self.asset, _scope_field, None)
        if finding_val is not None and asset_val is None:
            setattr(self.asset, _scope_field, finding_val)
        elif finding_val is None and asset_val is not None:
            setattr(self, _scope_field, asset_val)
    if self.vendor_severity is not None:
        self.vendor_severity = normalize_severity(self.vendor_severity)
    if self.cvss_severity is not None:
        self.cvss_severity = normalize_severity(self.cvss_severity)
    self.controls = _dedupe_control_tags(
        [
            *(tag if isinstance(tag, ControlTag) else ControlTag.from_dict(tag) for tag in self.controls),
            *self._legacy_control_tags(),
        ]
    )
    # Derive entity_type from asset_type aliases without mutating asset_type
    # (asset_type feeds stable_id / finding id — must stay byte-stable).
    if not self.entity_type:
        try:
            from agent_bom.graph.asset_entity import entity_type_for_asset_type

            mapped = entity_type_for_asset_type(self.asset.asset_type)
            if mapped is not None:
                self.entity_type = mapped.value
        except Exception:  # noqa: BLE001 — finding construction must stay resilient
            pass
    if not self.id:
        # Deterministic ID: same CVE on same asset always same ID
        cve_part = self.vulnerability_id or self.title
        pkg_name = ""
        pkg_version = ""
        if self.asset.asset_type == "package" and self.asset.identifier:
            # purl like "pkg:pypi/torch@2.3.0" — extract name/version
            purl = self.asset.identifier
            pkg_part = purl.split("/")[-1] if "/" in purl else purl
            if "@" in pkg_part:
                pkg_name, pkg_version = pkg_part.rsplit("@", 1)
        elif isinstance(self.evidence, dict):
            # Asset is a server/container/etc — the affected package lives in
            # evidence. Fold it into the discriminator so two CVEs on distinct
            # packages under one asset don't collide on the same id.
            pkg_name = str(self.evidence.get("package_name") or "")
            pkg_version = str(self.evidence.get("package_version") or "")
        self.id = canonical_finding_id(
            self.asset.stable_id,
            cve_part,
            pkg_name,
            pkg_version,
        )

normalized_controls ¶

normalized_controls() -> list[ControlTag]

Return deduplicated structured controls for this finding.

Source code in src/agent_bom/finding.py

def normalized_controls(self) -> list[ControlTag]:
    """Return deduplicated structured controls for this finding."""
    return _dedupe_control_tags([*self.controls, *self._legacy_control_tags()])

all_compliance_tags ¶

all_compliance_tags() -> list[str]

Return deduplicated union of all compliance tag lists.

Source code in src/agent_bom/finding.py

def all_compliance_tags(self) -> list[str]:
    """Return deduplicated union of all compliance tag lists."""
    seen: set[str] = set()
    result: list[str] = []
    for tag in (
        self.compliance_tags
        + self.owasp_tags
        + self.atlas_tags
        + self.attack_tags
        + self.nist_ai_rmf_tags
        + self.owasp_mcp_tags
        + self.owasp_agentic_tags
        + self.eu_ai_act_tags
        + self.nist_csf_tags
        + self.iso_27001_tags
        + self.soc2_tags
        + self.cis_tags
        + self.cmmc_tags
        + self.nist_800_53_tags
        + self.fedramp_tags
        + self.pci_dss_tags
        + [tag.control for tag in self.normalized_controls()]
    ):
        if tag not in seen:
            seen.add(tag)
            result.append(tag)
    return result

effective_severity ¶

effective_severity() -> str

Return the best severity value: vendor > cvss > base severity.

Source code in src/agent_bom/finding.py

def effective_severity(self) -> str:
    """Return the best severity value: vendor > cvss > base severity."""
    return self.vendor_severity or self.cvss_severity or self.severity

to_dict ¶

to_dict() -> dict

Return a JSON-serializable finding payload.

Source code in src/agent_bom/finding.py

def to_dict(self) -> dict:
    """Return a JSON-serializable finding payload."""
    return {
        "schema_version": FINDING_SCHEMA_VERSION,
        "id": self.id,
        "canonical_id": self.canonical_id,
        "finding_type": self.finding_type.value,
        "finding_category": self.finding_category,
        "source": self.source.value,
        "asset": {
            "name": self.asset.name,
            "asset_type": self.asset.asset_type,
            "identifier": self.asset.identifier,
            "location": self.asset.location,
            "stable_id": self.asset.stable_id,
            "canonical_id": self.asset.canonical_id,
            "source_ids": self.asset.source_ids,
            "provider": self.asset.provider,
            "account_ref": self.asset.account_ref,
            "region": self.asset.region,
            "environment": self.asset.environment,
        },
        # First-class scope + taxonomy (issue #3946)
        "provider": self.provider,
        "account_ref": self.account_ref,
        "region": self.region,
        "environment": self.environment,
        "security_domain": self.security_domain,
        "severity": self.severity,
        "effective_severity": self.effective_severity(),
        "vendor_severity": self.vendor_severity,
        "cvss_severity": self.cvss_severity,
        "title": self.title,
        "description": self.description,
        "cve_id": self.cve_id,
        "vulnerability_id": self.vulnerability_id,
        "advisory_ids": self.advisory_ids,
        "cve_ids": self.evidence.get("cve_ids") or ([self.cve_id] if self.cve_id else []),
        "match_confidence_tier": self.evidence.get("match_confidence_tier"),
        "advisory_aliases": self.evidence.get("advisory_aliases") or [],
        "cwe_ids": self.cwe_ids,
        "cvss_score": self.cvss_score,
        "cvss_vector": self.cvss_vector,
        "attack_vector": self.attack_vector,
        "attack_complexity": self.attack_complexity,
        "privileges_required": self.privileges_required,
        "user_interaction": self.user_interaction,
        "network_exploitable": self.network_exploitable,
        "epss_score": self.epss_score,
        "is_kev": self.is_kev,
        "is_malicious": self.is_malicious,
        "malicious_reason": self.malicious_reason,
        "fixed_version": self.fixed_version,
        "remediation_guidance": self.remediation_guidance,
        # Structured advisory remediation — emitted only when populated so
        # findings without it keep their existing serialization shape.
        **({"remediation": self.remediation.to_dict()} if self.remediation is not None else {}),
        "compliance_tags": self.all_compliance_tags(),
        "applicable_frameworks": list(self.applicable_frameworks),
        "controls": [tag.to_dict() for tag in self.normalized_controls()],
        "owasp_tags": self.owasp_tags,
        "atlas_tags": self.atlas_tags,
        "attack_tags": self.attack_tags,
        "nist_ai_rmf_tags": self.nist_ai_rmf_tags,
        "owasp_mcp_tags": self.owasp_mcp_tags,
        "owasp_agentic_tags": self.owasp_agentic_tags,
        "eu_ai_act_tags": self.eu_ai_act_tags,
        "nist_csf_tags": self.nist_csf_tags,
        "iso_27001_tags": self.iso_27001_tags,
        "soc2_tags": self.soc2_tags,
        "cis_tags": self.cis_tags,
        "cmmc_tags": self.cmmc_tags,
        "nist_800_53_tags": self.nist_800_53_tags,
        "fedramp_tags": self.fedramp_tags,
        "pci_dss_tags": self.pci_dss_tags,
        "related_findings": self.related_findings,
        "evidence": self.evidence,
        "node_id": self.node_id,
        "finding_node_id": self.finding_node_id,
        "entity_type": self.entity_type,
        "risk_score": self.risk_score,
        "reachability": self.reachability,
        "is_actionable": self.is_actionable,
        "impact_category": self.impact_category,
        # Suppression state — a suppressed finding must never surface as
        # unsuppressed downstream (mirrors BlastRadius / SARIF suppressions[]).
        "suppressed": self.suppressed,
        "suppression_id": self.suppression_id,
        "suppression_state": self.suppression_state,
        "suppression_reason": self.suppression_reason,
        "unsuppressed_risk_score": self.unsuppressed_risk_score,
        # AI-native risk context
        "ai_risk_context": self.ai_risk_context,
        "ai_summary": self.ai_summary,
        "attack_vector_summary": self.attack_vector_summary,
        # Structured reach / blast-radius lists (not collapsed to counts)
        "affected_servers": list(self.affected_servers),
        "affected_agents": list(self.affected_agents),
        "exposed_credentials": list(self.exposed_credentials),
        "exposed_tools": list(self.exposed_tools),
        # CWPP runtime/EDR — omit when unset so plain findings stay unchanged
        **(
            {"workload_runtime_evidence": dict(self.workload_runtime_evidence)}
            if isinstance(self.workload_runtime_evidence, dict)
            else {}
        ),
    }

AIBOMReport `dataclass` ¶

Complete AI-BOM report.

Source code in src/agent_bom/models.py

@dataclass
class AIBOMReport:
    """Complete AI-BOM report."""

    agents: list[Agent] = field(default_factory=list)
    blast_radii: list[BlastRadius] = field(default_factory=list)
    generated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    scan_id: str = ""  # Deterministic UUID v5 from scan inputs (set by CLI after discovery)
    tool_version: str = ""
    executive_summary: Optional[str] = None  # LLM-generated executive summary
    ai_threat_chains: list[str] = field(default_factory=list)  # LLM-generated threat chain analyses
    mcp_config_analysis: Optional[dict[str, Any]] = None  # LLM-powered MCP config security analysis
    ai_enrichment_metadata: Optional[dict[str, Any]] = None  # Non-secret provider/model provenance for AI-generated fields
    ai_finding_assessments: list[AIFindingAssessment] = field(default_factory=list)
    skill_audit_data: Optional[dict[str, Any]] = None  # Serialized SkillAuditResult (set by CLI)
    trust_assessment_data: Optional[dict[str, Any]] = None  # Serialized TrustAssessmentResult (set by CLI)
    prompt_scan_data: Optional[dict[str, Any]] = None  # Serialized PromptScanResult (set by CLI)
    model_files: list[dict[str, Any]] = field(default_factory=list)
    model_manifests: list[dict[str, Any]] = field(default_factory=list)
    model_provenance: list[dict[str, Any]] = field(default_factory=list)  # HuggingFace provenance results
    model_hash_verification_data: Optional[dict[str, Any]] = None  # Serialized model hash verification report
    model_supply_chain_data: Optional[dict[str, Any]] = None  # Consolidated model file/provenance/hash summary
    enforcement_data: Optional[dict[str, Any]] = None  # Serialized EnforcementReport (set by CLI)
    context_graph_data: Optional[dict[str, Any]] = None  # Serialized context graph (set by CLI)
    license_report: Optional[dict[str, Any]] = None  # Serialized license compliance report
    vex_data: Optional[dict[str, Any]] = None  # Serialized VEX document
    toxic_combinations: Optional[list[Any]] = None  # Serialized ToxicCombination list
    prioritized_findings: Optional[list[Any]] = None  # Priority-ordered findings
    sast_data: Optional[dict[str, Any]] = None  # Serialized SAST scan results (Semgrep)
    aws_organization_data: Optional[dict[str, Any]] = None  # AWS Organizations: org/OUs/accounts/SCPs hierarchy
    cis_benchmark_data: Optional[dict[str, Any]] = None  # Serialized CIS AWS Benchmark results
    snowflake_cis_benchmark_data: Optional[dict[str, Any]] = None  # Serialized CIS Snowflake Benchmark results
    snowflake_object_graph_data: Optional[dict[str, Any]] = None  # Snowflake tables/views + OBJECT_DEPENDENCIES lineage
    snowflake_login_anomalies_data: Optional[dict[str, Any]] = None  # Snowflake impossible-travel / login-anomaly detection
    snowflake_exfil_graph_data: Optional[dict[str, Any]] = None  # Snowflake egress: outbound shares, external stages, sensitive objects
    snowflake_auth_posture_data: Optional[dict[str, Any]] = (
        None  # Snowflake per-user auth matrix + network policies (MFA/key-pair/password)
    )
    snowflake_services_data: Optional[dict[str, Any]] = None  # Snowflake compute (warehouses) + database/schema containment hierarchy
    snowflake_pipeline_data: Optional[dict[str, Any]] = None  # Snowflake data-pipeline objects: tasks, streams, pipes
    snowflake_integrations_data: Optional[dict[str, Any]] = (
        None  # Snowflake account integrations: storage/API/external-access/security/catalog
    )
    snowflake_external_data_data: Optional[dict[str, Any]] = None  # Snowflake open-table-format + external data: iceberg + external tables
    snowflake_governance_data: Optional[dict[str, Any]] = (
        None  # Snowflake governance: ACCESS_HISTORY reads + Cortex agent telemetry + derived findings
    )
    snowflake_activity_data: Optional[dict[str, Any]] = (
        None  # Snowflake activity timeline: QUERY_HISTORY (365d) + AI observability events (summarized)
    )
    azure_cis_benchmark_data: Optional[dict[str, Any]] = None  # Serialized CIS Azure Benchmark results
    gcp_cis_benchmark_data: Optional[dict[str, Any]] = None  # Serialized CIS GCP Benchmark results
    databricks_security_data: Optional[dict[str, Any]] = None  # Serialized Databricks Security Best Practices results (canonical)
    aisvs_benchmark_data: Optional[dict[str, Any]] = None  # Serialized AISVS compliance results
    vector_db_scan_data: Optional[list[Any]] = None  # Serialized vector DB security assessments
    gpu_infra_data: Optional[dict[str, Any]] = None  # Serialized GPU/AI compute infra scan results
    # Serialized IaC misconfiguration findings. Set by the CLI scan path and by the
    # API repo_url tree scan (api/repo_tree_scan.py). NOT yet populated for API
    # local-estate/inventory scans or the MCP scan tool — those surfaces do not run
    # the IaC scanner, so /v1/findings shows the IaC category only for repo scans.
    iac_findings_data: Optional[dict[str, Any]] = None
    # Graph toxic-combination findings (serialized Finding dicts; set at the graph-build call site).
    # Rehydrated into the unified Finding stream by to_findings() so they reach --fail-on-severity.
    toxic_combination_findings_data: Optional[list[Any]] = None
    # NHI/CIEM governance findings (over-grant, dormant/orphaned, high-risk NHIs)
    # materialized from the unified graph at the scan call site. Held as Finding
    # objects (not serialized directly); folded into the unified stream by
    # to_findings() so the API path reaches CLI parity for identity governance.
    nhi_governance_findings: list["Finding"] = field(default_factory=list)
    # CIEM over-privilege findings (serialized Finding dicts; set at the graph-build call site).
    # Right-sizing from AWS Access-Advisor usage evidence; rehydrated by to_findings().
    ciem_over_privilege_findings_data: Optional[list[Any]] = None
    # Estate-wide cloud asset inventory; one provider payload or a per-provider list (opt-in AGENT_BOM_CLOUD_INVENTORY)
    cloud_inventory_data: Optional[Union[dict[str, Any], list[Any]]] = None
    identity_discovery_data: Optional[dict[str, Any]] = None  # Discovered non-human identities (opt-in AGENT_BOM_OKTA/ENTRA_DISCOVERY)
    # Cloud audit-trail behavioral payload(s); per-provider list of aggregated
    # (principal, resource, action) edges + findings (opt-in AGENT_BOM_AUDIT_TRAIL, read-only)
    cloud_audit_trail_data: Optional[Union[dict[str, Any], list[Any]]] = None
    runtime_correlation: Optional[dict[str, Any]] = None  # Runtime ↔ scan correlation (proxy audit vs CVE findings)
    delta_data: Optional[dict[str, Any]] = None  # Baseline/delta comparison metadata for CI gate outputs
    scan_performance_data: Optional[dict[str, Any]] = None  # Cache coverage / scan latency metadata
    vuln_data_freshness: Optional[dict[str, Any]] = None  # Vuln-data source/age/staleness snapshot (set by CLI; surfaced to API/MCP)
    training_pipelines: Optional[dict[str, Any]] = None  # Serialized TrainingPipelineScanResult
    dataset_cards: Optional[dict[str, Any]] = None  # Serialized DatasetScanResult
    serving_configs: Optional[list[Any]] = None  # Serialized ServingConfig list
    browser_extensions: Optional[dict[str, Any]] = None  # Serialized browser extension scan results
    ai_inventory_data: Optional[dict[str, Any]] = None  # AI component source scan results (SDK imports, models, keys)
    project_inventory_data: Optional[dict[str, Any]] = None  # Project manifest / lockfile inventory summary
    # Optional GitHub trust card from --repo / repo_url scans (stars, contributors,
    # license, pushed_at, …). Best-effort read-only API metadata — never required
    # for the security scan itself.
    repo_trust_data: Optional[dict[str, Any]] = None
    introspection_data: Optional[dict[str, Any]] = None  # Runtime MCP introspection results (tools, resources, drift)
    health_check_data: Optional[dict[str, Any]] = None  # MCP server reachability/health results
    runtime_session_graph: Optional[dict[str, Any]] = None  # Structured runtime session graph/timeline evidence

    # Unified Finding stream (issue #566 — Phase 1).
    # Populated alongside blast_radii for backward compatibility.
    # Future phases will migrate cloud_reports and proxy alerts here too.
    findings: list["Finding"] = field(default_factory=list)

    # Scan context metadata — what input sources were actually processed.
    # Populated by the CLI/API after scan completes. Consumers use this to
    # determine which UI panels, compliance frameworks, and graphs apply.
    scan_sources: list[str] = field(default_factory=list)  # e.g. ["agent_discovery", "image", "sbom"]

    # Execution quality is separate from security/policy verdicts. Consumers
    # must not interpret a complete scan with findings as an incomplete scan.
    scan_run: ScanRun = field(default_factory=ScanRun)

    # Per-release vulnerability-coverage gaps detected during the scan. Each item
    # is a serialized agent_bom.coverage.CoverageWarning (ecosystem, release,
    # reason, detail, package_count, advisory_rows). A non-empty list means the
    # data source does not carry advisories for an OS release present in the scan
    # target (typically end-of-life) — a low/zero count there must NOT be read as
    # a clean bill of health.
    coverage_warnings: list[dict[str, Any]] = field(default_factory=list)

    @property
    def has_mcp_context(self) -> bool:
        """True if scan discovered real MCP servers (not synthetic SBOM/image wrappers).

        Uses explicit server surface classification rather than command heuristics.
        """
        return any(s.is_mcp_surface for a in self.agents for s in a.mcp_servers)

    @property
    def has_agent_context(self) -> bool:
        """True if scan discovered real AI agents (not synthetic SBOM/image wrappers).

        Synthetic agents (SBOM ingest, image scan) use ``AgentType.CUSTOM`` with
        names prefixed by ``sbom:`` or ``image:``.  Real discovered agents have
        specific agent types (CLAUDE_DESKTOP, CURSOR, etc.).
        """
        return any(a.agent_type != AgentType.CUSTOM for a in self.agents)

    @property
    def agent_class_counts(self) -> dict[str, int]:
        """Real agents split by display class: AI clients vs background agents.

        Excludes synthetic SBOM/image wrappers. ``client`` = discovered AI
        host apps; ``background`` = framework/service agent definitions. See
        :func:`classify_agent_kind`.
        """
        counts = {"client": 0, "background": 0}
        for agent in self.agents:
            kind = classify_agent_kind(agent)
            if kind in counts:
                counts[kind] += 1
        return counts

    def __post_init__(self) -> None:
        if not self.tool_version:
            from agent_bom import __version__

            self.tool_version = __version__

    @property
    def databricks_cis_benchmark_data(self) -> Optional[dict[str, Any]]:
        """Deprecated alias for ``databricks_security_data``.

        Databricks has no official CIS benchmark; the canonical field is
        ``databricks_security_data`` (vendor security best practices). This
        CIS-named property is retained for backward compatibility with existing
        clients and delegates both read and write to the canonical field.
        """
        return self.databricks_security_data

    @databricks_cis_benchmark_data.setter
    def databricks_cis_benchmark_data(self, value: Optional[dict[str, Any]]) -> None:
        self.databricks_security_data = value

    @property
    def total_agents(self) -> int:
        return len(self.agents)

    @property
    def total_servers(self) -> int:
        return sum(len(a.mcp_servers) for a in self.agents)

    @property
    def total_packages(self) -> int:
        return sum(a.total_packages for a in self.agents)

    @property
    def total_vulnerabilities(self) -> int:
        return sum(a.total_vulnerabilities for a in self.agents)

    @property
    def critical_vulns(self) -> list[BlastRadius]:
        from agent_bom.vex import active_blast_radii

        return [br for br in active_blast_radii(self.blast_radii) if br.vulnerability.severity == Severity.CRITICAL]

    def _secret_findings(self) -> "list[Finding]":
        """Hardcoded-secret findings, lifted from ``ai_inventory_data['secrets']``.

        The secret scanner stores its results in a side block; surface them in the
        unified stream so they reach JSON/SARIF/CSV — redacted, never the value.
        """
        block = (self.ai_inventory_data or {}).get("secrets") if self.ai_inventory_data else None
        if not isinstance(block, dict):
            return []
        from agent_bom.finding import secret_dict_to_finding

        return [secret_dict_to_finding(s) for s in block.get("findings", []) if isinstance(s, dict)]

    def _toxic_combination_findings(self) -> "list[Finding]":
        """Graph toxic-combination findings, rehydrated from the side block.

        The graph evaluator (``graph.toxic_findings``) stores serialized Finding
        dicts on ``toxic_combination_findings_data`` at the graph-build call site.
        Surfacing them here routes them through ``--fail-on-severity`` and every
        machine output (JSON/SARIF/CSV), mirroring the secret-finding side block.
        """
        if not self.toxic_combination_findings_data:
            return []
        from agent_bom.graph.toxic_findings import toxic_combination_findings_from_data

        return toxic_combination_findings_from_data(self.toxic_combination_findings_data)

    def _nhi_governance_findings(self) -> "list[Finding]":
        """NHI/CIEM governance findings, surfaced from the graph-derived side block.

        The scan call site (CLI + API) computes these over the unified graph and
        stores the Finding objects on ``nhi_governance_findings``. Surfacing them
        here routes identity over-grant / dormant / orphaned risks through
        ``--fail-on-severity`` and every machine output, matching the CLI path.
        """
        return list(self.nhi_governance_findings or [])

    def _mcp_tool_rule_findings(self) -> "list[Finding]":
        """MCP tool-schema rule violations lifted into the unified stream.

        The MCP analyzer stores its violations on each ``MCPTool.schema_rule_findings``
        and previously only emitted them inside JSON tool blocks — so the severity
        gate and SARIF never saw them. Convert each stored rule dict to a Finding.
        Derived on the fly from the report's own tools, so it needs no side block
        and stays idempotent (stable ids from rule id + tool).
        """
        from agent_bom.mcp_tool_rules import mcp_rule_finding_to_finding

        findings: list[Finding] = []
        seen: set[str] = set()
        for agent in self.agents:
            for server in getattr(agent, "mcp_servers", []) or []:
                for tool in getattr(server, "tools", []) or []:
                    for raw in getattr(tool, "schema_rule_findings", []) or []:
                        if not isinstance(raw, dict):
                            continue
                        finding = mcp_rule_finding_to_finding(
                            raw,
                            tool_name=getattr(tool, "name", ""),
                            tool_stable_id=getattr(tool, "canonical_id", None),
                            server_name=getattr(server, "name", ""),
                            agent_name=getattr(agent, "name", ""),
                        )
                        if finding.id in seen:
                            continue
                        seen.add(finding.id)
                        findings.append(finding)
        return findings

    def _ciem_over_privilege_findings(self) -> "list[Finding]":
        """CIEM over-privilege findings, rehydrated from the side block.

        Access-Advisor right-sizing findings are serialized on
        ``ciem_over_privilege_findings_data`` at the graph-build call site;
        surfacing them here routes them through ``--fail-on-severity`` and every
        machine output (JSON/SARIF/CSV), mirroring the toxic-combination block.
        """
        if not self.ciem_over_privilege_findings_data:
            return []
        from agent_bom.graph.nhi_governance import ciem_over_privilege_findings_from_data

        return ciem_over_privilege_findings_from_data(self.ciem_over_privilege_findings_data)

    def to_findings(self) -> "list[Finding]":
        """Return the unified findings list, auto-populating from blast_radii if empty.

        Phase 1 shim: if ``self.findings`` is already populated (dual-write path),
        use it directly.  Otherwise convert ``blast_radii`` on the fly. Hardcoded-
        secret findings are always appended so machine consumers (JSON/SARIF/CSV)
        see them, not just the console.
        """
        from agent_bom.finding import blast_radius_to_finding

        # Keep explicit non-CVE findings even when the legacy blast-radius
        # projection is also present. API/VEX paths can update the projection
        # after the unified stream was built; dropping either side makes JSON
        # disagree with CSV/SARIF and hides policy findings from consumers.
        base = list(self.findings)
        existing_ids = {getattr(f, "canonical_id", getattr(f, "id", None)) for f in base}
        for br in self.blast_radii:
            finding = blast_radius_to_finding(br)
            finding_id = getattr(finding, "canonical_id", getattr(finding, "id", None))
            if finding_id not in existing_ids:
                base.append(finding)
                existing_ids.add(finding_id)
        # Avoid double-counting if a dual-write path ever adds the same secret
        # finding, but do not suppress unrelated secret findings in the side block.
        existing_ids = {getattr(f, "canonical_id", getattr(f, "id", None)) for f in base}
        base.extend(finding for finding in self._secret_findings() if finding.id not in existing_ids)
        cis_existing = existing_ids | {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._cloud_cis_findings() if finding.id not in cis_existing)
        toxic_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._toxic_combination_findings() if finding.id not in toxic_existing)
        nhi_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._nhi_governance_findings() if finding.id not in nhi_existing)
        ciem_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._ciem_over_privilege_findings() if finding.id not in ciem_existing)
        mcp_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._mcp_tool_rule_findings() if finding.id not in mcp_existing)
        iac_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._iac_findings() if finding.id not in iac_existing)
        gov_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._snowflake_governance_findings() if finding.id not in gov_existing)
        org_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._cloud_org_architecture_findings() if finding.id not in org_existing)
        malicious_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._malicious_package_findings() if finding.id not in malicious_existing)
        return base

    def _malicious_package_findings(self) -> "list[Finding]":
        """Malicious/typosquat packages with no CVE BlastRadius row."""
        from agent_bom.finding import malicious_package_to_finding

        covered: set[tuple[str, str, str]] = {
            (br.package.name, br.package.version or "", br.package.ecosystem or "") for br in self.blast_radii
        }
        grouped: dict[tuple[str, str, str], tuple[object, set[str], set[str]]] = {}
        for agent in self.agents:
            for server in agent.mcp_servers:
                for pkg in server.packages:
                    if not getattr(pkg, "is_malicious", False):
                        continue
                    key = (pkg.name, pkg.version or "", pkg.ecosystem or "")
                    if key in covered:
                        continue
                    if key not in grouped:
                        grouped[key] = (pkg, set(), set())
                    pkg_ref, agents, servers = grouped[key]
                    agents.add(agent.name)
                    servers.add(server.name)
                    grouped[key] = (pkg_ref, agents, servers)
        return [
            malicious_package_to_finding(
                pkg,
                affected_agents=sorted(agents),
                affected_servers=sorted(servers),
            )
            for pkg, agents, servers in grouped.values()
        ]

    def _snowflake_governance_findings(self) -> "list[Finding]":
        """Snowflake governance findings lifted into the unified findings stream.

        The derived governance findings live in a side block
        (``snowflake_governance_data['findings']``) and never reached the unified
        stream — so ``cloud`` scans exited 0 even on HIGH/CRITICAL access risks and
        ``--fail-on-severity`` was blind to governance posture. Convert each one to a
        Finding so the gate, SARIF, and severity rollups converge.
        """
        from agent_bom.finding import snowflake_governance_finding_to_finding

        data = self.snowflake_governance_data
        if not isinstance(data, dict):
            return []
        account = str(data.get("account", "") or "")
        findings: list[Finding] = []
        for raw in data.get("findings", []) or []:
            if isinstance(raw, dict):
                findings.append(snowflake_governance_finding_to_finding(raw, account))
        return findings

    def _cloud_org_architecture_findings(self) -> "list[Finding]":
        """AWS/GCP org-architecture findings (single-account / flat hierarchy)."""
        from agent_bom.finding import cloud_org_architecture_finding_to_finding

        payloads: list[tuple[str, dict[str, Any]]] = []
        aws = self.aws_organization_data
        if isinstance(aws, dict):
            payloads.append(("aws", aws))
        inventory = self.cloud_inventory_data
        inv_list = inventory if isinstance(inventory, list) else ([inventory] if isinstance(inventory, dict) else [])
        for entry in inv_list:
            if isinstance(entry, dict) and isinstance(entry.get("gcp_organization"), dict):
                payloads.append(("gcp", entry["gcp_organization"]))

        findings: list[Finding] = []
        for provider, payload in payloads:
            org_id = str(payload.get("org_id") or "")
            for raw in payload.get("findings", []) or []:
                if isinstance(raw, dict):
                    findings.append(cloud_org_architecture_finding_to_finding(raw, provider=provider, org_id=org_id))
        return findings

    def _iac_findings(self) -> "list[Finding]":
        """IaC misconfiguration findings lifted into the unified stream.

        The IaC scanner stores results in the ``iac_findings_data`` side block; they
        reached only JSON + SARIF's dedicated IaC loop, so exec ``total_findings``,
        ``--fail-on-severity``, and severity rollups under-counted them. Convert each
        to a Finding (mirrors the CIS / secret side-block pattern). SARIF continues
        to render IaC via its dedicated loop and skips these in the unified loop, so
        each IaC finding appears exactly once.
        """
        from agent_bom.finding import iac_finding_to_finding

        data = self.iac_findings_data
        if not isinstance(data, dict):
            return []
        findings: list[Finding] = []
        for raw in data.get("findings", []) or []:
            if isinstance(raw, dict):
                findings.append(iac_finding_to_finding(raw))
        return findings

    def _cloud_cis_findings(self) -> "list[Finding]":
        """Cloud CIS failures and evaluation errors lifted into findings.

        Each provider's CIS results live in a side block (``*_cis_benchmark_data``)
        and never reached the unified stream. Convert FAILED controls and ERROR
        controls to distinct CLOUD_CIS findings so severity gates fail closed on
        unevaluable high-risk controls. Genuine NOT_APPLICABLE controls remain
        outside the findings stream.
        """
        from agent_bom.finding import cloud_cis_check_to_finding

        findings: list[Finding] = []
        for provider, data in (
            ("aws", self.cis_benchmark_data),
            ("azure", self.azure_cis_benchmark_data),
            ("gcp", self.gcp_cis_benchmark_data),
            ("snowflake", self.snowflake_cis_benchmark_data),
            ("databricks", self.databricks_security_data),
        ):
            if not isinstance(data, dict):
                continue
            for check in data.get("checks", []) or []:
                if isinstance(check, dict) and str(check.get("status", "")).upper() in {"FAIL", "ERROR"}:
                    check_with_version = {**check, "benchmark_version": data.get("benchmark_version")}
                    findings.append(cloud_cis_check_to_finding(check_with_version, provider))
        return findings

    def cve_findings(self) -> "list[Finding]":
        """Return only CVE-type findings from the unified stream."""
        from agent_bom.finding import FindingType

        return [f for f in self.to_findings() if f.finding_type == FindingType.CVE]

has_mcp_context `property` ¶

has_mcp_context: bool

True if scan discovered real MCP servers (not synthetic SBOM/image wrappers).

Uses explicit server surface classification rather than command heuristics.

has_agent_context `property` ¶

has_agent_context: bool

True if scan discovered real AI agents (not synthetic SBOM/image wrappers).

Synthetic agents (SBOM ingest, image scan) use AgentType.CUSTOM with names prefixed by sbom: or image:. Real discovered agents have specific agent types (CLAUDE_DESKTOP, CURSOR, etc.).

agent_class_counts `property` ¶

agent_class_counts: dict[str, int]

Real agents split by display class: AI clients vs background agents.

Excludes synthetic SBOM/image wrappers. client = discovered AI host apps; background = framework/service agent definitions. See :func:classify_agent_kind.

databricks_cis_benchmark_data `property` `writable` ¶

databricks_cis_benchmark_data: Optional[dict[str, Any]]

Deprecated alias for databricks_security_data.

Databricks has no official CIS benchmark; the canonical field is databricks_security_data (vendor security best practices). This CIS-named property is retained for backward compatibility with existing clients and delegates both read and write to the canonical field.

to_findings ¶

to_findings() -> 'list[Finding]'

Return the unified findings list, auto-populating from blast_radii if empty.

Phase 1 shim: if self.findings is already populated (dual-write path), use it directly. Otherwise convert blast_radii on the fly. Hardcoded- secret findings are always appended so machine consumers (JSON/SARIF/CSV) see them, not just the console.

Source code in src/agent_bom/models.py

def to_findings(self) -> "list[Finding]":
    """Return the unified findings list, auto-populating from blast_radii if empty.

    Phase 1 shim: if ``self.findings`` is already populated (dual-write path),
    use it directly.  Otherwise convert ``blast_radii`` on the fly. Hardcoded-
    secret findings are always appended so machine consumers (JSON/SARIF/CSV)
    see them, not just the console.
    """
    from agent_bom.finding import blast_radius_to_finding

    # Keep explicit non-CVE findings even when the legacy blast-radius
    # projection is also present. API/VEX paths can update the projection
    # after the unified stream was built; dropping either side makes JSON
    # disagree with CSV/SARIF and hides policy findings from consumers.
    base = list(self.findings)
    existing_ids = {getattr(f, "canonical_id", getattr(f, "id", None)) for f in base}
    for br in self.blast_radii:
        finding = blast_radius_to_finding(br)
        finding_id = getattr(finding, "canonical_id", getattr(finding, "id", None))
        if finding_id not in existing_ids:
            base.append(finding)
            existing_ids.add(finding_id)
    # Avoid double-counting if a dual-write path ever adds the same secret
    # finding, but do not suppress unrelated secret findings in the side block.
    existing_ids = {getattr(f, "canonical_id", getattr(f, "id", None)) for f in base}
    base.extend(finding for finding in self._secret_findings() if finding.id not in existing_ids)
    cis_existing = existing_ids | {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._cloud_cis_findings() if finding.id not in cis_existing)
    toxic_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._toxic_combination_findings() if finding.id not in toxic_existing)
    nhi_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._nhi_governance_findings() if finding.id not in nhi_existing)
    ciem_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._ciem_over_privilege_findings() if finding.id not in ciem_existing)
    mcp_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._mcp_tool_rule_findings() if finding.id not in mcp_existing)
    iac_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._iac_findings() if finding.id not in iac_existing)
    gov_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._snowflake_governance_findings() if finding.id not in gov_existing)
    org_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._cloud_org_architecture_findings() if finding.id not in org_existing)
    malicious_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._malicious_package_findings() if finding.id not in malicious_existing)
    return base

cve_findings ¶

cve_findings() -> 'list[Finding]'

Return only CVE-type findings from the unified stream.

Source code in src/agent_bom/models.py

def cve_findings(self) -> "list[Finding]":
    """Return only CVE-type findings from the unified stream."""
    from agent_bom.finding import FindingType

    return [f for f in self.to_findings() if f.finding_type == FindingType.CVE]