Examples - Pipecat Cloud 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())

The bot entry-point examples below import from pipecat.runner.types, which pipecatcloud treats as an optional dependency. Outside an agent image, install it with uv add "pipecatcloud[pipecat]".

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