> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://developers.meshapi.ai/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://developers.meshapi.ai/_mcp/server.

# Realtime Audio

> Bidirectional speech-to-speech WebSocket sessions with the Python SDK.

# Realtime Audio

`client.realtime` opens a bidirectional WebSocket session to `wss://api.meshapi.ai/v1/realtime`. The wire format is identical to OpenAI's Realtime API.

Requires `websockets>=12.0`. Install with `pip install 'meshapi[realtime]'`.

## Connect and close (sync)

```python
from meshapi import MeshAPI

client = MeshAPI(base_url="https://api.meshapi.ai", token="rsk_...")

with client.realtime.connect(model="openai/gpt-realtime-mini") as session:
    pass  # clean close on exit
```

## Connect and close (async)

```python
from meshapi import AsyncMeshAPI

async with AsyncMeshAPI(base_url="https://api.meshapi.ai", token="rsk_...") as client:
    async with client.realtime.connect(model="openai/gpt-realtime-mini") as session:
        pass
```

## Configure the session

```python
session.send({
    "type": "session.update",
    "session": {
        "type":         "realtime",
        "modalities":   ["audio", "text"],
        "voice":        "alloy",
        "instructions": "You are a helpful assistant.",
    },
})
```

## Send audio

```python
# pcm_bytes is raw 16-bit PCM at 24 kHz mono
session.send_audio(pcm_bytes)
```

## Receive frames

```python
from meshapi.resources.realtime import RealtimeError

msg = session.receive()

if msg.audio:
    play_audio(msg.audio)
elif msg.event:
    print("event type:", msg.event.get("type"))
```

## Iterate over frames

```python
with client.realtime.connect(model="openai/gpt-realtime-mini") as session:
    session.send({"type": "session.update", "session": {"type": "realtime", "modalities": ["text"]}})
    session.send({"type": "response.create"})

    for msg in session:           # iterates until connection closes
        if msg.event:
            t = msg.event.get("type")
            if t == "response.text.delta":
                print(msg.event.get("delta"), end="", flush=True)
            elif t == "response.done":
                break
```

## Async iteration

```python
async with client.realtime.connect(model="openai/gpt-realtime-mini") as session:
    await session.send({"type": "session.update", "session": {"type": "realtime", "modalities": ["text"]}})
    await session.send({"type": "response.create"})

    async for msg in session:
        if msg.event and msg.event.get("type") == "response.text.delta":
            print(msg.event.get("delta"), end="", flush=True)
```

## Error handling

```python
from meshapi.resources.realtime import RealtimeError

try:
    msg = session.receive()
except RealtimeError as e:
    print("code:", e.code)       # "invalid_api_key", "insufficient_quota", …
    print("message:", str(e))
```

## Full voice agent example

```python
import threading

with client.realtime.connect(model="openai/gpt-realtime-mini") as session:
    session.send({
        "type": "session.update",
        "session": {"type": "realtime", "modalities": ["audio"], "voice": "alloy"},
    })

    def send_audio():
        for chunk in mic_stream():
            session.send_audio(chunk)
        session.send({"type": "input_audio_buffer.commit"})
        session.send({"type": "response.create"})

    threading.Thread(target=send_audio, daemon=True).start()

    for msg in session:
        if msg.audio:
            speaker.write(msg.audio)
```

## Supported models

| Model ID                   |
| -------------------------- |
| `openai/gpt-realtime-mini` |
| `openai/gpt-realtime`      |
| `openai/gpt-realtime-2`    |