params

Author: bcherry

examples/rpc.py

@@ -3,8 +3,7 @@
 import json
 import asyncio
 from dotenv import load_dotenv
-import time
-
+from livekit.rtc.rpc import RpcHandlerParams
 load_dotenv(dotenv_path=".env.local", override=False)
 LIVEKIT_API_KEY = os.getenv("LIVEKIT_API_KEY")
 LIVEKIT_API_SECRET = os.getenv("LIVEKIT_API_SECRET")
@@ -91,26 +90,20 @@ async def main():
 def register_receiver_methods(greeters_room: rtc.Room, math_genius_room: rtc.Room):
     @greeters_room.local_participant.rpc_method("arrival")
     async def arrival_method(
-        request_id: str,
-        caller_identity: str,
-        payload: str,
-        response_timeout: float,
+        params: RpcHandlerParams,
     ):
-        print(f'[Greeter] Oh {caller_identity} arrived and said "{payload}"')
+        print(f'[Greeter] Oh {params.caller_identity} arrived and said "{params.payload}"')
         await asyncio.sleep(2)
         return "Welcome and have a wonderful day!"
 
     @math_genius_room.local_participant.rpc_method("square-root")
     async def square_root_method(
-        request_id: str,
-        caller_identity: str,
-        payload: str,
-        response_timeout: float,
+        params: RpcHandlerParams,
     ):
-        json_data = json.loads(payload)
+        json_data = json.loads(params.payload)
         number = json_data["number"]
         print(
-            f"[Math Genius] I guess {caller_identity} wants the square root of {number}. I've only got {response_timeout} seconds to respond but I think I can pull it off."
+            f"[Math Genius] I guess {params.caller_identity} wants the square root of {number}. I've only got {params.response_timeout} seconds to respond but I think I can pull it off."
         )
 
         print("[Math Genius] *doing math*…")
