Akapulu Labs Video Avatar

uv add pipecat-akapulu

import aiohttp
from pipecat.pipeline.pipeline import Pipeline
from pipecat.transports.daily.transport import DailyParams, DailyTransport
from pipecat_akapulu import AkapuluAvatar, create_akapulu_call

async with aiohttp.ClientSession() as http:
    room_url, bot_token, browser_token, session_id = await create_akapulu_call(
        avatar_id="your-avatar-uuid",
        show_ai_label=True,
        session=http,
    )

    transport = DailyTransport(
        room_url=room_url,
        token=bot_token,
        bot_name="Assistant",
        params=DailyParams(
            audio_in_enabled=True,
            audio_out_enabled=False,
            video_out_enabled=False,
        ),
    )
    avatar = AkapuluAvatar(session_id=session_id)

    pipeline = Pipeline([
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        tts,
        avatar,  # Forwards assistant audio to the Akapulu Labs worker
        transport.output(),
        context_aggregator.assistant(),
    ])

@transport.event_handler("on_participant_joined")
async def on_participant_joined(_transport, participant):
    # Avatar publishes the spoken audio. Turn that microphone off for the voice bot.
    if participant["info"]["userName"] != "Avatar":
        return
    await transport.update_subscriptions(
        participant_settings={
            participant["id"]: {"media": {"microphone": "unsubscribed"}},
        },
    )