broswer-automation/USER_ID_METADATA_EXAMPLE.py

#!/usr/bin/env python3
"""
Example script showing how to pass user ID via LiveKit metadata
and how the LiveKit agent retrieves it with fallback options.
"""

import asyncio
import json
import os
from livekit import api, rtc

# Example of how to create a LiveKit room with user ID in metadata
async def create_room_with_user_id(user_id: str, room_name: str):
    """
    Create a LiveKit room with user ID in metadata
    """
    # Initialize LiveKit API client
    livekit_api = api.LiveKitAPI(
        url=os.getenv('LIVEKIT_URL', 'ws://localhost:7880'),
        api_key=os.getenv('LIVEKIT_API_KEY'),
        api_secret=os.getenv('LIVEKIT_API_SECRET')
    )
    
    # Create room with user ID in metadata
    room_metadata = {
        "userId": user_id,
        "createdBy": "chrome_extension",
        "timestamp": int(asyncio.get_event_loop().time())
    }
    
    try:
        room = await livekit_api.room.create_room(
            api.CreateRoomRequest(
                name=room_name,
                metadata=json.dumps(room_metadata),
                empty_timeout=300,  # 5 minutes
                max_participants=10
            )
        )
        print(f"✅ Room created: {room.name} with user ID: {user_id}")
        return room
    except Exception as e:
        print(f"❌ Failed to create room: {e}")
        return None

# Example of how to join a room and set participant metadata with user ID
async def join_room_with_user_id(user_id: str, room_name: str):
    """
    Join a LiveKit room and set participant metadata with user ID
    """
    # Create access token with user ID
    token = (
        api.AccessToken(
            api_key=os.getenv('LIVEKIT_API_KEY'),
            api_secret=os.getenv('LIVEKIT_API_SECRET')
        )
        .with_identity(f"user_{user_id}")
        .with_name(f"Chrome User {user_id[:8]}")
        .with_grants(api.VideoGrants(room_join=True, room=room_name))
        .with_metadata(json.dumps({
            "userId": user_id,
            "source": "chrome_extension",
            "capabilities": ["browser_automation", "voice_commands"]
        }))
        .to_jwt()
    )
    
    # Connect to room
    room = rtc.Room()
    
    try:
        await room.connect(
            url=os.getenv('LIVEKIT_URL', 'ws://localhost:7880'),
            token=token
        )
        print(f"✅ Connected to room: {room_name} as user: {user_id}")
        
        # Update participant metadata after connection
        await room.local_participant.update_metadata(json.dumps({
            "userId": user_id,
            "status": "active",
            "lastActivity": int(asyncio.get_event_loop().time())
        }))
        
        return room
    except Exception as e:
        print(f"❌ Failed to join room: {e}")
        return None

# Example usage functions
def example_user_id_from_chrome_extension():
    """Example of how Chrome extension generates user ID"""
    import time
    import random
    import string
    
    timestamp = int(time.time())
    random_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=12))
    return f"user_{timestamp}_{random_suffix}"

def example_user_id_from_environment():
    """Example of getting user ID from environment variable"""
    return os.getenv('CHROME_USER_ID', None)

def example_user_id_fallback():
    """Example of generating fallback user ID"""
    import time
    import random
    import string
    
    timestamp = int(time.time())
    random_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))
    return f"fallback_user_{timestamp}_{random_suffix}"

async def demonstrate_user_id_priority():
    """
    Demonstrate the priority order for getting user ID:
    1. From metadata (if available)
    2. From environment variable
    3. Generate random fallback
    """
    print("🔍 User ID Priority Demonstration")
    print("=" * 50)
    
    # 1. Check metadata (simulated - would come from LiveKit participant/room)
    metadata_user_id = None  # Would be extracted from LiveKit metadata
    if metadata_user_id:
        print(f"✅ Using user ID from metadata: {metadata_user_id}")
        return metadata_user_id
    else:
        print("❌ No user ID found in metadata")
    
    # 2. Check environment variable
    env_user_id = example_user_id_from_environment()
    if env_user_id:
        print(f"✅ Using user ID from environment: {env_user_id}")
        return env_user_id
    else:
        print("❌ No user ID found in environment variable")
    
    # 3. Generate fallback
    fallback_user_id = example_user_id_fallback()
    print(f"✅ Generated fallback user ID: {fallback_user_id}")
    return fallback_user_id

async def main():
    """Main demonstration function"""
    print("🚀 LiveKit User ID Metadata Example")
    print("=" * 60)
    
    # Demonstrate user ID priority
    user_id = await demonstrate_user_id_priority()
    print(f"\n📋 Final user ID: {user_id}")
    
    # Example room name (Chrome extension format)
    room_name = f"mcp-chrome-user-{user_id}"
    print(f"🏠 Room name: {room_name}")
    
    # Show how Chrome extension would generate user ID
    chrome_user_id = example_user_id_from_chrome_extension()
    print(f"🌐 Chrome extension user ID example: {chrome_user_id}")
    
    print("\n📝 Usage in LiveKit Agent:")
    print("   1. Agent checks participant metadata for 'userId' field")
    print("   2. If not found, checks room metadata for 'userId' field")
    print("   3. If not found, checks CHROME_USER_ID environment variable")
    print("   4. If not found, generates random user ID")
    
    print("\n🔧 To set user ID in metadata:")
    print("   - Room metadata: Include 'userId' in CreateRoomRequest metadata")
    print("   - Participant metadata: Include 'userId' in access token metadata")
    print("   - Environment: Set CHROME_USER_ID environment variable")

if __name__ == "__main__":
    # Set example environment variable for demonstration
    os.environ['CHROME_USER_ID'] = 'user_1704067200000_example123'
    
    # Run the demonstration
    asyncio.run(main())