Custom Messaging

The Pipecat JavaScript client can send and receive arbitrary messages to/from the server running the bot. This page outlines and demonstrates both client and server code for passing and responding to custom messages as well as providing arbitrary data at connection time.

Connection-Time Configuration

Oftentimes clients need to provide configuration data to the server when starting the bot. This can include things like preferred language, user preferences, initial messages, or any other data that the server needs to know about the client. This must occur before the bot is started and therefore is not part of the RTVI standard, but rather a custom implementation. That said, the PipecatClient makes it easy to send this data as part of the startBot() and startBotAndConnect() methods, by passing an API endpoint object with the requestData property. Your server endpoint can then handle this data as needed. In the example below, we demonstrate sending an initial prompt and preferred language to the server when connecting.

try {
    pcClient.startBotAndConnect({
        endpoint: '/api/start', // Your server endpoint to start the bot
        requestData: {
            initial_prompt: "You are a pirate captain",
            preferred_language: 'en-US'
        }
    });
} catch (error) {
    console.error("Error starting the bot:", error);
}

def validate_request_data(body: Dict[str, Any]) -> Tuple[str, str]:
    """Validate and extract prompt and language from request data."""
    if not isinstance(body, dict):
        raise ValueError("Request body must be a dictionary")

    prompt = body.get("initial_prompt", "You are a pirate captain")
    lang = body.get("preferred_language", "en-US")

    if not isinstance(prompt, str) or not isinstance(lang, str):
        raise ValueError("Both initial_prompt and preferred_language must be strings")

    return prompt, lang

@app.post("/api/start")
async def start(request: Request) -> Dict[Any, Any]:
    """Startup endpoint that creates a room, starts the bot, and returns
       connection credentials.

    This endpoint is called by clients to kick off a session and establish
    a connection.

    Returns:
        Dict[Any, Any]: Authentication bundle containing room_url and token

    Raises:
        HTTPException: If room creation, token generation, or bot startup fails
    """
    body = await request.json()
    try:
        prompt, lang = validate_request_data(body)
    except ValueError as e:
        raise HTTPException(status_code=400, info=f"Invalid request data: {e}")

    print("Creating room for RTVI connection", body)
    room_url, token = await create_room_and_token()
    print(f"Room URL: {room_url}")

    # Start the bot process
    try:
        proc = subprocess.Popen(
            [f"python3 -m bot -u {room_url} -t {token} -p {prompt} -l {lang}"],
            shell=True,
            bufsize=1,
            cwd=os.path.dirname(os.path.abspath(__file__)),
        )
        bot_procs[proc.pid] = (proc, room_url)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to start subprocess: {e}")

    # Return the authentication bundle in format expected by DailyTransport
    return {"url": room_url, "token": token}

import argparse
import asyncio

def extract_arguments():
    parser = argparse.ArgumentParser(description="Example")
    parser.add_argument(
        "-u", "--room-url", type=str, default=os.getenv("DAILY_SAMPLE_ROOM_URL", "")
    )
    parser.add_argument(
        "-t", "--token", type=str, default=os.getenv("DAILY_SAMPLE_ROOM_TOKEN", None)
    )
    parser.add_argument(
        "-p", "--prompt", type=str, default="You are a pirate captain"
    )
    parser.add_argument("-l", "--language", type=str, default="en-US")
    return parser.parse_args()

