Python SDK Reference

Package: cpod-sdk
PyPI: pypi.org/project/cpod-sdk
Source: github.com/cpod-ai/cpod-sdk/tree/main/sdks/python
Python: ≥ 3.9
Dependencies: pydantic>=2.0, httpx>=0.26

Clients

The Python SDK provides both synchronous and asynchronous clients with identical APIs.

from cpod import CpodClient, AsyncCpodClient
 
# Synchronous — for scripts, notebooks, Django sync views
client = CpodClient(
    api_key="cpod_live_xxx",   # or set CPOD_API_KEY env var
    base_url="https://api.cyberpod.app/v1",
    timeout=30.0,              # seconds
    max_retries=3,
)
 
# Asynchronous — for FastAPI, asyncio apps
async_client = AsyncCpodClient(
    api_key="cpod_live_xxx",
    timeout=30.0,
    max_retries=3,
)

The async client is a context manager and should be used with async with to ensure connection pools are properly cleaned up.

`PeopleService`

Accessed via client.people. Manages people entities — employees, contractors, and service accounts.

from cpod import AsyncCpodClient
from cpod.models import PersonType, PersonStatus
 
async with AsyncCpodClient() as client:
    # List people with optional filters
    result = await client.people.list(
        status=PersonStatus.ACTIVE,
        person_type=PersonType.EMPLOYEE,
        limit=50,
    )
    # result: PaginatedResponse[Person]
    # result.data: list[Person]
    # result.total: int
    # result.next_cursor: str | None
 
    # Get a single person
    person = await client.people.get("person_01jm8xk2y3z")
 
    # Create a person
    new_person = await client.people.create(
        email="alice@acme.com",
        first_name="Alice",
        last_name="Smith",
        person_type=PersonType.EMPLOYEE,
        department="Engineering",
        tags=["backend", "platform"],
    )
 
    # Update
    updated = await client.people.update(
        "person_01jm8xk2y3z",
        department="Platform Engineering",
    )
 
    # Deactivate (soft delete — sets status to inactive)
    await client.people.deactivate("person_01jm8xk2y3z")
 
    # Auto-paginated iterator
    async for person in client.people.list_all(status=PersonStatus.ACTIVE):
        print(f"{person.id}: {person.email}")

Sync usage

from cpod import CpodClient
 
with CpodClient() as client:
    page = client.people.list(person_type="contractor")
    for person in page.data:
        print(f"{person.first_name} {person.last_name} — {person.status}")

`GroupService`

Accessed via client.groups. Manages groups and team membership.

from cpod import AsyncCpodClient
 
async with AsyncCpodClient() as client:
    # List groups
    result = await client.groups.list(limit=25)
 
    # Get a group
    group = await client.groups.get("group_01abc")
 
    # Create a group
    new_group = await client.groups.create(
        name="Platform Team",
        description="Core platform engineering group",
        tags=["engineering", "platform"],
    )
 
    # Update
    await client.groups.update("group_01abc", name="Platform & Infra Team")
 
    # Member management
    await client.groups.add_member(
        "group_01abc",
        person_id="person_01jm8xk2y3z",
        role="member",  # 'owner' | 'member'
    )
    await client.groups.remove_member("group_01abc", "person_01jm8xk2y3z")
 
    # Delete
    await client.groups.delete("group_01abc")

`TechnologyService`

Accessed via client.technology. Manages technology assets (software, SaaS, infrastructure) and access entitlements.

from cpod import AsyncCpodClient
from cpod.models import AssetType, EntitlementStatus
 
async with AsyncCpodClient() as client:
    # List technology assets
    result = await client.technology.list(
        asset_type=AssetType.SAAS,
        limit=50,
    )
 
    # Get a technology asset
    asset = await client.technology.get("asset_01xyz")
 
    # Create a technology asset
    new_asset = await client.technology.create(
        name="GitHub Enterprise",
        asset_type=AssetType.SAAS,
        vendor="GitHub",
        description="Source code management platform",
        tags=["devtools", "vcs"],
    )
 
    # Update
    await client.technology.update("asset_01xyz", description="SCM and CI/CD platform")
 
    # Get entitlements for an asset
    entitlements = await client.technology.get_entitlements(
        "asset_01xyz",
        status=EntitlementStatus.ACTIVE,
    )
 
    # Grant an entitlement to a person
    await client.technology.grant_entitlement(
        asset_id="asset_01xyz",
        person_id="person_01jm8xk2y3z",
        role="member",
        expires_at="2027-01-01T00:00:00Z",
    )
 
    # Revoke an entitlement
    await client.technology.revoke_entitlement(
        asset_id="asset_01xyz",
        entitlement_id="ent_01abc",
    )

