Python SDK

pip install atlasent

import os
from atlasent import protect, AtlaSentDeniedError, AtlaSentError

# Picks up ATLASENT_API_KEY from the environment.
try:
    permit = protect(
        agent="agent-001",
        action="ehr.write",
        context={"environment": "production", "patient_id": "patient-42"},
    )
except AtlaSentDeniedError as exc:
    halt(exc.reason)               # policy denied / hold / escalate / verify-fail
except AtlaSentError:
    halt("fail-closed")            # transport / timeout / auth / 5xx

# Authorized end-to-end. Execute and persist permit.permit_id alongside
# your own record so the audit chain stays two-way navigable.
db.patients.update("patient-42", payload, atlasent_permit_id=permit.permit_id)

from atlasent import with_permit

result = with_permit(
    agent="agent-001",
    action="ehr.write",
    context={"environment": "production", "patient_id": "patient-42"},
    fn=lambda permit: db.patients.update(
        "patient-42", payload, atlasent_permit_id=permit.permit_id,
    ),
)

from atlasent import require_permit, ProtectedAction

await require_permit(
    ProtectedAction(
        action_type="database.table.drop",
        actor_id="agent:code-agent",
        resource_id="prod-db.users",
        environment="production",
        context={"reversibility": "irreversible"},
    ),
    execute=lambda: db.raw_async("DROP TABLE users"),
)

from atlasent import AtlaSentDeniedError, AtlaSentError
from atlasent.exceptions import RateLimitError, StreamTimeoutError

try:
    permit = protect(agent=..., action=..., context=...)
except AtlaSentDeniedError as e:
    halt(e.reason)                 # policy said no, or permit verify-fail
except RateLimitError as e:
    backoff(e.retry_after_s)       # 429
except AtlaSentError:
    halt("fail-closed")            # transport / auth / 5xx / timeout

from atlasent import evaluate

result = evaluate(action_type="ehr.write", actor_id="agent-001", context={...})
match result.decision:
    case "allow":    ...  # call verify_permit_by_id(), then execute
    case "deny":     ...  # surface result.deny_reason
    case "hold":     ...  # wait for the approval webhook
    case "escalate": ...  # routed to a human reviewer

from atlasent import AsyncAtlaSentClient
from atlasent.aio import with_permit as awith_permit

async with AsyncAtlaSentClient() as client:
    result = await awith_permit(
        client,
        agent="agent-001",
        action="ehr.write",
        context={"environment": "production"},
        fn=lambda permit: do_async_work(permit.permit_id),
    )

from atlasent import AtlaSentClient
from atlasent.scim_client import ScimClient

client = AtlaSentClient(api_key="ask_live_...")
scim = ScimClient(client)

# List provisioned users
page = scim.users.list(org_id="org_abc", count=50)
for user in page["Resources"]:
    print(user["userName"], user["active"])

# Provision a new user
new_user = scim.users.create(org_id="org_abc", user={
    "userName": "alice@example.com",
    "active": True,
    "name": {"givenName": "Alice", "familyName": "Smith"},
    "emails": [{"value": "alice@example.com", "primary": True}],
})

# Deprovision
scim.users.delete(org_id="org_abc", user_id=new_user["id"])

from atlasent import AtlaSentClient
from atlasent.auth import refresh_token, refresh_with_idp, list_idp_connections

client = AtlaSentClient(api_key="ask_live_...")

# List available IdP connections
connections = list_idp_connections(client)
# [{"id": "idp_okta_prod", "name": "Okta Production", "provider": "okta", ...}]

# Refresh using the default IdP
tokens = refresh_token(client, current_refresh_token)
# {"access_token": "at_...", "refresh_token": "rt_...", "expires_in": 3600}

# Refresh against a specific IdP (multi-IdP orgs)
tokens = refresh_with_idp(client, idp_id="idp_okta_prod",
                           refresh_token=current_refresh_token)

from atlasent.evidence_bundle import (
    create_evidence_bundle,
    get_evidence_bundle,
    download_evidence_bundle,
)

bundle = create_evidence_bundle(
    client,
    org_id="org_abc",
    incident_id="inc_001",
    include_overrides=True,
)

# Poll until ready
while bundle["status"] == "generating":
    import time; time.sleep(2)
    bundle = get_evidence_bundle(client, "org_abc", bundle["bundle_id"])

# Download as PDF
pdf_bytes = download_evidence_bundle(
    client, "org_abc", bundle["bundle_id"], format="pdf"
)
with open("evidence.pdf", "wb") as f:
    f.write(pdf_bytes)