async def main():
    args = extract_arguments()
    print(f"room_url: {args.room_url}")

    daily_transport = DailyTransport(
        args.room_url,
        args.token,
        "Chatbot",
        DailyParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
        ),
    )

    llm = GeminiLiveLLMService(
        api_key=os.getenv("GOOGLE_API_KEY"),
        settings=GeminiLiveLLMService.Settings(
            system_instruction=SYSTEM_INSTRUCTION,
            voice="Puck",
            language=args.language,  # Pass preferred language to LLM
        ),
    )

    messages = [ { role: "system", content: args.prompt } ]
    context = LLMContext(messages)
    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
        context,
        user_params=LLMUserAggregatorParams(
            vad_analyzer=SileroVADAnalyzer(),
        ),
    )

    pipeline = Pipeline(
        [
            daily_transport.input(),
            user_aggregator,
            llm,
            daily_transport.output(),
            assistant_aggregator,
        ]
    )

    task = PipelineTask(pipeline)

    @rtvi.event_handler("on_client_ready")
    async def on_client_ready(rtvi):
        logger.debug("Client ready")
        # Kick off the conversation
        await task.queue_frames([LLMRunFrame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        logger.debug(f"Client disconnected")
        await task.cancel()

    runner = PipelineRunner(handle_sigint=False)

    await runner.run(task)

if __name__ == "__main__":
    asyncio.run(main())

Sending Custom Messages to the Server

Once connected, you can send custom messages to the server using the sendClientMessage method. This is useful for triggering specific actions or sending data that the server needs to process.

try {
    pcClient.sendClientMessage('set-language', { language: 'en-US' });
} catch (error) {
    console.error("Error sending message to server:", error);
}

    task = PipelineTask(
        pipeline,
        params,
        observers=[RTVIObserver(rtvi)],
    )

    @rtvi.event_handler("on_client_message")
    async def on_client_message(rtvi, msg):
        print("RTVI client message:", msg.type, msg.data)
        if msg.type == "set-language":
            language = msg.data.get("language", "en-US")
            await task.queue_frames([STTUpdateSettingsFrame(language=language)])

### Alternatively, if your message requires asynchronous processing or storing of
### state, you may want to handle it from inside a FrameProcessor, listen for a
### RTVIClientMessageFrame and push a RTVIServerResponseFrame
class CustomFrameProcessor(FrameProcessor):
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, RTVIClientMessageFrame):
            print("RTVI client message:", frame.msg_id, frame.type, frame.data)
            if frame.type == "set-language":
                language = frame.data.get("language", "en-US")
                await self.push_frame(STTUpdateSettingsFrame(language=language))
                return
        await self.push_frame(frame, direction)

Requesting Information from the Server

You can also request information from the server using the sendClientRequest method. This is useful for querying the server for specific data or triggering and action and getting a success/failure response.

try {
    const response = await pcClient.sendClientRequest('get-language');
    console.log("Current language:", response.language);
} catch (error) {
    console.error("Error requesting data from server:", error);
}

@rtvi.event_handler("on_client_message")
async def on_client_message(rtvi, msg):
    print("RTVI client message:", msg.type, msg.data)
    if msg.type == "get-language":
        await rtvi.send_server_response(msg, {"language": get_current_language()})
    else:
        await rtvi.send_error_response(msg, "Unknown request type")

### Alternatively, if your message requires asynchronous processing or storing of
### state, you may want to handle it from inside a FrameProcessor, listen for a
### RTVIClientMessageFrame and push a RTVIServerResponseFrame
class CustomFrameProcessor(FrameProcessor):
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, RTVIClientMessageFrame):
            print("RTVI client message:", frame.msg_id, frame.type, frame.data)
            if frame.type == "get-language":
                data = {"language": get_current_language()}
                await self.push_frame(
                    RTVIServerResponseFrame(
                        client_msg=frame,
                        data=data,
                    ),
                )
                return
            else:
                await self.push_frame(
                    RTVIServerResponseFrame(
                        client_msg=frame,
                        error="Unknown request type"
                    )
                )
        await self.push_frame(frame, direction)

Handling Custom Messages from the Server

You can handle custom messages sent from the server using the onServerMessage callback. This allows you to process messages that the server sends back to the client, such as notifications or updates. For full details on sending server messages from your bot, see the RTVIProcessor custom messaging docs.

pcClient.onServerMessage((message) => {
    console.log("Received message from server:", message);
    if (message.data.msg === 'language-updated') {
        console.log("Language updated to:", message.data.language);
    }
});

## From inside an Observer, call `send_server_message` directly on your rtvi instance
class CustomObserver(BaseObserver):
    async def on_push_frame(self, data: FramePushed):
        if isinstance(frame, STTUpdateSettingsFrame):
            for key, value in settings.items():
                if key == "language":
                    await rtvi.send_server_message({
                        "msg": "language-updated",
                        "language": value
                    })

### Alternatively, from inside a FrameProcessor, push a RTVIServerMessageFrame
class CustomFrameProcessor(FrameProcessor):
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, STTUpdateSettingsFrame):
            for key, value in settings.items():
                if key == "language":
                    await self.push_frame(
                        RTVIServerMessageFrame(
                            data={
                                "msg": "language-updated",
                                "language": value
                            }
                        )
                    )
        await self.push_frame(frame, direction)

Introduction

JavaScript SDK

React SDK

React Native SDK

iOS SDK

Android SDK

C++ SDK

Voice UI Kit

Custom Messaging

Connection-Time Configuration

Sending Custom Messages to the Server

Requesting Information from the Server

Handling Custom Messages from the Server

​Connection-Time Configuration

​Sending Custom Messages to the Server

​Requesting Information from the Server

​Handling Custom Messages from the Server

Connection-Time Configuration

Sending Custom Messages to the Server

Requesting Information from the Server

Handling Custom Messages from the Server