Metadata-Version: 2.4
Name: openbmb-realtime
Version: 0.1.2
Summary: Async Python SDK for OpenBMB MiniCPM-o Realtime sessions
Author: OpenBMB
License-Expression: 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


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)


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.

## 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_video_frame(jpeg_bytes)
await session.send_audio(pcm_bytes)
await session.close()
```

`send_video_frame()` accepts JPEG bytes. Video and audio can be sent through the
same session.

## 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.

## 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.
