sdk reference / python

Python SDK Reference

The attestd package provides sync and async clients, typed response models, named exceptions, and testing utilities. Requires Python 3.10+.

bash
pip install attestd
client

attestd.Client

Synchronous client. Thread-safe; a single instance can be shared across threads.

ParameterDescription
api_keyRequired. Your Attestd API key.
base_urlBase URL override. Defaults to "https://api.attestd.io".
timeoutRequest timeout in seconds. Default: 10.0.
max_retriesRetry attempts on 5xx errors. Default: 3. Set to 0 to disable.
cache_policyResult cache policy: "development" (24h), "runtime" (5min, default), "ci" (never expire), or "none".
transportOptional httpx transport override. Used in tests (see attestd.testing).
check.py
import attestd

client = attestd.Client(api_key="YOUR_API_KEY")

result = client.check("nginx", "1.20.0")
print(result.risk_state)         # "high"
print(result.actively_exploited) # False
print(result.cve_ids)            # ["CVE-2021-23017"]

client.close()
gate.py
with attestd.Client(api_key="YOUR_API_KEY") as client:
    result = client.check("log4j", "2.14.1")
    if result.risk_state == "critical":
        raise SystemExit("Deployment blocked")

attestd.AsyncClient

Async client with identical parameters to Client, plus batch_window_ms (default 5) to coalesce concurrent check() calls into one batch request. Use in FastAPI, async agents, or any asyncio environment. Call await client.check().

async_check.py
import attestd
import asyncio

async def check_deps():
    async with attestd.AsyncClient(api_key="YOUR_API_KEY") as client:
        result = await client.check("redis", "6.0.9")
        return result

result = asyncio.run(check_deps())
cache

Cache policies and session stats

Results are cached in memory by product and version. Cache hits do not count against your monthly quota. Use invalidate_cache after a patch deploy or advisory webhook. Use stats() to inspect api_calls_made, cache_hits, batch_saves, and calls_saved. See Best Practices for CI vs production guidance.

cache.py
import attestd

# runtime (default): 5-minute TTL. development: 24h. ci: never expire. none: always hit API.
with attestd.Client(api_key="YOUR_API_KEY", cache_policy="runtime") as client:
    result = client.check("nginx", "1.20.0")
    result = client.check("nginx", "1.20.0")  # cache hit, no API call

    client.invalidate_cache("nginx", "1.20.0")
    result = client.check("nginx", "1.20.0")  # fetches again

    stats = client.stats()
    print(stats.api_calls_made, stats.cache_hits, stats.calls_saved)
fanout.py
import attestd
import asyncio

async def fanout(deps):
    async with attestd.AsyncClient(
        api_key="YOUR_API_KEY",
        batch_window_ms=5,  # coalesce concurrent check() into one batch
    ) as client:
        return await asyncio.gather(*(client.check(p, v) for p, v in deps))
batch

client.batch_check(items)

Check up to 100 product versions in one API call. Accepts a list of (product, version) tuples. Returns results in the same order. None means the product is not in Attestd's coverage. Each item counts as one API call against your quota. If the batch would exceed your quota, AttestdRateLimitError is raised before any results are delivered and no calls are billed.

batch.py
import attestd

deps = [
    ("log4j",   "2.14.1"),
    ("nginx",   "1.27.4"),
    ("openssl", "3.0.7"),
]

with attestd.Client(api_key="YOUR_API_KEY") as client:
    results = client.batch_check(deps)

for (product, version), result in zip(deps, results):
    if result is None:
        print(f"{product} {version}: not in coverage")
    elif result.risk_state in ("critical", "high"):
        raise SystemExit(f"Deployment blocked: {product} {version} is {result.risk_state}")
    else:
        print(f"{product} {version}: {result.risk_state}")

The async client exposes the same method as await client.batch_check(items):

async_batch.py
import attestd
import asyncio

async def check_sbom(deps):
    async with attestd.AsyncClient(api_key="YOUR_API_KEY") as client:
        return await client.batch_check(deps)

deps = [("nginx", "1.20.0"), ("redis", "6.0.9")]
results = asyncio.run(check_sbom(deps))
catalog and quota

client.products()

Returns the full Attestd product catalog: CVE-covered infrastructure slugs and monitored supply chain packages. Maps to GET /v1/products.

products.py
import attestd

with attestd.Client(api_key="YOUR_API_KEY") as client:
    catalog = client.products()
    print(catalog.total)
    print(catalog.cve_products[0].slug)

