Python SDK

Asynchrones Python-SDK für die Audact-Compliance-Validierungsplattform.

Voraussetzungen

• Python 3.11+
• httpx für asynchrones HTTP
• pydantic v2 für Typsicherheit

Installation

Das Python-SDK ist noch nicht veröffentlicht. Die Installationsanleitung erscheint hier, sobald es so weit ist.

Schnellstart

import asyncio
import uuid
from audact import AudactClient

async def main():
    async with AudactClient(api_key="aud_live_...") as client:
        # Validate a conversation turn
        result = await client.validate(
            call_id=uuid.uuid4(),
            tenant_id=uuid.UUID("..."),
            config_id=uuid.UUID("..."),
            turn=1,
            content="Hello, this is an automated message.",
        )
        print(result.verdict)  # "ALLOW" | "BLOCK" | "MODIFY"

asyncio.run(main())

Basis-URL

Den Client mit base_url auf einen einzelnen Host zeigen; jeder Dienst löst sich darüber auf, und das SDK fügt die Pfade /v1/... selbst an (ein abschließendes / oder /v1 in base_url wird entfernt):

client = AudactClient(api_key="aud_live_...", base_url="https://api.audact.ai")

Die Basis-URL kann ohne Codeänderung über die Umgebungsvariable überschrieben werden:

export AUDACT_BASE_URL=https://api.audact.ai

Auflösungsreihenfolge je Dienst: <service>_url=-Kwarg → AUDACT_<SERVICE>_URL-Umgebung → base_url=-Kwarg → AUDACT_BASE_URL-Umgebung. Dienstspezifische URLs werden nur für Deployments benötigt, bei denen jeder Dienst auf einem eigenen Host läuft (siehe Konfiguration).

Konfiguration

Dienst-Basis-URLs über Konstruktor-Schlüsselwortargumente oder Umgebungsvariablen überschreiben:

max_retries (Konstruktor-Kwarg, Standardwert 0) aktiviert einen begrenzten Wiederholungsversuch mit Backoff für idempotente GET-Anfragen und 429/5xx-Antworten — siehe Wiederholungsversuche & Rate-Limits weiter unten.

Abdeckung

Das SDK deckt die dokumentierte API vollständig ab: 209 von 211 Operationen sind abgedeckt. Die verbleibenden 2 sind empfangsseitige Provider-zu-Plattform-Webhooks — POST /telnyx/webhook und POST /webhooks/livekit — hierbei handelt es sich um Ingress-Endpunkte, die Audact empfängt, aber nie aufruft.

Das SDK ist monolithisch mit attributbasierten Sub-Clients: Jede bestehende flache Methode auf AudactClient (z. B. client.validate, client.evaluate, client.create_tenant) funktioniert weiterhin unverändert, und die großen neuen Oberflächen für die vollständige Abdeckung sind als client.<name>.*-Sub-Clients verfügbar:

Authentifizierungsabläufe

Die meisten Methoden senden Authorization: Bearer <api_key> mit dem api_key, der dem AudactClient-Konstruktor übergeben wurde. Zwei Oberflächen verwenden einen anderen Bearer-Token (Org und Plattform):

• OrgBearer — ein JWT, der von client.org.login() bezogen wird. Das SDK speichert ihn automatisch im Arbeitsspeicher des Clients; jeder nachfolgende client.org.*-Aufruf sendet ihn anstelle des API-Schlüssels.
• Plattform-Token — ein Super-Admin-Token, der für export_tenant / import_tenant erforderlich ist. Als platform_token= an den Konstruktor übergeben oder jederzeit mit client.set_platform_token(token) setzen.

async with AudactClient(api_key="aud_live_...") as client:
    # login() is public (no bearer required) and stores the returned access
    # token on client.org automatically.
    await client.org.login(OrgLoginRequest(email="admin@example.com", password="hunter2"))

    # Subsequent calls send `Authorization: Bearer <jwt>`.
    profile = await client.org.get_profile()
    tenants = await client.org.list_tenants()

    # Rotate the token before it expires, then clear it when done.
    await client.org.refresh()
    await client.org.logout()

# Oder vorher authentifiziert starten:
client = AudactClient(
    api_key="aud_live_...",
    org_token="<jwt from a prior login>",
)

Streaming

async with AudactClient(api_key="...") as client:
    async for chunk in client.proxy_stream(
        "/v1/proxy/openai/stream",
        json={"model": "gpt-4o", "messages": [...], "stream": True},
    ):
        print(chunk.decode(), end="", flush=True)

KYC-Dokumente & Rufnummernbereitstellung

Die Mandantenidentität wird immer aus dem API-Schlüssel entnommen — keine dieser Methoden nimmt einen tenant_id-Parameter entgegen. Dokument-Uploads erfolgen als multipart/form-data; Dokumentbytes werden von keinem Endpunkt zurückgegeben (nur Metadaten + sha256).

