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,
)
ArgDefaultDescription
api_keyresolved (env → config)cybros_sk_... key. See Authentication.
base_urlresolved → prod defaultAPI base URL.
timeout30.0Per-request timeout (seconds).
http_clienta new httpx.ClientInject 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

MethodReturnsEndpoint
create(repository_id, *, ref="HEAD", modules=None, frameworks=None, is_incremental=True)ScanPOST /scans
list(*, repository_id=None, limit=20, offset=0)Page[Scan]GET /scans
get(scan_id)ScanGET /scans/{id}
report(scan_id)ScanReportGET /scans/{id}/report
findings(scan_id, *, severity=None, limit=50, offset=0)Page[Finding]GET /findings?scan_id=
cancel(scan_id)ScanPOST /scans/{id}/cancel
wait(scan_id, *, timeout=600.0, poll_interval=2.0, on_progress=None)Scanpolls 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}/findings endpoint.

client.ai

MethodReturnsEndpoint
analyze(scan_id)AIAnalysisPOST /scans/{id}/ai-analysis
get(scan_id)AIAnalysisGET /scans/{id}/ai-analysis
patches(scan_id)dictGET /scans/{id}/ai-analysis/patches
wait(scan_id, *, timeout=600.0, poll_interval=3.0, on_progress=None)AIAnalysispolls 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

MethodReturnsEndpoint
create(finding_id)RemediationJobPOST /findings/{id}/remediate
get(job_id)RemediationJobGET /remediations/{id}
wait(job_id, *, timeout=600.0, poll_interval=3.0, on_progress=None)RemediationJobpolls 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

MethodReturnsEndpoint
list(*, scan_id=None, severity=None, limit=20, offset=0)Page[Finding]GET /findings
get(finding_id)FindingDetailGET /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

MethodReturnsEndpoint
overview(*, scan_id=None)ComplianceOverviewGET /compliance/overview
framework(framework_id, *, scan_id=None)FrameworkDetailGET /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

MethodReturnsEndpoint
overview(*, scan_id=None)AiSecurityOverviewGET /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

MethodReturnsEndpoint
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

MethodReturnsEndpoint
list(*, project_id=None, limit=100, offset=0)Page[Repository]GET /repositories
find_by_full_name(full_name)`RepositoryNone`
repo = client.repositories.find_by_full_name("acme/api")
if repo:
    scan = client.scans.create(repo.id, ref=repo.default_branch)

client.api_keys

MethodReturnsEndpoint
create(name, *, role="member")ApiKeyCreatedPOST /api-keys
list()list[ApiKey]GET /api-keys
revoke(key_id)NoneDELETE /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

MethodReturnsEndpoint
usage()UsageOverviewGET /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:

ExceptionHTTP
AuthError401 / 403
NotFoundError404
ValidationError400 / 422
RateLimitError429
ServerError5xx
ConfigErrorlocal (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