Python SDK

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

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

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

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

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)

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

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)

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)

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

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)