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

> Learn how to use RPC for client-agent communication in LiveKit Agents. Register RPC methods to query agent state, fetch chat history, or send messages using Pydantic models.

- Repository: [LiveKit/agents](https://github.com/livekit/agents)
- Tags: how-to-guide
- Published: 2026-03-06

---

**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`](https://github.com/livekit/agents/blob/main/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`](https://github.com/livekit/agents/blob/main/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`](https://github.com/livekit/agents/blob/main/livekit/agents/voice/client_events.py) (lines 84-94), the registration looks like this:

```python
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`](https://github.com/livekit/agents/blob/main/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`](https://github.com/livekit/agents/blob/main/client_events.py) (lines 284-291) implements the session state query:

```python
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:

```python
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`](https://github.com/livekit/agents/blob/main/client_events.py) (lines 302-306), the handler registers for stream requests:

```python
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`:

```python

# 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:

```python
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"`:

```python
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`](https://github.com/livekit/agents/blob/main/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`](https://github.com/livekit/agents/blob/main/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).