Skip to main content
The Pipecat client can send and receive arbitrary messages to and from the server running the bot. This is useful for passing configuration at startup, triggering actions mid-session, querying server state, and receiving notifications from the bot.

​
Connection-time configuration

Often you need to pass configuration to the server when starting the bot — a system prompt, preferred language, user preferences, or any other data the server needs before the pipeline runs. Pass it via requestData in startBotAndConnect(). On the server, your endpoint receives this data as the request body and can forward relevant values to the bot process or use them to configure the pipeline: 
@app.post("/api/start")
async def start(request: Request) -> Dict[Any, Any]:
    body = await request.json()
    prompt = body.get("initial_prompt", "You are a pirate captain")
    lang = body.get("preferred_language", "en-US")

    room_url, token = await create_room_and_token()

    proc = subprocess.Popen(
        [f"python3 -m bot -u {room_url} -t {token} -p {prompt} -l {lang}"],
        shell=True,
        cwd=os.path.dirname(os.path.abspath(__file__)),
    )
    bot_procs[proc.pid] = (proc, room_url)

    return {"url": room_url, "token": token}


def extract_arguments():
    parser = argparse.ArgumentParser()
    parser.add_argument("-u", "--room-url", type=str)
    parser.add_argument("-t", "--token", type=str)
    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()

    llm = GeminiLiveLLMService(
        api_key=os.getenv("GOOGLE_API_KEY"),
        settings=GeminiLiveLLMService.Settings(
            system_instruction=args.prompt,
            language=args.language,
        ),
    )

​
Sending messages to the server

Use sendClientMessage() to send a fire-and-forget message to the bot. The server handles it and is not expected to send a direct response. On the server, handle it via the on_client_message event handler or from inside a FrameProcessor: 
@rtvi.event_handler("on_client_message")
async def on_client_message(rtvi, msg):
    if msg.type == "set-language":
        language = msg.data.get("language", "en-US")
        await worker.queue_frames([STTUpdateSettingsFrame(language=language)])

class CustomFrameProcessor(FrameProcessor):
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, RTVIClientMessageFrame):
            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 data from the server

Use sendClientRequest() to send a message and wait for a response. Useful for querying server state or triggering an action that needs to confirm success or failure. On the server, respond with send_server_response() or push a RTVIServerResponseFrame: 
@rtvi.event_handler("on_client_message")
async def on_client_message(rtvi, msg):
    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")

class CustomFrameProcessor(FrameProcessor):
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, RTVIClientMessageFrame):
            if frame.type == "get-language":
                await self.push_frame(
                    RTVIServerResponseFrame(
                        client_msg=frame,
                        data={"language": get_current_language()},
                    )
                )
                return
            else:
                await self.push_frame(
                    RTVIServerResponseFrame(
                        client_msg=frame,
                        error="Unknown request type",
                    )
                )
        await self.push_frame(frame, direction)

​
Receiving messages from the server

The server can push unsolicited messages to the client at any time — for example, to notify the client that a setting has changed. On the server, send messages via send_server_message() or by pushing a RTVIServerMessageFrame: 
class CustomObserver(BaseObserver):
    async def on_push_frame(self, data: FramePushed):
        if isinstance(data.frame, STTUpdateSettingsFrame):
            for key, value in data.frame.settings.items():
                if key == "language":
                    await rtvi.send_server_message({
                        "msg": "language-updated",
                        "language": value,
                    })

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 frame.settings.items():
                if key == "language":
                    await self.push_frame(
                        RTVIServerMessageFrame(
                            data={"msg": "language-updated", "language": value}
                        )
                    )
        await self.push_frame(frame, direction)

​
API reference

Client Methods

sendClientMessage, sendClientRequest, registerFunctionCallHandler

Callbacks & Events

onServerMessage, onMessageError, and related events