Metadata-Version: 2.5
Name: pyai-sdk
Version: 0.4.0
Summary: Official Python SDK for PyAI: speech-to-text (Hear), text-to-speech (Speak), realtime voice agents (Omni), and call compliance (Trace).
Project-URL: Homepage, https://pyai.com
Project-URL: Documentation, https://docs.pyai.com
Project-URL: Repository, https://github.com/atomsai/pyai-platform-backend
Author: PyAI
License: MIT
Keywords: compliance,openai-compatible,pyai,realtime,speech,stt,transcription,tts,voice-agents,voice-ai
Requires-Python: >=3.9
Requires-Dist: httpx[http2]<1,>=0.28
Description-Content-Type: text/markdown

# pyai-sdk (Python SDK)

Official Python SDK for [PyAI](https://pyai.com), 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](https://pyai.com/models/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](https://pyai.com/models/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](https://pyai.com/models/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](https://pyai.com/models/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](https://pyai.com/models/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](https://pyai.com/models/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

```bash
pip install pyai-sdk
```

## Quickstart

```python
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`](https://www.npmjs.com/package/@pyai/mcp)) so
the agent can mint a free key and call PyAI as tools, no endpoint guessing, no
human setup step:

```jsonc
// .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`](https://github.com/atomsai/pyai-examples/tree/main/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:

```python
# 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
```

| `response_format` | sample rates (Hz) | Content-Type |
|---|---|---|
| `wav` (default) | 8000 / 16000 / 24000 / 48000 | `audio/wav` |
| `mp3` | 8000 / 16000 / 24000 / 48000 | `audio/mpeg` |
| `opus` | 8000 / 16000 / 24000 / 48000 | `audio/ogg` |
| `aac` | 8000 / 16000 / 24000 / 48000 | `audio/aac` |
| `flac` | 8000 / 16000 / 24000 / 48000 | `audio/flac` |
| `pcm` (raw int16 LE, no header) | 8000 / 16000 / 24000 / 48000 | `audio/pcm` |
| `g711_ulaw` | 8000 (forced) | `audio/basic` |
| `g711_alaw` | 8000 (forced) | `audio/basic` |

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`](../../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`).

```python
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:

```python
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`:

```python
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

```python
# 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:

```python
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:

```bash
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):

```python
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

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