WebSocket

OverviewCopy Anchor Link

@pipedog/transport-websocket carries frames over a WebSocket, encoded by a serializer. 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).

InstallationCopy Anchor Link

bun add @pipedog/transport-websocket

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

npm install ws

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.

PrerequisitesCopy Anchor Link

• A serializer that matches the other end: ProtobufFrameSerializer for Pipecat's WebSocket clients, a telephony serializer such as Telnyx for phone calls.
• For allowedOrigins, the list of origins allowed to connect, or the PIPECAT_ALLOWED_ORIGINS environment variable (comma-separated).

FastAPIWebsocketTransportCopy Anchor Link

ConfigurationCopy Anchor Link

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

options:

UsageCopy Anchor Link

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

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:

// 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:

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

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:

// 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:

// 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 });

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. See also Frameworks.

Event handlersCopy Anchor Link

NotesCopy Anchor Link

• 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.

SingleClientWebsocketServerTransportCopy Anchor Link

ConfigurationCopy Anchor Link

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

options:

UsageCopy Anchor Link

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 handlersCopy Anchor Link

NotesCopy Anchor Link

• 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.

WebsocketClientTransportCopy Anchor Link

ConfigurationCopy Anchor Link

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

UsageCopy Anchor Link

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 handlersCopy Anchor Link

NotesCopy Anchor Link

• 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.

RTVIClientTransportCopy Anchor Link

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.

UsageCopy Anchor Link

// 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 handlersCopy Anchor Link

NotesCopy Anchor Link

• serializer defaults to RTVIClientSerializer.
• 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.

WebsocketConnectionCopy Anchor Link

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:

// 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:

Options for websocketServe:

Runtime notesCopy Anchor Link

Generated from pipecat.runtimes and pipecat.compat in packages/transport-websocket/package.json: ✅ means the compat smoke test passes on that runtime.

• 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).