From 403d22e62c6ecce9831788081be4c209e2da9549 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 26 Sep 2025 10:28:19 -0300 Subject: [PATCH 1/3] Creating the whatsapp routes inside the runner. --- src/pipecat/runner/run.py | 134 +++++++++++++++++++++++++++++++++++++- 1 file changed, 133 insertions(+), 1 deletion(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index dcf0d05ad..6672f9b37 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -70,7 +70,9 @@ import asyncio import os import sys from contextlib import asynccontextmanager +from typing import Optional +import aiohttp from loguru import logger from pipecat.runner.types import ( @@ -82,7 +84,7 @@ from pipecat.runner.types import ( try: import uvicorn from dotenv import load_dotenv - from fastapi import BackgroundTasks, FastAPI, Request, WebSocket + from fastapi import BackgroundTasks, FastAPI, HTTPException, Request, WebSocket from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, RedirectResponse except ImportError as e: @@ -166,6 +168,7 @@ def _create_server_app( # Set up transport-specific routes if transport_type == "webrtc": _setup_webrtc_routes(app, esp32_mode=esp32_mode, host=host) + _setup_whatsapp_routes(app) elif transport_type == "daily": _setup_daily_routes(app) elif transport_type in ["twilio", "telnyx", "plivo", "exotel"]: @@ -229,6 +232,135 @@ def _setup_webrtc_routes(app: FastAPI, esp32_mode: bool = False, host: str = "lo app.router.lifespan_context = lifespan +def _setup_whatsapp_routes(app: FastAPI): + """Set up WebRTC-specific routes.""" + try: + from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI + + from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection + from pipecat.transports.smallwebrtc.request_handler import ( + SmallWebRTCRequest, + SmallWebRTCRequestHandler, + ) + from pipecat.transports.whatsapp.api import WhatsAppWebhookRequest + from pipecat.transports.whatsapp.client import WhatsAppClient + except ImportError as e: + logger.error(f"WebRTC transport dependencies not installed: {e}") + return + + WHATSAPP_TOKEN = os.getenv("WHATSAPP_TOKEN") + WHATSAPP_PHONE_NUMBER_ID = os.getenv("WHATSAPP_PHONE_NUMBER_ID") + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN = os.getenv("WHATSAPP_WEBHOOK_VERIFICATION_TOKEN") + + if not all( + [ + WHATSAPP_TOKEN, + WHATSAPP_PHONE_NUMBER_ID, + WHATSAPP_WEBHOOK_VERIFICATION_TOKEN, + ] + ): + logger.debug( + "Missing required environment variables for WhatsApp transport. Keeping it disabled." + ) + return + + # Global WhatsApp client instance + whatsapp_client: Optional[WhatsAppClient] = None + + @app.get( + "/whatsapp", + summary="Verify WhatsApp webhook", + description="Handles WhatsApp webhook verification requests from Meta", + ) + async def verify_webhook(request: Request): + """Verify WhatsApp webhook endpoint. + + This endpoint is called by Meta's WhatsApp Business API to verify + the webhook URL during setup. It validates the verification token + and returns the challenge parameter if successful. + """ + params = dict(request.query_params) + logger.debug(f"Webhook verification request received with params: {list(params.keys())}") + + try: + result = await whatsapp_client.handle_verify_webhook_request( + params=params, expected_verification_token=WHATSAPP_WEBHOOK_VERIFICATION_TOKEN + ) + logger.info("Webhook verification successful") + return result + except ValueError as e: + logger.warning(f"Webhook verification failed: {e}") + raise HTTPException(status_code=403, detail="Verification failed") + + @app.post( + "/whatsapp", + summary="Handle WhatsApp webhook events", + description="Processes incoming WhatsApp messages and call events", + ) + async def whatsapp_webhook(body: WhatsAppWebhookRequest, background_tasks: BackgroundTasks): + """Handle incoming WhatsApp webhook events. + + For call events, establishes WebRTC connections and spawns bot instances + in the background to handle real-time communication. + """ + # Validate webhook object type + if body.object != "whatsapp_business_account": + logger.warning(f"Invalid webhook object type: {body.object}") + raise HTTPException(status_code=400, detail="Invalid object type") + + logger.debug(f"Processing WhatsApp webhook: {body.model_dump()}") + + async def connection_callback(connection: SmallWebRTCConnection): + """Handle new WebRTC connections from WhatsApp calls. + + Called when a WebRTC connection is established for a WhatsApp call. + Spawns a bot instance to handle the conversation. + + Args: + connection: The established WebRTC connection + """ + bot_module = _get_bot_module() + runner_args = SmallWebRTCRunnerArguments(webrtc_connection=connection) + background_tasks.add_task(bot_module.bot, runner_args) + + try: + # Process the webhook request + result = await whatsapp_client.handle_webhook_request(body, connection_callback) + logger.debug(f"Webhook processed successfully: {result}") + return {"status": "success", "message": "Webhook processed successfully"} + except ValueError as ve: + logger.warning(f"Invalid webhook request format: {ve}") + raise HTTPException(status_code=400, detail=f"Invalid request: {str(ve)}") + except Exception as e: + logger.error(f"Internal error processing webhook: {e}") + raise HTTPException(status_code=500, detail="Internal server error processing webhook") + + @asynccontextmanager + async def lifespan(app: FastAPI): + """Manage FastAPI application lifecycle and cleanup connections.""" + global whatsapp_client + + # Initialize WhatsApp client with persistent HTTP session + async with aiohttp.ClientSession() as session: + whatsapp_client = WhatsAppClient( + whatsapp_token=WHATSAPP_TOKEN, + phone_number_id=WHATSAPP_PHONE_NUMBER_ID, + session=session, + ) + logger.info("WhatsApp client initialized successfully") + + try: + yield # Run the application + finally: + # Cleanup all active calls on shutdown + logger.info("Cleaning up WhatsApp client resources...") + if whatsapp_client: + await whatsapp_client.terminate_all_calls() + logger.info("Cleanup completed") + + app.router.lifespan_context = lifespan + + def _setup_daily_routes(app: FastAPI): """Set up Daily-specific routes.""" From b3a84fc7729bcad707d8d790a3c714916c511f2c Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 26 Sep 2025 10:47:04 -0300 Subject: [PATCH 2/3] Refactoring how we are handling the lifespan inside the runner. --- src/pipecat/runner/run.py | 49 ++++++++++++++++++++++++++++++++------- 1 file changed, 41 insertions(+), 8 deletions(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 6672f9b37..80a3c5771 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -224,12 +224,36 @@ def _setup_webrtc_routes(app: FastAPI, esp32_mode: bool = False, host: str = "lo return answer @asynccontextmanager - async def lifespan(app: FastAPI): + async def smallwebrtc_lifespan(app: FastAPI): """Manage FastAPI application lifecycle and cleanup connections.""" yield await small_webrtc_handler.close() - app.router.lifespan_context = lifespan + # Add the SmallWebRTC lifespan to the app + _add_lifespan_to_app(app, smallwebrtc_lifespan) + + +def _add_lifespan_to_app(app: FastAPI, new_lifespan): + """Add a new lifespan context manager to the app, combining with existing if present. + + Args: + app: The FastAPI application instance + new_lifespan: The new lifespan context manager to add + """ + if hasattr(app.router, 'lifespan_context') and app.router.lifespan_context is not None: + # If there's already a lifespan context, combine them + existing_lifespan = app.router.lifespan_context + + @asynccontextmanager + async def combined_lifespan(app: FastAPI): + async with existing_lifespan(app): + async with new_lifespan(app): + yield + + app.router.lifespan_context = combined_lifespan + else: + # No existing lifespan, use the new one + app.router.lifespan_context = new_lifespan def _setup_whatsapp_routes(app: FastAPI): @@ -279,6 +303,10 @@ def _setup_whatsapp_routes(app: FastAPI): the webhook URL during setup. It validates the verification token and returns the challenge parameter if successful. """ + if whatsapp_client is None: + logger.error("WhatsApp client is not initialized") + raise HTTPException(status_code=503, detail="Service unavailable") + params = dict(request.query_params) logger.debug(f"Webhook verification request received with params: {list(params.keys())}") @@ -303,6 +331,10 @@ def _setup_whatsapp_routes(app: FastAPI): For call events, establishes WebRTC connections and spawns bot instances in the background to handle real-time communication. """ + if whatsapp_client is None: + logger.error("WhatsApp client is not initialized") + raise HTTPException(status_code=503, detail="Service unavailable") + # Validate webhook object type if body.object != "whatsapp_business_account": logger.warning(f"Invalid webhook object type: {body.object}") @@ -336,9 +368,9 @@ def _setup_whatsapp_routes(app: FastAPI): raise HTTPException(status_code=500, detail="Internal server error processing webhook") @asynccontextmanager - async def lifespan(app: FastAPI): - """Manage FastAPI application lifecycle and cleanup connections.""" - global whatsapp_client + async def whatsapp_lifespan(app: FastAPI): + """Manage WhatsApp client lifecycle and cleanup connections.""" + nonlocal whatsapp_client # Initialize WhatsApp client with persistent HTTP session async with aiohttp.ClientSession() as session: @@ -356,9 +388,10 @@ def _setup_whatsapp_routes(app: FastAPI): logger.info("Cleaning up WhatsApp client resources...") if whatsapp_client: await whatsapp_client.terminate_all_calls() - logger.info("Cleanup completed") + logger.info("WhatsApp cleanup completed") - app.router.lifespan_context = lifespan + # Add the WhatsApp lifespan to the app + _add_lifespan_to_app(app, whatsapp_lifespan) def _setup_daily_routes(app: FastAPI): @@ -654,4 +687,4 @@ def main(): if __name__ == "__main__": - main() + main() \ No newline at end of file From 4d1915eb41ba1f0453a3c18e50404dbca2a55703 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 26 Sep 2025 10:49:52 -0300 Subject: [PATCH 3/3] Fixing ruff format. --- src/pipecat/runner/run.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 80a3c5771..edcfbe9c4 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -235,21 +235,21 @@ def _setup_webrtc_routes(app: FastAPI, esp32_mode: bool = False, host: str = "lo def _add_lifespan_to_app(app: FastAPI, new_lifespan): """Add a new lifespan context manager to the app, combining with existing if present. - + Args: app: The FastAPI application instance new_lifespan: The new lifespan context manager to add """ - if hasattr(app.router, 'lifespan_context') and app.router.lifespan_context is not None: + if hasattr(app.router, "lifespan_context") and app.router.lifespan_context is not None: # If there's already a lifespan context, combine them existing_lifespan = app.router.lifespan_context - + @asynccontextmanager async def combined_lifespan(app: FastAPI): async with existing_lifespan(app): async with new_lifespan(app): yield - + app.router.lifespan_context = combined_lifespan else: # No existing lifespan, use the new one @@ -306,7 +306,7 @@ def _setup_whatsapp_routes(app: FastAPI): if whatsapp_client is None: logger.error("WhatsApp client is not initialized") raise HTTPException(status_code=503, detail="Service unavailable") - + params = dict(request.query_params) logger.debug(f"Webhook verification request received with params: {list(params.keys())}") @@ -334,7 +334,7 @@ def _setup_whatsapp_routes(app: FastAPI): if whatsapp_client is None: logger.error("WhatsApp client is not initialized") raise HTTPException(status_code=503, detail="Service unavailable") - + # Validate webhook object type if body.object != "whatsapp_business_account": logger.warning(f"Invalid webhook object type: {body.object}") @@ -687,4 +687,4 @@ def main(): if __name__ == "__main__": - main() \ No newline at end of file + main()