Metadata-Version: 2.4
Name: openbmb-realtime
Version: 0.1.0
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

OpenBMB MiniCPM-o Realtime 的异步 Python SDK。SDK 封装 WebSocket 连接、排队、
`session.init`、事件归一化、音频/视频帧编码、打断、暂停、恢复和幂等关闭。

## 安装

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

## 音频会话

```python
import asyncio
import os

from openbmb_realtime import AudioDelta, RealtimeClient, TextDelta


async def microphone_stream():
    # 返回 16 kHz、单声道、float32 PCM bytes。
    raise NotImplementedError


async def play_audio(audio: bytes):
    # 将模型输出的音频 bytes 交给播放器。
    raise NotImplementedError


async def main():
    client = RealtimeClient(
        api_key=os.environ["MODELBEST_API_KEY"],
        base_url="https://api.modelbest.cn/dev",
    )
    session = await client.audio.connect(
        model="minicpm-o-4.5-realtime",
        system_prompt="你是一个有帮助的语音助手",
    )

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

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


asyncio.run(main())
```

`send_audio()` 的输入必须是 16 kHz、单声道、float32 PCM 字节流。SDK 会负责
Base64 编码和 `input.append` 帧构造。

## 视频会话

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

## 会话控制

```python
session.pause()
session.resume()
session.interrupt()
await session.close(reason="user_stop")
```

`connect()` 只有在服务端返回 `session.created` 后才会返回。连接过程中的排队、
初始化和错误由 SDK 处理；`close()` 可以重复调用。

## 鉴权与安全

Python 客户端使用 `Authorization: Bearer` 请求头和 `map.realtime` WebSocket 子协议。
API Key 不会进入 URL，也不会被 SDK 写入日志。生产环境应通过环境变量或密钥管理系统
注入 `MODELBEST_API_KEY`。

## 本地测试

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

测试使用 fake WebSocket，不需要真实 API Key、模型服务或网络连接。
