Tavus Transport

pip install pipecat-ai[tavus]

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.transports.services.tavus import TavusParams, TavusTransport

async def run_bot():
    async with aiohttp.ClientSession() as session:
        # Create the Tavus transport
        transport = TavusTransport(
            bot_name="Pipecat bot",
            api_key=os.getenv("TAVUS_API_KEY"),
            replica_id=os.getenv("TAVUS_REPLICA_ID"),
            session=session,
            params=TavusParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                microphone_out_enabled=False,
                vad_analyzer=SileroVADAnalyzer(),
            ),
        )

        # Set up your services and context

        # Create the pipeline
        pipeline = Pipeline([
            transport.input(),              # Receive audio from client
            stt,                            # Convert speech to text
            context_aggregator.user(),      # Add user messages to context
            llm,                            # Process text with LLM
            tts,                            # Convert text to speech
            transport.output(),             # Send audio responses to client
            context_aggregator.assistant(), # Add assistant responses to context
        ])

        # Register event handlers
        @transport.event_handler("on_client_connected")
        async def on_client_connected(transport, client):
            logger.info("Client connected")
            # Start the conversation when client connects
            await task.queue_frames([context_aggregator.user().get_context_frame()])

        @transport.event_handler("on_client_disconnected")
        async def on_client_disconnected(transport, client):
            logger.info("Client disconnected")