SDKs

Python SDK

Full reference for the official perceive8 PyPI package, including sync and async clients.

Python SDK

The perceive8 PyPI package is the official Python client for the Perceive8 Audio Intelligence API. It bundles a synchronous client (Perceive8Client) and an asynchronous client (AsyncPerceive8Client).

Installation
Client Initialization
Resources Overview
Analyses
Speakers
Alerts
Stream
Pagination
Exception Handling
Models

Installation

pip install perceive8
# or with Poetry
poetry add perceive8

Requires: Python 3.10+

Dependencies:

Package	Purpose
`httpx`	HTTP client (sync + async, HTTP/2 enabled)
`tenacity`	Automatic retry with exponential back-off
`pydantic`	Data model validation
`websockets`	WebSocket support for live streaming

Client Initialization

Synchronous client

from perceive8 import Perceive8Client

# API key
client = Perceive8Client(api_key="pk_live_xxxxxxxxxxxx")

# Bearer JWT
client = Perceive8Client(token="<supabase-jwt>")

# Custom base URL (e.g. local dev)
client = Perceive8Client(api_key="...", base_url="http://localhost:8000")

The synchronous client supports the context manager protocol for automatic cleanup:

with Perceive8Client(api_key="pk_live_xxxx") as client:
    analyses = client.analyses.list()
# HTTP connections are closed on __exit__

Asynchronous client

from perceive8 import AsyncPerceive8Client

async with AsyncPerceive8Client(api_key="pk_live_xxxx") as client:
    analyses = await client.analyses.list()

Constructor Parameters

Parameter	Type	Default	Description
`api_key`	`str \| None`	`None`	API key → `X-API-Key` header
`token`	`str \| None`	`None`	JWT → `Authorization: Bearer` header
`base_url`	`str`	`https://api.perceive8.com`	API base URL

You must supply either api_key or token. Supplying neither raises ValueError.

Resources Overview

Both clients expose identical resource namespaces:

Attribute	Type	Description
`client.analyses`	`AnalysesResource`	Upload, fetch, list, delete analyses
`client.speakers`	`SpeakersResource`	Manage enrolled speaker profiles
`client.alerts`	`AlertsResource`	List, acknowledge, and stream alerts
`client.stream`	`StreamResource`	Open live WebSocket sessions (async only)

Note: stream and alerts are only available on AsyncPerceive8Client. The synchronous Perceive8Client does not include these resources because WebSocket and SSE connections require an async runtime.

Analyses

All methods are available on both Perceive8Client (sync) and AsyncPerceive8Client (async).

`client.analyses.upload(file, filename, language="en")`

Upload an audio file to create a new analysis job.

with open("meeting.mp3", "rb") as f:
    result = client.analyses.upload(f, "meeting.mp3", language="en")

analysis_id = result["data"]["id"]
print("Created:", analysis_id, "status:", result["data"]["status"])

Parameters

Parameter	Type	Default	Description
`file`	`BinaryIO`	—	File-like object opened in binary mode
`filename`	`str`	—	Filename sent as the multipart field name
`language`	`str`	`"en"`	ISO 639-1 language code

Returns: dict — { "data": { "id": ..., "status": "pending", ... } }

`client.analyses.list(limit=20, cursor=None)`

List analyses for the authenticated user.

result = client.analyses.list(limit=10)
for analysis in result.get("data", []):
    print(analysis["id"], analysis["filename"], analysis["status"])

Parameters

Parameter	Type	Default	Description
`limit`	`int`	`20`	Items per page
`cursor`	`str \| None`	`None`	Pagination cursor

Returns: dict — { "data": [...], "meta": { "next_cursor": ... } }

`client.analyses.get(analysis_id)`

Retrieve a single analysis by ID.

result = client.analyses.get("uuid-here")
data = result["data"]

if data["status"] == "completed":
    print(data["transcript"])

Returns: dict — { "data": { ... analysis fields ... } }

`client.analyses.delete(analysis_id)`

Delete an analysis.

client.analyses.delete("uuid-here")

Returns: None

`await client.analyses.stream_progress(analysis_id)` (async only)

An async generator that yields pipeline progress events over SSE for the given analysis.

async with AsyncPerceive8Client(api_key="...") as client:
    async for event in client.analyses.stream_progress("uuid-here"):
        print(event["step"], event.get("progress_pct"), "%")
        if event.get("progress_pct") == 100:
            break

Yields: dict with keys:

Key	Type	Description
`step`	`str`	Pipeline step name
`status`	`str`	`"started"` \| `"completed"` \| `"failed"`
`progress_pct`	`int`	0–100
`message`	`str`	Human-readable status message
`timestamp`	`str`	ISO 8601

Poll-until-complete helper (sync)

The SDK does not include a built-in polling helper, but the pattern is straightforward:

