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
    graph_reachable: Optional[bool] = None
    graph_min_hop_distance: Optional[int] = None
    graph_reachable_from_agents: list[str] = field(default_factory=list)
    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

    # Ownership + remediation SLA (additive, optional). ``first_seen`` anchors
    # the SLA window (scan-observation time for a fresh scan); ``owner`` surfaces
    # the triage assignee when one exists; ``sla_due_at`` is severity-derived
    # from ``first_seen`` with a KEV override. All left None by default so
    # existing findings serialize unchanged and to_dict() derives on demand.
    first_seen: Optional[str] = None
    owner: Optional[str] = None
    sla_due_at: Optional[str] = None
    sla_due_at_source: Optional[str] = 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="")

    # Appended after the long-standing explicit-ID slot so additive workflow
    # state does not shift legacy positional construction of this public model.
    lifecycle_status: Optional[str] = None

    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 occurrence ID: the same issue at the same location
            # on one asset stays stable, while distinct source-native locations
            # cannot silently collapse into one workflow record.
            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 self.evidence.get("package") or "")
                pkg_version = str(self.evidence.get("package_version") or "")
                # Several scanners expose the affected component as the compact
                # ``package=name@version`` form. Normalise that public evidence
                # shape before deriving the occurrence identity; otherwise two
                # packages attached to one container/server silently collapse.
                if pkg_name and not pkg_version and "@" in pkg_name:
                    pkg_name, pkg_version = pkg_name.rsplit("@", 1)
            qualifiers: list[object] = [pkg_name, pkg_version]
            occurrence_qualifier = _occurrence_evidence_qualifier(self.evidence)
            if occurrence_qualifier:
                qualifiers.append(occurrence_qualifier)
            self.id = canonical_finding_id(self.asset.stable_id, cve_part, *qualifiers)

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

    @property
    def occurrence_id(self) -> str:
        """Workflow identity for this finding on this exact asset occurrence."""
        return self.id

    @property
    def finding_group_key(self) -> str:
        """Asset-independent issue identity used only for aggregate/expand views.

        The occurrence id remains authoritative for triage, lifecycle, evidence,
        and persistence.  Vulnerability groups intentionally omit the asset and
        installed version so repeated occurrences can be presented together
        without destroying their distinct workflow records.
        """
        vulnerability_id = str(self.vulnerability_id or "").strip().lower()
        if self.finding_type is not FindingType.CVE or not vulnerability_id:
            return f"occurrence:{self.occurrence_id}"
        package_name = self.asset.name if self.asset.asset_type == "package" else ""
        if not package_name and isinstance(self.evidence, dict):
            package_name = str(self.evidence.get("package_name") or "")
        package_name = package_name.strip().lower()
        return f"vulnerability:{vulnerability_id}:{package_name}"

    @property
    def finding_group_id(self) -> str:
        """Deterministic identifier for a grouped issue queue row."""
        return stable_id("finding-group", self.finding_group_key)

    @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:
                    if field_name == "compliance_tags":
                        tags.append(_control_tag_from_compliance_tag(str(value)))
                        continue
                    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."""
        from agent_bom.graph.sla import finding_owner, finding_sla_fields

        sla = finding_sla_fields(
            {
                "severity": self.effective_severity(),
                "first_seen": self.first_seen,
                "evidence": self.evidence,
                "sla_due_at": self.sla_due_at,
                "sla_due_at_source": self.sla_due_at_source or ("explicit" if self.sla_due_at is not None else None),
            }
        )
        return {
            "schema_version": FINDING_SCHEMA_VERSION,
            "id": self.id,
            "canonical_id": self.canonical_id,
            "finding_id": self.occurrence_id,
            "occurrence_id": self.occurrence_id,
            "finding_group_id": self.finding_group_id,
            "finding_group_key": self.finding_group_key,
            "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,
            "graph_reachable": self.graph_reachable,
            "graph_min_hop_distance": self.graph_min_hop_distance,
            "graph_reachable_from_agents": list(self.graph_reachable_from_agents),
            "is_actionable": self.is_actionable,
            "impact_category": self.impact_category,
            # Ownership + remediation SLA (derived, single source of truth in
            # agent_bom.graph.sla). ``owner`` is an explicit None when nobody is
            # assigned (an honest absence for the API/exports; the CLI/UI render
            # it as "Unassigned"); ``sla_due_at`` is None when no deadline can be
            # derived (unrated severity + no anchor/KEV date).
            "first_seen": self.first_seen,
            "owner": finding_owner(self.owner),
            **sla,
            "status": self.lifecycle_status,
            "lifecycle_status": self.lifecycle_status,
            # 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.

occurrence_id property

occurrence_id: str

Workflow identity for this finding on this exact asset occurrence.

finding_group_key property

finding_group_key: str

Asset-independent issue identity used only for aggregate/expand views.

The occurrence id remains authoritative for triage, lifecycle, evidence, and persistence. Vulnerability groups intentionally omit the asset and installed version so repeated occurrences can be presented together without destroying their distinct workflow records.

finding_group_id property

finding_group_id: str

Deterministic identifier for a grouped issue queue row.

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 occurrence ID: the same issue at the same location
        # on one asset stays stable, while distinct source-native locations
        # cannot silently collapse into one workflow record.
        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 self.evidence.get("package") or "")
            pkg_version = str(self.evidence.get("package_version") or "")
            # Several scanners expose the affected component as the compact
            # ``package=name@version`` form. Normalise that public evidence
            # shape before deriving the occurrence identity; otherwise two
            # packages attached to one container/server silently collapse.
            if pkg_name and not pkg_version and "@" in pkg_name:
                pkg_name, pkg_version = pkg_name.rsplit("@", 1)
        qualifiers: list[object] = [pkg_name, pkg_version]
        occurrence_qualifier = _occurrence_evidence_qualifier(self.evidence)
        if occurrence_qualifier:
            qualifiers.append(occurrence_qualifier)
        self.id = canonical_finding_id(self.asset.stable_id, cve_part, *qualifiers)

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."""
    from agent_bom.graph.sla import finding_owner, finding_sla_fields

    sla = finding_sla_fields(
        {
            "severity": self.effective_severity(),
            "first_seen": self.first_seen,
            "evidence": self.evidence,
            "sla_due_at": self.sla_due_at,
            "sla_due_at_source": self.sla_due_at_source or ("explicit" if self.sla_due_at is not None else None),
        }
    )
    return {
        "schema_version": FINDING_SCHEMA_VERSION,
        "id": self.id,
        "canonical_id": self.canonical_id,
        "finding_id": self.occurrence_id,
        "occurrence_id": self.occurrence_id,
        "finding_group_id": self.finding_group_id,
        "finding_group_key": self.finding_group_key,
        "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,
        "graph_reachable": self.graph_reachable,
        "graph_min_hop_distance": self.graph_min_hop_distance,
        "graph_reachable_from_agents": list(self.graph_reachable_from_agents),
        "is_actionable": self.is_actionable,
        "impact_category": self.impact_category,
        # Ownership + remediation SLA (derived, single source of truth in
        # agent_bom.graph.sla). ``owner`` is an explicit None when nobody is
        # assigned (an honest absence for the API/exports; the CLI/UI render
        # it as "Unassigned"); ``sla_due_at`` is None when no deadline can be
        # derived (unrated severity + no anchor/KEV date).
        "first_seen": self.first_seen,
        "owner": finding_owner(self.owner),
        **sla,
        "status": self.lifecycle_status,
        "lifecycle_status": self.lifecycle_status,
        # 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
    endpoint_inventory_data: Optional[dict[str, Any]] = None  # Bounded workstation app/process/service/runtime inventory
    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
    # Ownership parsed from the repository's CODEOWNERS file, carried as
    # provenance so graph overlays never need filesystem access. Two shapes are
    # accepted and round-trip unchanged: ordered ``{"pattern", "owners"}`` rules
    # (full CODEOWNERS precedence, used to assign source-finding owners) and the
    # flattened ``{path_prefix: owner}`` map the repo-tree scan produces.
    codeowners: Union[list[dict[str, Any]], dict[str, str]] = field(default_factory=list)
    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:
        # Package manifests, SBOMs, and container layers reuse MCPServer as a
        # dependency-bearing surface model. They are not MCP servers and must
        # not inflate the public MCP inventory count.
        return sum(1 for agent in self.agents for server in agent.mcp_servers if server.is_mcp_surface)

    @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 _ast_flow_findings(self) -> "list[Finding]":
        """Native AST flow risks promoted from the source-analysis side block."""
        inventory = self.ai_inventory_data or {}
        block = inventory.get("ast_analysis") if isinstance(inventory, dict) else None
        if not isinstance(block, dict):
            return []
        from agent_bom.finding import ast_flow_dict_to_finding, is_ast_security_flow

        severity_rank = {"low": 1, "medium": 2, "high": 3, "critical": 4}
        by_sink: dict[tuple[object, ...], Finding] = {}
        for raw in block.get("flow_findings", []):
            if not isinstance(raw, dict) or not is_ast_security_flow(raw):
                continue
            finding = ast_flow_dict_to_finding(raw)
            key = (
                finding.asset.location,
                finding.evidence.get("line"),
                finding.evidence.get("entrypoint"),
                finding.evidence.get("sink") or finding.evidence.get("category"),
            )
            current = by_sink.get(key)
            categories = {str(finding.evidence.get("category", ""))}
            if current is not None:
                categories.add(str(current.evidence.get("category", "")))
                categories.update(str(item) for item in current.evidence.get("detector_categories", []))
            categories.discard("")
            if current is None or severity_rank.get(finding.severity, 0) > severity_rank.get(current.severity, 0):
                finding.evidence["detector_categories"] = sorted(categories)
                by_sink[key] = finding
            else:
                current.evidence["detector_categories"] = sorted(categories)
        return list(by_sink.values())

    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 _enforcement_findings(self) -> "list[Finding]":
        """MCP description/enforcement findings promoted from the side block."""
        if not isinstance(self.enforcement_data, dict):
            return []
        from agent_bom.finding import enforcement_dict_to_finding

        return [enforcement_dict_to_finding(raw) for raw in self.enforcement_data.get("findings", []) or [] if isinstance(raw, dict)]

    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 FindingType, 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)

        def _materialized_cve_key(finding: "Finding") -> tuple[str, str, str, str, tuple[str, ...], tuple[str, ...]]:
            evidence = finding.evidence if isinstance(finding.evidence, dict) else {}
            return (
                str(finding.vulnerability_id or ""),
                str(evidence.get("package_name") or ""),
                str(evidence.get("package_version") or ""),
                str(evidence.get("ecosystem") or ""),
                tuple(sorted(finding.affected_agents or [])),
                tuple(sorted(finding.affected_servers or [])),
            )

        materialized_cve_keys = sorted(_materialized_cve_key(finding) for finding in base if finding.finding_type is FindingType.CVE)
        blast_radius_keys = sorted(
            (
                str(br.vulnerability.id or ""),
                str(br.package.name or ""),
                str(br.package.version or ""),
                str(br.package.ecosystem or ""),
                tuple(sorted(str(getattr(agent, "name", "") or "") for agent in br.affected_agents)),
                tuple(sorted(str(getattr(server, "name", "") or "") for server in br.affected_servers)),
            )
            for br in self.blast_radii
        )
        if materialized_cve_keys != blast_radius_keys:
            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)
        ast_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._ast_flow_findings() if finding.id not in ast_existing)
        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)
        enforcement_existing = {getattr(f, "id", None) for f in base}
        base.extend(finding for finding in self._enforcement_findings() if finding.id not in enforcement_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)
        if self.codeowners:
            from agent_bom.graph.codeowners import apply_codeowners

            apply_codeowners(base, self.codeowners)
        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 FindingType, 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)

    def _materialized_cve_key(finding: "Finding") -> tuple[str, str, str, str, tuple[str, ...], tuple[str, ...]]:
        evidence = finding.evidence if isinstance(finding.evidence, dict) else {}
        return (
            str(finding.vulnerability_id or ""),
            str(evidence.get("package_name") or ""),
            str(evidence.get("package_version") or ""),
            str(evidence.get("ecosystem") or ""),
            tuple(sorted(finding.affected_agents or [])),
            tuple(sorted(finding.affected_servers or [])),
        )

    materialized_cve_keys = sorted(_materialized_cve_key(finding) for finding in base if finding.finding_type is FindingType.CVE)
    blast_radius_keys = sorted(
        (
            str(br.vulnerability.id or ""),
            str(br.package.name or ""),
            str(br.package.version or ""),
            str(br.package.ecosystem or ""),
            tuple(sorted(str(getattr(agent, "name", "") or "") for agent in br.affected_agents)),
            tuple(sorted(str(getattr(server, "name", "") or "") for server in br.affected_servers)),
        )
        for br in self.blast_radii
    )
    if materialized_cve_keys != blast_radius_keys:
        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)
    ast_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._ast_flow_findings() if finding.id not in ast_existing)
    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)
    enforcement_existing = {getattr(f, "id", None) for f in base}
    base.extend(finding for finding in self._enforcement_findings() if finding.id not in enforcement_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)
    if self.codeowners:
        from agent_bom.graph.codeowners import apply_codeowners

        apply_codeowners(base, self.codeowners)
    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]

