Quickstart

# Install the Pipecat CLI
uv tool install "pipecat-ai[cli]"

# Scaffold the quickstart project (also writes AGENTS.md + CLAUDE.md)
pipecat init quickstart

# Change to the project directory
cd pipecat-quickstart

cp .env.example .env

DEEPGRAM_API_KEY=your_deepgram_api_key
OPENAI_API_KEY=your_openai_api_key
CARTESIA_API_KEY=your_cartesia_api_key

uv sync

uv run bot.py

🚀 WebRTC server starting at http://localhost:7860/client
   Open this URL in your browser to connect!

pipecat cloud auth login

agent_name = "pipecat-quickstart"
secret_set = "pipecat-quickstart-secrets"

[scaling]
  min_agents = 1

pipecat cloud secrets set pipecat-quickstart-secrets --file .env

pipecat cloud deploy

# Create AI Services
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
    api_key=os.getenv("CARTESIA_API_KEY"),
    settings=CartesiaTTSService.Settings(
        voice=os.getenv("CARTESIA_VOICE_ID", "71a7ad14-091c-4e8e-a314-022ece01c121"),
    ),
)
llm = OpenAIRealtimeLLMService(
    api_key=os.getenv("OPENAI_API_KEY"),
    settings=OpenAIRealtimeLLMService.Settings(
        model=os.getenv("OPENAI_MODEL", "gpt-4.1"),
        system_instruction="You are a helpful assistant.",
    ),
)

context = LLMContext()
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        vad_analyzer=SileroVADAnalyzer(),
    ),
)

# Create the pipeline with the processors
pipeline = Pipeline([
    transport.input(),     # Receive audio from browser
    stt,                   # Speech-to-text (Deepgram)
    user_aggregator,       # Add user message to context
    llm,                   # Language model (OpenAI)
    tts,                   # Text-to-speech (Cartesia)
    transport.output(),    # Send audio back to browser
    assistant_aggregator,  # Add bot response to context
])

# Create a PipelineWorker to manage the pipeline execution
worker = PipelineWorker(
    pipeline,
    params=PipelineParams(
        enable_metrics=True,
        enable_usage_metrics=True,
    ),
)

# Event handler for when a client connects
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
    logger.info(f"Client connected")
    # Add a greeting message to the context
    context.add_message("developer", "Say hello and briefly introduce yourself.")
    # Prompt the bot to start talking when the client connects
    await worker.queue_frames([LLMRunFrame()])

# Event handler for when a client disconnects
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
    logger.info(f"Client disconnected")
    # Cancel the worker when the client disconnects
    # This stops the pipeline and all processors, cleaning up resources
    await worker.cancel()

# Create a WorkerRunner to run the worker
runner = WorkerRunner(handle_sigint=False)

# Finally, run the worker using the runner
# This will start the pipeline and begin processing frames
await runner.run(worker)

async def bot(runner_args: RunnerArguments):
    """Main bot entry point."""

    # Configure transport parameters for different environments
    transport_params = {
        "daily": lambda: DailyParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
        ),
        "webrtc": lambda: TransportParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
        ),
    }

    transport = await create_transport(runner_args, transport_params)
    await run_bot(transport, runner_args)

if __name__ == "__main__":
    from pipecat.runner.run import main
    main()