rstream Python SDK

rstreamlabs-rstream is the Python SDK for rstream tunnels and webhook receivers. Python code imports it as rstream.

The SDK is async-first, framework-neutral, and uses the same runtime protocol and CLI-compatible configuration model as the Go, JavaScript, Java, and C++ SDKs. It is designed for Python services that need to publish local HTTP applications, dial private tunnels, accept bytestream tunnel connections directly, or verify rstream webhook deliveries in a backend framework.

SDK Surface

This repository targets Python 3.10 and newer.

The first public release focuses on the Python-native surface that is ready for production use:

Datagram tunnels, QUIC runtime transport, HTTP/3 tunnel creation, custom transport proxies, and external credential stores are rejected explicitly by this SDK version instead of being ignored.

Install

pip install rstreamlabs-rstream

For ASGI/FastAPI helpers:

pip install "rstreamlabs-rstream[asgi]"

For WSGI/Flask/Django helpers:

pip install "rstreamlabs-rstream[wsgi]"

For managed project endpoint discovery through the Control plane API:

pip install "rstreamlabs-rstream[api]"

For local example applications:

pip install "rstreamlabs-rstream[examples]"

Configuration

The SDK reads the same config file as the CLI by default:

~/.rstream/config.yaml

Configuration is resolved in this order:

1. Explicit Client(...) options.
2. Environment variables.
3. The selected context in the config file.
4. SDK defaults.

Environment variables

RSTREAM_ENGINE_ADDRESS is also accepted for compatibility with older local SDK workflows. Prefer RSTREAM_ENGINE in new code.

See docs/001-configuration.md for supported YAML fields and error behavior.

FastAPI tunnel

import asyncio

from fastapi import FastAPI

import rstream

app = FastAPI()


@app.get("/")
async def root() -> dict[str, str]:
    return {"status": "ok"}


async def main() -> None:
    async with (
        rstream.Client.from_env() as client,
        await client.connect() as control,
    ):
        tunnel = await control.create_tunnel(
            protocol="http",
            http_version="http/1.1",
            publish=True,
        )
        print("Forwarding address:", tunnel.forwarding_address)
        await rstream.asgi.serve(app, tunnel)


asyncio.run(main())

The ASGI helper parses accepted rstream streams and dispatches them to the app in-process. It does not start a loopback server.

Local forwarding

forward_to() remains available for existing services that already listen on a local TCP port:

await tunnel.forward_to("127.0.0.1", 8000)

It keeps accepting rstream streams and relays them to the local TCP service until the tunnel or control channel is closed.

The control channel uses a negotiated heartbeat deadline. If that deadline expires, or the control transport ends unexpectedly, the SDK immediately stops accepting new streams but lets already accepted streams and active forward_to() relays drain normally. This avoids interrupting an established application session because of a transient or asymmetric control-path outage. An explicit tunnel/control close or a protocol violation remains a hard close and terminates active local forwarding. Applications should use each stream's EOF/error as its payload-lifecycle signal; control.done() only describes the control plane.

Private dial

import asyncio

import rstream


async def main() -> None:
    async with (
        rstream.Client.from_env() as client,
        await client.dial("private-api") as stream,
    ):
        stream.write(b"ping")
        await stream.drain()
        print(await stream.read(1024))


asyncio.run(main())

Private tunnels are addressed by name or ID. They do not expose a public forwarding address.

Live inventory and real-time events

The engine exposes its live state through the same client. list_tunnels and list_clients return the current inventory, optionally filtered, and watch streams lifecycle events (tunnel.created, tunnel.updated, tunnel.deleted, and the client equivalents) as an async iterator. Label filters make the inventory usable as a service registry: tag tunnels at creation time, then discover them by label.

import asyncio

import rstream


async def main() -> None:
    async with rstream.Client.from_env() as client:
        filters = rstream.TunnelFilters(labels={"role": "inference"})
        for tunnel in await client.list_tunnels(filters=filters):
            print(tunnel.properties.name, tunnel.status)
        async for event in client.watch(tunnels=filters, transport="websocket"):
            print(event.type, event.object)


asyncio.run(main())

Listing and the SSE transport require the api extra (pip install "rstreamlabs-rstream[api]"); the WebSocket transport requires the realtime extra, which adds the websockets dependency. The watch connection uses the runtime token and ends when the surrounding task is cancelled.

Webhook receiver

from fastapi import Request

import rstream


async def handle_webhook(request: Request) -> None:
    payload = await request.body()
    signature = request.headers["rstream-signature"]
    secret = "whsec_..."

    event = rstream.verify_event(payload, signature, secret)
    resource_id = event.object.get("id")
    if event.type == "tunnel.created":
        print("Tunnel is online:", resource_id)

event.id is suitable for idempotency. Keep the raw request body unchanged when verifying the signature.

See docs/003-webhooks.md for the payload shape and headers.

Examples

Development

python -m venv .venv
. .venv/bin/activate
pip install -e ".[dev]"
ruff check .
ruff format --check .
mypy
pytest
python -m build

Real-engine tests are opt-in:

RSTREAM_PYTHON_E2E=1 pytest tests/e2e

See docs/004-testing.md for local-engine and managed-environment test commands.

Repository setup and release

The intended GitHub repository is rstreamlabs/rstream-python. CI requires no secret for normal pull request checks. Release automation uses release-please and requires the maintainer-managed RELEASE_PLEASE_TOKEN secret plus the CI_ALLOWED_ACTOR repository variable.

See docs/006-github-setup.md before creating or publishing the repository.

License

Apache-2.0. See LICENSE.

Engine HTTP access with certificates

The existing list_clients, list_tunnels and watch methods support enrolled file-based mTLS identities. Keep one Client in an async with block. It discovers the dedicated project API through anonymous, verified HTTPS and caches the address in memory, normally for one hour. Concurrent callers share discovery; cancellation of one waiter does not cancel others, and closing the client cancels pending lookup work. No YAML field, persistent cache or new login is required.

Discovery never sends the certificate, a token, cookies or Control plane headers, refuses redirects and verifies the server even when runtime TLS is marked insecure. The ordinary Engine hostname never requests a certificate. Runtime requests use the discovered authority with the existing identity and project permissions. Both SSE and WebSocket certificate watches use TLS/TCP. Python does not implement HTTP/3 or a TURN generation helper; use Go/JavaScript or the documented Engine API for TURN issuance. Control plane administration and region selectors still require a token; provision an explicit Engine address for a certificate-only device.

Tests in tests/integration/test_engine_mtls.py use real TLS servers for discovery, inventory, both watch transports, trust validation and cancellation. Opt-in real-Engine inventory checks use RSTREAM_PYTHON_E2E_ENGINE_API=1 with the existing E2E credential configuration.