@@ -121,32 +114,26 @@ async def square_root_method(
         return json.dumps({"result": result})
 
     @math_genius_room.local_participant.rpc_method("divide")
-    def divide_method(
-        request_id: str,
-        caller_identity: str,
-        payload: str,
-        response_timeout: float,
+    async def divide_method(
+        params: RpcHandlerParams,
     ):
-        json_data = json.loads(payload)
+        json_data = json.loads(params.payload)
         dividend = json_data["dividend"]
         divisor = json_data["divisor"]
         print(
-            f"[Math Genius] {caller_identity} wants to divide {dividend} by {divisor}."
+            f"[Math Genius] {params.caller_identity} wants to divide {dividend} by {divisor}."
         )
 
         result = dividend / divisor
         return json.dumps({"result": result})
 
     @math_genius_room.local_participant.rpc_method("long-calculation")
     async def long_calculation_method(
-        request_id: str,
-        caller_identity: str,
-        payload: str,
-        response_timeout: float,
+        params: RpcHandlerParams,
     ):
-        print(f"[Math Genius] Starting a very long calculation for {caller_identity}")
+        print(f"[Math Genius] Starting a very long calculation for {params.caller_identity}")
         print(
-            f"[Math Genius] This will take 30 seconds even though you're only giving me {response_timeout} seconds"
+            f"[Math Genius] This will take 30 seconds even though you're only giving me {params.response_timeout} seconds"
         )
         await asyncio.sleep(30)
         return json.dumps({"result": "Calculation complete!"})

livekit-rtc/livekit/rtc/participant.py

@@ -40,6 +40,7 @@
 from .log import logger
 import asyncio
 
+from .rpc import RpcHandlerParams
 
 class PublishTrackError(Exception):
     def __init__(self, message: str) -> None:
@@ -118,7 +119,7 @@ def __init__(
         self._room_queue = room_queue
         self._track_publications: dict[str, LocalTrackPublication] = {}  # type: ignore
         self._rpc_handlers: Dict[
-            str, Callable[[str, str, str, float], Union[Awaitable[str], str]]
+            str, Callable[[RpcHandlerParams], Union[Awaitable[str], str]]
         ] = {}
 
     @property
@@ -291,7 +292,7 @@ async def perform_rpc(
     def register_rpc_method(
         self,
         method: str,
-        handler: Callable[[str, str, str, float], Union[Awaitable[str], str]],
+        handler: Callable[[RpcHandlerParams], Union[Awaitable[str], str]],
     ) -> None:
         """
         Establishes the participant as a receiver for calls of the specified RPC method.
@@ -308,20 +309,15 @@ def register_rpc_method(
             RpcError: On failure. Details in `message`.
 
         Example:
-            async def greet_handler(request_id: str, caller_identity: str, payload: str, response_timeout: float) -> str:
-                print(f"Received greeting from {caller_identity}: {payload}")
-                return f"Hello, {caller_identity}!"
+            async def greet_handler(params: RpcHandlerParams) -> str:
+                print(f"Received greeting from {params.caller_identity}: {params.payload}")
+                return f"Hello, {params.caller_identity}!"
 
             await room.local_participant.register_rpc_method('greet', greet_handler)
 
-        The handler receives the following parameters:
-        - `request_id`: A unique identifier for this RPC request
-        - `caller_identity`: The identity of the RemoteParticipant who initiated the RPC call
-        - `payload`: The data sent by the caller (as a string)
-        - `response_timeout`: The maximum time available to return a response
-
         The handler should return a string or a coroutine that resolves to a string.
-        If unable to respond within `response_timeout`, the request will result in an error on the caller's side.
+        
+        If unable to respond within `response_timeout`, the caller will hang up and receive an error on their side.
 
         You may raise errors of type `RpcError` with a string `message` in the handler,
         and they will be received on the caller's side with the message intact.
@@ -344,16 +340,16 @@ def rpc_method(self, method: str):
 
         Example:
             @local_participant.rpc_method("greet")
-            async def greet_handler(request_id: str, caller_identity: str, payload: str, response_timeout: float) -> str:
-                print(f"Received greeting from {caller_identity}: {payload}")
-                return f"Hello, {caller_identity}!"
+            async def greet_handler(params: RpcHandlerParams) -> str:
+                print(f"Received greeting from {params.caller_identity}: {params.payload}")
+                return f"Hello, {params.caller_identity}!"
 
         See Also:
             `register_rpc_method` for more details
         """
 
         def decorator(
-            handler: Callable[[str, str, str, float], Union[Awaitable[str], str]],
+            handler: Callable[[RpcHandlerParams], Union[Awaitable[str], str]],
         ):
             self.register_rpc_method(method, handler)
             return handler
@@ -386,6 +382,8 @@ async def _handle_rpc_method_invocation(
     ) -> None:
         response_error: Optional[RpcError] = None
         response_payload: Optional[str] = None
+        
+        params = RpcHandlerParams(request_id, caller_identity, payload, response_timeout)
 
         handler = self._rpc_handlers.get(method)
 
@@ -394,13 +392,11 @@ async def _handle_rpc_method_invocation(
         else:
             try:
                 if asyncio.iscoroutinefunction(handler):
-                    async_handler = handler  # type: Callable[[str, str, str, float], Awaitable[str]]
+                    async_handler = handler  # type: Callable[[RpcHandlerParams], Awaitable[str]]
 
                     async def run_handler():
                         try:
-                            return await async_handler(
-                                request_id, caller_identity, payload, response_timeout
-                            )
+                            return await async_handler(params)
                         except asyncio.CancelledError:
                             # This will be caught by the outer try-except if it's due to timeout
                             raise
@@ -416,10 +412,8 @@ async def run_handler():
                             RpcError.ErrorCode.RECIPIENT_DISCONNECTED
                         )
                 else:
-                    sync_handler = handler  # type: Callable[[str, str, str, float], str]
-                    response_payload = sync_handler(
-                        request_id, caller_identity, payload, response_timeout
-                    )
+                    sync_handler = handler  # type: Callable[[RpcHandlerParams], str]
+                    response_payload = sync_handler(params)
             except RpcError as error:
                 response_error = error
             except Exception as error:

livekit-rtc/livekit/rtc/rpc.py

@@ -15,6 +15,22 @@
 from typing import Optional, Dict, Union, ClassVar
 from enum import IntEnum
 from ._proto import rpc_pb2 as proto_rpc
+from dataclasses import dataclass
+
+@dataclass
+class RpcHandlerParams:
+    """Parameters passed to method handler for incoming RPC invocations
+    
+    Attributes:
+        request_id (str): The unique request ID. Will match at both sides of the call, useful for debugging or logging.
+        caller_identity (str): The unique participant identity of the caller.
+        payload (str): The payload of the request. User-definable format, typically JSON.
+        response_timeout (float): The maximum time the caller will wait for a response.
+    """
+    request_id: str
+    caller_identity: str
+    payload: str
+    response_timeout: float
 
 
 class RpcError(Exception):