`LicenseService`

Accessed via client.licenses. Manages software licenses and seat assignments.

from cpod import AsyncCpodClient
 
async with AsyncCpodClient() as client:
    # List licenses
    result = await client.licenses.list(limit=50)
 
    # Get a license
    license = await client.licenses.get("lic_01abc")
 
    # Create a license
    new_license = await client.licenses.create(
        name="Datadog Pro",
        vendor="Datadog",
        seat_count=150,
        expires_at="2027-06-30T00:00:00Z",
        cost_usd=45000.00,
    )
 
    # Update
    await client.licenses.update("lic_01abc", seat_count=200)
 
    # Get current assignments
    assignments = await client.licenses.get_assignments("lic_01abc")
    # assignments.data: list[LicenseAssignment]
    # assignments.seats_used: int
    # assignments.seats_total: int
 
    # Assign a seat to a person
    await client.licenses.assign(
        license_id="lic_01abc",
        person_id="person_01jm8xk2y3z",
    )
 
    # Revoke a seat
    await client.licenses.revoke(
        license_id="lic_01abc",
        assignment_id="asgn_01xyz",
    )

`AssetService`

Accessed via client.assets. Manages physical assets and cloud resources.

from cpod import AsyncCpodClient
 
async with AsyncCpodClient() as client:
    # List physical assets
    physical = await client.assets.list_physical(
        location="HQ-NYC",
        limit=50,
    )
    # physical.data: list[PhysicalAsset]
 
    # Get a physical asset
    asset = await client.assets.get_physical("asset_01abc")
 
    # Create a physical asset
    new_asset = await client.assets.create_physical(
        name="MacBook Pro 16\" — Alice Smith",
        asset_tag="HWNY-2024-0042",
        serial_number="C02ZF1XKMD6T",
        location="HQ-NYC",
        assigned_to="person_01jm8xk2y3z",
        purchased_at="2024-03-15",
        tags=["laptop", "apple"],
    )
 
    # List cloud resources
    cloud = await client.assets.list_cloud(
        provider="aws",
        region="us-east-1",
        limit=100,
    )
    # cloud.data: list[CloudResource]
 
    # Get a cloud resource
    resource = await client.assets.get_cloud("cloud_01xyz")
 
    # Update an asset (works for both PhysicalAsset and CloudResource)
    await client.assets.update("asset_01abc", location="HQ-SF")

`RiskService`

Accessed via client.risk. Query vulnerabilities, compliance controls, and risk items across the EDM.

from cpod import AsyncCpodClient
from cpod.models import Severity, VulnerabilityStatus, Framework
 
async with AsyncCpodClient() as client:
    # List vulnerabilities with severity filter
    vulns = await client.risk.list_vulnerabilities(
        severity=Severity.CRITICAL,
        status=VulnerabilityStatus.OPEN,
        limit=50,
    )
    # vulns.data: list[Vulnerability]
 
    # Get a single vulnerability
    vuln = await client.risk.get_vulnerability("vuln_01abc")
    print(f"CVE: {vuln.cve_id} — {vuln.severity} — {vuln.status}")
    print(f"Affected assets: {vuln.affected_asset_ids}")
 
    # Update vulnerability status (e.g. after remediation)
    await client.risk.update_vulnerability_status(
        "vuln_01abc",
        status=VulnerabilityStatus.REMEDIATED,
        notes="Patched via Dependabot PR #412",
    )
 
    # List compliance controls by framework
    controls = await client.risk.list_controls(
        framework=Framework.SOC2,
        limit=100,
    )
    # controls.data: list[ComplianceControl]
 
    # Get a single compliance control
    control = await client.risk.get_control("ctrl_01xyz")
    print(f"{control.framework} / {control.control_id}: {control.title}")
    print(f"Status: {control.status} — Owner: {control.owner_id}")
 
    # List risk items
    risk_items = await client.risk.list_risk_items(
        severity=Severity.HIGH,
        limit=50,
    )
    # risk_items.data: list[RiskItem]
 
    # Get a single risk item
    item = await client.risk.get_risk_item("risk_01abc")
    print(f"Risk: {item.title} — Score: {item.risk_score}")

Key models

from cpod.models import (
    Vulnerability, VulnerabilityStatus,
    Severity,        # critical | high | medium | low
    ComplianceControl,
    Framework,       # soc2 | iso27001 | nist_csf | pci_dss | hipaa | gdpr | cis
    RiskItem,
)
 
