refined-hydrogen-81988
12/17/2025, 5:08 PMtall-airline-16756
12/17/2025, 5:08 PMtall-airline-16756
12/17/2025, 5:08 PMimport json
import asyncio
from livekit.agents import get_job_context, function_tool
@function_tool()
async def send_rpc_to_frontend(self, context, message: str, method: str = "showNotification"):
"""
Send an RPC from an agent (local participant) to a remote frontend participant.
"""
job_ctx = get_job_context()
room = job_ctx.room
# pick a remote participant (adjust selection as needed)
remote_participants = room.remote_participants
if not remote_participants:
raise Exception("No remote participants in the room")
remote_participant = next(iter(remote_participants.values()))
payload = json.dumps({
"message": message,
"sent_at": asyncio.get_event_loop().time()
})
try:
response = await room.local_participant.perform_rpc(
destination_identity=remote_participant.identity,
method=method,
payload=payload,
response_timeout=10.0
)
return {"status": "sent", "response": response}
except Exception as e:
raise Exception(f"Failed to send RPC: {e}")
Notes:
• method must match the RPC handler registered in the frontend.
• payload must be a JSON string; adjust its structure to your frontend's expectations.
• Change how you select the target participant if you need a specific user (by identity or metadata).
Would you like an example of the corresponding frontend RPC handler (JS) or help selecting a specific participant?
Sources: External data and RAG | LiveKit Docs | Remote method calls | LiveKit docs