How to Use RPC for Client-Agent Communication in LiveKit Agents

LiveKit Agents expose a typed RPC API via register_rpc_method that allows any room participant to query agent state, fetch chat history, or send messages using Pydantic models for request/response serialization.

The livekit/agents repository provides a robust framework for building voice and multimodal AI agents. One of its core features is the ability to handle RPC for client-agent communication, enabling bidirectional request-response patterns between participants and the agent beyond simple audio/video streams.

Architecture of RPC for Client-Agent Communication

The RPC system in LiveKit Agents is built on top of the LiveKit RTC room primitives. It leverages the local participant's ability to register named methods that can be invoked by any other participant in the same room.

Core Components

Component	Location	Purpose
ClientEventsHandler	livekit/agents/voice/client_events.py	Manages RPC registration and implements request handlers
register_rpc_method	room.local_participant	Binds a method name to a handler coroutine
unregister_rpc_method	room.local_participant	Cleans up RPC bindings during shutdown
RPC_* constants	livekit/agents/types.py	Canonical identifiers for built-in RPC methods
rtc.TextStream	TOPIC_AGENT_REQUEST / TOPIC_AGENT_RESPONSE	Alternative stream-based RPC for payloads exceeding 4 KB

Registering RPC Methods on the Agent

When an AgentSession initializes, it creates a ClientEventsHandler instance that automatically registers the built-in RPC methods. This occurs in the _register_rpc_handlers method of ClientEventsHandler.

In livekit/agents/voice/client_events.py (lines 84-94), the registration looks like this:

self._room.local_participant.register_rpc_method(
    RPC_GET_SESSION_STATE, self._rpc_get_session_state
)
self._room.local_participant.register_rpc_method(
    RPC_GET_CHAT_HISTORY, self._rpc_get_chat_history
)
self._room.local_participant.register_rpc_method(
    RPC_GET_AGENT_INFO, self._rpc_get_agent_info
)
self._room.local_participant.register_rpc_method(
    RPC_SEND_MESSAGE, self._rpc_send_message
)

Each registration binds a constant (defined in livekit/agents/types.py) to an instance method of the handler. The framework ensures these registrations occur only once per session.

Implementing RPC Handlers

RPC handlers are async methods that accept a Pydantic request model and return a Pydantic response model. This provides type safety and automatic JSON serialization.

For example, the _rpc_get_session_state handler in client_events.py (lines 284-291) implements the session state query:

async def _rpc_get_session_state(self, req: GetSessionStateRequest) -> GetSessionStateResponse:
    agent = self._session.current_agent
    return GetSessionStateResponse(
        agent_state=self._session.agent_state,
        user_state=self._session.user_state,
        agent_id=agent.id,
        options=asdict(self._session.options),
        created_at=self._session._started_at or time.time(),
    )

The handler accesses the current AgentSession state, constructs a GetSessionStateResponse, and returns it. The framework handles the JSON encoding and transmission back to the caller.

Invoking RPC Methods from the Client

Any participant in the room can invoke these registered methods using invoke_rpc_method on their local participant instance. The client must use the same RPC constants and Pydantic models as the agent.

Here is an example of client-side invocation:

from livekit import rtc
from livekit.agents.types import (
    RPC_GET_SESSION_STATE,
    RPC_GET_CHAT_HISTORY,
    RPC_SEND_MESSAGE,
)
from livekit.agents.voice.client_events import (
    GetSessionStateRequest,
    GetChatHistoryRequest,
    SendMessageRequest,
)

async def query_agent(room: rtc.Room):
    # Fetch current agent state

    state = await room.local_participant.invoke_rpc_method(
        RPC_GET_SESSION_STATE, GetSessionStateRequest()
    )
    print(f"Agent state: {state.agent_state}")
    
    # Retrieve chat history

    history = await room.local_participant.invoke_rpc_method(
        RPC_GET_CHAT_HISTORY, GetChatHistoryRequest()
    )
    for msg in history.items:
        print(f"{msg.role}: {msg.content}")
    
    # Send a message to the agent

    response = await room.local_participant.invoke_rpc_method(
        RPC_SEND_MESSAGE, SendMessageRequest(text="Hello from client!")
    )

The invoke_rpc_method call automatically serializes the Pydantic request to JSON, transmits it via the LiveKit signaling channel, awaits the response, and deserializes it back into the appropriate Pydantic model.

Handling Large Payloads with Text Streams

