""" Pipecat Flow Handlers for RTVI Integration This module provides handlers for integrating Pipecat flows with RTVI punt system, allowing client-side code execution while maintaining server-side flow management. """ from datetime import datetime from typing import Callable, Optional from pipecat.processors.frameworks.rtvi import RTVIProcessor, RTVIServerMessageFrame from pipecat_flows import FlowConfig, FlowManager, FlowsFunctionSchema from loguru import logger # Define the set of action types that should get no-op handlers NO_OP_ACTION_TYPES = { "save_summary_conversation", "node_summary_conversation", "save_html_conversation", } def create_stub_handler( node_name: str, function_name: str, rtvi_client: RTVIProcessor, transition_to: Optional[str] = None, ) -> Callable: """ Create a stub handler that forwards function calls to client via RTVI. This enables "RTVI stubs" - when server-side flow encounters a client function, it sends the request to the client for execution while maintaining flow state. Args: node_name: Name of the node that contains the function function_name: Name of the function to punt to client rtvi_client: RTVI client for sending messages transition_to: Optional node to transition to after function execution Returns: Async handler function for use with FlowManager """ async def stub_handler(params, flow_manager: FlowManager): """Fire-and-forget stub handler""" try: # Create function request message message = { "type": "function_request", "function_name": function_name, "params": [params], "timestamp": datetime.now().isoformat(), } # Send to client via RTVI server message frame frame = RTVIServerMessageFrame(data=message) await flow_manager.task.queue_frame(frame) logger.info(f"🚀 Punted function '{function_name}' to client") if transition_to: state_message = { "type": "flow_state_sync", "current_node": transition_to, "previous_node": node_name, "timestamp": datetime.now().isoformat(), } frame = RTVIServerMessageFrame(data=state_message) await flow_manager.task.queue_frame(frame) logger.info(f"🔄 State sync: {node_name} → {transition_to}") return {"status": "success"} except Exception as e: logger.error(f"❌ Failed to punt function '{function_name}': {e}") return {"status": "error", "error": str(e)} return stub_handler async def send_state_sync(flow_manager: FlowManager, node_name: str): state_message = { "type": "flow_state_sync", "current_node": node_name, "timestamp": datetime.now().isoformat(), } frame = RTVIServerMessageFrame(data=state_message) await flow_manager.task.queue_frame(frame) logger.info(f"🔄 State sync: {node_name}") def create_no_op_action_handler() -> Callable: """ Create a no-op handler for actions that should be ignored on server-side. Returns: Async handler function that does nothing """ async def no_op_handler(action: dict, flow_manager: FlowManager): """No-op handler for client-side only actions""" logger.debug( f"No-op handler called for action type: {action.get('type', 'unknown')}" ) pass return no_op_handler async def end_conversation_handler(action: dict, flow_manager: FlowManager): """Handler for end_conversation action - works for both Daily and Twilio""" try: # Import here to avoid circular imports from utils.twilio import terminate_call, get_call_context # Determine call source if not action or action == {}: call_source = "function_call" else: call_source = "post_action" logger.info(f"🏁 End conversation called with source {call_source}") # Create end_conversation message with source information message = { "type": "end_conversation", "timestamp": datetime.now().isoformat(), "call_source": call_source, "immediate_disconnect": call_source == "function_call", } # Try to send RTVI message (works for Daily/WebRTC clients) try: frame = RTVIServerMessageFrame(data=message) await flow_manager.task.queue_frame(frame) logger.info("🏁 Sent end_conversation message to client") except Exception as rtvi_error: logger.debug(f"RTVI message not sent (expected for Twilio): {rtvi_error}") # For Twilio calls, terminate via REST API call_context = get_call_context() if call_context and call_context.get("call_sid"): call_sid = call_context["call_sid"] logger.info(f"🏁 Terminating Twilio call via REST API: {call_sid}") await terminate_call(call_sid) # Cancel the task to end the call (works for both Daily and Twilio) # This ensures the call is terminated regardless of transport type if call_source == "function_call": logger.info("🏁 Cancelling task to end call") await flow_manager.task.cancel() except Exception as e: logger.error(f"❌ Failed to end conversation: {e}") def process_flow_config_handlers( llm_flow: FlowConfig, rtvi_client: RTVIProcessor, ) -> FlowConfig: """ Process flow configuration and replace string handlers with RTVI punt handlers. This is the core of the RTVI punt system - it finds string handlers like "__function__:handleTransition" and replaces them with lambda functions that send function requests to the client. Args: llm_flow: LLM flow configuration rtvi_client: RTVI client for sending messages Returns: Processed flow config with punt handlers """ processed_config = llm_flow.copy() # Get the handlers once no_op_handler = create_no_op_action_handler() # Process each node for node_name, node_config in processed_config.get("nodes", {}).items(): # Process functions functions = node_config.get("functions", []) # Inject end_conversation function at every node end_conversation_function = FlowsFunctionSchema( name="end_conversation", description="Call this function when the user explicitly asks to stop, end, or terminate the current interview or conversation. This should be called when user clearly expresses they want to finish the interview session.", properties={}, required=[], handler=end_conversation_handler, ) functions.append(end_conversation_function) logger.info(f"💬 Injected end_conversation function in node: {node_name}") for i, func_outer in enumerate(functions): # Handle both dict format and FlowsFunctionSchema if isinstance(func_outer, FlowsFunctionSchema): func = func_outer func_name = func.name else: func = func_outer["function"] func_name = func["name"] # Skip processing if this is the end_conversation function we just added if func_name == "end_conversation": continue # Convert to FlowsFunctionSchema if it's not already if isinstance(func_outer, dict): func_data = func_outer["function"] name = func_data["name"] description = func_data.get("description", "") parameters = func_data.get("parameters", {}) or {} properties = parameters.get("properties", {}) required = parameters.get("required", []) # Extract Flows-specific fields transition_to = func_data.get("transition_to") # Create RTVI stub handler handler = create_stub_handler( node_name, func_name, rtvi_client, transition_to=transition_to ) logger.info(f"📤 Converted {func_name} to RTVI stub") new_func = FlowsFunctionSchema( name=name, description=description, properties=properties, required=required, handler=handler, transition_to=transition_to, ) functions[i] = new_func # Process pre_actions pre_actions = node_config.get("pre_actions", []) for action in pre_actions: action_type = action.get("type") if action_type in NO_OP_ACTION_TYPES: action["handler"] = no_op_handler logger.info( f"Added no-op handler to pre_action: {action_type} in node: {node_name}" ) # Process post_actions post_actions = node_config.get("post_actions", []) for action in post_actions: action_type = action.get("type") if action_type in NO_OP_ACTION_TYPES: action["handler"] = no_op_handler logger.info( f"Added no-op handler to post_action: {action_type} in node: {node_name}" ) logger.info("✅ Flow config handlers processed for RTVI stubs and actions") return processed_config