import time
from perceive8 import Perceive8Client, Perceive8Error

def wait_for_analysis(client: Perceive8Client, analysis_id: str, poll_interval: float = 3.0) -> dict:
    while True:
        data = client.analyses.get(analysis_id)["data"]
        if data["status"] in ("completed", "failed"):
            return data
        time.sleep(poll_interval)

with Perceive8Client(api_key="pk_live_xxxx") as client:
    with open("audio.mp3", "rb") as f:
        result = client.analyses.upload(f, "audio.mp3")
    analysis = wait_for_analysis(client, result["data"]["id"])
    print(analysis["status"], analysis.get("transcript"))

Speakers

`client.speakers.list()`

result = client.speakers.list()
for speaker in result.get("data", []):
    print(speaker["id"], speaker["name"])

Returns: dict — { "data": [ { "id": ..., "name": ..., ... } ] }

`client.speakers.get(speaker_id)`

result = client.speakers.get("speaker-uuid")
speaker = result["data"]
print(speaker["name"])

Returns: dict — { "data": { ... } }

`client.speakers.create(name)`

result = client.speakers.create("Alice")
speaker_id = result["data"]["id"]

Returns: dict — { "data": { "id": ..., "name": ..., ... } }

`client.speakers.delete(speaker_id)`

client.speakers.delete("speaker-uuid")

Returns: None

Alerts

Async only. AlertsResource is only attached to AsyncPerceive8Client.

`await client.alerts.list(analysis_id, acknowledged=None)`

List alerts for an analysis.

async with AsyncPerceive8Client(api_key="...") as client:
    result = await client.alerts.list("analysis-uuid", acknowledged=False)
    for alert in result.get("alerts", []):
        print(alert["severity_level"], alert["message"])

Parameters

Parameter	Type	Default	Description
`analysis_id`	`str`	—	The analysis to query
`acknowledged`	`bool \| None`	`None`	Filter to unacknowledged (`False`) or acknowledged (`True`)

Returns: dict — { "alerts": [...], "total": int }

`await client.alerts.acknowledge(alert_id)`

Mark an alert as acknowledged.

updated = await client.alerts.acknowledge("alert-uuid")
print(updated["acknowledged"])  # True

Returns: dict — updated alert object

`client.alerts.stream()` (async generator)

SSE stream yielding real-time alerts as they fire during live sessions.

async with AsyncPerceive8Client(api_key="...") as client:
    async for alert in client.alerts.stream():
        print(alert["severity"], alert["message"])

Yields: dict — alert payload (same shape as list response items)

Stream

Async only. Available on AsyncPerceive8Client.

`client.stream.live(language="en", sample_rate=16000, scenario_id=None)`

Returns a LiveSessionContext (async context manager) that connects to the WebSocket, sends the start action, and returns a LiveSession.

import asyncio
from perceive8 import AsyncPerceive8Client

async def main():
    async with AsyncPerceive8Client(api_key="pk_live_xxxx") as client:
        async with client.stream.live(language="en", sample_rate=16000) as session:
            print("Session:", session.session_id)

            # Send PCM audio bytes (Int16, mono, 16 kHz)
            with open("audio_pcm.raw", "rb") as f:
                while chunk := f.read(3200):  # 100ms of audio at 16 kHz
                    await session.send_audio(chunk)

            await session.stop()

            async for event in session.events():
                if event.get("event") == "transcript_final":
                    print("Final:", event["data"]["text"])
                if event.get("event") == "session_ended":
                    break

asyncio.run(main())

Parameters

Parameter	Type	Default	Description
`language`	`str`	`"en"`	ISO 639-1 code
`sample_rate`	`int`	`16000`	Hz — must match the audio you send
`scenario_id`	`str \| None`	`None`	Enable real-time alert rules

Returns: LiveSessionContext (use as async with)

`LiveSession`

`session.session_id`

Server-assigned session ID (str).

`await session.send_audio(pcm_bytes)`

Send raw PCM audio data.

await session.send_audio(pcm_bytes)  # bytes, Int16 little-endian, mono

Parameters: pcm_bytes: bytes

`await session.stop()`

Send { "action": "stop" } to end the session gracefully.

`session.events()` (async generator)

Yields WebSocket event dicts until session_ended or error.

async for event in session.events():
    match event.get("event"):
        case "transcript_partial":
            print("[partial]", event["data"]["text"])
        case "transcript_final":
            print("[final]  ", event["data"]["text"])
        case "alert":
            print("[ALERT]  ", event["data"])
        case "session_ended":
            break

WebSocket event types

`event` value	Description
`session_started`	Session active; `session_id` in payload
`transcript_partial`	In-progress transcript text
`transcript_final`	Committed transcript with word timestamps
`alert`	Alert from scenario rules engine
`analysis_chunk`	Per-segment real-time analysis result
`alert_triggered`	Real-time alert from analysis engine
`analysis_started`	Post-stream analysis pipeline starting
`session_ended`	Session ended; optional `analysis_id`
`error`	Server error; `message` field

