Python SDK
The official Python client for the Cybros API. Sync and async, fully typed with pydantic models, with polling helpers for scans, AI analyses, and remediations.
Install
pip install cybros # Python 3.10+
Installing the package also puts the cybros CLI on your PATH.
Client
from cybros import Cybros
client = Cybros(api_key="cybros_sk_...") # or resolved from env / config
me = client.me()
print(me.email, me.active_org_id, me.role)
Constructor
Cybros(
api_key: str | None = None,
base_url: str | None = None,
*,
timeout: float = 30.0,
http_client: httpx.Client | None = None,
)
| Arg | Default | Description |
|---|---|---|
api_key | resolved (env → config) | cybros_sk_... key. See Authentication. |
base_url | resolved → prod default | API base URL. |
timeout | 30.0 | Per-request timeout (seconds). |
http_client | a new httpx.Client | Inject your own httpx client (proxies, retries…). |
Cybros is a context manager — use with to close the underlying HTTP client:
with Cybros() as client:
scan = client.scans.create("repo_123", ref="main")
Authentication
Credentials resolve explicit arg → env (CYBROS_API_KEY / CYBROS_API_URL) →
~/.cybros/config.toml → prod default:
Cybros(api_key="cybros_sk_...") # explicit
# CYBROS_API_KEY=... in the environment # env
# after `cybros login` # config file
The default base URL is https://backend.cybros.hacktigerlabs.com/api/v1. See
Authentication for the complete precedence table.
Top-level
client.me() -> MeResponse
Verify auth and return the current principal (user, active org, role, permissions, organizations).
me = client.me()
print(me.user_id, me.role, me.permissions)
for org in me.organizations:
print(org.name, org.plan)
Resources
Every method returns a typed pydantic model. Models set extra="allow", so new
server fields never break the client.
client.scans
| Method | Returns | Endpoint |
|---|---|---|
create(repository_id, *, ref="HEAD", modules=None, frameworks=None, is_incremental=True) | Scan | POST /scans |
list(*, repository_id=None, limit=20, offset=0) | Page[Scan] | GET /scans |
get(scan_id) | Scan | GET /scans/{id} |
report(scan_id) | ScanReport | GET /scans/{id}/report |
findings(scan_id, *, severity=None, limit=50, offset=0) | Page[Finding] | GET /findings?scan_id= |
cancel(scan_id) | Scan | POST /scans/{id}/cancel |
wait(scan_id, *, timeout=600.0, poll_interval=2.0, on_progress=None) | Scan | polls GET /scans/{id} |
scan = client.scans.create("repo_123", ref="main", modules=["sast", "secrets"])
scan = client.scans.wait(scan.id)
report = client.scans.report(scan.id)
print(report.total_findings, report.risk_score, report.severity_breakdown)
for f in client.scans.findings(scan.id, severity="high"):
print(f.severity.value, f.title, f.file_path, f.line)
Findings for a scan are fetched via
GET /findings?scan_id=— there is no/scans/{id}/findingsendpoint.
client.ai
| Method | Returns | Endpoint |
|---|---|---|
analyze(scan_id) | AIAnalysis | POST /scans/{id}/ai-analysis |
get(scan_id) | AIAnalysis | GET /scans/{id}/ai-analysis |
patches(scan_id) | dict | GET /scans/{id}/ai-analysis/patches |
wait(scan_id, *, timeout=600.0, poll_interval=3.0, on_progress=None) | AIAnalysis | polls the GET endpoint |
analysis = client.ai.analyze(scan.id)
analysis = client.ai.wait(scan.id)
print("risk:", analysis.risk_score, "confidence:", analysis.overall_confidence)
for run in sorted(analysis.agent_runs, key=lambda r: r.sequence):
print(run.sequence, run.agent_id, run.status, run.confidence)
client.remediation
| Method | Returns | Endpoint |
|---|---|---|
create(finding_id) | RemediationJob | POST /findings/{id}/remediate |
get(job_id) | RemediationJob | GET /remediations/{id} |
wait(job_id, *, timeout=600.0, poll_interval=3.0, on_progress=None) | RemediationJob | polls GET /remediations/{id} |
job = client.remediation.create(finding_id)
job = client.remediation.wait(job.id)
if job.pr_url:
print("PR:", job.pr_url, "branch:", job.branch)
client.findings
| Method | Returns | Endpoint |
|---|---|---|
list(*, scan_id=None, severity=None, limit=20, offset=0) | Page[Finding] | GET /findings |
get(finding_id) | FindingDetail | GET /findings/{id} |
detail = client.findings.get(finding_id)
print(detail.owasp, detail.cwe, detail.cvss_score)
if detail.fix and detail.fix.unified_diff:
print(detail.fix.unified_diff)
client.compliance
| Method | Returns | Endpoint |
|---|---|---|
overview(*, scan_id=None) | ComplianceOverview | GET /compliance/overview |
framework(framework_id, *, scan_id=None) | FrameworkDetail | GET /compliance/{id} |
overview = client.compliance.overview()
print(overview.overall_score)
soc2 = client.compliance.framework("soc2")
for c in soc2.controls:
print(c.control_id, c.status, c.title)
client.ai_security
| Method | Returns | Endpoint |
|---|---|---|
overview(*, scan_id=None) | AiSecurityOverview | GET /ai-security/overview |
posture = client.ai_security.overview()
print(posture.summary.checks_failed, "/", posture.summary.checks_total)
for chk in posture.checks:
print(chk.name, chk.owasp_llm, chk.status)
client.audit
| Method | Returns | Endpoint |
|---|---|---|
list(*, action=None, actor_id=None, target_type=None, limit=20, offset=0) | Page[AuditLog] | GET /audit-logs |
for entry in client.audit.list(action="scan.created", limit=50):
print(entry.created_at, entry.action, entry.actor_id)
client.repositories
| Method | Returns | Endpoint |
|---|---|---|
list(*, project_id=None, limit=100, offset=0) | Page[Repository] | GET /repositories |
find_by_full_name(full_name) | `Repository | None` |
repo = client.repositories.find_by_full_name("acme/api")
if repo:
scan = client.scans.create(repo.id, ref=repo.default_branch)
client.api_keys
| Method | Returns | Endpoint |
|---|---|---|
create(name, *, role="member") | ApiKeyCreated | POST /api-keys |
list() | list[ApiKey] | GET /api-keys |
revoke(key_id) | None | DELETE /api-keys/{id} |
created = client.api_keys.create("ci-token", role="member")
print(created.token) # shown once
client.api_keys.revoke("key_1a2b...")
client.billing
| Method | Returns | Endpoint |
|---|---|---|
usage() | UsageOverview | GET /billing/usage |
usage = client.billing.usage()
print(usage.remaining_credits)
Waiting on long-running jobs
scans.wait, ai.wait, and remediation.wait poll until the job reaches a
terminal state, calling an optional on_progress callback on each poll:
def on_progress(scan):
done = sum(1 for m in scan.modules if m.status in {"completed", "failed"})
print(f"{scan.status} — {done}/{len(scan.modules)} modules")
scan = client.scans.wait(scan.id, timeout=600, poll_interval=2.0, on_progress=on_progress)
If the timeout elapses before the job is terminal, a CybrosError is raised.
Pagination
List endpoints return a Page[T] — iterable, with total, limit, and
offset:
page = client.findings.list(severity="high", limit=50)
print(len(page), "of", page.total)
for finding in page: # Page is directly iterable
...
# Manual paging:
offset = 0
while True:
page = client.scans.list(limit=100, offset=offset)
for scan in page.items:
...
offset += page.limit
if offset >= page.total:
break
Error handling
Every failure raises a subclass of cybros.CybrosError:
| Exception | HTTP |
|---|---|
AuthError | 401 / 403 |
NotFoundError | 404 |
ValidationError | 400 / 422 |
RateLimitError | 429 |
ServerError | 5xx |
ConfigError | local (e.g. no API key configured) |
from cybros import Cybros, AuthError, NotFoundError, CybrosError
try:
client.scans.get("nope")
except NotFoundError:
...
except AuthError as e:
print("re-authenticate:", e)
except CybrosError as e:
print(e.status_code, e.request_id, e.body)
CybrosError carries .message, .status_code, .body, and .request_id.
Async
AsyncCybros mirrors the sync client method-for-method; await every call and
use async with:
import asyncio
from cybros import AsyncCybros
async def main():
async with AsyncCybros(api_key="cybros_sk_...") as client:
me = await client.me()
scan = await client.scans.create("repo_123", ref="main")
scan = await client.scans.wait(scan.id)
page = await client.scans.findings(scan.id, severity="high")
for f in page.items:
print(f.title)
asyncio.run(main())
on_progress callbacks may be sync or async in the async client.
Types
Response models live in cybros.models: MeResponse, Repository, Scan,
ScanReport, Finding, FindingDetail, AIAnalysis, AgentRun,
RemediationJob, ComplianceOverview, FrameworkDetail, AiSecurityOverview,
AuditLog, ApiKey, ApiKeyCreated, UsageOverview, and Page[T]. The
Severity enum has critical/high/medium/low/info.
See also
- CLI reference — the same surface from your shell.
- TypeScript SDK.
- Concepts.