AgentBomSDKError

Bases: RuntimeError

Raised when the public Python API cannot complete a requested operation.

Source code in src/agent_bom/sdk.py

class AgentBomSDKError(RuntimeError):
    """Raised when the public Python API cannot complete a requested operation."""

PackageCheckResult dataclass

Typed result returned by :func:check and :func:async_check.

Source code in src/agent_bom/sdk.py

@dataclass(frozen=True)
class PackageCheckResult:
    """Typed result returned by :func:`check` and :func:`async_check`."""

    package: str
    version: str
    ecosystem: str
    status: str
    vulnerabilities: int
    details: list[dict[str, Any]] = field(default_factory=list)
    message: str = ""

    @property
    def is_clean(self) -> bool:
        return self.status == "clean"

    def to_dict(self) -> dict[str, Any]:
        return {
            "package": self.package,
            "version": self.version,
            "ecosystem": self.ecosystem,
            "status": self.status,
            "vulnerabilities": self.vulnerabilities,
            "details": self.details,
            "message": self.message,
        }

InventoryResult dataclass

Typed inventory wrapper returned by :func:inventory.

Source code in src/agent_bom/sdk.py

@dataclass(frozen=True)
class InventoryResult:
    """Typed inventory wrapper returned by :func:`inventory`."""

    data: dict[str, Any]
    agent_count: int
    server_count: int
    package_count: int

    def to_dict(self) -> dict[str, Any]:
        return {
            "data": self.data,
            "agent_count": self.agent_count,
            "server_count": self.server_count,
            "package_count": self.package_count,
        }

