feat(rtc): add participant actions (register, discover, call)

Author: anunaym14

livekit-rtc/livekit/rtc/__init__.py

@@ -115,6 +115,12 @@
 except Exception:  # pragma: no cover - optional dependency (sounddevice)
     _HAS_MEDIA_DEVICES = False
 from .utils import combine_audio_frames
+from .actions import (
+    ActionContext,
+    ActionDeclinedError,
+    ActionEntry,
+    ActionRegistration,
+)
 from .rpc import RpcCallInfo, RpcError, RpcInterceptor, RpcInvocationData
 from .synchronizer import AVSynchronizer
 from .data_stream import (
@@ -209,6 +215,10 @@
     "AudioMixer",
     "AudioResampler",
     "AudioResamplerQuality",
+    "ActionContext",
+    "ActionDeclinedError",
+    "ActionEntry",
+    "ActionRegistration",
     "RpcError",
     "RpcInvocationData",
     "RpcCallInfo",

livekit-rtc/livekit/rtc/actions.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import inspect
+import json
+from dataclasses import asdict, dataclass, field
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Union
+
+from .rpc import RpcError
+
+ACTIONS_ATTRIBUTE = "lk.actions"
+ACTION_METHOD_PREFIX = "action:"
+ACTION_DECLINED_CODE = 1710
+
+
+@dataclass
+class ActionEntry:
+    name: str
+    description: str
+    parameters: Dict[str, Any] = field(default_factory=dict)
+    consent: str = "none"
+
+
+@dataclass
+class ActionContext:
+    caller_identity: str
+
+
+class ActionDeclinedError(Exception):
+    def __init__(self, message: str = "action declined") -> None:
+        super().__init__(message)
+
+
+ActionHandler = Callable[[Dict[str, Any], ActionContext], Union[Any, Awaitable[Any]]]
+
+
+class ActionRegistration:
+    def __init__(self, unregister: Callable[[], Awaitable[None]]) -> None:
+        self._unregister = unregister
+
+    async def unregister(self) -> None:
+        await self._unregister()
+
+
+def parse_actions(raw: Optional[str]) -> List[ActionEntry]:
+    if not raw:
+        return []
+    try:
+        items = json.loads(raw)
+    except ValueError:
+        return []
+    if not isinstance(items, list):
+        return []
+    return [
+        ActionEntry(
+            name=item["name"],
+            description=item.get("description", ""),
+            parameters=item.get("parameters") or {},
+            consent=item.get("consent", "none"),
+        )
+        for item in items
+        if isinstance(item, dict) and isinstance(item.get("name"), str)
+    ]
+
+
+def serialize_actions(entries: List[ActionEntry]) -> str:
+    return json.dumps([asdict(e) for e in entries])
+
+
+async def invoke_handler(handler: ActionHandler, payload: str, caller_identity: str) -> str:
+    args = json.loads(payload) if payload else {}
+    try:
+        result = handler(args, ActionContext(caller_identity=caller_identity))
+        if inspect.isawaitable(result):
+            result = await result
+    except ActionDeclinedError as e:
+        raise RpcError(ACTION_DECLINED_CODE, str(e)) from e
+    return json.dumps(result)

livekit-rtc/livekit/rtc/participant.py

@@ -18,12 +18,13 @@
 import asyncio
 import functools
 import inspect
+import json
 import datetime
 import enum
 import os
 import mimetypes
 import weakref
-from typing import List, Union, Callable, Dict, Awaitable, Optional, Mapping, cast, TypeVar
+from typing import Any, List, Union, Callable, Dict, Awaitable, Optional, Mapping, cast, TypeVar
 from abc import abstractmethod, ABC
 
 from ._ffi_client import FfiClient, FfiHandle
@@ -59,6 +60,18 @@
     _chain_outgoing,
 )
 from ._proto.rpc_pb2 import RpcMethodInvocationResponseRequest