client.cve(cve_id)

Returns CVSS, EPSS, KEV status, and affected products for a single CVE. Raises AttestdAPIError with status_code=404 when the CVE is not in Attestd's database.

cve.py
import attestd
from attestd import AttestdAPIError

with attestd.Client(api_key="YOUR_API_KEY") as client:
    try:
        detail = client.cve("CVE-2021-44228")
        print(detail.cvss_score, detail.epss_score)
    except AttestdAPIError as e:
        if e.status_code == 404:
            print("CVE not in database")

client.usage()

Returns API call quota for the authenticated key: calls used this month, included cap, billing period start and end, and overage estimate. Use for quota monitoring in CI or agent loops.

usage.py
import attestd

with attestd.Client(api_key="YOUR_API_KEY") as client:
    usage = client.usage()
    print(usage.tier)
    print(usage.key_calls_this_month, "/", usage.included_calls)
    print(usage.billing_period_end)

ProductsResult, CveDetail, UsageResult

python
@dataclass(frozen=True, slots=True)
class ProductEntry:
    slug: str
    display_name: str

@dataclass(frozen=True, slots=True)
class SupplyChainEntry:
    package: str
    ecosystem: str
    display_name: str | None = None

@dataclass(frozen=True, slots=True)
class ProductsResult:
    cve_products: list[ProductEntry]
    supply_chain_packages: list[SupplyChainEntry]
    total: int
python
@dataclass(frozen=True, slots=True)
class CveDetail:
    cve_id: str
    description: str | None
    cvss_score: float | None
    cvss_vector: str | None
    actively_exploited: bool
    remote_exploitable: bool
    authentication_required: bool
    affected_products: list[str]
    epss_score: float | None
    epss_percentile: float | None
    source_published_at: datetime | None
    last_checked_at: datetime | None
python
@dataclass(frozen=True, slots=True)
class UsageResult:
    tier: str
    key_calls_this_month: int
    account_calls_this_month: int
    included_calls: int
    billing_period_start: datetime
    billing_period_end: datetime
    overage_calls: int
    estimated_overage_usd: float
models

attestd.RiskResult

Frozen dataclass returned by client.check(). All fields are read-only. See the Response Field Reference for field semantics.

python
@dataclass(frozen=True, slots=True)
class RiskResult:
    product: str
    version: str
    risk_state: RiskState        # "critical" | "high" | "elevated" | "low" | "none"
    risk_factors: list[str]
    actively_exploited: bool
    remote_exploitable: bool
    authentication_required: bool
    patch_available: bool
    fixed_version: str | None
    confidence: float
    cve_ids: list[str]
    max_epss: float | None = None
    cves: list[CveSummary] = field(default_factory=list)
    supply_chain: SupplyChainSignal | None   # None for CVE-only products
    typosquat: TyposquatSignal | None        # None when no resemblance detected
    last_updated: datetime
response

SupplyChainSignal

Supply chain integrity data for monitored PyPI packages. Present on RiskResult as the supply_chain field. None for CVE-only products (nginx, PostgreSQL, etc.).

python
@dataclass(frozen=True, slots=True)
class SupplyChainSignal:
    compromised: bool
    sources: list[str]              # e.g., ["registry", "osv"]
    malware_type: str | None
    description: str | None
    advisory_url: str | None
    compromised_at: datetime | None
    removed_at: datetime | None
    provenance: bool | None         # true / false / None (npm only)

Usage example: check for compromised PyPI packages:

check_supply_chain.py
import attestd

client = attestd.Client(api_key="YOUR_API_KEY")

# Check a monitored PyPI package
result = client.check("langchain", "0.1.0")

if result.supply_chain and result.supply_chain.compromised:
    print(f"WARNING: {result.product} {result.version} is compromised")
    print(f"Detected by: {', '.join(result.supply_chain.sources)}")
    raise SystemExit("Do not deploy")

# CVE-only products have supply_chain = None
result = client.check("nginx", "1.20.0")
assert result.supply_chain is None  # Not in supply chain monitoring

client.close()

See the Supply Chain Integrity guide for detailed semantics and monitoring examples.

response

TyposquatSignal

Package name integrity signal for typosquats and AI-hallucinated package names. Present on RiskResult as the typosquat field when populated. For unsupported products, the signal is attached to AttestdUnsupportedProductError.typosquat instead of raising a result. kind is "typosquat" or "hallucination". See Response Fields for API semantics.

