pyai-sdk (Python SDK)

Official Python SDK for PyAI, the all-in-one voice AI platform: lightning-fast speech-to-text, ultra-realistic text-to-speech, end-to-end realtime voice agents, and automatic call compliance. Python 3.9+; installs HTTPX with HTTP/2 support.

PyAI products

• Hear, Lightning-fast, telephony-native speech-to-text. Whisper-compatible transcription tuned for real phone-call audio, with live streaming partials so your app reacts mid-sentence, plus async batch transcription for big archives. POST /v1/audio/transcriptions
• Speak, Ultra-realistic text-to-speech that starts speaking in tens of milliseconds. Stream lifelike, expressive voices, choose from 144 stock voices, or clone any voice instantly, for free. POST /v1/audio/speech
• Omni (flagship), One API for a complete, end-to-end voice AI agent. A single WebSocket where your agent listens, thinks, and speaks, grounded in your knowledge bases and tools, with human-like turn-taking and instant barge-in, no STT, LLM, or TTS to stitch together yourself. wss://api.pyai.com/v1/omni
• Trace (flagship), The compliance API that keeps your AI agents safe. Trace automatically checks every call for HIPAA, TCPA, and PII risks (plus your own brand-voice rules), flags the exact rule broken, redacts sensitive data, and seals each call with a tamper-evident audit trail, so a risky conversation never slips through. GET /v1/trace/interactions
• Cue, reserved turn-detection and grounding fields on Hear streaming. Grounding is not active on the serving route yet.
• AMD, Answering-machine detection that tells your dialer who or what answered, human, voicemail, IVR, iPhone/Google screening, dead number, fax, in a fraction of Twilio's dead-air dwell, with the reason. A one-line-TwiML Twilio Media Streams drop-in; billed per answered call (first 5,000/month free). wss://api.pyai.com/v1/amd/stream
• Telephony, Instant managed phone numbers for your voice agents. Provision a US number and route live calls straight into an Omni agent, no carrier contracts, no telephony glue. POST /v1/telephony/numbers

The contract is https://api.pyai.com/openapi.json. This SDK wraps it with typed errors, automatic retries, and realtime URL helpers.

audio.speech_stream() provides true first-byte streaming with the SDK's default pooled HTTP transport. The response closes when iteration is exhausted or the iterator's close() method is called after stopping early. The legacy transport= hook remains compatible, but its contract returns a complete bytes body; speech_stream() can only slice that buffered value locally and cannot provide network-level streaming with such a custom transport.

Reuse one PyAI client across requests to reuse connections. The synchronous client uses pooled HTTP/1.1 by default. Use with PyAI(api_key=...) as pyai: or call pyai.close() during shutdown. Close a partially consumed audio iterator explicitly before closing the client. http2=True opts into HTTP/2 negotiation, with HTTP/1.1 fallback when unsupported. Keep the default for clients shared across threads: a concurrent HTTP/2 test on macOS/Python 3.14 hit a socket read error, so this mode needs validation in your runtime before use. timeout=30.0 sets a per-operation timeout in seconds (the default remains no timeout); trust_env=False opts out of environment proxy and certificate settings. Transport exceptions, including truncated audio, propagate without automatically replaying the request. The legacy custom transport keeps its existing behavior. This HTTP transport change does not alter Hear or Omni WebSocket connections.

Install

pip install pyai-sdk

Quickstart

import os
from pyai import PyAI, new_idempotency_key

pyai = PyAI(api_key=os.environ["PYAI_API_KEY"])

# Text-to-speech
audio = pyai.audio.speech(input="Hello from PyAI.", voice="stock_emma_en_gb")
open("hello.wav", "wb").write(audio)

# Incremental TTS: the default transport yields network bytes as they arrive.
with open("hello-stream.pcm", "wb") as output:
    for chunk in pyai.audio.speech_stream(
        input="Hello from PyAI.",
        voice="stock_emma_en_gb",
        response_format="pcm",
    ):
        output.write(chunk)

# Voices
voices = pyai.voices.list(gender="female")

# Async transcription (safe retry with an idempotency key)
job = pyai.transcription_jobs.create(
    audio_url="https://example.com/call.wav",
    diarize=True,
    idempotency_key=new_idempotency_key(),
)
done = pyai.transcription_jobs.get(job["job_id"])

Use with MCP (AI coding agents)

Building this SDK with an AI coding agent (Cursor, Claude Code, Codex)? Add the PyAI MCP server (@pyai/mcp) so the agent can mint a free key and call PyAI as tools, no endpoint guessing, no human setup step:

// .cursor/mcp.json  ·  or:  claude mcp add pyai -- npx -y @pyai/mcp
{ "mcpServers": { "pyai": { "command": "npx", "args": ["-y", "@pyai/mcp"] } } }