The built-in RPC channel has a payload limit of approximately 4 KB. For operations that may return large datasets (such as extensive chat histories or large metadata objects), LiveKit Agents provides an alternative text-stream RPC mechanism.

This system uses rtc.TextStream with specific topics: TOPIC_AGENT_REQUEST for incoming requests and TOPIC_AGENT_RESPONSE for outgoing responses.

In client_events.py (lines 302-306), the handler registers for stream requests:

self._room.register_text_stream_handler(
    TOPIC_AGENT_REQUEST, self._on_stream_request
)

The _on_stream_request method parses the incoming JSON as a StreamRequest, dispatches to the same internal logic as the RPC handlers, and sends the result back via send_text on TOPIC_AGENT_RESPONSE:

# Excerpt from stream handling logic

request = StreamRequest.model_validate_json(data)

# ... dispatch to internal handler ...

response = StreamResponse(
    request_id=request.request_id,
    payload=response_payload,
    error=error,
)
await self._room.local_participant.send_text(
    response.model_dump_json(),
    topic=TOPIC_AGENT_RESPONSE,
    destination_identities=[participant_identity],
)

Client implementations should check payload sizes and use the text-stream alternative when exceeding the 4 KB threshold.

Adding Custom RPC Methods

You can extend the ClientEventsHandler to expose domain-specific functionality beyond the built-in methods.

Here is a complete example of adding a custom "reset_agent" RPC:

from pydantic import BaseModel
from livekit.agents.voice.client_events import ClientEventsHandler

# 1. Define request/response models

class ResetAgentRequest(BaseModel):
    force: bool = False

class ResetAgentResponse(BaseModel):
    ok: bool
    message: str

# 2. Extend ClientEventsHandler

class CustomClientHandler(ClientEventsHandler):
    def _register_rpc_handlers(self):
        super()._register_rpc_handlers()  # Register built-ins

        # Register custom method

        self._room.local_participant.register_rpc_method(
            "reset_agent", self._rpc_reset_agent
        )
    
    async def _rpc_reset_agent(self, req: ResetAgentRequest) -> ResetAgentResponse:
        if req.force:
            await self._session.hard_reset()
            return ResetAgentResponse(ok=True, message="Agent hard reset completed")
        await self._session.soft_reset()
        return ResetAgentResponse(ok=True, message="Agent soft reset completed")

Clients invoke this custom method using the string identifier "reset_agent":

response = await room.local_participant.invoke_rpc_method(
    "reset_agent", ResetAgentRequest(force=True)
)

Summary

• RPC for client-agent communication in LiveKit Agents is implemented via ClientEventsHandler in livekit/agents/voice/client_events.py.
• Built-in methods (RPC_GET_SESSION_STATE, RPC_SEND_MESSAGE, etc.) are registered automatically using room.local_participant.register_rpc_method.
• Handlers use Pydantic models for type-safe request/response serialization.
• Clients invoke methods via room.local_participant.invoke_rpc_method, which handles JSON encoding and transport.
• For payloads exceeding 4 KB, use the text-stream alternative with topics TOPIC_AGENT_REQUEST and TOPIC_AGENT_RESPONSE.
• Custom RPC methods can be added by extending ClientEventsHandler and registering additional method names.

Frequently Asked Questions

What is the maximum payload size for RPC calls in LiveKit Agents?

The built-in RPC channel has a payload limit of approximately 4 KB. For larger data transfers, such as extensive chat histories or large metadata objects, you should use the text-stream RPC mechanism via TOPIC_AGENT_REQUEST and TOPIC_AGENT_RESPONSE topics, which removes this size restriction.

How do I add a custom RPC method to my LiveKit Agent?

Extend the ClientEventsHandler class and override the _register_rpc_handlers method to call super()._register_rpc_handlers() followed by self._room.local_participant.register_rpc_method("your_method_name", self._your_handler). Implement your handler as an async method that accepts a Pydantic request model and returns a Pydantic response model.

Can any participant in the room invoke RPC methods on the agent?

Yes, any participant that has joined the room can invoke the RPC methods registered by the agent using room.local_participant.invoke_rpc_method. The agent registers methods on its local participant, making them available to all other participants in the same room.

What built-in RPC methods are available in LiveKit Agents?

The framework provides four built-in RPC methods defined in livekit/agents/types.py: RPC_GET_SESSION_STATE (retrieves agent and user state), RPC_GET_CHAT_HISTORY (fetches conversation history), RPC_GET_AGENT_INFO (returns agent metadata), and RPC_SEND_MESSAGE (allows clients to send text messages to the agent).

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}