Transport (TypeScript)

The SDK separates session logic from delivery. RealtimeClient depends only on the RealtimeTransport interface; the default implementation uses LiveKit. You would swap the transport to test without a live server, to use a different SFU, or to inject a mock for integration testing.

`RealtimeTransport` interface

Import it from the cosmo-ai/transport/types subpath. Eleven members are required; the rest are optional so test fakes can opt out — call sites feature-detect before invoking them.

interface RealtimeTransport {
  // Session lifecycle
  connect(options: RealtimeConnectOptions): Promise<void>;
  disconnect(options?: { sendEndFrame?: boolean }): Promise<void>;

  // Control messages
  send(message: RealtimeClientMessage): Promise<void>;
  sendBytes?(data: Uint8Array, topic: string): Promise<void>;

  // Microphone
  setMicMuted(muted: boolean): Promise<void>;

  // Video
  addVideoStream?(stream: MediaStream, options?: VideoStreamOptions): Promise<VideoStreamHandle>;
  removeVideoStream?(streamId: VideoStreamHandle): Promise<void>;

  // Caller-owned audio
  startAudioStream?(stream: MediaStream): Promise<void>;
  stopAudioStream?(): Promise<void>;

  // Audio elements
  getInputStream(): MediaStream | null;
  getOutputAudioElement(): HTMLAudioElement | null;
  getOutputStream?(): MediaStream | null;
  onOutputStreamChanged?(cb: () => void): Unsubscribe;
  attachAudioElement(el: HTMLAudioElement | null): void;
  resumeAudioPlayback?(): Promise<void>;

  // Subscriptions
  onMessage(cb: (msg: RealtimeServerMessage) => void): Unsubscribe;
  onClose(cb: (info?: RealtimeCloseInfo) => void): Unsubscribe;
  onReconnecting(cb: () => void): Unsubscribe;
  onReconnected(cb: () => void): Unsubscribe;

  // Client-tool RPC
  registerRpcMethod?(
    name: string,
    handler: (invocation: RpcInvocation) => Promise<string>,
  ): Unsubscribe;
}

Each member, its signature, and when the client calls it:

Member	Description
`connect`	Open the session. Rejects on auth failure, mic denial, or transport failure; on any throw during setup, all allocations are released.
`disconnect`	Tear down. Idempotent and always resolves. `sendEndFrame: false` skips the graceful wire `end` frame (an abrupt local close).
`send`	Send one logical client message; the implementation owns any chunking. Resolves once the bytes hit the wire; rejects on publish failure.
`sendBytes?`	Send a large binary payload to the agent participant only, on a named topic — for payloads too large for the reliable data channel (~15 KiB), such as a screenshot plus accessibility dump.
`setMicMuted`	Toggle the local mic track and send the matching `mute` frame so the agent's voice-activity detection respects user intent.
`addVideoStream?`	Publish a `MediaStream` as a video track; returns the id for `removeVideoStream`. Screen-share publishes go through here with `{ kind: 'screen' }`.
`removeVideoStream?`	Unpublish a stream. Idempotent — a missing id is a no-op.
`startAudioStream?`	Take the session's voice with a `MediaStream`. Also sends `bind-input` and clears the mute gate, so the agent listens to the new track. A session carries one voice, so the microphone steps aside for its lifetime.
`stopAudioStream?`	Unpublish the running audio stream and hand the voice back to the microphone. Idempotent.
`getInputStream`	The mic media stream, for building an `AnalyserNode` without re-requesting permission. `null` until connected.
`getOutputAudioElement`	The remote audio element. `null` until connected. Do not build an output analyser from it: `createMediaElementSource` claims an element for the life of the page and routes its sound through the graph, so the tap cannot be rebuilt for a later session and closing its context leaves the element silent. Use `getOutputStream`.
`getOutputStream?`	The agent's audio as a `MediaStream`, for building an output `AnalyserNode`. `null` until a remote audio track is subscribed. Stable by reference between track changes. Omit it and the session reports no output level.
`onOutputStreamChanged?`	Fires when `getOutputStream` starts returning a different stream — the track arriving after connect, being replaced, or going away. Omit it and the output analyser is built once, at connect.
`attachAudioElement`	Hand the transport a host-owned `<audio>` element for the remote track; `null` detaches back to the auto-created one. Idempotent; callable before or after `connect`.
`resumeAudioPlayback?`	Replay remote audio from a user gesture to clear a browser autoplay block.
`onMessage`	Subscribe to inbound server messages. Multiple subscribers each receive every message.
`onClose`	Unsolicited disconnects only — a close caused by the local `disconnect()` does not fire it.
`onReconnecting`	Transient transport-level recovery began (ICE restart, signal reconnect). Informational; followed by `onReconnected` or `onClose`.
`onReconnected`	Recovery after `onReconnecting` succeeded.
`registerRpcMethod?`	Register an RPC handler for server-invoked client tools. Callable before `connect()`; every method registered pre-connect is live before any inbound invocation is delivered. The handler receives an `RpcInvocation` and returns a JSON-encoded result envelope.

`RealtimeConnectOptions`