With no key set, the server exposes create_sandbox_key, calls it, and adopts the minted key for the session, then get_started, list_voices, synthesize_speech, and the transcription tools work immediately. Full setup + a runnable client: the mcp-quickstart example.

Speak audio formats (incl. telephony G.711)

audio.speech encodes server-side into any of eight formats via response_format, so telephony callers no longer hand-roll a resampler + μ-law encoder, the audio comes back already in the shape you need:

# Twilio/SIP-ready in one param: raw 8 kHz mono μ-law, no client-side DSP.
ulaw = pyai.audio.speech(
    input="Your appointment is confirmed.",
    voice="stock_emma_en_gb",
    response_format="g711_ulaw",   # -> audio/basic, forced 8 kHz
)
import base64
media_frame_payload = base64.b64encode(ulaw).decode()  # straight into Twilio

The accepted set is exported as SPEECH_FORMATS / SPEECH_SAMPLE_RATES (and a SpeechFormat Literal for type-checkers). Any other value is a 400 unsupported_format. sample_rate is optional, omit it for the engine's native 24 kHz (g711_* is always 8 kHz); omit response_format for the default wav. See examples/speak-telephony-formats for the full before/after.

Realtime (Omni)

The standard library has no production-grade WebSocket client, so the SDK gives you the connect URL (realtime_url), the subprotocol helper, and the omni_configure_frame() builder; pair them with your preferred WS library (e.g. websockets).

⚠️ Frame-key asymmetry, the #1 Omni bug. Your outbound control frames (configure, dtmf) are keyed on type; the inbound server frames (hello, session_started, configured, …) are keyed on event. The gateway is transparent, so a mis-keyed {"event":"configure"} is acked but silently dropped, you get a connected session with zero turns and no error. omni_configure_frame() guarantees the correct {"type":"configure",...} envelope; parse inbound frames on event (see OMNI_EVENTS).

import asyncio, json, websockets
from pyai import OMNI_EVENTS

# Omni is zero-state: the key's org authorizes the session, nothing to create
# first. session_label is an optional opaque tag echoed to your kb_endpoint.
url = pyai.realtime_url(session_label="support", query={"rate": "16000"})

async def main():
    async with websockets.connect(url, subprotocols=[pyai.realtime_subprotocol()]) as ws:
        # Send the agent config FIRST, 0x03-tagged and type-keyed (the builder
        # guarantees the key):
        await ws.send(b"\x03" + json.dumps(pyai.omni_configure_frame(
            voice_id="stock_emma_en_gb", persona="You are a receptionist.")).encode())
        # ...then stream 0x01-tagged PCM16 audio continuously (server-side VAD).
        await ws.send(b"\x01" + pcm16_bytes)
        async for frame in ws:
            if not isinstance(frame, (bytes, bytearray)):
                raise RuntimeError("Omni server frames must be binary")
            tag, body = frame[0], frame[1:]
            if tag == 0x01:
                play_agent_audio(body)
            elif tag == 0x02:
                transcript = pyai.omni_transcript_body(body)
                if transcript is None:
                    raise RuntimeError("Invalid Omni transcript body")
                print(transcript)                   # normalized live UTF-8 caller delta
            elif tag == 0x03:
                event = json.loads(body)            # keyed on `event`
                print(event["event"])

asyncio.run(main())

Live 0x02 bodies are plain UTF-8 caller-text deltas, not JSON. omni_transcript_body() normalizes them to {event, role, text, final, mode} and retains bounded direct-JSON support for older bridges. rate=16000 configures caller input but agent output remains 24 kHz; read hello.audio_out. Omni has no commit frame, so keep streaming silence during caller pauses.

Omni connects only to wss://api.pyai.com/v1/omni and is zero-state. No Agent is required. session_label is an optional opaque tag (never required). The raw URL helper accepts canonical format, rate, and api_key query parameters; retired connect aliases, model selectors, and token query names raise ValueError instead of being translated.

An optional managed Agent can store a custom vocabulary list:

agent = pyai.agents.create(
    name="Front desk",
    vocabulary=["Nguyen", "Acme Dental", "SKU-99"],
)

pyai.agents.update(agent["agent_id"], vocabulary=[])

The Agent list is sanitized to at most five effective terms and fixed when a new session starts. An empty list turns the feature off. Organization Hear vocabulary is never applied to Omni.

Streaming speech-to-text (Hear / Cue)

The standard library has no production-grade WebSocket client, so the SDK gives you a URL builder (hear_stream_url) plus the subprotocol helper; pair them with websockets (or websocket-client). The wire protocol: stream binary PCM16/opus frames, send {"type":"commit"} to force-finalize, and read JSON frames of type config_ack / partial / partial_stable / speech_final / final / error:

import asyncio, json, websockets