class Severity(str, Enum):
    CRITICAL = "critical"
    HIGH     = "high"
    MEDIUM   = "medium"
    LOW      = "low"
 
class Framework(str, Enum):
    SOC2      = "soc2"
    ISO27001  = "iso27001"
    NIST_CSF  = "nist_csf"
    PCI_DSS   = "pci_dss"
    HIPAA     = "hipaa"
    GDPR      = "gdpr"
    CIS       = "cis"

`RelationshipService`

Accessed via client.relationships. Manage typed graph edges between any two EDM entities.

from cpod import AsyncCpodClient
 
async with AsyncCpodClient() as client:
    # List relationships for an entity
    rels = await client.relationships.list(
        source_id="person_01jm8xk2y3z",
        limit=50,
    )
    # rels.data: list[Relationship]
 
    # Create a relationship
    rel = await client.relationships.create(
        source_id="person_01jm8xk2y3z",
        source_type="person",
        target_id="group_01abc",
        target_type="group",
        relationship_type="member_of",
        metadata={"since": "2024-01-15"},
    )
 
    # Delete a relationship
    await client.relationships.delete("rel_01xyz")
 
    # Fetch the full entity graph (N hops)
    graph = await client.relationships.get_graph(
        entity_id="person_01jm8xk2y3z",
        depth=2,
    )
    # graph.nodes: list[GraphNode]
    # graph.edges: list[Relationship]

`DataSourceService`

Accessed via client.data_sources. Manage data source integrations and trigger syncs.

from cpod import AsyncCpodClient
 
async with AsyncCpodClient() as client:
    # List data sources
    sources = await client.data_sources.list(limit=50)
    # sources.data: list[DataSource]
 
    # Get a single data source
    source = await client.data_sources.get("ds_01abc")
    print(f"{source.name} ({source.connector_type}) — {source.status}")
 
    # Create a data source
    new_source = await client.data_sources.create(
        name="Okta Production",
        connector_type="okta",
        config={
            "domain": "acme.okta.com",
            "api_token_secret": "vault:secret/okta/token",
        },
        sync_schedule="0 */6 * * *",  # every 6 hours
    )
 
    # Trigger a manual sync
    sync_job = await client.data_sources.sync("ds_01abc")
    print(f"Sync job started: {sync_job.job_id}")
 
    # Poll sync status
    status = await client.data_sources.get_status("ds_01abc")
    print(f"Last sync: {status.last_synced_at} — {status.records_synced} records")

Pydantic Models

All SDK response objects are Pydantic v2 models. You get:

.model_dump() — serialize to dict
.model_dump_json() — serialize to JSON string
Full type annotations and IDE autocomplete
Runtime validation on deserialization

from cpod.models import (
    # People
    Person, PersonType, PersonStatus,
    # Technology
    TechnologyAsset, AssetType, AccessEntitlement, EntitlementStatus,
    # Licenses
    SoftwareLicense, LicenseAssignment,
    # Assets
    PhysicalAsset, CloudResource,
    # Risk
    Vulnerability, Severity, VulnerabilityStatus,
    ComplianceControl, Framework, RiskItem,
    # Common
    Relationship, DataSource, Tag,
    ApiResponse, PaginatedResponse,
)
 
person: Person
print(person.id)            # str
print(person.email)         # str
print(person.status)        # PersonStatus.ACTIVE
print(person.person_type)   # PersonType.EMPLOYEE
print(person.created_at)    # datetime (UTC-aware)
print(person.model_dump())  # {'id': ..., 'status': 'active', ...}

Key enums

class PersonType(str, Enum):
    EMPLOYEE        = "employee"
    CONTRACTOR      = "contractor"
    SERVICE_ACCOUNT = "service_account"
 
class PersonStatus(str, Enum):
    ACTIVE   = "active"
    INACTIVE = "inactive"
    PENDING  = "pending"

Error Handling

from cpod.errors import (
    CpodError,
    NotFoundError,
    ValidationError,
    AuthError,
    RateLimitError,
)
 
async with AsyncCpodClient() as client:
    try:
        person = await client.people.get("person_nonexistent")
    except NotFoundError as e:
        print(f"404: {e.message}")
    except ValidationError as e:
        for field, msg in e.fields.items():
            print(f"  {field}: {msg}")
    except AuthError as e:
        print(f"Auth error ({e.status}): {e.message}")
    except RateLimitError as e:
        print(f"Rate limited. Retry in {e.retry_after}s")
    except CpodError as e:
        print(f"API error {e.status} [{e.code}]: {e.message}")

TypeScript Go