python
@dataclass(frozen=True, slots=True)
class TyposquatSignal:
    detected: bool
    resembles: str | None
    confidence: float        # 0.0–1.0
    ecosystem: str           # "pypi" | "npm"
    kind: Literal["typosquat", "hallucination"] = "typosquat"
    likely_intended: tuple[str, ...] = ()
response

CveSummary

Per-CVE detail records returned on RiskResult.cves when the API request includes include=cves. Empty list on the default response. See Response Fields for subfield semantics.

python
@dataclass(frozen=True, slots=True)
class CveSummary:
    cve_id: str
    cvss_score: float | None
    actively_exploited: bool
    remote_exploitable: bool
    epss_score: float | None      # EPSS probability (0.0–1.0)
    epss_percentile: float | None # EPSS percentile rank (0.0–1.0)
errors

Error types

ExceptionWhen raised
AttestdAuthErrorHTTP 401. Invalid or missing API key.
AttestdRateLimitErrorHTTP 429. Rate limit exceeded. Has retry_after attribute (seconds).
AttestdUnsupportedProductErrorsupported: false. Product is outside coverage.
AttestdAPIErrorHTTP 5xx, connection errors, or unexpected responses after all retries
AttestdErrorBase class for all above. Catch this for a single broad handler.

Note on retry behaviour: 401 and 429 errors are never retried. Only transient 5xx errors and connection failures are retried (up to max_retries).

errors.py
import attestd

client = attestd.Client(api_key="YOUR_API_KEY")

try:
    result = client.check("nginx", "1.20.0")
except attestd.AttestdAuthError:
    print("Invalid API key")
except attestd.AttestdRateLimitError as e:
    print(f"Rate limited. Retry in {e.retry_after}s")
except attestd.AttestdUnsupportedProductError as e:
    # Product is outside coverage. Not a safety clearance.
    # Check typosquat first: the name itself may be the risk.
    if e.typosquat and e.typosquat.detected:
        print(
            f"{e.product} fails name integrity "
            f"(kind={e.typosquat.kind}). Prefer {e.typosquat.resembles}."
        )
    else:
        print(f"{e.product} is outside attestd coverage")
    raise  # or handle according to your policy
except attestd.AttestdAPIError as e:
    print(f"API error: {e}")
testing

attestd.testing

The attestd.testing module provides httpx transport classes for injecting mock API responses without making real network calls. Use these to test your security branching logic against controlled, reproducible responses.

MockTransport / MockAsyncTransport

Returns the same response for every request.

test_gate.py
import attestd
from attestd.testing import MockTransport, NGINX_VULNERABLE, NGINX_SAFE

def test_deployment_blocked_on_high_risk():
    transport = MockTransport(200, NGINX_VULNERABLE)
    client = attestd.Client(api_key="test", transport=transport)

    result = client.check("nginx", "1.20.0")
    assert result.risk_state == "high"

def test_deployment_allowed_when_safe():
    transport = MockTransport(200, NGINX_SAFE)
    client = attestd.Client(api_key="test", transport=transport)

    result = client.check("nginx", "1.27.4")
    assert result.risk_state == "none"

SequentialMockTransport / SequentialMockAsyncTransport

Returns responses from a pre-defined sequence. Use for testing retry logic where the first one or two requests fail before a success.

test_retry.py
from attestd.testing import SequentialMockTransport, NGINX_SAFE

def test_retries_on_503():
    transport = SequentialMockTransport([
        (503, {}),         # first attempt fails
        (503, {}),         # second attempt fails
        (200, NGINX_SAFE), # third attempt succeeds
    ])
    client = attestd.Client(api_key="test", transport=transport, max_retries=2)
    result = client.check("nginx", "1.27.4")
    assert result.risk_state == "none"
    assert transport.call_count == 3

Canned response bodies

Ready-made response dicts for common test scenarios. All are plain dicts and can be merged with | to override individual fields.

test_coverage.py
from attestd.testing import (
    NGINX_SAFE,       # risk_state: "none"
    NGINX_VULNERABLE, # risk_state: "high"
    LOG4J_CRITICAL,   # risk_state: "critical", actively_exploited: true
    UNSUPPORTED,      # supported: false
)

# Override individual fields
transport = MockTransport(200, NGINX_VULNERABLE | {"risk_state": "critical"})
related