# WebSocket

> WebSocket transports with FastAPIWebsocketTransport (bring your own socket), SingleClientWebsocketServerTransport, WebsocketClientTransport and RTVIClientTransport, plus endpoint helpers for each runtime.

Source: https://polvallverdu.github.io/pipedog/docs/transports/websocket/



## Overview [#overview]

`@pipedog/transport-websocket` carries frames over a WebSocket, encoded by a
[serializer](https://polvallverdu.github.io/pipedog/docs/serializers/protobuf.md). It ports Pipecat's four WebSocket transports, and adds
the helpers that accept a socket on each runtime:

* **`FastAPIWebsocketTransport`**: runs a pipeline on a socket your server already accepted.
  Upstream takes the socket a FastAPI endpoint accepted; here it takes a `WebsocketConnection`
  from any runtime. This is the one to use for a web server, a Worker or telephony.
* **`SingleClientWebsocketServerTransport`**: starts its own WebSocket server and serves one
  client at a time. For local development and single-session bots.
* **`WebsocketClientTransport`**: connects out to a WebSocket server.
* **`RTVIClientTransport`**: a `WebsocketClientTransport` that speaks RTVI as a client to a bot.
* **`WebsocketConnection`**: the runtime-agnostic socket all four transports talk to.
* Endpoint helpers that produce a `WebsocketConnection`: `websocketFetchHandler` (fetch
  handlers, with `upgradeWebSocket` and `bunWebSocket`), `serveFetch` (serve a fetch handler with
  WebSocket upgrades on Bun, Node and Deno), `websocketServe` (listen on a port) and
  `honoWebsocket` (Hono).

<Callout type="info">
  For a browser voice client, WebRTC handles network loss and jitter better. See
  [SmallWebRTC](https://polvallverdu.github.io/pipedog/docs/transports/smallwebrtc.md). WebSockets fit server-to-server links, telephony and
  runtimes without UDP, such as Cloudflare Workers.
</Callout>

<Cards>
  <Card title="Upstream: FastAPI WebSocket" href="https://docs.pipecat.ai/api-reference/server/services/transport/fastapi-websocket">
    Pipecat's page for `FastAPIWebsocketTransport`.
  </Card>

  <Card title="Upstream: WebSocket server" href="https://docs.pipecat.ai/api-reference/server/services/transport/websocket-server">
    Pipecat's page for the WebSocket server transport.
  </Card>

  <Card title="Source" href="https://github.com/polvallverdu/pipedog/tree/main/packages/transport-websocket/src">
    `fastapi.ts`, `server.ts`, `client.ts`, `rtvi_client.ts`, `socket.ts`, `fetch.ts`, `serve.ts`
    and `hono.ts`.
  </Card>

  <Card title="Cloudflare Workers" href="/docs/deployment/cloudflare-workers">
    Running a voice agent in a Durable Object over a WebSocket.
  </Card>
</Cards>

## Installation [#installation]

```bash
bun add @pipedog/transport-websocket
```

To use `serveFetch`, `websocketServe` or `SingleClientWebsocketServerTransport` on Node, also
install the `ws` package (an optional peer dependency):

```bash
npm install ws
```

```ts
import {
  FastAPIWebsocketParams,
  FastAPIWebsocketTransport,
} from "@pipedog/transport-websocket/fastapi";
import {
  SingleClientWebsocketServerParams,
  SingleClientWebsocketServerTransport,
} from "@pipedog/transport-websocket/server";
import { WebsocketClientParams, WebsocketClientTransport } from "@pipedog/transport-websocket/client";
import { RTVIClientTransport } from "@pipedog/transport-websocket/rtvi_client";
import { WebsocketConnection } from "@pipedog/transport-websocket/socket";
import { bunWebSocket, upgradeWebSocket, websocketFetchHandler } from "@pipedog/transport-websocket/fetch";
import { serveFetch, websocketServe } from "@pipedog/transport-websocket/serve";
import { honoWebsocket } from "@pipedog/transport-websocket/hono";
```

The package depends on `@pipedog/serializer-protobuf` and `@pipedog/serializer-rtvi-client`, the
default serializers of the two client transports.

## Prerequisites [#prerequisites]

* A serializer that matches the other end: `ProtobufFrameSerializer` for Pipecat's WebSocket
  clients, a telephony serializer such as [Telnyx](https://polvallverdu.github.io/pipedog/docs/serializers/telnyx.md) for phone calls.
* For `allowedOrigins`, the list of origins allowed to connect, or the `PIPECAT_ALLOWED_ORIGINS`
  environment variable (comma-separated).

## FastAPIWebsocketTransport [#fastapiwebsockettransport]

### Configuration [#configuration]

`new FastAPIWebsocketTransport(websocket, params, options?)`. `websocket` is any object that
implements the `FastAPIWebSocket` interface; a `WebsocketConnection` does. `params` is a
`FastAPIWebsocketParams`:

| Option | Type | Required | Description |
|---|---|---|---|
| `audioOutDestinations` | `string[] \| undefined` | no | List of audio output destination identifiers. |
| `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. |
| `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. |
| `addWavHeader` | `boolean \| undefined` | no | Whether to add WAV headers to audio frames. |
| `serializer` | `FrameSerializer \| null \| undefined` | no | Frame serializer for encoding/decoding messages. |
| `sessionTimeout` | `number \| null \| undefined` | no | Session timeout in seconds, null for no timeout. |
| `fixedAudioPacketSize` | `number \| null \| undefined` | no | Optional fixed-size packetization for raw PCM audio payloads. Useful when the remote WebSocket media endpoint requires strict audio framing. |
| `allowedOrigins` | `string[] \| undefined` | no | List of allowed origins. Empty list allows all origins. When set, connections with a missing or disallowed Origin header are rejected. Defaults to `PIPECAT_ALLOWED_ORIGINS` env var (comma-separated). |
| `wsCloseTimeout` | `number \| undefined` | no | Maximum time, in seconds, to wait for the WebSocket close handshake during disconnect. See `FastAPIWebsocketParams`. |
| `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. |
| `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<string, VideoInSourceParams \| VideoInSourceParamsOptions> \| 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 |
|---|---|---|---|
| `inputName` | `string \| null \| undefined` | no | Optional name for the input processor. |
| `outputName` | `string \| null \| undefined` | no | Optional name for the output processor. |

### Usage [#usage]

Build the transport inside a function that takes the accepted connection. The endpoint helpers
below call it once per connection:

```ts title="bot.ts"
import { createTask } from "@pipedog/asyncio";
import { Pipeline } from "@pipedog/core/pipeline/pipeline";
import { PipelineWorker } from "@pipedog/core/pipeline/worker";
import { WorkerRunner } from "@pipedog/core/workers/runner";
import { ProtobufFrameSerializer } from "@pipedog/serializer-protobuf/protobuf";
import {
  FastAPIWebsocketParams,
  FastAPIWebsocketTransport,
} from "@pipedog/transport-websocket/fastapi";
import { websocketFetchHandler } from "@pipedog/transport-websocket/fetch";
import { honoWebsocket } from "@pipedog/transport-websocket/hono";
import { serveFetch, websocketServe } from "@pipedog/transport-websocket/serve";
import { WebsocketConnection } from "@pipedog/transport-websocket/socket";

/** Runs one bot on an accepted WebSocket connection. */
async function runBot(connection: WebsocketConnection): Promise<void> {
  const transport = new FastAPIWebsocketTransport(
    connection,
    new FastAPIWebsocketParams({
      audioInEnabled: true,
      audioOutEnabled: true,
      serializer: new ProtobufFrameSerializer(),
      sessionTimeout: 300,
    }),
  );

  const pipeline = new Pipeline([
    transport.input(),
    // ... STT, LLM context aggregators, LLM, TTS
    transport.output(),
  ]);
  const worker = new PipelineWorker(pipeline);
  const runner = new WorkerRunner({ handleSigint: false });

  transport.eventHandler("on_client_connected", async () => {
    console.log("client connected");
  });
  transport.eventHandler("on_client_disconnected", async () => {
    await runner.cancel();
  });

  await runner.addWorkers(worker);
  await runner.run();
}
```

`websocketFetchHandler` turns it into a fetch handler,
`(request, server?) => Promise<Response>`. It answers 426 to a request that is not a WebSocket
upgrade. Cloudflare Workers, Durable Objects and Deno upgrade from the request alone:

```ts
// Cloudflare Workers (and Durable Objects) and Deno upgrade the request from fetch.
const handler = websocketFetchHandler(runBot);

const workerEntry = {
  fetch(request: Request): Promise<Response> {
    return handler(request);
  },
};
```

Bun upgrades with its server, so pass the `server` Bun gives `fetch` and serve with the
`bunWebSocket` handlers. `serveFetch` serves a fetch handler with WebSocket upgrades on Bun, Node
(with the `ws` package) and Deno, so one entry point runs on all three. `upgradeWebSocket(request, { server })`, from the same module, is the upgrade on its own, for your own handlers:

```ts
// Bun, Node (with the `ws` package) and Deno: serve the same fetch handler, with upgrades.
const fetchServer = await serveFetch(handler, { host: "0.0.0.0", port: 8765 });

// Later: stop listening.
await fetchServer.close();
```

```ts title="Bun.serve"
Bun.serve({
  fetch: (request, server) => handler(request, server),
  websocket: bunWebSocket, // from @pipedog/transport-websocket/fetch
});
```

`websocketServe` listens on a host and port with the runtime's own server, on Bun, Node and Deno,
and takes the connection handler directly:

```ts
// Bun, Node (with the `ws` package) and Deno: listen on a port.
const server = await websocketServe({
  handler: runBot,
  host: "0.0.0.0",
  port: 8765,
  origins: ["https://app.example.com"],
});

// Later: stop listening, close the open connections and wait for their handlers.
await server.close();
```

With Hono, `honoWebsocket` returns the events that Hono's `upgradeWebSocket()` expects, on any
Hono adapter. Pass the upgrade request's headers so `allowedOrigins` can be checked:

```ts
// Pass the result to Hono's upgradeWebSocket() (hono/bun, hono/deno,
// hono/cloudflare-workers or @hono/node-ws).
const honoEvents = (request: Request) => honoWebsocket(runBot, { headers: request.headers });
```

```ts title="app.ts"
app.get(
  "/ws",
  upgradeWebSocket((c) => honoEvents(c.req.raw)),
);
```

A full Hono app, and the same bot mounted in Express, Next.js, SvelteKit and other frameworks, is
in [`examples/framework-snippets`](https://github.com/polvallverdu/pipedog/tree/main/examples/framework-snippets).
See also [Frameworks](https://polvallverdu.github.io/pipedog/docs/frameworks.md).

### Event handlers [#event-handlers]

| Event                    | Description                                                                         |
| ------------------------ | ----------------------------------------------------------------------------------- |
| `on_client_connected`    | The input transport started. Receives the socket.                                   |
| `on_client_disconnected` | The peer closed the socket (not when the transport closes it). Receives the socket. |
| `on_session_timeout`     | `sessionTimeout` seconds passed since the transport started. Receives the socket.   |

### Notes [#notes]

* The constructor throws if `allowedOrigins` is non-empty and the socket's `Origin` header is
  missing or not listed. The caller closes the socket in that case; `websocketFetchHandler`,
  `websocketServe` and `honoWebsocket` log the error and close it for you.
* When the input starts it pushes a `ClientConnectedFrame`, then reads messages and deserializes
  them with `serializer`. Without a serializer, incoming messages are dropped and nothing is sent.
* Output audio is paced to real time. `fixedAudioPacketSize` cuts raw PCM into fixed-size
  packets, for media endpoints that need strict framing. `addWavHeader` wraps each chunk in a WAV
  header.
* `wsCloseTimeout` (default 0.5 seconds) bounds how long a disconnect waits for the peer to
  acknowledge the close, so a dead peer does not stall shutdown.

## SingleClientWebsocketServerTransport [#singleclientwebsocketservertransport]

### Configuration [#configuration-1]

`new SingleClientWebsocketServerTransport(params, options?)`. `params` is a
`SingleClientWebsocketServerParams`:

| Option | Type | Required | Description |
|---|---|---|---|
| `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. |
| `addWavHeader` | `boolean \| undefined` | no | Whether to add WAV headers to audio frames. |
| `serializer` | `FrameSerializer \| null \| undefined` | no | Frame serializer for message encoding/decoding. |
| `sessionTimeout` | `number \| null \| undefined` | no | Timeout in seconds for client sessions. |
| `allowedOrigins` | `string[] \| undefined` | no | List of allowed origins. Empty list allows all origins. When set, connections with a missing or disallowed Origin header are rejected. Defaults to `PIPECAT_ALLOWED_ORIGINS` env var (comma-separated). |
| `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. |
| `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. |
| `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<string, VideoInSourceParams \| VideoInSourceParamsOptions> \| 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 |
|---|---|---|---|
| `host` | `string \| undefined` | no | Host address to bind the server to. Defaults to "localhost". |
| `port` | `number \| undefined` | no | Port number to bind the server to. Defaults to 8765. |
| `inputName` | `string \| null \| undefined` | no | Optional name for the input processor. |
| `outputName` | `string \| null \| undefined` | no | Optional name for the output processor. |
| `serve` | `WebsocketServe \| null \| undefined` | no | The function that starts the server. Defaults to `websocketServe` (the runtime's own server). |

### Usage [#usage-1]

```ts title="server.ts"
import { Pipeline } from "@pipedog/core/pipeline/pipeline";
import { PipelineWorker } from "@pipedog/core/pipeline/worker";
import { WorkerRunner } from "@pipedog/core/workers/runner";
import { ProtobufFrameSerializer } from "@pipedog/serializer-protobuf/protobuf";
import {
  SingleClientWebsocketServerParams,
  SingleClientWebsocketServerTransport,
} from "@pipedog/transport-websocket/server";

const transport = new SingleClientWebsocketServerTransport(
  new SingleClientWebsocketServerParams({
    audioInEnabled: true,
    audioOutEnabled: true,
    serializer: new ProtobufFrameSerializer(),
  }),
  { host: "localhost", port: 8765 },
);

transport.eventHandler("on_websocket_ready", async () => {
  console.log("listening on ws://localhost:8765");
});
transport.eventHandler("on_client_connected", async () => {
  console.log("client connected");
});

const pipeline = new Pipeline([
  transport.input(),
  // ... STT, LLM context aggregators, LLM, TTS
  transport.output(),
]);

const runner = new WorkerRunner();
await runner.addWorkers(new PipelineWorker(pipeline));
await runner.run();
```

### Event handlers [#event-handlers-1]

| Event                    | Description                                                                                     |
| ------------------------ | ----------------------------------------------------------------------------------------------- |
| `on_websocket_ready`     | The server is listening.                                                                        |
| `on_client_connected`    | A client connected. Receives the `WebsocketConnection`.                                         |
| `on_client_disconnected` | The client disconnected. Receives the `WebsocketConnection`.                                    |
| `on_session_timeout`     | `sessionTimeout` seconds passed since the client connected. Receives the `WebsocketConnection`. |

### Notes [#notes-1]

* One client at a time. While a client is connected, a new connection is closed with code 1013
  and the existing client is kept.
* The server starts when the pipeline starts, with `options.serve` (default `websocketServe`).
  Pass your own `WebsocketServe` function to listen some other way.
* A non-empty `allowedOrigins` is passed to the server as `origins`; other origins get HTTP 403.
* The server closes only after both the input and the output have finished, so output still
  being flushed after the `EndFrame` is not cut off.
* The deprecated `WebsocketServer*` aliases are not ported. Use the `SingleClientWebsocketServer*`
  names.

## WebsocketClientTransport [#websocketclienttransport]

### Configuration [#configuration-2]

`new WebsocketClientTransport(uri, params?)`. `params` is a `WebsocketClientParams`:

| Option | Type | Required | Description |
|---|---|---|---|
| `audioOutEnabled` | `boolean \| undefined` | no | Enable audio output streaming. |
| `audioOutSampleRate` | `number \| null \| undefined` | no | Output audio sample rate in Hz. |
| `addWavHeader` | `boolean \| undefined` | no | Whether to add WAV headers to audio frames. |
| `additionalHeaders` | `Record<string, string> \| null \| undefined` | no | Extra headers for the WebSocket handshake. |
| `serializer` | `FrameSerializer \| null \| undefined` | no | Frame serializer for encoding/decoding messages. |
| `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. |
| `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. |
| `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<string, VideoInSourceParams \| VideoInSourceParamsOptions> \| 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. |

### Usage [#usage-2]

```ts title="client.ts"
import { Pipeline } from "@pipedog/core/pipeline/pipeline";
import {
  WebsocketClientParams,
  WebsocketClientTransport,
} from "@pipedog/transport-websocket/client";
import { RTVIClientTransport } from "@pipedog/transport-websocket/rtvi_client";

// The serializer defaults to ProtobufFrameSerializer.
const client = new WebsocketClientTransport(
  "ws://localhost:8765",
  new WebsocketClientParams({
    audioInEnabled: true,
    audioOutEnabled: true,
    additionalHeaders: { Authorization: `Bearer ${process.env.BOT_TOKEN ?? ""}` },
  }),
);

client.eventHandler("on_connected", async () => {
  console.log("connected to the bot");
});

const pipeline = new Pipeline([client.input(), client.output()]);
```

### Event handlers [#event-handlers-2]

| Event             | Description                                                  |
| ----------------- | ------------------------------------------------------------ |
| `on_connected`    | Connected to the server. Receives the `WebsocketConnection`. |
| `on_disconnected` | The connection closed. Receives the `WebsocketConnection`.   |

### Notes [#notes-2]

* `serializer` defaults to `ProtobufFrameSerializer`. `addWavHeader` defaults to `true` here,
  unlike the server transports.
* The connection opens with `websocketConnect`, which uses the global `WebSocket`, and on
  Cloudflare Workers opens the socket through `fetch` so `additionalHeaders` are sent there too.
  Browsers can't send handshake headers.

## RTVIClientTransport [#rtviclienttransport]

`new RTVIClientTransport(uri, params?)`, with the same `WebsocketClientParams`. It connects to a
bot that runs an RTVI server transport, such as an eval harness, and exchanges frames with it.

### Usage [#usage-3]

```ts title="rtvi.ts"
// Talks RTVI to a bot: sends client-ready on connect and waits for bot-ready.
const rtvi = new RTVIClientTransport("ws://localhost:7860");

rtvi.eventHandler("on_bot_ready", async () => {
  console.log("bot ready:", rtvi.botReady);
});
```

### Event handlers [#event-handlers-3]

| Event             | Description                                                   |
| ----------------- | ------------------------------------------------------------- |
| `on_connected`    | Connected to the bot.                                         |
| `on_disconnected` | The connection closed.                                        |
| `on_bot_ready`    | The bot answered the `client-ready` message with `bot-ready`. |

### Notes [#notes-3]

* `serializer` defaults to [`RTVIClientSerializer`](https://polvallverdu.github.io/pipedog/docs/serializers/rtvi-client.md).
* `addWavHeader` is always set to `false`: an RTVI audio message carries bare PCM.
* On connect it sends an RTVI `client-ready` message. `botReady` turns `true` once `bot-ready`
  arrives.

## WebsocketConnection [#websocketconnection]

The socket the transports talk to. It exposes both APIs the upstream transports use: the
`websockets` one (`state`, `recv()`, `send()`, `close()`, async iteration) and the Starlette one
(`clientState`, `applicationState`, `receive()`, `sendBytes()`, `sendText()`, `headers`). Build one
in either of two ways:

* `WebsocketConnection.fromWebSocket(ws, options)` wraps a standard `WebSocket`: the global
  client, the server side of a Cloudflare `WebSocketPair`, Deno's upgraded socket or a Node `ws`
  socket.
* `new WebsocketConnection(sink, options)` wraps anything else that can send and close. Report
  what arrives with `feedMessage()` and `feedClose()`. This is what `bunWebSocket` does for you;
  to wire Bun's `ServerWebSocket` by hand, inside your own `Bun.serve` handlers:

```ts title="bun.ts"
// Bun's ServerWebSocket reports its events to Bun.serve, so feed them to a
// WebsocketConnection built on a sink.
type SocketData = { connection: WebsocketConnection | null };

Bun.serve<SocketData>({
  port: 3000,
  fetch(request, bunServer) {
    if (bunServer.upgrade(request, { data: { connection: null } })) return undefined;
    return new Response("not found", { status: 404 });
  },
  websocket: {
    open(ws) {
      const connection = new WebsocketConnection({
        send: (message) => ws.send(message),
        close: (code, reason) => ws.close(code, reason),
      });
      ws.data.connection = connection;
      createTask(() => runBot(connection), "bun-socket");
    },
    message(ws, message) {
      ws.data.connection?.feedMessage(message);
    },
    close(ws, code, reason) {
      ws.data.connection?.feedClose(code, reason);
    },
  },
});
```

Options for both:

| Option | Type | Required | Description |
|---|---|---|---|
| `headers` | `HeadersInit \| null \| undefined` | no | The handshake request's headers (server side). |
| `remoteAddress` | `string \| null \| undefined` | no | The peer's address, for logging. |
| `path` | `string \| null \| undefined` | no | The request path (server side). |
| `state` | `State \| undefined` | no | The initial state. Defaults to `State.OPEN`. |
| `closeTimeout` | `number \| undefined` | no | Seconds `close()` waits for the closing handshake. Defaults to 10. |

Options for `websocketServe`:

| Option | Type | Required | Description |
|---|---|---|---|
| `handler` | `(websocket: WebsocketConnection) => Promise<void>` | yes | Runs for every accepted connection. |
| `host` | `string` | yes | The host to listen on. |
| `port` | `number` | yes | The port to listen on. |
| `origins` | `string[] \| null \| undefined` | no | Accepted `Origin` header values; `null` accepts any. A connection whose origin isn't listed is rejected with HTTP 403. |

## 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-websocket/package.json`: ✅ means the compat smoke test passes on that runtime.

|                                                   | Bun                                              | Node                    | Deno                          | Cloudflare Workers    |
| ------------------------------------------------- | ------------------------------------------------ | ----------------------- | ----------------------------- | --------------------- |
| `FastAPIWebsocketTransport`                       | yes                                              | yes                     | yes                           | yes                   |
| `websocketFetchHandler`, `upgradeWebSocket`       | yes (pass the server, serve with `bunWebSocket`) | yes (with `serveFetch`) | yes (`Deno.upgradeWebSocket`) | yes (`WebSocketPair`) |
| `serveFetch`                                      | yes (`Bun.serve`)                                | yes (`ws` peer)         | yes (`Deno.serve`)            | no                    |
| `websocketServe`                                  | yes (`Bun.serve`)                                | yes (`ws` peer)         | yes (`Deno.serve`)            | no                    |
| `honoWebsocket`                                   | yes                                              | yes (`@hono/node-ws`)   | yes                           | yes                   |
| `SingleClientWebsocketServerTransport`            | yes                                              | yes (`ws` peer)         | yes                           | no                    |
| `WebsocketClientTransport`, `RTVIClientTransport` | yes                                              | yes                     | yes                           | yes                   |

* `websocketFetchHandler` needs the upgrading server on Bun (the second argument of `fetch`) and on
  Node (from `serveFetch`); without one it throws, naming the fix.
* `websocketServe` and `serveFetch` throw on Cloudflare Workers, which can't listen on a port. Closing the server
  closes the open connections with code 1001 and waits for their handlers.
* Both fetch and Hono helpers copy the request headers during the upgrade: on Deno the request is
  closed once upgraded, and reading it then throws.
* Each connection's handler runs in its own task, and the connection closes when the handler
  returns or throws.

Not ported yet: the `WebsocketServer*` deprecated aliases (by design) and the eval harness's
FastAPI transport (`pipecat.evals.fastapi_transport`).