+from .actions import (
+    ACTION_DECLINED_CODE,
+    ACTION_METHOD_PREFIX,
+    ACTIONS_ATTRIBUTE,
+    ActionDeclinedError,
+    ActionEntry,
+    ActionHandler,
+    ActionRegistration,
+    invoke_handler,
+    parse_actions,
+    serialize_actions,
+)
 from .log import logger
 
 from .data_stream import (
@@ -173,6 +186,11 @@ def attributes(self) -> dict[str, str]:
         """Custom attributes associated with the participant."""
         return dict(self._info.attributes)
 
+    @property
+    def actions(self) -> List[ActionEntry]:
+        """Snapshot of the actions this participant exposes."""
+        return parse_actions(self._info.attributes.get(ACTIONS_ATTRIBUTE))
+
     @property
     def kind(self) -> proto_participant.ParticipantKind.ValueType:
         """Participant's kind (e.g., regular participant, ingress, egress, sip, agent)."""
@@ -276,6 +294,7 @@ def __init__(
         self._track_publications: dict[str, LocalTrackPublication] = {}
         self._rpc_handlers: Dict[str, RpcHandler] = {}
         self._rpc_interceptors: List[RpcInterceptor] = []
+        self._action_catalog: Dict[str, ActionEntry] = {}
         # Handles of data stream writers that have been opened but not yet
         # closed, so the room can drop them at disconnect. The FFI close
         # request consumes the handle (take_handle), so an entry is removed as
@@ -584,6 +603,74 @@ def unregister_rpc_method(self, method: str) -> None:
 
         FfiClient.instance.request(req)
 
+    async def register_action(
+        self,
+        name: str,
+        description: str,
+        parameters: Dict[str, Any],
+        handler: ActionHandler,
+        *,
+        consent: str = "none",
+    ) -> ActionRegistration:
+        """
+        Expose an action other participants can discover and call.
+
+        The handler receives the parsed arguments and an `ActionContext`, may be sync or async,
+        and returns a JSON-serializable result. Raise `ActionDeclinedError` to decline.
+        """
+        method = ACTION_METHOD_PREFIX + name
+
+        async def rpc_handler(data: RpcInvocationData) -> str:
+            return await invoke_handler(handler, data.payload, data.caller_identity)
+
+        self.register_rpc_method(method, rpc_handler)
+        self._action_catalog[name] = ActionEntry(name, description, parameters, consent)
+        await self._publish_actions()
+
+        async def unregister() -> None:
+            self.unregister_rpc_method(method)
+            self._action_catalog.pop(name, None)
+            await self._publish_actions()
+
+        return ActionRegistration(unregister)
+
+    async def call_action(
+        self,
+        destination_identity: str,
+        name: str,
+        args: Optional[Dict[str, Any]] = None,
+        *,
+        response_timeout: Optional[float] = None,
+    ) -> Any:
+        """
+        Call an action on another participant and return its parsed result.
+
+        Raises:
+            ActionDeclinedError: The participant declined the call.
+            RpcError: Unknown action, timeout, or other transport failure.
+        """
+        try:
+            response = await self.perform_rpc(
+                destination_identity=destination_identity,
+                method=ACTION_METHOD_PREFIX + name,
+                payload=json.dumps(args or {}),
+                response_timeout=response_timeout,
+            )
+        except RpcError as e:
+            if e.code == ACTION_DECLINED_CODE:
+                raise ActionDeclinedError(e.message) from e
+            raise
+        return json.loads(response) if response else None
+
+    async def _publish_actions(self) -> None:
+        await self.set_attributes(
+            {ACTIONS_ATTRIBUTE: serialize_actions(list(self._action_catalog.values()))}
+        )
+
+    def _republish_actions(self) -> None:
+        if self._action_catalog:
+            asyncio.ensure_future(self._publish_actions())
+
     def set_track_subscription_permissions(
         self,
         *,

livekit-rtc/livekit/rtc/room.py

@@ -33,6 +33,7 @@
 from ._proto.rpc_pb2 import RpcMethodInvocationEvent
 from ._utils import BroadcastQueue
 from .e2ee import E2EEManager, E2EEOptions
+from .actions import ACTIONS_ATTRIBUTE
 from .log import logger
 from .participant import (
     LocalParticipant,
@@ -72,6 +73,7 @@
     "participant_metadata_changed",
     "participant_name_changed",
     "participant_attributes_changed",
+    "participant_actions_changed",
     "connection_quality_changed",
     "participant_encryption_status_changed",
     "participant_permissions_changed",
@@ -403,6 +405,8 @@ def on(self, event: EventTypes, callback: Optional[Callable] = None) -> Callable
                 - Arguments: `participant` (Participant), `old_name` (str), `new_name` (str)
             - **"participant_attributes_changed"**: Called when a participant's attributes change.
                 - Arguments: `changed_attributes` (dict), `participant` (Participant)
+            - **"participant_actions_changed"**: Called when a participant's exposed actions change.
+                - Arguments: `participant` (Participant), `actions` (list[ActionEntry])
             - **"participant_encryption_status_changed"**: Called when a participant's encryption status changes.
                 - Arguments `is_encrypted` (bool), `participant` (Participant)
             - **"connection_quality_changed"**: Called when a participant's connection quality changes.
@@ -1030,6 +1034,8 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None:
                 changed_attributes,
                 participant,
             )
+            if ACTIONS_ATTRIBUTE in changed_attributes:
+                self.emit("participant_actions_changed", participant.actions, participant)
         elif which == "participant_encryption_status_changed":
             identity = event.participant_encryption_status_changed.participant_identity
             participant = self._retrieve_participant(identity)
@@ -1130,6 +1136,8 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None:
         elif which == "reconnecting":
             self.emit("reconnecting")
         elif which == "reconnected":
+            if self._local_participant is not None:
+                self._local_participant._republish_actions()
             self.emit("reconnected")
         elif which == "text_stream_opened":
             self._handle_text_stream_opened(event.text_stream_opened)