type RealtimeConnectOptions = {
  /** The external-protocol `session-config` body, built by the agent
   *  mapping. The transport POSTs it verbatim and never inspects it. */
  config: SessionConfig;
  /** Absolute session-start URL, composed by the SDK from `baseUrl`
   *  (or the page origin), so the transport holds no URL layout itself. */
  sessionStartUrl: string;
  /** Publish the local microphone after joining. Default true. False joins
   *  as a silent observer (outbound-phone sessions) and also skips the
   *  voice-binding frame, so the agent listens to the dialed party. */
  publishMicrophone?: boolean;
  /** Resolve extra request headers for the session-start POST.
   *  Content-Type is set by the transport and cannot be overridden. */
  getAuthHeaders?: () => Record<string, string> | Promise<Record<string, string>>;
  /** Fired with the server-minted session_id the instant the session-start
   *  POST returns — well before the ready data-channel message. */
  onSessionStarted?: (sessionId: string) => void;
  /** Fired with the client-measured connection phase breakdown. */
  onConnectTimings?: (timings: SessionConnectTimings) => void;
  /** Report whether the browser blocked remote-audio playback. */
  onOutputBlocked?: (blocked: boolean) => void;
};

`VideoStreamOptions`

type VideoStreamOptions = {
  id?: string;    // caller-supplied handle; omitted → the transport picks one
  fps?: number;   // capture ceiling; defaults low (~1 FPS), sized for vision input
  kind?: 'camera' | 'screen';
};

kind: 'screen' tells the transport the stream came from getDisplayMedia, so it publishes as Track.Source.ScreenShare and the server can distinguish camera from screen.

`RealtimeCloseInfo`

type RealtimeCloseInfo = {
  reason?: string; // e.g. 'livekit:CLIENT_INITIATED'
  code?: string;
};

RealtimeClient strips the vendor prefix (for example, livekit:) before surfacing reason in ErrorEvent.message, so consumers never see transport-specific prefixes.

`RpcInvocation`

One inbound RPC invocation, narrowed at the transport boundary so consumers never import vendor types.

type RpcInvocation = {
  payload: string;        // JSON-encoded args object
  callerIdentity: string; // room identity of the invoking participant
  callerIsAgent: boolean; // true when the caller is the session's agent
};

Built-in transports

`LiveKitTransport`

The default transport. Connects using the session token returned by POST /api/v1/external/realtime/session/start, joins the LiveKit room, and routes JSON control messages over the reliable data channel. Audio and video ride separate RTP tracks.

LiveKitTransport is internal — the package doesn't export it, and there is no import that reaches it. RealtimeClient instantiates it for you unless transportFactory is overridden, so the default needs no import:

import { RealtimeClient } from 'cosmo-ai';

const client = new RealtimeClient({ getAuthHeaders: ... });

For the one-process local OSS server, pass transport: "websocket". That built-in adapter carries PCM audio, control frames and ordinary client-tool RPC on one browser WebSocket. It has no reconnection, video, screen share, byte streams or background client tools.

To run against something else, pass a transportFactory returning your own RealtimeTransport — see Swapping the transport. transport and transportFactory are mutually exclusive.

What LiveKitTransport does

POSTs the SessionConfig body to the session endpoint with the caller's auth headers.
Receives { livekit_url, token, room_name, session_id }.
Joins the LiveKit room at livekit_url with token.
Publishes the default mic as a Track.Source.Microphone track (unless publishMicrophone: false).
Routes outbound JSON via LocalParticipant.publishData(reliable: true).
Subscribes to dataReceived for inbound JSON messages.
Routes oversized messages through the envelope chunking layer, which splits anything over 12,000 bytes into 8,000-byte chunks. See Envelope chunking.

Transport replacement

Test fake

Every required member must be present — the optional members can be left off:

import { RealtimeClient } from 'cosmo-ai';
import type { RealtimeTransport } from 'cosmo-ai/transport/types';

const fakeTransport: RealtimeTransport = {
  connect: async () => {},
  disconnect: async () => {},
  send: async () => {},
  setMicMuted: async () => {},
  getInputStream: () => null,
  getOutputAudioElement: () => null,
  attachAudioElement: () => {},
  onMessage: (cb) => {
    // simulate ready
    setTimeout(() => cb({ type: 'ready', session_id: 'test', version: '1.0' }), 10);
    return () => {};
  },
  onClose: () => () => {},
  onReconnecting: () => () => {},
  onReconnected: () => () => {},
};

const client = new RealtimeClient({
  transportFactory: () => fakeTransport,
});

When to swap

Unit tests — fake resolves immediately, no network.
Custom auth layer — override getAuthHeaders instead; no transport swap needed.
Alternative SFU — implement RealtimeTransport for a different WebRTC stack and pass it via transportFactory.

addVideoStream / removeVideoStream and startAudioStream / stopAudioStream are optional on the interface. If your custom transport doesn't implement them, RealtimeClient.addVideoStream() throws "Active transport does not support video streams." and RealtimeClient.startAudioStream() throws "Active transport does not support audio streams.".

Envelope chunking

The LiveKit reliable data channel has a 12,000-byte chunking threshold. Both client and server transparently chunk oversized messages using the envelope protocol:

Client → server: type: "envelope-chunk" with envelope_id, seq, total, base64-encoded fragment.
Server → client: type: "server-envelope-chunk" with identical fields.

The transport and client handle reassembly automatically. Callers never need to deal with chunking directly.

Transport (TypeScript)

On this page