Pagination

For paginating through analysis lists manually:

def iter_all_analyses(client, limit=50):
    cursor = None
    while True:
        result = client.analyses.list(limit=limit, cursor=cursor)
        items = result.get("data", [])
        yield from items
        cursor = result.get("meta", {}).get("next_cursor")
        if not cursor:
            break

with Perceive8Client(api_key="...") as client:
    for analysis in iter_all_analyses(client):
        print(analysis["id"])

The Page dataclass is exported for use when building custom pagination helpers:

from perceive8.pagination import Page

page: Page[dict] = Page(items=[...], next_cursor="abc", has_more=True)
for item in page:
    print(item)

Exception Handling

All SDK calls raise subclasses of Perceive8Error:

Exception	HTTP Status	Description
`Perceive8Error`	any >= 400	Base class
`AuthenticationError`	401	Invalid or missing credentials
`NotFoundError`	404	Resource not found
`RateLimitError`	429	Rate limit exceeded
`ValidationError`	422	Request body validation failed

The synchronous client auto-retries on RateLimitError up to 3 attempts with exponential back-off (using tenacity).

Exception Properties

class Perceive8Error(Exception):
    status_code: int | None
    response: dict | None

Example

from perceive8 import Perceive8Client
from perceive8.exceptions import (
    AuthenticationError,
    NotFoundError,
    RateLimitError,
    ValidationError,
    Perceive8Error,
)

with Perceive8Client(api_key="pk_live_xxxx") as client:
    try:
        result = client.analyses.get("bad-id")
    except NotFoundError:
        print("Analysis not found")
    except AuthenticationError as e:
        print(f"Auth failed ({e.status_code})")
    except RateLimitError:
        print("Rate limit hit — retries exhausted")
    except ValidationError as e:
        print("Validation error:", e.response)
    except Perceive8Error as e:
        print(f"API error {e.status_code}: {e}")

Models

Models live in perceive8.models and are built with Pydantic v2.

`Analysis`

from perceive8.models.analysis import Analysis, AnalysisStatus

class AnalysisStatus(str, Enum):
    pending    = "pending"
    processing = "processing"
    completed  = "completed"
    failed     = "failed"

class Analysis(BaseModel):
    id: str
    user_id: str
    filename: str
    status: AnalysisStatus
    created_at: datetime
    completed_at: datetime | None = None
    transcript: Any = None

`Speaker`

from perceive8.models.speaker import Speaker

class Speaker(BaseModel):
    id: str
    user_id: str
    name: str
    created_at: datetime

`Page[T]`

from perceive8.pagination import Page

@dataclass
class Page(Generic[T]):
    items: list[T]
    next_cursor: str | None = None
    has_more: bool = False

Page is iterable — for item in page: yields from items.

← Back to docs

Python SDK

Table of Contents

Installation

Client Initialization

Synchronous client

Asynchronous client

Constructor Parameters

Resources Overview

Analyses

client.analyses.upload(file, filename, language="en")

client.analyses.list(limit=20, cursor=None)

client.analyses.get(analysis_id)

client.analyses.delete(analysis_id)

await client.analyses.stream_progress(analysis_id) (async only)

Poll-until-complete helper (sync)

Speakers

client.speakers.list()

client.speakers.get(speaker_id)

client.speakers.create(name)

client.speakers.delete(speaker_id)

Alerts

await client.alerts.list(analysis_id, acknowledged=None)

await client.alerts.acknowledge(alert_id)

client.alerts.stream() (async generator)

Stream

client.stream.live(language="en", sample_rate=16000, scenario_id=None)

LiveSession

session.session_id

await session.send_audio(pcm_bytes)

await session.stop()

session.events() (async generator)

Pagination

Exception Handling

Exception Properties

Example

Models

Analysis

Speaker

Page[T]

`client.analyses.upload(file, filename, language="en")`

`client.analyses.list(limit=20, cursor=None)`

`client.analyses.get(analysis_id)`

`client.analyses.delete(analysis_id)`

`await client.analyses.stream_progress(analysis_id)` (async only)

`client.speakers.list()`

`client.speakers.get(speaker_id)`

`client.speakers.create(name)`

`client.speakers.delete(speaker_id)`

`await client.alerts.list(analysis_id, acknowledged=None)`

`await client.alerts.acknowledge(alert_id)`

`client.alerts.stream()` (async generator)

`client.stream.live(language="en", sample_rate=16000, scenario_id=None)`

`LiveSession`

`session.session_id`

`await session.send_audio(pcm_bytes)`

`await session.stop()`

`session.events()` (async generator)

`Analysis`

`Speaker`

`Page[T]`