Examples

Common usage patterns for the Pipecat Cloud Python SDK

This page provides examples of common tasks with the Pipecat Cloud Python SDK.

Starting an Agent Session

This example shows how to start a session with various configurations:

import asyncio

from pipecatcloud.exception import AgentStartError
from pipecatcloud.session import Session, SessionParams


async def main():
    try:
        # Create session object
        session = Session(
            agent_name="my-first-agent",
            api_key=API_KEY,  # Replace with your actual API key
            params=SessionParams(
                use_daily=True,  # Optional: Creates a Daily room
                daily_room_properties={"start_video_off": False},
                data={"key": "value"},
            ),
        )

        # Start the session
        response = await session.start()

        # Get Daily room URL
        daily_url = f"{response['dailyRoom']}?t={response['dailyToken']}"
        print(f"Join Daily room: {daily_url}")

    except AgentStartError as e:
        print(f"Error starting agent: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")


# Run the async function
if __name__ == "__main__":
    asyncio.run(main())

Building a Bot Entry Point with Daily Arguments

from loguru import logger
from pipecat.runner.types import DailyRunnerArguments

async def bot(args: DailyRunnerArguments):
    """Main bot entry point compatible with the FastAPI route handler.

    Args:
        room_url: The Daily room URL
        token: The Daily room token
        body: The configuration object from the request body
    """
    logger.info(f"Bot process initialized {args.room_url} {args.token}")

    try:
        await main(args.room_url, args.token)
        logger.info("Bot process completed")
    except Exception as e:
        logger.exception(f"Error in bot process: {str(e)}")
        raise

Building a Bot Entry Point with WebSocket Arguments

from loguru import logger
from pipecat.runner.types import WebSocketRunnerArguments

async def bot(args: WebSocketRunnerArguments):
    """Main bot entry point for WebSocket connections.

    Args:
        ws: The WebSocket connection
    """
    logger.info("WebSocket bot process initialized")

    try:
        await main(args.websocket)
        logger.info("WebSocket bot process completed")
    except Exception as e:
        logger.exception(f"Error in WebSocket bot process: {str(e)}")
        raise