# Cloudflare Realtime > CloudflareRealtimeTransport connects a bot to the Cloudflare Realtime SFU over its WebSocket adapters, so users talk over WebRTC while the bot runs in a Worker, a Durable Object or any server. Source: https://polvallverdu.github.io/pipedog/docs/transports/cloudflare-realtime/ ## Overview [#overview] `@pipedog/transport-cloudflare-realtime` connects a bot to [Cloudflare Realtime](https://developers.cloudflare.com/realtime/sfu/), Cloudflare's WebRTC SFU (formerly Calls). The user talks over WebRTC from a browser or an app, with the SFU handling the network. The bot never speaks WebRTC: the SFU's [WebSocket adapters](https://developers.cloudflare.com/realtime/sfu/features/media-transport-adapters/websocket-adapter/) convert its tracks to and from PCM. That makes WebRTC users possible where the bot can't open UDP sockets, in particular [Cloudflare Workers and Durable Objects](https://polvallverdu.github.io/pipedog/docs/deployment/cloudflare-workers.md). This package is pipedog's own; Pipecat has no counterpart. It contains: * **`CloudflareRealtimeTransport`**: the transport. It takes the sockets the SFU dials, and the browser's messages socket, from your server. * **`CloudflareRealtimeAPI`**: a small client for the Realtime HTTPS API: sessions, tracks and WebSocket adapters. * **`CloudflareRealtimeCall`**: the API calls for one user talking to one bot, in order, keeping what it allocated so it can release it. * The adapter's packet codec (`/packet`). A Worker, a Durable Object per call, and a browser page. `transport.ts`, `api.ts` and `packet.ts`. Cloudflare's documentation for sessions, tracks and adapters. ## Installation [#installation] ```bash bun add @pipedog/transport-cloudflare-realtime ``` ```ts import { CloudflareRealtimeParams, CloudflareRealtimeTransport, } from "@pipedog/transport-cloudflare-realtime/transport"; import { CloudflareRealtimeAPI, CloudflareRealtimeCall } from "@pipedog/transport-cloudflare-realtime/api"; ``` ## Prerequisites [#prerequisites] * A Realtime SFU app: its App ID and API token (the app secret), from the Cloudflare dashboard (Realtime > SFU). Keep the secret on your backend. * A public `wss://` URL for the bot: the SFU dials it, and can't reach `localhost`. For local development, use a tunnel such as `cloudflared tunnel`. ## How it works [#how-it-works] ```text browser ──WebRTC (Opus)──► Realtime SFU ──WebSocket adapter (PCM)──► bot: input socket browser ◄──WebRTC (Opus)── Realtime SFU ◄──WebSocket adapter (PCM)── bot: output socket browser ◄────────────── WebSocket (RTVI JSON) ─────────────────────► bot: messages socket ``` * **The SFU dials the bot.** Each adapter connects one track to a WebSocket URL you choose, in one direction. An adapter with `location: "remote"` streams the user's microphone to the bot (the bot's **input**); one with `location: "local"` takes the bot's audio and publishes it as a new track (the bot's **output**). The SFU can't send headers, so authenticate them with a per-call token in the URL. * **Audio** on both adapters is 48 kHz stereo 16-bit PCM, one protobuf packet per WebSocket message, at most 32 KB. The transport converts to and from the pipeline's sample rates and channels, and paces the bot's audio in real time, so an interruption stops it at once. * **App messages** can't ride the adapters, and SFU DataChannels need a WebRTC peer on both ends, so RTVI goes over a third WebSocket that the browser opens to the bot: the **messages** socket. It carries the same JSON messages SmallWebRTC sends over its data channel, one per text frame. * **Reconnects.** The SFU reconnects a dropped input adapter for up to 15 seconds; the transport waits for it (`inputReconnectTimeout`). The output adapter doesn't reconnect. The order of a call, which `CloudflareRealtimeCall` follows: 1. The browser publishes its microphone: it sends an SDP offer; your backend creates a session, publishes the track (`publishUser`) and returns the SFU's answer. 2. Once the browser is connected, your backend creates both adapters (`attachBot`). The SFU dials both endpoints before the request returns, so the transport must already exist. 3. Your backend subscribes the browser's session to the bot's track (`subscribeUser`, or `connectBot` for steps 2 and 3) and returns the SFU's offer. The browser answers, and your backend submits the answer (`renegotiate`). 4. At the end, `close()` closes the adapters and force-closes the browser's tracks. What it can't close stays in `call.state`, plain JSON you can store and retry from. ## Configuration [#configuration] `new CloudflareRealtimeTransport(params?, options?)`. `params` is a `CloudflareRealtimeParams`, with the usual transport parameters (`audioInEnabled`, `audioOutEnabled`, sample rates, channels, filters) plus: | Option | Type | Required | Description | |---|---|---|---| | `audioOutEndSilenceSecs` | `number \| undefined` | no | How much silence (seconds) to send after an EndFrame (0 for no silence). | | `audioOutAutoSilence` | `boolean \| undefined` | no | Insert silence frames when the audio output queue is empty. When false, the transport will wait for audio data instead of inserting silence. | | `audioOutWriteTimeoutSecs` | `number \| undefined` | no | How long (seconds) a single write to the transport may take before the peer is considered gone. A client that stops reading leaves the write waiting with nothing to fail, so it would otherwise never return. | | `inputReconnectTimeout` | `number \| undefined` | no | Seconds to wait for the SFU to reconnect the input adapter after its socket drops, before the client counts as disconnected. The SFU retries the same endpoint for up to 15 seconds. Defaults to 15. | | `disconnectOnMessagesClose` | `boolean \| undefined` | no | Whether the messages socket closing counts as the client disconnecting. With a browser client it is the most reliable signal that the user left, since the SFU keeps the adapters open. Defaults to true. | | `audioOutEnabled` | `boolean \| undefined` | no | Enable audio output streaming. | | `audioOutSampleRate` | `number \| null \| undefined` | no | Output audio sample rate in Hz. | | `audioOutChannels` | `number \| undefined` | no | Number of output audio channels. | | `audioOutBitrate` | `number \| undefined` | no | Output audio bitrate in bits per second. | | `audioOut10msChunks` | `number \| undefined` | no | Number of 10ms chunks to buffer for output. | | `audioOutFilter` | `BaseAudioFilter \| null \| undefined` | no | Audio filter to apply to output audio of the default destination, before the mixer. Audio the transport generates itself (mixer audio, end silence, DTMF tones) is not filtered. | | `audioOutVolume` | `number \| undefined` | no | Volume of the output audio, as a multiplier on its samples, applied after the filter and before the mixer. A `VolumeFrame` changes it while the transport runs, and a `VolumeGainFrame` adjusts it for some audio. | | `audioOutMixer` | `AudioOutMixer \| null \| undefined` | no | Audio mixer instance or destination mapping. | | `audioOutDestinations` | `string[] \| undefined` | no | List of audio output destination identifiers. | | `audioInEnabled` | `boolean \| undefined` | no | Enable audio input streaming. | | `audioInSampleRate` | `number \| null \| undefined` | no | Input audio sample rate in Hz. | | `audioInChannels` | `number \| undefined` | no | Number of input audio channels. | | `audioInFilter` | `BaseAudioFilter \| null \| undefined` | no | Audio filter to apply to input audio. | | `audioInStreamOnStart` | `boolean \| undefined` | no | Start audio streaming immediately on transport start. | | `audioInPassthrough` | `boolean \| undefined` | no | Pass through input audio frames downstream. | | `videoInEnabled` | `boolean \| undefined` | no | Enable video input streaming. | | `videoInSources` | `Record \| undefined` | no | Video sources to capture from each user as they connect, keyed by source (`"camera"`, `"screenVideo"`, or a transport-specific custom source). Sources not listed here are captured only when the application asks for them, e.g. with the transport's `captureParticipantVideo()`. Requires `videoInEnabled`. Transports that don't support it log a warning and ignore it, so check the logs if video doesn't arrive as configured. | | `videoOutEnabled` | `boolean \| undefined` | no | Enable video output streaming. | | `videoOutIsLive` | `boolean \| undefined` | no | Enable real-time video output streaming. | | `videoOutWidth` | `number \| undefined` | no | Video output width in pixels. | | `videoOutHeight` | `number \| undefined` | no | Video output height in pixels. | | `videoOutFramerate` | `number \| undefined` | no | Video output frame rate in FPS. | | `videoOutColorFormat` | `string \| undefined` | no | Video output color format string. | | `videoOutCodec` | `string \| null \| undefined` | no | Preferred video codec for output (e.g., 'VP8', 'H264', 'H265'). | | `videoOutDestinations` | `string[] \| undefined` | no | List of video output destination identifiers. | `options`: | Option | Type | Required | Description | |---|---|---|---| | `clientId` | `string \| null \| undefined` | no | The client's id, as `getClientId` returns it. Defaults to a random UUID. | | `name` | `string \| null \| undefined` | no | Optional name for the transport. | | `inputName` | `string \| null \| undefined` | no | Optional name for the input processor. | | `outputName` | `string \| null \| undefined` | no | Optional name for the output processor. | `CloudflareRealtimeAPI` takes: | Option | Type | Required | Description | |---|---|---|---| | `appId` | `string` | yes | The Realtime SFU app ID. | | `appSecret` | `string` | yes | The app secret (the API token shown with the app). Keep it on the backend. | | `baseUrl` | `string \| undefined` | no | The API base URL. Defaults to `https://rtc.live.cloudflare.com/v1`. | | `fetch` | `FetchFn \| undefined` | no | The `fetch` to use. Defaults to the global `fetch`. | | `timeout` | `number \| undefined` | no | Seconds before a request times out. Defaults to 15. | ## Usage [#usage] The bot is an ordinary pipeline around the transport: ```ts title="bot.ts" import { Pipeline } from "@pipedog/core/pipeline/pipeline"; import { PipelineParams, PipelineWorker } from "@pipedog/core/pipeline/worker"; import { WorkerRunner } from "@pipedog/core/workers/runner"; import { CloudflareRealtimeAPI, CloudflareRealtimeCall, type SessionDescription, } from "@pipedog/transport-cloudflare-realtime/api"; import { CloudflareRealtimeParams, CloudflareRealtimeTransport, } from "@pipedog/transport-cloudflare-realtime/transport"; /** Starts a bot on `transport`; it runs until the user leaves. */ async function runBot(transport: CloudflareRealtimeTransport): Promise { const pipeline = new Pipeline([ transport.input(), // ... STT, LLM context aggregators, LLM, TTS transport.output(), ]); const worker = new PipelineWorker(pipeline, { // The transport converts to and from the SFU's 48 kHz stereo. params: new PipelineParams({ audioInSampleRate: 16000, audioOutSampleRate: 24000 }), }); const runner = new WorkerRunner({ handleSigint: false, handleSigterm: false }); transport.eventHandler("on_client_connected", async () => { console.log("the SFU attached both adapters"); }); transport.eventHandler("on_client_disconnected", async () => { await runner.cancel(); }); await runner.addWorkers(worker); await runner.run(); } ``` Your server does the signaling for the browser and hands the transport the sockets. As a Durable Object, or any fetch handler that keeps one object per call: ```ts title="voice_call.ts" /** One call, e.g. a Durable Object: signaling, the sockets the SFU dials, and the bot. */ class VoiceCall { readonly #env: Env; #call: CloudflareRealtimeCall | null = null; #transport: CloudflareRealtimeTransport | null = null; #token = crypto.randomUUID(); constructor(env: Env) { this.#env = env; } async fetch(request: Request): Promise { const url = new URL(request.url); const action = url.pathname.split("/").slice(4).join("/"); // /api/call// if (action === "offer") { // 1. The browser's microphone offer: publish it, and start the bot. const { sessionDescription, mid } = (await request.json()) as { sessionDescription: SessionDescription; mid: string; }; const api = new CloudflareRealtimeAPI({ appId: this.#env.CLOUDFLARE_REALTIME_APP_ID, appSecret: this.#env.CLOUDFLARE_REALTIME_APP_SECRET, }); this.#call = new CloudflareRealtimeCall(api); this.#transport = new CloudflareRealtimeTransport( new CloudflareRealtimeParams({ audioInEnabled: true, audioOutEnabled: true }), ); void runBot(this.#transport).finally(() => this.#call?.close()); const answer = await this.#call.publishUser(sessionDescription, mid); return Response.json({ sessionDescription: answer.sessionDescription }); } if (action === "connect" && this.#call) { // 2. The browser is connected: the SFU dials both endpoints during this call. const media = `wss://${url.host}${url.pathname.replace(/connect$/, "media")}`; const offer = await this.#call.connectBot({ input: `${media}/input?token=${this.#token}`, output: `${media}/output?token=${this.#token}`, }); return Response.json({ sessionDescription: offer.sessionDescription }); } if (action === "answer" && this.#call) { // 3. The browser's answer to the bot's track. const { sessionDescription } = (await request.json()) as { sessionDescription: SessionDescription; }; await this.#call.renegotiate(sessionDescription); return Response.json({ ok: true }); } if (this.#transport && (action === "media/input" || action === "media/output")) { // The SFU can't send headers: check the token from the URL. if (url.searchParams.get("token") !== this.#token) { return new Response("forbidden", { status: 403 }); } return this.#transport.acceptSocket(action === "media/input" ? "input" : "output", request); } if (this.#transport && action === "messages") { // The browser's RTVI messages. return this.#transport.acceptSocket("messages", request); } return new Response("not found", { status: 404 }); } } ``` `acceptSocket(role, request)` upgrades the request and attaches the socket. It upgrades from the request alone on Workers, Durable Objects and Deno; on Bun, pass `{ server }` and serve with `bunWebSocket`, and on Node serve with `serveFetch()`, as for the [WebSocket transport](https://polvallverdu.github.io/pipedog/docs/transports/websocket.md). With a socket you upgraded yourself, call `attachSocket(role, WebsocketConnection.fromWebSocket(ws))`. ### The browser [#the-browser] There's no Pipecat client SDK transport for Cloudflare Realtime yet, so the browser side is plain WebRTC: publish the microphone, play the bot's track, and speak RTVI over the messages socket: ```ts title="client.ts" /** The browser: publish the microphone, play the bot, and speak RTVI over the messages socket. */ async function call(base: string): Promise<{ pc: RTCPeerConnection; messages: WebSocket }> { const post = async (action: string, body: unknown = {}) => { const response = await fetch(`${base}/${action}`, { method: "POST", body: JSON.stringify(body), }); return (await response.json()) as { sessionDescription: RTCSessionDescriptionInit }; }; const gathered = (pc: RTCPeerConnection) => new Promise((resolve) => { if (pc.iceGatheringState === "complete") return resolve(); pc.addEventListener("icegatheringstatechange", () => { if (pc.iceGatheringState === "complete") resolve(); }); setTimeout(resolve, 2000); }); const mic = await navigator.mediaDevices.getUserMedia({ audio: true }); const pc = new RTCPeerConnection({ iceServers: [{ urls: "stun:stun.cloudflare.com:3478" }], bundlePolicy: "max-bundle", }); const audio = new Audio(); audio.autoplay = true; pc.addEventListener("track", (event) => { audio.srcObject = new MediaStream([event.track]); }); // 1. Publish the microphone. const transceiver = pc.addTransceiver(mic.getAudioTracks()[0]!, { direction: "sendonly" }); await pc.setLocalDescription(await pc.createOffer()); await gathered(pc); const published = await post("offer", { sessionDescription: pc.localDescription, mid: transceiver.mid, }); await pc.setRemoteDescription(published.sessionDescription); await new Promise((resolve) => { pc.addEventListener("connectionstatechange", () => { if (pc.connectionState === "connected") resolve(); }); }); // 2. The messages socket (RTVI), then the bot's track. const messages = new WebSocket(`${base.replace(/^http/, "ws")}/messages`); await new Promise((resolve) => messages.addEventListener("open", resolve, { once: true })); const subscription = await post("connect"); await pc.setRemoteDescription(subscription.sessionDescription); await pc.setLocalDescription(await pc.createAnswer()); await gathered(pc); await post("answer", { sessionDescription: pc.localDescription }); // 3. Ready: the bot can greet the user now. messages.send( JSON.stringify({ label: "rtvi-ai", type: "client-ready", id: crypto.randomUUID(), data: { version: "2.0.0", about: { library: "my-app" } }, }), ); return { pc, messages }; } ``` RTVI works as with any transport, since the bot's `RTVIProcessor` sends and receives through the transport's message frames: the bot answers `client-ready` with `bot-ready`, and sends transcriptions and bot output. A Pipecat client SDK transport would need to do the signaling above and carry RTVI over the messages socket. ### Event handlers [#event-handlers] | Event | Description | | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `on_client_connected` | Both adapter sockets are attached. Receives the client (`getClientId` gives its id). | | `on_client_disconnected` | The input adapter didn't reconnect within `inputReconnectTimeout`, the output adapter closed, or the messages socket closed (with `disconnectOnMessagesClose`, the default). Not when the transport itself closes. | | `on_app_message` | A JSON message arrived on the messages socket. Receives the message and the client's id. | ### Notes [#notes] * Greet the user when their browser can hear the bot: the bot's track only plays once the browser has answered the subscription offer. The example greets on RTVI's `on_client_ready`, which the browser triggers after `renegotiate`. * The transport doesn't end the call at the SFU: call `CloudflareRealtimeCall.close()` when the bot stops, or the SFU keeps redialing a closed input adapter for 15 seconds. * Input is mono at the pipeline's rate; with `audioInChannels: 2` and a 48 kHz pipeline it stays stereo. Set `audioOutSampleRate: 48000` to skip output resampling. * Audio that arrives before the pipeline starts is dropped; bot audio sent while the output adapter isn't attached is dropped. ## RealtimeKit [#realtimekit] RealtimeKit, Cloudflare's meeting SDK, is out of scope for now: its agent participant can't send or receive audio yet (it is "planned" in Cloudflare's capability matrix). This transport uses the SFU directly. ## Verification [#verification] The transport, the API helper and the example are tested against fakes built from Cloudflare's documentation: a fake SFU that dials the bot's endpoints as the adapters do, and a fake API. They haven't been run against the real Realtime service yet. A live API test runs when `CLOUDFLARE_REALTIME_APP_ID` and `CLOUDFLARE_REALTIME_APP_SECRET` are set. ## Runtime notes [#runtime-notes] | Runtime | Works | What it needs | | ------------------ | ----- | ------------- | | Bun | ✅ yes | Nothing. | | Node.js | ✅ yes | Nothing. | | Deno | ✅ yes | Nothing. | | Cloudflare Workers | ✅ yes | Nothing. | Generated from `pipecat.runtimes` and `pipecat.compat` in `packages/transport-cloudflare-realtime/package.json`: ✅ means the compat smoke test passes on that runtime.