DiffResult dataclass

Typed report diff wrapper returned by :func:diff.

Source code in src/agent_bom/sdk.py

@dataclass(frozen=True)
class DiffResult:
    """Typed report diff wrapper returned by :func:`diff`."""

    data: dict[str, Any]

    @property
    def summary(self) -> dict[str, Any]:
        summary = self.data.get("summary", {})
        return summary if isinstance(summary, dict) else {}

    @property
    def new_findings(self) -> int:
        return int(self.summary.get("new_findings", 0) or 0)

    def to_dict(self) -> dict[str, Any]:
        return self.data

scan

scan(*, config_path: str | Path | None = None, project: str | Path | None = None, demo: bool = False, offline: bool = False, enrich: bool = False, compliance: bool = False, transitive: bool = False, max_depth: int = 3, blast_radius_depth: int = 2) -> AIBOMReport

Run the standard local scan pipeline and return a typed report.

project and config_path both point at a local project/config scope. The name config_path is kept for MCP/API users who are already familiar with that argument; internally this delegates to the same simple scan runner used by CLI commands.

Source code in src/agent_bom/sdk.py

def scan(
    *,
    config_path: str | Path | None = None,
    project: str | Path | None = None,
    demo: bool = False,
    offline: bool = False,
    enrich: bool = False,
    compliance: bool = False,
    transitive: bool = False,
    max_depth: int = 3,
    blast_radius_depth: int = 2,
) -> AIBOMReport:
    """Run the standard local scan pipeline and return a typed report.

    ``project`` and ``config_path`` both point at a local project/config scope.
    The name ``config_path`` is kept for MCP/API users who are already familiar
    with that argument; internally this delegates to the same simple scan runner
    used by CLI commands.
    """

    if project is not None and config_path is not None and Path(project).expanduser() != Path(config_path).expanduser():
        raise ValueError("project and config_path refer to different scan scopes")

    from rich.console import Console

    from agent_bom.cli._scan_runner import ScanConfig, run_default_scan

    output = io.StringIO()
    console = Console(file=output, force_terminal=False, no_color=True, width=120)
    scan_scope = project if project is not None else config_path
    result = run_default_scan(
        ScanConfig(
            project=str(scan_scope) if scan_scope is not None else None,
            demo=demo,
            offline=offline,
            enrich=enrich,
            compliance=compliance,
            resolve_transitive=transitive,
            max_depth=max_depth,
            blast_radius_depth=blast_radius_depth,
            quiet=True,
        ),
        console,
    )
    if result.report is None:
        raise AgentBomSDKError("scan completed without a report")
    return result.report

