Common usage patterns for the Pipecat Cloud Python SDK
This page provides examples of common tasks with the Pipecat Cloud Python SDK.
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())
from loguru import logger
from pipecatcloud.agent import DailySessionArguments
async def bot(args: DailySessionArguments):
"""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
from loguru import logger
from pipecatcloud.agent import WebSocketSessionArguments
async def bot(args: WebSocketSessionArguments):
"""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