Getting Started

Five minutes from pip install to two peers talking.

Build with AI Tools

Use our Python SDK reference file with Claude Code, Cursor, or any AI coding tool so it writes code against the current metered-realtime API. Copy-paste prompts available on the Build with AI page. Never paste a secret key into an AI tool: keep it in an environment variable on your server and tell the tool the variable's name.

Install

pip install "metered-realtime[webrtc]"   # MeteredPeer + media
# or, pub/sub only (no WebRTC stack):
pip install metered-realtime

Requires Python 3.10+. The webrtc extra pulls in aiortc + av (a native build); the pub/sub-only install needs just websockets.

Get a key from the dashboard

1. Sign up (skip if you already have an account).
2. Dashboard → Realtime Messaging → API Keys → Create key.
3. Pick a key type:

Key type	When to use	Where it goes
pk_live_… publishable	Prototypes, trusted server-side processes, no per-user scoping	Directly in your code / env
Secret key (sk_id_… + sk_secret_…)	Per-user permissions, custom peer_id, embedded TURN creds	Server-side only — your backend mints JWTs

For the rest of this page we use the pk_live_ path. See Authentication when you're ready for JWT minting (and to give a peer Send permission, which publishable keys don't grant by default).

Your first connection — pub/sub

SignallingClient is the smaller surface: pub/sub, no WebRTC. Good for telemetry, coordination, anywhere you don't need peer-to-peer media.

import asyncio
from metered_realtime import SignallingClient, Connected, Message

async def main() -> None:
    async with SignallingClient(api_key="pk_live_…") as client:
        @client.on(Connected)
        def _(ev: Connected) -> None:
            print("connected as", ev.peer_id)

        @client.on(Message)
        def _(ev: Message) -> None:
            print(f"{ev.sender_peer_id} → {ev.channel}:", ev.data)

        await client.subscribe("room-42")
        await client.publish("room-42", {"hello": "world"})
        await asyncio.sleep(5)

asyncio.run(main())

Run two copies side by side. Each gets a different peer_id; one's publish shows up as a Message event in the other. That's the whole pub/sub model — subscribe to channels, publish to channels, receive Message events.

Add peers (WebRTC)

MeteredPeer is the same connection plus channel-driven peer discovery, a per-peer connection lifecycle, and media fan-out. Use it when peers exchange audio/video or low-latency P2P data.

import asyncio
from metered_realtime import MeteredPeer, PeerJoined, Track, Data

async def main() -> None:
    async with MeteredPeer(api_key="pk_live_…") as peer:
        @peer.on(PeerJoined)
        def _(ev: PeerJoined) -> None:
            print("someone joined:", ev.peer.id)

            @ev.peer.on(Track)
            def _(t: Track) -> None:
                # t.track is an aiortc MediaStreamTrack — record it, transcribe it, …
                print("receiving", t.track.kind)

        @peer.on(Data)
        def _(ev: Data) -> None:
            print(ev.sender_peer_id, "said:", ev.data)

        await peer.join("my-room")
        await peer.send({"hi": "everyone"})
        await asyncio.sleep(30)

asyncio.run(main())

What this does:

1. join("my-room") connects, subscribes you to my-room, and asks the server who else is here.
2. For every peer the server reports, PeerJoined fires and the SDK opens a peer-to-peer connection under the hood.
3. You receive each peer's media on the Track event; anything you attached with add_track is sent to every peer.
4. peer.send(data) broadcasts to everyone in the channel; peer.send_to(remote_id, data) targets one peer.

Peers don't arrive synchronously with join. The server's first presence for the channel lands shortly after the subscribe ack, so peer.remote_peers is typically still empty right after await peer.join(...). Populate your state from the PeerJoined handler, not a snapshot.

Sending media

There's no getUserMedia on the server, so you supply the track. For an AI agent streaming synthesized speech:

from metered_realtime import AudioSource, MediaStream

source = AudioSource(input_rate=16_000)        # 16 kHz mono PCM in
peer.add_track(source, MediaStream(id="agent-voice"))
await source.push(pcm_bytes)                    # emitted as real-time 48 kHz audio
source.end()

Or feed from a file / IP camera (.audio / .video is None if the source lacks it):

from metered_realtime import from_file, from_rtsp

peer.add_track(from_file("clip.mp4").audio)
peer.add_track(from_rtsp("rtsp://cam:554/stream").video)

See the Media reference for AudioSource, iter_frames (consuming a peer's audio for STT), and all the source helpers.

Messages are server-routed by default

peer.send(data) / peer.send_to(id, data) are server-routed, not P2P over a data channel. They work before ICE completes, but go Peer → Metered server → Peer, so each message counts against your signalling quota. For low-latency P2P data, open a real data channel:

from metered_realtime import DataChannel, DCMessage

raw = ev.peer.create_data_channel("game-state")
dc = DataChannel(raw)

@dc.on(DCMessage)
def _(m: DCMessage) -> None:
    handle(m.data)

See Data Channels for the backpressure-aware wrapper and the reconnect gotcha.

Next step depends on what you're building

You're building…	Read
An AI voice agent	Guide: AI Agent Communication
An IoT / edge camera bridge	Guide: IoT Telemetry
Anything going to production	Guide: Reconnect Best Practices — required reading
Per-user JWTs (custom peer_id, channel permissions, TURN creds)	Guide: Authentication