Python SDK

The Python SDK is built for backend services that should own their rstream tunnel lifecycle from application code. It reads the same config file as the CLI and other SDKs, opens the engine control channel, creates bytestream tunnels, serves ASGI or WSGI applications directly from accepted streams, forwards streams to existing local services when needed, and dials private tunnels by name or ID.

This page focuses on the Python runtime surface. For raw HTTP routes, see APIs. For webhook delivery semantics, see Webhooks.

The Go SDK remains the reference SDK and covers the broadest protocol surface today. The Python SDK is intentionally narrower: it is designed for Python services, FastAPI applications, workers, device agents, notebooks, and backend automation where tunnel creation or private dialing should happen inside the process rather than through a shell command.

Use it when the Python process is already the natural owner of the service lifecycle. A FastAPI application can publish itself during startup, a device agent can maintain a private tunnel for command traffic, and a backend worker can dial a private resource without spawning the CLI or passing sockets through another supervisor.

SDK Surface

The first Python SDK surface focuses on async bytestream workflows:

Unsupported transport and credential-storage settings fail during configuration resolution instead of being ignored. This keeps Python integrations predictable when they share the same ~/.rstream/config.yaml file as the CLI or another SDK.

Choosing a Pattern

The SDK exposes a few small primitives rather than a large framework. Pick the path that matches the ownership boundary in your application.

The common thread is that rstream remains part of the application lifecycle. The process decides when the tunnel exists, which labels describe it, how shutdown is handled, and whether public access is published or kept private.

Runtime Model

The runtime path is async-first. A Client resolves configuration, connects to the engine, and opens a control channel. The control channel creates published or private tunnels. Published tunnels accept streams from the rstream edge and can serve ASGI or WSGI applications in-process, or forward to a local TCP service that already exists. Private tunnels are consumed by Client.dial() and return an async byte stream directly to application code.

The control channel is long-lived. Keep it open for as long as the tunnels it owns should remain available. Closing the channel or the tunnel tears down the runtime state in the engine; it does not depend on a separate CLI process staying alive. Use async context managers for clients, control channels, and streams so shutdown remains explicit and cancellation paths close sockets cleanly:

import asyncio
 
import rstream
 
 
async def main() -> None:
    async with (
        rstream.Client.from_env() as client,
        await client.connect() as control,
    ):
        tunnel = await control.create_tunnel(
            name="api",
            protocol="http",
            http_version="http/1.1",
            publish=True,
        )
        print("Forwarding address:", tunnel.forwarding_address)
        await tunnel.close()
 
 
asyncio.run(main())

When the engine and token are already known, create the client directly instead of reading the local config file:

import os
 
import rstream
 
client = rstream.Client(
    engine="project-endpoint.cluster.example.rstream.test:443",
    read_config_file=False,
    token=os.environ["RSTREAM_AUTHENTICATION_TOKEN"],
)

The low-level tunnel object also supports direct stream acceptance. Most Python web applications should use the ASGI or WSGI helper; existing port-bound services can use forward_to(). Advanced integrations can iterate over accepted streams and attach their own framing, metrics, or admission logic.

Install

pip install rstreamlabs-rstream

For ASGI helpers:

pip install "rstreamlabs-rstream[asgi]"

For WSGI helpers:

pip install "rstreamlabs-rstream[wsgi]"

For managed project endpoint discovery through the Control plane API:

pip install "rstreamlabs-rstream[api]"

Extras are intentionally separate. The base runtime stays small for agents and workers that already know their engine address. Add asgi or wsgi only when the process serves a Python web application, and add api only when it should resolve a hosted project endpoint through the Control plane API.

The PyPI distribution name is rstreamlabs-rstream; Python code imports the package as rstream.

Authentication and Project Setup

The Python SDK uses the same runtime config model as the rstream CLI.

Typical developer-machine setup:

rstream login
rstream project use <project-endpoint>

The SDK then reads the same ~/.rstream/config.yaml file and context selection model as the CLI. Explicit client options are read first, then environment variables, then the selected context from the config file.

Token authentication and mTLS are mutually exclusive for the engine control channel. If an explicit engine override is set, stored credentials from another context are refused unless the credential is also provided explicitly. That fail-closed behavior prevents a token from a selected CLI context from being sent to an unrelated engine by accident.

