SDK Python

SDK Python asynchrone pour la plateforme de validation de conformité Audact.

Prérequis

• Python 3.11+
• httpx pour le HTTP asynchrone
• pydantic v2 pour le typage

Installation

Le SDK Python n’est pas encore publié. Les instructions d’installation figureront ici lorsqu’il le sera.

Démarrage rapide

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())

URL de base

Pointez le client vers un seul hôte avec base_url ; tous les services y sont résolus et le SDK ajoute lui-même les chemins /v1/... (un / ou un /v1 final dans base_url est supprimé) :

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

Remplacez l’URL de base via la variable d’environnement, sans changer le code :

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

Ordre de résolution par service : argument nommé <service>_url= → variable d’environnement AUDACT_<SERVICE>_URL → argument nommé base_url= → variable d’environnement AUDACT_BASE_URL. Les URL par service ne sont nécessaires que pour les déploiements où chaque service a son propre hôte (voir Configuration).

Configuration

Remplacez les URL de base des services via les arguments nommés du constructeur ou via des variables d’environnement :

max_retries (argument nommé du constructeur, 0 par défaut) active, sur demande, une reprise bornée avec temporisation progressive pour les GET idempotents et les réponses 429/5xx — voir Nouvelles tentatives et limites de débit ci-dessous.

Couverture

Le SDK est à parité complète avec l’API documentée : 209 des 211 opérations sont couvertes. Les 2 restantes sont des webhooks côté réception, du fournisseur vers la plateforme — POST /telnyx/webhook et POST /webhooks/livekit — qui sont des points d’entrée qu’Audact reçoit, sans jamais les appeler.

Le SDK est monolithique avec des sous-clients en attributs : toutes les méthodes plates préexistantes de AudactClient (par exemple client.validate, client.evaluate, client.create_tenant) continuent de fonctionner sans changement, et les nouvelles grandes surfaces ajoutées pour la parité sont exposées sous forme de sous-clients client.<name>.* :

Flux d’authentification

La plupart des méthodes envoient Authorization: Bearer <api_key> avec la api_key transmise au constructeur AudactClient. Deux surfaces utilisent un autre jeton Bearer (organisation et plateforme) :

• OrgBearer — un JWT obtenu auprès de client.org.login(). Le SDK le conserve automatiquement en mémoire sur le client ; chaque appel client.org.* suivant l’envoie à la place de la clé API.
• Jeton de plateforme — un jeton de super-administrateur exigé pour export_tenant / import_tenant. Transmettez-le au constructeur sous la forme platform_token= ou appelez client.set_platform_token(token) à tout moment.

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()

# Or start pre-authenticated:
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)

Documents KYC et provisionnement de numéros

L’identité du tenant est toujours tirée de la clé API — aucune de ces méthodes n’accepte de paramètre tenant_id. Les envois de documents se font en multipart/form-data ; les octets des documents ne sont jamais renvoyés par aucun point de terminaison (métadonnées + sha256 uniquement).

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

Téléphonie bring-your-own (BYO)

L’identité du tenant est toujours tirée de la clé API. Le BYO se fait en JSON pur — sans multipart. Les identifiants SIP (sip_username/sip_password) sont en écriture seule : stockés de manière sécurisée et jamais renvoyés dans une réponse (has_credentials est uniquement un indicateur booléen).

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)

Un nouveau sous-client : les 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)

Vérifiez les livraisons entrantes avec l’assistant de signature fourni (en temps constant, tolérant à une période de grâce de rotation du secret) :

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,
)

Exemples

Un script d’exemple exécutable se trouve sous examples/ :

• examples/end_to_end_flow.py — le parcours courant : créer un agent, provisionner et affecter un numéro de téléphone via le parcours de commande, passer un appel de test depuis le navigateur, puis lister/lire les appels. Exécutez-le avec AUDACT_API_KEY défini, ou en simulation avec --check (simule tous les appels HTTP dans le processus, aucun accès réseau n’est requis).

Gestion des erreurs

Les erreurs d’API héritent de AudactAPIError ; les délais dépassés (AudactTimeoutError) et les échecs de connexion (AudactConnectionError) héritent de AudactError :

Le serveur enveloppe chaque erreur dans une enveloppe imbriquée, {"error": {"code", "message", "hint", "details", "request_id"}}. AudactAPIError l’analyse et expose code, message, details, hint et request_id (ce dernier se rabat aussi sur l’en-tête de réponse X-Request-ID) — indiquez request_id lorsque vous contactez le support :

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)

Nouvelles tentatives et limites de débit

• AudactRateLimitError (HTTP 429) expose en plus retry_after (analysé à partir de l’en-tête Retry-After) et rate_limit (un instantané RateLimitInfo de X-RateLimit-Limit/-Remaining/-Reset). client.last_rate_limit conserve le dernier instantané X-RateLimit-* vu sur n’importe quelle réponse, en succès comme en erreur.
• Les nouvelles tentatives sont facultatives et désactivées par défaut — transmettez max_retries=N au constructeur pour activer une reprise bornée avec temporisation progressive pour les GET idempotents et les réponses 429/5xx (en respectant Retry-After). Les nouvelles tentatives ne sont jamais appliquées aux appels proches du Validator (/v1/validate), qui disposent d’un budget de 40 ms sur le chemin critique.

Licence

MIT