Initial commit: Adding support for new API to send files

1. adds new RTVI types
2. adds new `sendFile()` method
3. adds better error handling for sending messages that are too large
4. Currently, only sending images as bytes is tested and working

Author: mattieruth

client-js/client/client.ts

@@ -23,6 +23,7 @@ import {
   PipecatMetricsData,
   RTVIEvent,
   RTVIEvents,
+  RTVIFile,
   RTVIMessage,
   RTVIMessageType,
   SendTextOptions,
@@ -182,9 +183,17 @@ export class PipecatClient extends RTVIEventEmitter {
         options?.callbacks?.onError?.(message);
         try {
           this.emit(RTVIEvent.Error, message);
-          // eslint-disable-next-line @typescript-eslint/no-unused-vars
         } catch (e) {
-          logger.debug("Could not emit error", message);
+          if (e instanceof Error && e.message.includes("Unhandled error")) {
+            if (!options?.callbacks?.onError) {
+              logger.debug(
+                "No onError callback registered to handle error",
+                message
+              );
+            }
+          } else {
+            logger.debug("Could not emit error", message, e);
+          }
         }
         const data = message.data as ErrorData;
         if (data?.fatal) {
@@ -489,10 +498,30 @@ export class PipecatClient extends RTVIEventEmitter {
 
     // Create a new message dispatch queue for async message handling
     this._messageDispatcher = new MessageDispatcher(
-      this._transport.sendMessage.bind(this._transport)
+      this._sendMessage.bind(this)
     );
   }
 
+  /**
+   * Internal wrapper around the transport's sendMessage method
+   */
+  private _sendMessage(message: RTVIMessage): void {
+    try {
+      this._transport.sendMessage(message);
+    } catch (error) {
+      if (error instanceof Error) {
+        this._options.callbacks?.onError?.(
+          RTVIMessage.error(error.message, false)
+        );
+      } else {
+        this._options.callbacks?.onError?.(
+          RTVIMessage.error("Unknown error sending message", false)
+        );
+      }
+      throw error;
+    }
+  }
+
   /**
    * Get the current state of the transport
    */
@@ -588,7 +617,7 @@ export class PipecatClient extends RTVIEventEmitter {
    */
   @transportReady
   public sendClientMessage(msgType: string, data?: unknown): void {
-    this._transport.sendMessage(
+    this._sendMessage(
       new RTVIMessage(RTVIMessageType.CLIENT_MESSAGE, {
         t: msgType,
         d: data,
@@ -637,7 +666,7 @@ export class PipecatClient extends RTVIEventEmitter {
   @transportReady
   public async appendToContext(context: LLMContextMessage) {
     logger.warn("appendToContext() is deprecated. Use sendText() instead.");
-    await this._transport.sendMessage(
+    await this._sendMessage(
       new RTVIMessage(RTVIMessageType.APPEND_TO_CONTEXT, {
         role: context.role,
         content: context.content,
@@ -649,22 +678,35 @@ export class PipecatClient extends RTVIEventEmitter {
 
   @transportReady
   public async sendText(content: string, options: SendTextOptions = {}) {
-    await this._transport.sendMessage(
+    await this._sendMessage(
       new RTVIMessage(RTVIMessageType.SEND_TEXT, {
         content,
         options,
       })
     );
   }
 
+  @transportReady
+  public async sendFile(
+    file: RTVIFile,
+    content: string,
+    options: SendTextOptions = {}
+  ) {
+    await this._sendMessage(
+      new RTVIMessage(RTVIMessageType.SEND_FILE, {
+        file,
+        content,
+        options,
+      })
+    );
+  }
+
   /**
    * Disconnects the bot, but keeps the session alive
    */
   @transportReady
   public disconnectBot(): void {
-    this._transport.sendMessage(
-      new RTVIMessage(RTVIMessageType.DISCONNECT_BOT, {})
-    );
+    this._sendMessage(new RTVIMessage(RTVIMessageType.DISCONNECT_BOT, {}));
   }
 
   protected handleMessage(ev: RTVIMessage): void {
@@ -772,7 +814,7 @@ export class PipecatClient extends RTVIEventEmitter {
             if (result == undefined) {
               return;
             }
-            this._transport.sendMessage(
+            this._sendMessage(
               new RTVIMessage(RTVIMessageType.LLM_FUNCTION_CALL_RESULT, {
                 function_name: data.function_name,
                 tool_call_id: data.tool_call_id,

client-js/rtvi/errors.ts

@@ -76,6 +76,14 @@ export class UnsupportedFeatureError extends RTVIError {
   }
 }
 
+export class MessageTooLargeError extends RTVIError {
+  constructor(message?: string | undefined) {
+    super(
+      message ?? "Message size exceeds the maximum allowed limit for transport."
+    );
+  }
+}
+
 export type DeviceArray = Array<"cam" | "mic" | "speaker">;
 export type DeviceErrorType =
   | "in-use"

client-js/rtvi/messages.ts

@@ -25,6 +25,7 @@ export enum RTVIMessageType {
   // Client-to-server messages
   CLIENT_MESSAGE = "client-message",
   SEND_TEXT = "send-text",
+  SEND_FILE = "send-file",
   // DEPRECATED
   APPEND_TO_CONTEXT = "append-to-context",
 
@@ -187,6 +188,75 @@ export type SendTextOptions = {
   audio_response?: boolean;
 };
 
+type Serializable =
+  | string
+  | number
+  | boolean
+  | null
+  | Serializable[]
+  | { [key: number | string]: Serializable };
+
+export type RTVIImageFormat =
+  | "png"
+  | "jpg"
+  | "jpeg"
+  | "webp"
+  | "gif"
+  | "heic"
+  | "hief";
+export type RTVIDocFormat =
+  | "pdf"
+  | "csv"
+  | "txt"
+  | "md"
+  | "doc"
+  | "docx"
+  | "xls"
+  | "xlsx"
+  | "json"
+  | "html"
+  | "css"
+  | "javascript";
+export type RTVIMediaFormat =
+  | "mp3"
+  | "wav"
+  | "ogg"
+  | "aac"
+  | "mp4"
+  | "webm"
+  | "ogg"
+  | "avi";
+export type RTVIFileFormat = RTVIImageFormat | RTVIDocFormat | RTVIMediaFormat;
+
+export type FileSourceType = "bytes" | "url" | "id";
+
+export type FileBytes = {
+  type: Extract<FileSourceType, "bytes">;
+  bytes: string;
+  width?: number;
+  height?: number;
+};
+export type FileUrl = {
+  type: Extract<FileSourceType, "url">;
+  url: string | URL;
+};
+export type FileId = {
+  type: Extract<FileSourceType, "id">;
+  id: string;
+};
+
+export type RTVIFile = {
+  format: RTVIFileFormat;
+  source: FileBytes | FileUrl | FileId;
+  customOpts: { [key: number | string]: Serializable }; // for things like 'detail' in openAI or 'citations' in Bedrock
+};
+
+export type FileSupport = {
+  formats: RTVIFileFormat[];
+  sources: FileSourceType[];
+  maxSize: number; // bytes
+};
+
 /** DEPRECATED */
 export type LLMContextMessage = {
   role: "user" | "assistant";