streamcore (Python)

English | 简体中文

Python SDK for connecting to a streamcore server via WebRTC + WHIP, powered by aiortc.

Requirements

• Python 3.10+

Installation

pip install streamcore

Or install from source:

cd python-sdk
pip install -e .

Quick Start

import asyncio
import numpy as np
import streamcore


async def main():
    def on_transcript(entry, all_entries):
        print(f"[{entry.role}] {entry.text}")

    client = streamcore.Client(
        config=streamcore.Config(whip_endpoint="http://localhost:8080/whip"),
        events=streamcore.EventHandler(
            on_transcript=on_transcript,
            on_error=lambda err: print(f"Error: {err}"),
        ),
    )

    await client.connect()

    # Send a 20 ms frame of silence
    pcm = np.zeros(streamcore.FRAME_SIZE, dtype=np.int16)
    await client.send_pcm(pcm)

    # Receive decoded audio from the agent
    audio = await client.recv_pcm()  # numpy int16 array

    await client.disconnect()


asyncio.run(main())

API

Client(config?, events?)

Creates a new voice agent client.

Config

EventHandler

All callbacks are optional.

Methods

Audio Constants

Reconnection

A network change mid-call is recovered automatically, and the conversation survives it: the agent still knows who you are and does not replay its greeting.

The mechanism differs from the other SDKs. They keep the transport alive with an ICE restart; aiortc cannot do that (RTCPeerConnection.createOffer() takes no options, aioice fixes its ICE credentials at construction, and aiortc has no disconnected state to act on — it goes straight from connected to failed). So Python recovers by redialling with a resume token: a brand new peer connection, reattached server-side to the session it was already running. The transport is new; the conversation is not.

Status goes connected → reconnecting → connected:

import streamcore

client = streamcore.Client(
    streamcore.Config(
        whip_endpoint="http://localhost:8080/whip",
        reconnect_attempts=3,
        reconnect_delay=2.0,
    ),
    streamcore.EventHandler(
        on_reconnect=lambda e: print(f"redial {e.attempt}/{e.max_attempts}: {e.outcome}"),
    ),
)

One outcome deserves handling rather than logging:

• RECOVERED — reattached; history intact.
• RECOVERED_WITHOUT_HISTORY — the call works but the agent has forgotten everything. The server had already reaped the session, so the redial started a fresh conversation. Surface this to the user rather than letting them discover it by being asked their name again.
• FAILED — every attempt failed; status becomes disconnected.

The window is the server's session_grace_ms (30s by default), measured from when the connection dropped — not the ~25s ICE deadline the other SDKs work against. Keep reconnect_attempts × reconnect_delay (doubling each retry) inside it. Set reconnect_attempts=0 to disable and handle drops yourself.

Two caveats specific to this stack:

• Realtime (speech-to-speech) sessions cannot be resumed. Their history lives inside the provider, so the server issues no token and a redial starts a new conversation.
• A caller-supplied user_track is reused across the redial. The built-in track (the send_pcm path) is rebuilt automatically; if you pass your own track to connect(), make sure it is still live, or reconnect yourself.

The wire-format helpers for ICE restart still ship in streamcore.icerestart for callers driving another WebRTC stack:

from streamcore import whip_restart_ice, ice_fragment_from_sdp, apply_ice_fragment

Audio I/O

The SDK handles aiortc track management, av.AudioFrame construction, and resampling internally. Callers only deal with raw PCM data as numpy int16 arrays:

# Send microphone audio (960 samples = 20 ms at 48 kHz)
await client.send_pcm(pcm_int16)

# Receive agent audio
audio = await client.recv_pcm()

If you need direct track access (e.g. for a custom aiortc pipeline), pass your own AudioStreamTrack via client.connect(user_track=my_track).

Dependencies

License

Apache-2.0