Python SDK
The ComplianceLayer Python SDK provides synchronous and asynchronous clients for the API, with automatic retries on transient failures, typed response models, and helpers for polling scans to completion.
Overview
The compliancelayer package exposes two client classes:
ComplianceLayer— synchronous client for scripts, notebooks, and simple integrationsAsyncComplianceLayer— async client for concurrent workloads usingasyncio
Both expose the same surface and the same constructor options; the async client returns awaitables. The SDK handles authentication, response parsing, error mapping, and retry logic.
- Python 3.9+ — tested on 3.9 through 3.13
- Type hints — ships a
py.typedmarker; responses are dataclasses - Auto-retry — exponential backoff on 429 and 5xx responses, up to
max_retries - One dependency —
httpx(installed automatically)
Installation
pip install compliancelayerRequires version 0.1.1 or newer — earlier releases used an outdated API path and do not work. httpx is installed automatically as a dependency. Requires Python 3.9 or newer.
Quickstart
Create a client with your API key and scan a domain. scan() submits the job, polls until it finishes, and returns the completed report.
from compliancelayer import ComplianceLayer
client = ComplianceLayer("cl_your_api_key_here")
report = client.scan("example.com")
print(report.domain, report.grade, report.score)
print(f"{report.total_issues} issues ({report.critical_issues} critical)")
for issue in report.issues[:5]:
print(f"[{issue.severity}] {issue.finding}")
client.close()Client Configuration
api_key is a required positional argument. Everything else is keyword-only.
| Parameter | Type | Default | Description |
|---|---|---|---|
api_key | str | — (required) | Your API key (starts with cl_) or a JWT token |
base_url | str | https://api.compliancelayer.net | Override the API base URL, e.g. to point at a local server. Give the host only — the SDK appends /v1/... itself |
timeout | float | 30.0 | Per-request HTTP timeout, in seconds |
poll_interval | float | 3.0 | Seconds between status polls in scan() |
poll_timeout | float | 120.0 | Maximum seconds to wait for a scan to complete |
max_retries | int | 3 | Retry attempts on 429 and 5xx responses |
from compliancelayer import ComplianceLayer
client = ComplianceLayer(
"cl_your_api_key_here",
timeout=60.0,
poll_interval=5.0,
poll_timeout=300.0,
max_retries=5,
)/v1 or /api to base_url. The client builds paths like /v1/scan/ on top of whatever you pass, so adding a prefix yourself produces URLs such as /v1/v1/scan/ and every call 404s.Scanning
Blocking scan
scan(domain, *, poll_interval=None, poll_timeout=None) returns a ScanReport. The two keyword arguments override the client defaults for this call only. Raises ScanTimeoutError if the scan does not finish in time, or ScanError if it fails server-side.
report = client.scan("example.com", poll_interval=2.0, poll_timeout=180.0)
print(report.score) # 82
print(report.grade) # "B"
print(report.risk_level) # "low"
print(report.total_issues) # 7
# Per-module breakdown
for name, module in report.modules.items():
print(f"{name}: {module.grade} ({module.score})")Non-blocking scan
scan_async(domain) submits the job and returns a ScanJob immediately, without polling. Despite the name it is a normal synchronous method — it is the sync client's non-blocking submit. Call job.refresh() to update the status and job.get_report() once complete.
import time
job = client.scan_async("example.com")
print(job.job_id, job.status)
while not job.is_complete:
time.sleep(3)
job.refresh()
if job.is_failed:
print("Scan failed:", job.failure_reason)
else:
report = job.get_report()
print(report.grade, report.score)Fetching jobs and reports directly
job = client.get_scan_job(12345)
print(job.status)
# Raises ScanError (HTTP 409) if the scan has not finished yet
report = client.get_scan_report(12345)
print(report.grade)Scan history
# limit accepts 1-100, defaults to 50
for entry in client.scan_history(limit=20):
print(entry.scanned_at, entry.domain, entry.grade, entry.total_issues)Free scan
free_scan(domain) calls the unauthenticated, rate-limited endpoint and returns a FreeScanReport with the grade and top findings rather than a full report.
result = client.free_scan("example.com")
print(result.grade, result.score, result.risk_level)
for issue in result.top_issues:
print(f"[{issue.severity}] {issue.finding}")Batch Operations
batch_scan() scans up to 50 domains in a single request. compare() ranks several domains against each other. Both require a plan with batch access — see Batch Operations.
DomainScanResult entries carrying per-module scores, not the full issue list of a ScanReport. Call get_scan_report() when you need the detail for a specific domain.# sort_by is "risk" (default) or "domain"
batch = client.batch_scan(
["example.com", "example.org", "example.net"],
sort_by="risk",
)
print(f"{batch.total} domains scanned")
print(batch.summary) # aggregate counts returned by the API
for result in batch.results:
if result.error:
print(f"{result.domain}: failed -- {result.error}")
else:
print(result.domain, result.overall_grade, result.overall_score)Domain Monitoring
Domain operations live under client.domains.
| Method | Returns | Description |
|---|---|---|
domains.list() | DomainList | All monitored domains, plus usage against your domain limit |
domains.add(domain, ...) | MonitoredDomain | Register a domain for continuous monitoring |
domains.remove(domain_id) | None | Stop monitoring a domain, by numeric ID |
domains.scan(domain_id) | ScanJob | Trigger an immediate scan of a monitored domain |
domains.alerts(...) | List[Alert] | Fetch alerts across monitored domains |
Adding a domain
alert_threshold is the minimum score decrease that triggers an alert — a delta, not an absolute score. It defaults to 10, meaning “alert when the score drops by 10 points or more”. scan_interval accepts "hourly", "daily", or "weekly".
domain = client.domains.add(
"example.com",
scan_interval="daily", # "hourly" | "daily" | "weekly"
alert_on_score_drop=True,
alert_threshold=5, # alert on a drop of 5+ points
)
print(domain.id, domain.domain, domain.scan_frequency)Webhooks
Webhook operations live under client.webhooks. See the Webhooks guide for event payloads and signature verification.
| Method | Returns | Description |
|---|---|---|
webhooks.create(url, enabled_events, description="") | WebhookEndpoint | Create an endpoint; the response includes the signing secret |
webhooks.list() | List[WebhookEndpoint] | All endpoints on the account |
webhooks.get(endpoint_id) | WebhookEndpoint | Fetch a single endpoint by ID |
webhooks.update(endpoint_id, *, ...) | WebhookEndpoint | Update url, enabled_events, description, or is_active |
webhooks.delete(endpoint_id) | None | Delete an endpoint |
webhooks.test(endpoint_id) | WebhookTestResult | Send a test payload to the endpoint |
webhooks.deliveries(endpoint_id, limit=50) | List[WebhookDelivery] | Delivery history for an endpoint |
webhooks.event_types() | List[str] | Event types available to subscribe to |
endpoint = client.webhooks.create(
url="https://example.com/hooks/compliancelayer",
enabled_events=["scan.completed", "alert.triggered"],
description="Production listener",
)
# The signing secret is only returned on creation -- store it now.
print(endpoint.id, endpoint.secret)
# endpoint_id is the numeric ID
result = client.webhooks.test(endpoint.id)
print(result.success, result.status_code, result.response_time_ms)
for delivery in client.webhooks.deliveries(endpoint.id, limit=10):
print(delivery.created_at, delivery.event_type, delivery.status)
print(" HTTP", delivery.http_status_code, "retries:", delivery.retry_count)
if not delivery.success:
print(" error:", delivery.error_message)
client.webhooks.update(endpoint.id, is_active=False)Badges
badge_url() builds a badge URL without making a request; badge_svg() fetches the SVG markup. Both take a style keyword; the SDK defaults to "flat", the compact strip — the API's own default (no style parameter) is the larger ComplianceLayer card. Badges are opt-in: the URL returns 404 until the domain's owner has published a badge. See Security Badges.
url = client.badge_url("example.com")
print(url) # embed this in your README
svg = client.badge_svg("example.com")
with open("badge.svg", "w") as f:
f.write(svg)Account Usage
usage() reads GET /v1/auth/me and returns a Usage object describing your current quota. raw holds the unmodified response if you need a field the dataclass does not expose.
usage = client.usage()
print(usage.plan) # "professional"
print(usage.email)
print(usage.scans_used) # scans consumed this month
print(usage.scans_limit) # monthly scan allowance
print(usage.scans_remaining) # reported by the API, not computed
print(usage.domains_limit) # maximum monitored domains
print(usage.subscription_status)Error Handling
Every SDK exception inherits from APIError, which carries status_code and response_body. Catching APIError catches all of them.
| Exception | Raised when |
|---|---|
APIError | Base class; also raised directly for unmapped status codes and transport errors |
AuthenticationError | HTTP 401 — the API key is missing or invalid |
ForbiddenError | HTTP 403 — your plan does not allow this, or a limit was reached |
NotFoundError | HTTP 404 — the resource does not exist |
ValidationError | HTTP 400 or 422 — the request payload failed validation |
QuotaExceededError | HTTP 429 where the response indicates the scan quota is exhausted |
RateLimitError | HTTP 429 after retries are exhausted; exposes retry_after |
ScanError | The scan failed server-side, or a report was requested before completion (HTTP 409) |
ScanTimeoutError | Polling exceeded poll_timeout |
from compliancelayer import ComplianceLayer
from compliancelayer.exceptions import (
APIError,
AuthenticationError,
QuotaExceededError,
RateLimitError,
ScanError,
ScanTimeoutError,
ValidationError,
)
client = ComplianceLayer("cl_your_api_key_here")
try:
report = client.scan("example.com")
except AuthenticationError:
print("Check your API key.")
except QuotaExceededError as exc:
print("Monthly scan quota exhausted:", exc)
except RateLimitError as exc:
print(f"Rate limited; retry after {exc.retry_after}s")
except ValidationError as exc:
print("Invalid request:", exc)
except ScanTimeoutError:
print("Scan did not finish in time -- poll the job later.")
except ScanError as exc:
print("Scan failed:", exc)
except APIError as exc:
# Catch-all for anything else the API returned
print(f"API error {exc.status_code}: {exc}")max_retries. RateLimitError is only raised once those retries are exhausted; QuotaExceededError is raised immediately, since waiting will not help.Async Client
AsyncComplianceLayer takes the same constructor arguments and exposes the same methods as awaitables. It is an async context manager, so async with closes the connection pool for you.
import asyncio
from compliancelayer import AsyncComplianceLayer
async def main():
async with AsyncComplianceLayer("cl_your_api_key_here") as client:
report = await client.scan("example.com")
print(report.grade, report.score)
usage = await client.usage()
print(usage.scans_used, "/", usage.scans_limit)
asyncio.run(main())badge_url() is a plain method, not a coroutine, because it only builds a string. Everything that performs I/O — including badge_svg()— must be awaited.Response Models
Responses are dataclasses, not dictionaries. The most commonly used fields:
| Model | Key fields |
|---|---|
ScanReport | job_id, domain, score, grade, risk_level, modules, issues (aliased as findings), total_issues, critical_issues, high_issues, medium_issues, low_issues, recommendations, compliance |
ScanJob | job_id, domain, status, failure_reason, is_complete, is_failed, refresh(), get_report() |
Issue | severity, finding, remediation |
ModuleResult | score, grade, weight, issues, detail |
MonitoredDomain | id, domain, scan_frequency, alert_threshold, last_score, last_grade |
Alert | id, alert_type, severity, title, message, is_read, old_value, new_value |
DomainScanResult | domain, overall_score, overall_grade, modules, error — the per-domain entry in batch and compare results |
RankedDomain | rank, domain, overall_score, overall_grade |
Usage | scans_used, scans_limit, scans_remaining, domains_limit, plan, email, subscription_status, domains_used (usually None), raw |
WebhookEndpoint | id, url, enabled_events, is_active, secret (creation only), consecutive_failures |
WebhookDelivery | id, event_id, event_type, status, retry_count, http_status_code, error_message, sent_at, delivered_at, failed_at, and a success property (true when status is "delivered") |