async_check async

async_check(package: str, *, ecosystem: str = 'npm', offline: bool = False) -> PackageCheckResult

Check one package spec and return a typed vulnerability result.

Source code in src/agent_bom/sdk.py

async def async_check(package: str, *, ecosystem: str = "npm", offline: bool = False) -> PackageCheckResult:
    """Check one package spec and return a typed vulnerability result."""

    eco = validate_ecosystem(ecosystem, SUPPORTED_PACKAGE_ECOSYSTEM_SET)
    name, version = _parse_package_spec(package)
    if eco in {"deb", "apk", "rpm"} and version in {"", "latest"}:
        raise ValueError(f"explicit version required for {eco} packages")

    from agent_bom.scanners import ScanOptions, scan_packages

    pkg = Package(name=name, version=version, ecosystem=eco)
    await scan_packages([pkg], options=ScanOptions(offline=offline))
    details = _vulnerability_details(pkg)
    status = "vulnerable" if details else "clean"
    return PackageCheckResult(
        package=pkg.name,
        version=pkg.version,
        ecosystem=eco,
        status=status,
        vulnerabilities=len(details),
        details=details,
        message=f"No known vulnerabilities in {pkg.name}@{pkg.version}" if status == "clean" else "",
    )

check

check(package: str, *, ecosystem: str = 'npm', offline: bool = False) -> PackageCheckResult

