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. Zero third-party dependencies (standard library only); Python 3.9+.

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 standard-library 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.

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 to create. 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.

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