async with AudactClient(api_key="aud_live_...") as client:
    # Upload a KYC document (raw bytes or an open file handle both work).
    doc = await client.upload_kyc_document(
        open("passport.pdf", "rb"), doc_type="passport"
    )

    # Discover what Telnyx requires, then create + submit a document set.
    requirements = await client.get_kyc_requirements(country="NL", number_type="mobile")
    doc_set = await client.create_kyc_document_set(
        CreateKYCDocumentSetRequest(
            name="NL mobile KYC",
            country_code="NL",
            number_type="mobile",
            items=[
                KYCDocumentSetItemRequest(requirement_id=req.id, document_id=doc.id)
                for req in requirements
            ],
        )
    )

    # Once doc_set.status == "approved" (poll get_kyc_document_set), place an order.
    order = await client.create_number_order(
        CreateNumberOrderRequest(country_code="NL", number_type="mobile", set_id=doc_set.id)
    )
    order = await client.wait_for_order(order.id)  # polls until success/failure/cancelled

Bring-your-own (BYO)-Telefonie

Die Mandantenidentität wird immer aus dem API-Schlüssel entnommen. BYO ist reines JSON — kein Multipart erforderlich. SIP-Anmeldedaten (sip_username/sip_password) sind schreibgeschützt: sie werden sicher gespeichert und in keiner Antwort zurückgegeben (has_credentials ist ausschließlich ein boolesches Flag).

async with AudactClient(api_key="aud_live_...") as client:
    # Fetch this deployment's SIP ingress host before configuring any trunk.
    config = await client.get_byo_config()
    print(config.sip_ingress_host, config.signaling_ips)

    # Register a BYO SIP trunk connection (either creds or an IP allowlist
    # is required) — runs a connect-time health probe before returning.
    conn = await client.create_byo_connection(
        CreateBYOConnectionRequest(
            name="Acme HQ Twilio Trunk",
            provider="twilio",
            sip_host="acme.pstn.twilio.com",
            sip_username="acme-trunk-user",
            sip_password="correct-horse-battery-staple",
        )
    )
    print(conn.connection_health)  # "healthy" | "unreachable"

    # Add a number over that connection — origin-aware create_number.
    number = await client.create_number(
        CreateNumberRequest(
            did="+12125551234",
            country_code="US",
            origin="byo",
            byo_connection_id=conn.id,
        )
    )
    print(number.origin, number.provider, number.connection_health)

Ein neuer Sub-Client: webhooks

async with AudactClient(api_key="aud_live_...") as client:
    endpoint = await client.webhooks.create_webhook(
        CreateWebhookRequest(
            scope="account",
            url="https://example.com/hooks/audact",
            events=["call_ended", "compliance.violation_detected"],
        )
    )
    # `endpoint.signing_secret` is returned ONLY on this first call — store it.

    await client.webhooks.test_webhook(endpoint.id)
    deliveries = await client.webhooks.list_webhook_deliveries(endpoint.id)

Eingehende Zustellungen mit dem mitgelieferten Signatur-Hilfsprogramm verifizieren (zeitkonstant, tolerant gegenüber einer Geheimnis-Rotations-Karenzzeit):

from audact import verify_webhook_signature

is_valid = verify_webhook_signature(
    payload_bytes=raw_request_body,          # the exact, unparsed bytes
    signature_header=request.headers["X-Audact-Signature"],
    secret=stored_signing_secret,
)

Beispiele

Ein ausführbares Beispielskript liegt unter examples/:

• examples/end_to_end_flow.py — der typische Ablauf: einen Agenten erstellen, eine Rufnummer über den Bestellvorgang bereitstellen und zuweisen, einen Browser-Testanruf tätigen, dann Anrufe auflisten/lesen. Mit gesetzter Umgebungsvariable AUDACT_API_KEY ausführen oder mit --check als Dry-Run starten (jeder HTTP-Aufruf wird prozessintern simuliert, kein Netzwerkzugriff erforderlich).

Fehlerbehandlung

API-Fehler erben von AudactAPIError; Timeouts (AudactTimeoutError) und Verbindungsfehler (AudactConnectionError) erben von AudactError:

Der Server verpackt jeden Fehler in ein verschachteltes Envelope: {"error": {"code", "message", "hint", "details", "request_id"}}. AudactAPIError analysiert dieses und stellt code, message, details, hint und request_id bereit (letzteres greift auch auf den X-Request-ID-Antwort-Header zurück) — geben Sie request_id beim Support an:

from audact import AudactAPIError, AudactRateLimitError

try:
    await client.validate(...)
except AudactRateLimitError as exc:
    print("retry in", exc.retry_after, "seconds", exc.request_id)
except AudactAPIError as exc:
    print(exc.code, exc.message, exc.hint, exc.request_id)

Wiederholungsversuche & Rate-Limits

• AudactRateLimitError (HTTP 429) stellt zusätzlich retry_after bereit (geparst aus dem Retry-After-Header) sowie rate_limit (ein RateLimitInfo-Snapshot von X-RateLimit-Limit/-Remaining/-Reset). client.last_rate_limit enthält den aktuellsten X-RateLimit-*-Snapshot von jeder Antwort, ob Erfolg oder Fehler.
• Wiederholungsversuche sind opt-in und standardmäßig deaktiviert — max_retries=N an den Konstruktor übergeben, um einen begrenzten Wiederholungsversuch mit Backoff für idempotente GET-Anfragen und 429/5xx-Antworten zu aktivieren (unter Beachtung von Retry-After). Wiederholungsversuche werden nie auf Validator-nahe Aufrufe (/v1/validate) angewendet, die ein 40-ms-Hotpath-Budget haben.

Lizenz

MIT