DocumentationSDK ReferenceOverview

SDK Reference

The SDK is a REST client for cpod-backend. It does not talk to the sidecar directly — the sidecar is an internal implementation detail of the backend. Configure three environment variables and call the API.

The SDK auto-manages OAuth tokens: it fetches a client_credentials JWT on first use, caches it, and refreshes at 80% of the TTL. You never call the token endpoint manually.

Initialization

import { CpodClient } from '@cpod/sdk'
 
// Reads CPOD_API_KEY (and optionally CPOD_BASE_URL) from environment
const client = CpodClient.fromEnv()
 
// Or configure explicitly:
const client = new CpodClient({
  apiKey: 'cpod_xxxxxxxxxxxxxxxxxxxx',
  baseUrl: 'https://api.yourdomain.com', // optional — default: https://api.cyberpod.app
})

Auth

Validates a Bearer token and checks policy. Internally cpod-backend calls AuthService.Authorize on the sidecar — the SDK calls the backend REST endpoint.

authorize(token, action, resource)

Validates the JWT against your IdP and evaluates Rego policy. Returns an AuthDecision.

decision = await sdk.authorize(
    "Bearer eyJhbGci...",
    action="read",
    resource="/api/orders",
)
 
print(decision.allowed)   # bool
print(decision.claims)    # dict — decoded JWT claims
print(decision.reason)    # str — why allowed/denied

validate_token(token) / revoke_token(token)

# Validate without policy evaluation
claims = await sdk.validate_token("Bearer eyJhbGci...")
 
# Revoke a token (adds to deny-list in sidecar)
await sdk.revoke_token("Bearer eyJhbGci...")

JobService

Maps to the JobService gRPC service. Key methods: SubmitJob, GetJob, ListJobs, CancelJob, WatchJob (server-stream), GetJobLogs (server-stream), GetJobOutput.

The Job service runs arbitrary containerized workloads asynchronously. Submit a job with an image and command, stream progress events, then retrieve output files via presigned URLs.

run_job() / jobs.submit() + jobs.watch()

result = await sdk.run_job(
    kind="hello",
    image="busybox:1.36",
    command=["sh", "-lc"],
    args=['echo "hello"'],
    env={"GREETING": "howdy"},
    inline_files={"prompt.txt": b"hello from the caller"},
    user_id="alice@example.com",
    timeout_seconds=30,
    on_progress=lambda ev: print(f"  • {ev.kind} {ev.stage}"),
)
print(result.status, result.exit_code)
for f in result.output.files:
    print(f.key, f.presigned_url)

With secrets injection — inject a secret bundle as environment variables:

result = await sdk.run_job(
    kind="hello",
    image="busybox:1.36",
    command=["sh", "-lc"],
    args=['echo "hello"'],
    user_id="alice@example.com",
    timeout_seconds=30,
    on_progress=lambda ev: print(f"  • {ev.kind} {ev.stage}"),
    secret_bundles=["demo"],  # secrets from the "demo" bundle injected as env vars
)

WatchJob and GetJobLogs are server-streaming gRPC calls — the sidecar pushes events to the SDK as the job progresses. on_progress in Python wraps this stream behind a callback for convenience.

PolicyService

Maps to the PolicyService gRPC service. Key methods: Evaluate, DryRun, WatchPolicyUpdates (server-stream).

evaluate_policy(path, input)

Evaluate a specific Rego rule directly. Useful for custom policy checks beyond the standard auth flow.

result = await sdk.evaluate_policy(
    path="authz/feature_enabled",
    input={
        "user_id": "u_123",
        "feature": "dark_mode",
        "roles": ["viewer"],
    },
)
 
print(result.result)   # Any — the value returned by the Rego rule
print(result.allowed)  # bool — True if result is truthy

SecretsService

Maps to the SecretsService gRPC service. Key method: ResolveSecret.

resolve_secret(name)

Resolve a secret by name from the Vault-backed secrets store. Used at startup or in job setup — prefer secret_bundles in run_job() for injecting secrets into containerized workloads.

secret = await sdk.resolve_secret("db/password")
print(secret.value)   # str — the resolved secret value

MaskingService

Maps to the MaskingService gRPC service. Key methods: Mask, MaskString.

mask() / mask_string()

Mask PII and sensitive fields before logging, storing, or returning data to clients.

# Mask a structured object
masked = await sdk.mask({
    "name": "Alice Smith",
    "email": "alice@example.com",
    "ssn": "123-45-6789",
})
# masked["email"] -> "a***@example.com"
# masked["ssn"]   -> "***-**-****"
 
# Mask a raw string
safe = await sdk.mask_string("Call Alice at 555-867-5309")
# safe -> "Call [NAME] at [PHONE]"

AuditService

Maps to the AuditService gRPC service. Key methods: EmitAuditEvent, QueryAudit.

emit_audit_event(event)

Emit a structured audit event. The sidecar persists and forwards it to your configured audit sink.

await sdk.emit_audit_event({
    "actor": "alice@example.com",
    "action": "delete",
    "resource": "/api/documents/doc_123",
    "outcome": "allow",
    "metadata": {"doc_id": "doc_123", "ip": "10.0.0.1"},
})

FlagService

Maps to the FlagService gRPC service. Key method: EvaluateFlag.

evaluate_flag(flag, context)

Evaluate a feature flag against the given context. Returns a boolean decision from the sidecar.

enabled = await sdk.evaluate_flag("dark_mode", {
    "user_id": "alice@example.com",
    "plan": "pro",
})
if enabled:
    return render_dark_theme()

RateLimitService

Maps to the RateLimitService gRPC service. Key methods: CheckRateLimit, ResetRateLimit.

check_rate_limit(key, limit, window)

Check a token-bucket rate limit for the given key. Returns whether the request is allowed and the remaining quota.

result = await sdk.check_rate_limit(
    key=f"user:{user_id}:api",
    limit=100,
    window_seconds=60,
)
if not result.allowed:
    raise HTTPException(status_code=429, detail="Rate limit exceeded")
print(result.remaining)  # int — remaining tokens in this window

Environment Variables

These are the env vars your application sets to configure the SDK (i.e., where to find cpod-backend):

VariableDefaultDescription
CPOD_API_URLhttp://localhost:8000cpod-backend REST API base URL
CPOD_CLIENT_ID—OAuth client_id for your app
CPOD_CLIENT_SECRET—OAuth client_secret (store in secrets manager, never in code)
CPOD_TENANT_IDdefaultTenant identifier

The following are internal cpod-backend config — set by ops, not by SDK consumers:

VariableDescription
CORESDK_SIDECAR_ADDRgRPC address of the sidecar (cpod-backend → sidecar, loopback: [::1]:50051)
CORESDK_ENVSet to development to disable JWT validation (never in production)
CORESDK_JWKS_URIIdP JWKS endpoint for production JWT validation
CORESDK_CONTROL_PLANE_URLControl plane REST API — internal admin only

Error Types

from cpod.errors import (
    CpodError,        # Base error class
    NotFoundError,    # 404 — entity not found
    AuthError,        # JWT invalid or expired
    ValidationError,  # 422 — invalid input fields
    RateLimitError,   # 429 — rate limit exceeded
)
 
try:
    vuln = await sdk.risk.get_vulnerability("vuln-nonexistent")
except NotFoundError as e:
    print(f"Not found: {e.message}")
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}")

PolicyDeniedError is only raised when the sidecar returns a hard denial. If AuthDecision.allowed is false without an error, the policy evaluated successfully and returned deny — inspect decision.reason for details.

Security ChecklistStorage