Metadata-Version: 2.4
Name: openbmb-realtime
Version: 0.1.5
Summary: Async Python SDK for OpenBMB MiniCPM-o Realtime sessions
Author: OpenBMB
License: Apache-2.0
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Multimedia :: Sound/Audio
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: websockets<18,>=16
Provides-Extra: test
Requires-Dist: pytest>=8; extra == "test"
Requires-Dist: pytest-asyncio>=0.24; extra == "test"

# openbmb-realtime

An asynchronous Python SDK for OpenBMB MiniCPM-o Realtime sessions.

The SDK provides a high-level interface for WebSocket connection management,
server-side queueing, session initialization, normalized events, audio and video
frame encoding, interruption, pause/resume, and idempotent session shutdown.
Application code does not need to construct protocol frames such as
`session.init`, `input.append`, or `session.close`.

## Installation

```bash
pip install openbmb-realtime
```

Supported Python versions: 3.10 and later.

## Audio Sessions

```python
import asyncio
import os

from openbmb_realtime import AudioDelta, RealtimeClient, TextDelta, UsageEvent


async def microphone_stream():
    """Yield 16 kHz, mono, Float32 PCM audio chunks."""
    raise NotImplementedError


async def play_audio(audio: bytes):
    """Send model audio bytes to an audio output device."""
    raise NotImplementedError


async def consume_events(session):
    async for event in session:
        if isinstance(event, TextDelta):
            print(event.text, end="", flush=True)
        elif isinstance(event, AudioDelta):
            await play_audio(event.audio)
        elif isinstance(event, UsageEvent):
            print("Usage:", event.usage)


async def main():
    client = RealtimeClient(
        api_key=os.environ["MODELBEST_API_KEY"],
        base_url="https://api.modelbest.cn",
    )

    session = await client.audio.connect(
        model="minicpm-o-4.5-realtime",
        system_prompt="You are a helpful voice assistant.",
    )
    receiver = asyncio.create_task(consume_events(session))

    try:
        async for chunk in microphone_stream():
            await session.send_audio(chunk)
    finally:
        await session.close()
        await receiver


asyncio.run(main())
```

`send_audio()` accepts 16 kHz, mono, Float32 PCM bytes. The SDK handles Base64
encoding and constructs the corresponding `input.append` protocol frame.
`UsageEvent` contains the current usage snapshot returned by the server.
`session.usage` exposes the same snapshot directly. The SDK never estimates
tokens from audio duration, media size, or video frame count.

## Video Sessions

```python
import os

from openbmb_realtime import RealtimeClient


client = RealtimeClient(
    api_key=os.environ["MODELBEST_API_KEY"],
    base_url="https://api.modelbest.cn",
)
session = await client.video.connect(model="minicpm-o-4.5-realtime")

await session.send_audio_video(pcm_bytes, jpeg_bytes)
await session.close()
```

`send_audio_video()` accepts one 16 kHz mono Float32 PCM chunk and one JPEG
frame, and sends them in the same `input.append` frame. The MiniCPM-o
full-duplex video protocol requires every video input to include audio.
`send_video_frame()` therefore raises `RealtimeError` instead of sending an
invalid empty-audio frame.

## Session Control

```python
session.pause()                       # Stop sending microphone input
session.resume()                      # Resume microphone input
session.interrupt()                   # Interrupt the current model response
await session.close(reason="user_stop")
```

`connect()` returns only after the server has created the session. Queueing,
initialization, protocol errors, and connection failures are handled by the SDK.
`close()` is safe to call more than once and returns `None`; read
`session.usage` after it completes when the server has sent official usage.
Usage in `response.done.response.usage` is aggregated across response turns,
while top-level usage is treated as a cumulative session snapshot.

## Authentication and Security

The client sends the API key in an `Authorization: Bearer` header and uses the
`map.realtime` WebSocket subprotocol. The API key is never placed in the URL or
written to SDK logs.

For production deployments, inject `MODELBEST_API_KEY` through an environment
variable or a managed secret provider.

## Development and Testing

Run the test suite from the package directory:

```bash
cd python
PYTHONPATH=. python -m unittest discover -s openbmb_realtime/tests -v
```

The tests use a fake WebSocket implementation and do not require a real API
key, model service, or network connection.
