Python SDK Reference
The attestd package provides sync and async clients, typed response models, named exceptions, and testing utilities. Requires Python 3.10+.
pip install attestdattestd.Client
Synchronous client. Thread-safe; a single instance can be shared across threads.
| Parameter | Description |
|---|---|
api_key | Required. Your Attestd API key. |
base_url | Base URL override. Defaults to "https://api.attestd.io". |
timeout | Request timeout in seconds. Default: 10.0. |
max_retries | Retry attempts on 5xx errors. Default: 3. Set to 0 to disable. |
cache_policy | Result cache policy: "development" (24h), "runtime" (5min, default), "ci" (never expire), or "none". |
transport | Optional httpx transport override. Used in tests (see attestd.testing). |
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()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().
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 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.
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)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))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.
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):
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))client.products()
Returns the full Attestd product catalog: CVE-covered infrastructure slugs and monitored supply chain packages. Maps to GET /v1/products.
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.
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.
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
@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@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@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: floatattestd.RiskResult
Frozen dataclass returned by client.check(). All fields are read-only. See the Response Field Reference for field semantics.
@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: datetimeSupplyChainSignal
Supply chain integrity data for monitored PyPI packages. Present on RiskResult as the supply_chain field. None for CVE-only products (nginx, PostgreSQL, etc.).
@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:
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.
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.
@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, ...] = ()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.
@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)Error types
| Exception | When raised |
|---|---|
AttestdAuthError | HTTP 401. Invalid or missing API key. |
AttestdRateLimitError | HTTP 429. Rate limit exceeded. Has retry_after attribute (seconds). |
AttestdUnsupportedProductError | supported: false. Product is outside coverage. |
AttestdAPIError | HTTP 5xx, connection errors, or unexpected responses after all retries |
AttestdError | Base 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).
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}")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.
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.
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 == 3Canned response bodies
Ready-made response dicts for common test scenarios. All are plain dicts and can be merged with | to override individual fields.
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"})