url = pyai.hear_stream_url(sample_rate=16000, endpointing_ms=800)

async def transcribe(pcm_chunks):
    async with websockets.connect(url, subprotocols=[pyai.realtime_subprotocol()]) as ws:
        async for pcm16 in pcm_chunks:
            await ws.send(pcm16)  # keep sending silence through pauses
        await ws.send(json.dumps({"type": "commit"}))
        async for frame in ws:
            event = json.loads(frame)
            if event["type"] == "config_ack" and event["warnings"]:
                raise RuntimeError(event["warnings"])
            if event["type"] == "speech_final":
                print("endpoint reason:", event["endpoint_reason"])
            print(event)

asyncio.run(transcribe(mic_source()))

For Cue (turn detection + KB context), send {"type": "config", "grounding": true} as the first text frame after connecting; final/speech_final frames then carry a grounding list of top KB passages.

Sync STT, telephony output, and more APIs

# Synchronous speech-to-text
text = pyai.audio.transcriptions.create(file=open("call.wav", "rb"), language="en")["text"]

# Telephony-ready TTS: raw 8 kHz G.711 for Twilio/SIP, encoded server-side, # no client-side resampler or μ-law encoder. Just base64 it into a media frame.
ulaw = pyai.audio.speech(input="Hi there", response_format="g711_ulaw")

# Voice clones (Speak)
clone = pyai.clones.create(name="Brand VO", file=open("ref.wav", "rb"))
pyai.clones.delete(clone["id"])

# Managed phone numbers (Telephony)
avail = pyai.telephony.numbers.available(area_code="415")["data"]
num = pyai.telephony.numbers.buy(phone_number=avail[0]["phone_number"], agent_id="agent_123")
pyai.telephony.numbers.assign(num["id"], "agent_123")
pyai.telephony.numbers.release(num["id"])

# Answering-machine detection (AMD). Already on Twilio? One line of TwiML points
# the call's media at wss://api.pyai.com/v1/amd/stream, no client code. From the
# SDK you set the operating-point dial and read decisions back:
pyai.amd.config.set(aggressiveness=0.25, webhook_url="https://you/amd-events")
decision = pyai.amd.calls.get("C_123")
# decision["answered_by"]        -> human | voicemail | screening | sit_invalid | ...
# decision["answered_by_twilio"] -> Twilio's exact enum, for drop-in routing parity
# decision["reason"]             -> "machine phrase: 'leave a message' @1.2s"
ws_url = pyai.amd_stream_url(aggressiveness=0.25)  # feed into websockets + realtime_subprotocol()

# Compliance (Trace)
fails = pyai.trace.interactions.list(verdict="FAIL")["data"]
pyai.trace.config.set(agent_id="agent_123", enabled=True)
exposure = pyai.trace.exposure(window_days=30)

# Per-call eval scorecard (timeline + quality metrics). Additive and forward-
# compatible, present once the engine emits them, so reading is always safe
# (call_timeline returns [] until then).
timeline = pyai.trace.call_timeline(fails[0]["id"])              # list[dict] of turns
quality = pyai.trace.interactions.get(fails[0]["id"]).get("quality_metrics")

Reproducible runs (evals)

audio.speech and audio.transcriptions.create take optional seed and temperature for deterministic eval runs. They're forward-compatible, honored once the engine supports them and otherwise ignored, so it's always safe to pass:

pyai.audio.speech(input="Hello", voice="stock_emma_en_gb", seed=42, temperature=0)
pyai.audio.transcriptions.create(file=open("call.wav", "rb"), seed=42)

CLI (pyai)

The package installs a pyai command (also python -m pyai). pyai doctor introspects your key/scopes via GET /v1/me (skipped gracefully if the route isn't deployed yet), checks endpoint liveness, runs a Speak→Hear round-trip, and prints remediation hints:

export PYAI_API_KEY=pyai_test_...
pyai doctor
# PASS  key (/v1/me), env=test; 3 scope(s): hear:transcribe, speak:synthesize, hear:stream
# PASS  speak→hear round-trip, synth 45210 bytes → "the quick brown fox…"
# Diagnosis: healthy. Key, endpoint, and a Speak→Hear round-trip all work.

pyai smoke   # lighter: models + voices + speak

Errors

Failures raise PyAIError with a stable code (branch on it, not the message):

from pyai import PyAIError

try:
    pyai.audio.speech(input="hi")
except PyAIError as err:
    if err.code == "credit_exhausted":
        ...  # out of prepaid credit, add credit or use a sandbox key

Common codes: unauthorized, forbidden, credit_exhausted, rate_limit_exceeded, concurrency_limit_exceeded, idempotency_conflict. 429/5xx are retried automatically (honoring Retry-After); tune with PyAI(api_key, max_retries=...).

Develop

python -m unittest discover -s tests -v   # no network; transport injected