For hosted projects, pass project_endpoint or set it in the selected config context. The api extra lets the SDK resolve the managed project through the Control plane API before opening the runtime connection.

For automation and containers, pass engine, token, and read_config_file=False explicitly when the runtime should not depend on a developer-machine config file. For local development, Client.from_env() keeps Python behavior aligned with rstream login and rstream project use.

FastAPI and ASGI

The ASGI helper is the shortest path for FastAPI, Starlette, and other ASGI applications. It parses accepted rstream streams and dispatches requests to the app in-process. No loopback HTTP server is started.

Application routing, middleware, authentication, and observability remain normal ASGI concerns. The SDK owns the remote tunnel lifecycle and the HTTP/1.1 stream adapter.

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

Flask, Django, and WSGI

The WSGI helper follows the same direct model for Flask, Django, and other WSGI applications.

import asyncio
 
from flask import Flask, jsonify
 
import rstream
 
app = Flask(__name__)
 
 
@app.get("/")
def root() -> object:
    return jsonify(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.wsgi.serve(app, tunnel)
 
 
asyncio.run(main())

Managed Local Forwarding

Use this path when a Python process should publish an existing local service. The call to forward_to() is intentionally long-running: it keeps accepting remote streams until the tunnel is closed or the task is cancelled.

import asyncio
 
import rstream
 
 
async def main() -> None:
    async with (
        rstream.Client.from_env() as client,
        await client.connect() as control,
    ):
        tunnel = await control.create_tunnel(
            name="python-api",
            protocol="http",
            http_version="http/1.1",
            labels={"service": "api", "runtime": "python"},
            publish=True,
            auth=rstream.TunnelAuth(token=True, rstream=True),
        )
        print("Forwarding address:", tunnel.forwarding_address)
        await tunnel.forward_to("127.0.0.1", 8000)
 
 
asyncio.run(main())

Private Dial

Private tunnels are dialed by tunnel name or ID and return an async byte stream. Application code can layer HTTP, a custom binary protocol, or another bytestream protocol on top of the returned stream.

By default, private dialing uses the SDK's configured zero-RTT behavior so the first bytes can be sent as soon as the tunnel stream is opened. Pass zero_rtt=False when the caller should wait for an explicit engine response before writing application data.

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

Webhook Receivers

The SDK also includes helpers for Python backends that receive rstream webhooks. Verify the raw request body against the rstream-signature header before parsing or storing the event. The returned event id is stable and can be used as the receiver-side idempotency key.

Webhook payloads use the same project lifecycle events exposed by the platform. They are useful when an external system needs durable state around tunnels that are otherwise runtime resources: last seen timestamps, device online/offline status, inventory reconciliation, or automation triggered by tunnel creation and deletion.

import os
 
from fastapi import FastAPI, Header, HTTPException, Request
 
import rstream
 
app = FastAPI()
 
 
@app.post("/webhooks/rstream")
async def rstream_webhook(
    request: Request,
    rstream_signature: str = Header(alias="rstream-signature"),
) -> dict[str, bool]:
    secret = os.environ["RSTREAM_WEBHOOK_SECRET"]
    payload = await request.body()
    try:
        event = rstream.verify_event(payload, rstream_signature, secret)
    except rstream.RstreamError as error:
        raise HTTPException(status_code=400, detail=str(error)) from error
 
    resource_id = event.object.get("id")
    print(event.type, event.id, resource_id)
    return {"received": True}

Use tunnel labels when webhook receivers need to build persistent application state around runtime resources that may appear and disappear frequently.

Examples

The repository includes runnable examples for the common integration paths:

Each example is deliberately small and uses the same configuration model as production code. Start with the one that matches the integration boundary, then move engine, token, labels, and endpoint policy into explicit application configuration as the deployment matures.

Environment variables

The Python SDK follows the runtime resolution model used by the CLI. Explicit SDK options are read first, then environment variables, then the selected context from RSTREAM_CONFIG or the default ~/.rstream/config.yaml file.

RSTREAM_AUTHENTICATION_TOKEN and the mTLS certificate/key pair are mutually exclusive for the agent control-channel connection. When both mTLS variables are set, config-derived tokens are not used for that connection.