Example: Self-Hosted LiveKit Agent with Gestures

from livekit.plugins import bithuman

bithuman_avatar = bithuman.AvatarSession(
    api_secret=os.getenv("BITHUMAN_API_SECRET"),
    model_path=os.getenv("BITHUMAN_MODEL_PATH"),
)

import requests

agent_id = "A31KJC8622"
url = f"https://api.bithuman.ai/v1/dynamics/{agent_id}"
headers = {"api-secret": "YOUR_API_SECRET"}

response = requests.get(url, headers=headers)
dynamics_data = response.json()

if dynamics_data.get("success"):
    gestures_dict = dynamics_data["data"].get("gestures", {})
    available_gestures = list(gestures_dict.keys())
    print(f"Available gestures: {available_gestures}")
    # Example: ["mini_wave_hello", "talk_head_nod_subtle", "laugh_react"]

from livekit.agents import AgentSession, JobContext, UserInputTranscribedEvent
from livekit.plugins import bithuman
from bithuman.api import VideoControl
import asyncio
import os

KEYWORD_ACTION_MAP = {
    "laugh": "laugh_react",
    "laughing": "laugh_react",
    "haha": "laugh_react",
    "funny": "laugh_react",
    "hello": "mini_wave_hello",
    "hi": "mini_wave_hello",
}

async def entrypoint(ctx: JobContext):
    await ctx.connect()
    await ctx.wait_for_participant()

    bithuman_avatar = bithuman.AvatarSession(
        api_secret=os.getenv("BITHUMAN_API_SECRET"),
        model_path=os.getenv("BITHUMAN_MODEL_PATH"),
    )

    session = AgentSession(...)
    await bithuman_avatar.start(session, room=ctx.room)

    @session.on("user_input_transcribed")
    def on_user_input_transcribed(event: UserInputTranscribedEvent):
        if not event.is_final:
            return

        transcript = event.transcript.lower()
        for keyword, action in KEYWORD_ACTION_MAP.items():
            if keyword in transcript:
                asyncio.create_task(
                    bithuman_avatar.runtime.push(VideoControl(action=action))
                )
                break