Synchronous wrapper around :func:async_check.

Async applications should call :func:async_check directly to avoid nesting event loops.

Source code in src/agent_bom/sdk.py

def check(package: str, *, ecosystem: str = "npm", offline: bool = False) -> PackageCheckResult:
    """Synchronous wrapper around :func:`async_check`.

    Async applications should call :func:`async_check` directly to avoid nesting
    event loops.
    """

    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(async_check(package, ecosystem=ecosystem, offline=offline))
    raise AgentBomSDKError("check() cannot run inside an active event loop; use async_check() instead")

inventory

inventory(source: str | Path) -> InventoryResult

Load a JSON/CSV/NDJSON inventory artifact and return typed counts.

Source code in src/agent_bom/sdk.py

def inventory(source: str | Path) -> InventoryResult:
    """Load a JSON/CSV/NDJSON inventory artifact and return typed counts."""

    from agent_bom.inventory import load_inventory

    data = load_inventory(str(source))
    agents = data.get("agents", [])
    agent_count = len(agents) if isinstance(agents, list) else 0
    server_count = 0
    package_count = 0
    if isinstance(agents, list):
        for agent in agents:
            servers = agent.get("mcp_servers", []) if isinstance(agent, dict) else []
            if not isinstance(servers, list):
                continue
            server_count += len(servers)
            for server in servers:
                packages = server.get("packages", []) if isinstance(server, dict) else []
                if isinstance(packages, list):
                    package_count += len(packages)
    return InventoryResult(
        data=data,
        agent_count=agent_count,
        server_count=server_count,
        package_count=package_count,
    )

diff

diff(baseline: str | Path | Mapping[str, Any], current: str | Path | Mapping[str, Any]) -> DiffResult

Diff two agent-bom reports or SBOM documents.

Source code in src/agent_bom/sdk.py

def diff(baseline: str | Path | Mapping[str, Any], current: str | Path | Mapping[str, Any]) -> DiffResult:
    """Diff two agent-bom reports or SBOM documents."""

    from agent_bom.history import diff_reports

    return DiffResult(diff_reports(_coerce_report_input(baseline), _coerce_report_input(current)))