#!/usr/bin/env python3 """ AGILEADAPT AUTOMATION WORKFLOW BUILDER ====================================== Build and execute automation workflows for AgileAdapt clients. Features: - Missed call text-back workflow - Quote follow-up sequence (5 touches) - Review request automation - Appointment reminder sequence - Client-customizable workflows Storage: PostgreSQL (Elestio Core) VERIFICATION_STAMP Story: STORY-4.2 Verified By: CLAUDE Verified At: 2026-01-24 """ import sys import json import logging import re from abc import ABC, abstractmethod from dataclasses import dataclass, field, asdict from datetime import datetime, timedelta from enum import Enum from typing import Any, Callable, Dict, List, Optional, Tuple from uuid import uuid4 # Add genesis-memory for Elestio config sys.path.insert(0, '/mnt/e/genesis-system/data/genesis-memory') try: from elestio_config import PostgresConfig import psycopg2 import psycopg2.extras HAS_POSTGRES = True except ImportError: HAS_POSTGRES = False PostgresConfig = None # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # ============================================================================= # ENUMS AND CONSTANTS # ============================================================================= class TriggerType(Enum): """Types of workflow triggers.""" MISSED_CALL = "missed_call" QUOTE_SENT = "quote_sent" JOB_COMPLETE = "job_complete" APPOINTMENT_SCHEDULED = "appointment_scheduled" MANUAL = "manual" WEBHOOK = "webhook" SCHEDULED = "scheduled" REPLY_RECEIVED = "reply_received" NO_REPLY = "no_reply" class ActionType(Enum): """Types of workflow actions.""" SEND_SMS = "send_sms" SEND_EMAIL = "send_email" WAIT = "wait" CONDITION = "condition" NOTIFY_OWNER = "notify_owner" ROUTE_TO_AI = "route_to_ai" UPDATE_CRM = "update_crm" WEBHOOK = "webhook" END_WORKFLOW = "end_workflow" class WorkflowStatus(Enum): """Status of a workflow execution.""" PENDING = "pending" RUNNING = "running" WAITING = "waiting" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" class MessageChannel(Enum): """Message delivery channels.""" SMS = "sms" EMAIL = "email" VOICE = "voice" WEBHOOK = "webhook" # ============================================================================= # DATA CLASSES # ============================================================================= @dataclass class WorkflowCondition: """A condition that controls workflow branching.""" field: str operator: str # eq, ne, gt, lt, contains, is_empty value: Any def evaluate(self, context: Dict[str, Any]) -> bool: """Evaluate the condition against a context.""" actual_value = context.get(self.field) if self.operator == "eq": return actual_value == self.value elif self.operator == "ne": return actual_value != self.value elif self.operator == "gt": return actual_value is not None and actual_value > self.value elif self.operator == "lt": return actual_value is not None and actual_value < self.value elif self.operator == "contains": return self.value in str(actual_value or "") elif self.operator == "is_empty": return not actual_value elif self.operator == "is_not_empty": return bool(actual_value) return False @dataclass class WorkflowAction: """A single action in a workflow.""" id: str action_type: ActionType config: Dict[str, Any] = field(default_factory=dict) conditions: List[WorkflowCondition] = field(default_factory=list) next_action_id: Optional[str] = None on_failure_action_id: Optional[str] = None def to_dict(self) -> Dict[str, Any]: """Convert action to dictionary.""" return { "id": self.id, "action_type": self.action_type.value, "config": self.config, "conditions": [{"field": c.field, "operator": c.operator, "value": c.value} for c in self.conditions], "next_action_id": self.next_action_id, "on_failure_action_id": self.on_failure_action_id } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'WorkflowAction': """Create action from dictionary.""" conditions = [WorkflowCondition(**c) for c in data.get("conditions", [])] return cls( id=data["id"], action_type=ActionType(data["action_type"]), config=data.get("config", {}), conditions=conditions, next_action_id=data.get("next_action_id"), on_failure_action_id=data.get("on_failure_action_id") ) @dataclass class WorkflowTrigger: """A trigger that starts a workflow.""" trigger_type: TriggerType config: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: return { "trigger_type": self.trigger_type.value, "config": self.config } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'WorkflowTrigger': return cls( trigger_type=TriggerType(data["trigger_type"]), config=data.get("config", {}) ) @dataclass class WorkflowTemplate: """A workflow template that can be instantiated.""" id: str name: str description: str trigger: WorkflowTrigger actions: List[WorkflowAction] variables: Dict[str, str] = field(default_factory=dict) # Default variable values client_customizable: bool = True is_active: bool = True created_at: datetime = field(default_factory=datetime.utcnow) updated_at: datetime = field(default_factory=datetime.utcnow) def to_dict(self) -> Dict[str, Any]: return { "id": self.id, "name": self.name, "description": self.description, "trigger": self.trigger.to_dict(), "actions": [a.to_dict() for a in self.actions], "variables": self.variables, "client_customizable": self.client_customizable, "is_active": self.is_active, "created_at": self.created_at.isoformat(), "updated_at": self.updated_at.isoformat() } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'WorkflowTemplate': trigger = WorkflowTrigger.from_dict(data["trigger"]) actions = [WorkflowAction.from_dict(a) for a in data["actions"]] return cls( id=data["id"], name=data["name"], description=data["description"], trigger=trigger, actions=actions, variables=data.get("variables", {}), client_customizable=data.get("client_customizable", True), is_active=data.get("is_active", True), created_at=datetime.fromisoformat(data["created_at"]) if "created_at" in data else datetime.utcnow(), updated_at=datetime.fromisoformat(data["updated_at"]) if "updated_at" in data else datetime.utcnow() ) @dataclass class WorkflowInstance: """An instance of a running workflow.""" id: str template_id: str client_id: str contact_id: str status: WorkflowStatus current_action_id: Optional[str] context: Dict[str, Any] # Variables and state history: List[Dict[str, Any]] = field(default_factory=list) scheduled_at: Optional[datetime] = None started_at: Optional[datetime] = None completed_at: Optional[datetime] = None error_message: Optional[str] = None def to_dict(self) -> Dict[str, Any]: return { "id": self.id, "template_id": self.template_id, "client_id": self.client_id, "contact_id": self.contact_id, "status": self.status.value, "current_action_id": self.current_action_id, "context": self.context, "history": self.history, "scheduled_at": self.scheduled_at.isoformat() if self.scheduled_at else None, "started_at": self.started_at.isoformat() if self.started_at else None, "completed_at": self.completed_at.isoformat() if self.completed_at else None, "error_message": self.error_message } @dataclass class MessageDeliveryResult: """Result of a message delivery attempt.""" success: bool channel: MessageChannel recipient: str message_id: Optional[str] = None error: Optional[str] = None timestamp: datetime = field(default_factory=datetime.utcnow) # ============================================================================= # TEMPLATE VARIABLE PROCESSOR # ============================================================================= class TemplateProcessor: """Processes template variables like {{first_name}}.""" VARIABLE_PATTERN = re.compile(r'\{\{(\w+)\}\}') @classmethod def process(cls, template: str, context: Dict[str, Any]) -> str: """Replace template variables with values from context.""" def replace_var(match): var_name = match.group(1) value = context.get(var_name, match.group(0)) # Keep original if not found return str(value) if value is not None else "" return cls.VARIABLE_PATTERN.sub(replace_var, template) @classmethod def extract_variables(cls, template: str) -> List[str]: """Extract all variable names from a template.""" return cls.VARIABLE_PATTERN.findall(template) @classmethod def validate_context(cls, template: str, context: Dict[str, Any]) -> Tuple[bool, List[str]]: """Validate that context contains all required variables.""" required = cls.extract_variables(template) missing = [v for v in required if v not in context] return len(missing) == 0, missing # ============================================================================= # MESSAGE HANDLERS (Abstract interfaces for SMS/Email providers) # ============================================================================= class MessageHandler(ABC): """Abstract base class for message handlers.""" @abstractmethod def send(self, recipient: str, message: str, **kwargs) -> MessageDeliveryResult: """Send a message to a recipient.""" pass class SMSHandler(MessageHandler): """SMS message handler (placeholder for Telnyx/Twilio integration).""" def __init__(self, config: Optional[Dict[str, Any]] = None): self.config = config or {} def send(self, recipient: str, message: str, **kwargs) -> MessageDeliveryResult: """Send SMS message.""" # In production, integrate with Telnyx API logger.info(f"[SMS] To: {recipient} | Message: {message[:50]}...") # Placeholder for actual implementation return MessageDeliveryResult( success=True, channel=MessageChannel.SMS, recipient=recipient, message_id=f"sms_{uuid4().hex[:8]}" ) class EmailHandler(MessageHandler): """Email message handler (placeholder for email integration).""" def __init__(self, config: Optional[Dict[str, Any]] = None): self.config = config or {} def send(self, recipient: str, message: str, **kwargs) -> MessageDeliveryResult: """Send email message.""" subject = kwargs.get("subject", "Notification") logger.info(f"[EMAIL] To: {recipient} | Subject: {subject}") # Placeholder for actual implementation return MessageDeliveryResult( success=True, channel=MessageChannel.EMAIL, recipient=recipient, message_id=f"email_{uuid4().hex[:8]}" ) # ============================================================================= # WORKFLOW STORAGE (PostgreSQL) # ============================================================================= class WorkflowStorage: """PostgreSQL storage for workflows.""" def __init__(self): self.conn = None self._ensure_tables() def _get_connection(self): """Get database connection.""" if not HAS_POSTGRES: raise RuntimeError("PostgreSQL not available. Install psycopg2.") if self.conn is None or self.conn.closed: self.conn = psycopg2.connect(**PostgresConfig.get_connection_params()) return self.conn def _ensure_tables(self): """Create tables if they don't exist.""" if not HAS_POSTGRES: logger.warning("PostgreSQL not available, using in-memory storage") self._memory_templates = {} self._memory_instances = {} self._memory_client_configs = {} return try: conn = self._get_connection() cursor = conn.cursor() # Workflow templates table cursor.execute(""" CREATE TABLE IF NOT EXISTS workflow_templates ( id VARCHAR(64) PRIMARY KEY, name VARCHAR(255) NOT NULL, description TEXT, trigger_config JSONB NOT NULL, actions JSONB NOT NULL, variables JSONB DEFAULT '{}', client_customizable BOOLEAN DEFAULT TRUE, is_active BOOLEAN DEFAULT TRUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Workflow instances table cursor.execute(""" CREATE TABLE IF NOT EXISTS workflow_instances ( id VARCHAR(64) PRIMARY KEY, template_id VARCHAR(64) REFERENCES workflow_templates(id), client_id VARCHAR(64) NOT NULL, contact_id VARCHAR(64) NOT NULL, status VARCHAR(32) NOT NULL, current_action_id VARCHAR(64), context JSONB DEFAULT '{}', history JSONB DEFAULT '[]', scheduled_at TIMESTAMP, started_at TIMESTAMP, completed_at TIMESTAMP, error_message TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Client workflow configurations (customizations) cursor.execute(""" CREATE TABLE IF NOT EXISTS client_workflow_configs ( id VARCHAR(64) PRIMARY KEY, client_id VARCHAR(64) NOT NULL, template_id VARCHAR(64) REFERENCES workflow_templates(id), custom_variables JSONB DEFAULT '{}', custom_actions JSONB, is_enabled BOOLEAN DEFAULT TRUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(client_id, template_id) ) """) # Index for faster lookups cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_workflow_instances_client ON workflow_instances(client_id, status) """) cursor.execute(""" CREATE INDEX IF NOT EXISTS idx_workflow_instances_scheduled ON workflow_instances(scheduled_at) WHERE status = 'waiting' """) conn.commit() cursor.close() logger.info("Workflow tables initialized") except Exception as e: logger.error(f"Failed to create workflow tables: {e}") # Fallback to in-memory storage self._memory_templates = {} self._memory_instances = {} self._memory_client_configs = {} def save_template(self, template: WorkflowTemplate) -> bool: """Save a workflow template.""" if not HAS_POSTGRES: self._memory_templates[template.id] = template return True try: conn = self._get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO workflow_templates (id, name, description, trigger_config, actions, variables, client_customizable, is_active, created_at, updated_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, description = EXCLUDED.description, trigger_config = EXCLUDED.trigger_config, actions = EXCLUDED.actions, variables = EXCLUDED.variables, client_customizable = EXCLUDED.client_customizable, is_active = EXCLUDED.is_active, updated_at = CURRENT_TIMESTAMP """, ( template.id, template.name, template.description, json.dumps(template.trigger.to_dict()), json.dumps([a.to_dict() for a in template.actions]), json.dumps(template.variables), template.client_customizable, template.is_active, template.created_at, template.updated_at )) conn.commit() cursor.close() return True except Exception as e: logger.error(f"Failed to save template: {e}") return False def get_template(self, template_id: str) -> Optional[WorkflowTemplate]: """Get a workflow template by ID.""" if not HAS_POSTGRES: return self._memory_templates.get(template_id) try: conn = self._get_connection() cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cursor.execute(""" SELECT * FROM workflow_templates WHERE id = %s """, (template_id,)) row = cursor.fetchone() cursor.close() if not row: return None return WorkflowTemplate( id=row['id'], name=row['name'], description=row['description'], trigger=WorkflowTrigger.from_dict(row['trigger_config']), actions=[WorkflowAction.from_dict(a) for a in row['actions']], variables=row['variables'] or {}, client_customizable=row['client_customizable'], is_active=row['is_active'], created_at=row['created_at'], updated_at=row['updated_at'] ) except Exception as e: logger.error(f"Failed to get template: {e}") return None def list_templates(self, active_only: bool = True) -> List[WorkflowTemplate]: """List all workflow templates.""" if not HAS_POSTGRES: templates = list(self._memory_templates.values()) if active_only: templates = [t for t in templates if t.is_active] return templates try: conn = self._get_connection() cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) if active_only: cursor.execute("SELECT * FROM workflow_templates WHERE is_active = TRUE") else: cursor.execute("SELECT * FROM workflow_templates") rows = cursor.fetchall() cursor.close() templates = [] for row in rows: templates.append(WorkflowTemplate( id=row['id'], name=row['name'], description=row['description'], trigger=WorkflowTrigger.from_dict(row['trigger_config']), actions=[WorkflowAction.from_dict(a) for a in row['actions']], variables=row['variables'] or {}, client_customizable=row['client_customizable'], is_active=row['is_active'], created_at=row['created_at'], updated_at=row['updated_at'] )) return templates except Exception as e: logger.error(f"Failed to list templates: {e}") return [] def save_instance(self, instance: WorkflowInstance) -> bool: """Save a workflow instance.""" if not HAS_POSTGRES: self._memory_instances[instance.id] = instance return True try: conn = self._get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO workflow_instances (id, template_id, client_id, contact_id, status, current_action_id, context, history, scheduled_at, started_at, completed_at, error_message) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, current_action_id = EXCLUDED.current_action_id, context = EXCLUDED.context, history = EXCLUDED.history, scheduled_at = EXCLUDED.scheduled_at, started_at = EXCLUDED.started_at, completed_at = EXCLUDED.completed_at, error_message = EXCLUDED.error_message """, ( instance.id, instance.template_id, instance.client_id, instance.contact_id, instance.status.value, instance.current_action_id, json.dumps(instance.context), json.dumps(instance.history), instance.scheduled_at, instance.started_at, instance.completed_at, instance.error_message )) conn.commit() cursor.close() return True except Exception as e: logger.error(f"Failed to save instance: {e}") return False def get_instance(self, instance_id: str) -> Optional[WorkflowInstance]: """Get a workflow instance by ID.""" if not HAS_POSTGRES: return self._memory_instances.get(instance_id) try: conn = self._get_connection() cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cursor.execute("SELECT * FROM workflow_instances WHERE id = %s", (instance_id,)) row = cursor.fetchone() cursor.close() if not row: return None return WorkflowInstance( id=row['id'], template_id=row['template_id'], client_id=row['client_id'], contact_id=row['contact_id'], status=WorkflowStatus(row['status']), current_action_id=row['current_action_id'], context=row['context'] or {}, history=row['history'] or [], scheduled_at=row['scheduled_at'], started_at=row['started_at'], completed_at=row['completed_at'], error_message=row['error_message'] ) except Exception as e: logger.error(f"Failed to get instance: {e}") return None def get_waiting_instances(self, before: datetime = None) -> List[WorkflowInstance]: """Get all waiting workflow instances ready for execution.""" if before is None: before = datetime.utcnow() if not HAS_POSTGRES: return [ i for i in self._memory_instances.values() if i.status == WorkflowStatus.WAITING and i.scheduled_at and i.scheduled_at <= before ] try: conn = self._get_connection() cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cursor.execute(""" SELECT * FROM workflow_instances WHERE status = 'waiting' AND scheduled_at <= %s ORDER BY scheduled_at ASC """, (before,)) rows = cursor.fetchall() cursor.close() instances = [] for row in rows: instances.append(WorkflowInstance( id=row['id'], template_id=row['template_id'], client_id=row['client_id'], contact_id=row['contact_id'], status=WorkflowStatus(row['status']), current_action_id=row['current_action_id'], context=row['context'] or {}, history=row['history'] or [], scheduled_at=row['scheduled_at'], started_at=row['started_at'], completed_at=row['completed_at'], error_message=row['error_message'] )) return instances except Exception as e: logger.error(f"Failed to get waiting instances: {e}") return [] def get_client_config(self, client_id: str, template_id: str) -> Optional[Dict[str, Any]]: """Get client's custom workflow configuration.""" if not HAS_POSTGRES: key = f"{client_id}:{template_id}" return self._memory_client_configs.get(key) try: conn = self._get_connection() cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cursor.execute(""" SELECT * FROM client_workflow_configs WHERE client_id = %s AND template_id = %s """, (client_id, template_id)) row = cursor.fetchone() cursor.close() if not row: return None return { "custom_variables": row['custom_variables'], "custom_actions": row['custom_actions'], "is_enabled": row['is_enabled'] } except Exception as e: logger.error(f"Failed to get client config: {e}") return None def save_client_config(self, client_id: str, template_id: str, custom_variables: Dict[str, Any], is_enabled: bool = True) -> bool: """Save client's custom workflow configuration.""" if not HAS_POSTGRES: key = f"{client_id}:{template_id}" self._memory_client_configs[key] = { "custom_variables": custom_variables, "is_enabled": is_enabled } return True try: conn = self._get_connection() cursor = conn.cursor() cursor.execute(""" INSERT INTO client_workflow_configs (id, client_id, template_id, custom_variables, is_enabled) VALUES (%s, %s, %s, %s, %s) ON CONFLICT (client_id, template_id) DO UPDATE SET custom_variables = EXCLUDED.custom_variables, is_enabled = EXCLUDED.is_enabled, updated_at = CURRENT_TIMESTAMP """, ( f"config_{uuid4().hex[:8]}", client_id, template_id, json.dumps(custom_variables), is_enabled )) conn.commit() cursor.close() return True except Exception as e: logger.error(f"Failed to save client config: {e}") return False # ============================================================================= # WORKFLOW ENGINE # ============================================================================= class WorkflowEngine: """ Core workflow execution engine. Handles: - Workflow instantiation from templates - Action execution - State management - Scheduling """ def __init__(self, storage: Optional[WorkflowStorage] = None): self.storage = storage or WorkflowStorage() self.sms_handler = SMSHandler() self.email_handler = EmailHandler() self._action_handlers: Dict[ActionType, Callable] = { ActionType.SEND_SMS: self._handle_send_sms, ActionType.SEND_EMAIL: self._handle_send_email, ActionType.WAIT: self._handle_wait, ActionType.CONDITION: self._handle_condition, ActionType.NOTIFY_OWNER: self._handle_notify_owner, ActionType.ROUTE_TO_AI: self._handle_route_to_ai, ActionType.UPDATE_CRM: self._handle_update_crm, ActionType.END_WORKFLOW: self._handle_end_workflow, } def create_instance( self, template_id: str, client_id: str, contact_id: str, context: Dict[str, Any] ) -> Optional[WorkflowInstance]: """Create a new workflow instance from a template.""" template = self.storage.get_template(template_id) if not template: logger.error(f"Template not found: {template_id}") return None # Get client customizations client_config = self.storage.get_client_config(client_id, template_id) # Merge default variables with client customizations merged_context = {**template.variables} if client_config and client_config.get("custom_variables"): merged_context.update(client_config["custom_variables"]) merged_context.update(context) # Get first action first_action_id = template.actions[0].id if template.actions else None instance = WorkflowInstance( id=f"wf_{uuid4().hex[:12]}", template_id=template_id, client_id=client_id, contact_id=contact_id, status=WorkflowStatus.PENDING, current_action_id=first_action_id, context=merged_context, history=[] ) self.storage.save_instance(instance) logger.info(f"Created workflow instance: {instance.id}") return instance def start_workflow(self, instance_id: str) -> bool: """Start executing a workflow instance.""" instance = self.storage.get_instance(instance_id) if not instance: logger.error(f"Instance not found: {instance_id}") return False if instance.status != WorkflowStatus.PENDING: logger.warning(f"Instance {instance_id} is not in pending state") return False instance.status = WorkflowStatus.RUNNING instance.started_at = datetime.utcnow() self.storage.save_instance(instance) return self._execute_next_action(instance) def resume_workflow(self, instance_id: str) -> bool: """Resume a waiting workflow instance.""" instance = self.storage.get_instance(instance_id) if not instance: return False if instance.status != WorkflowStatus.WAITING: return False instance.status = WorkflowStatus.RUNNING self.storage.save_instance(instance) return self._execute_next_action(instance) def handle_reply(self, instance_id: str, reply_content: str) -> bool: """Handle a reply received during a workflow.""" instance = self.storage.get_instance(instance_id) if not instance: return False # Add reply to context instance.context["last_reply"] = reply_content instance.context["reply_received"] = True instance.context["reply_timestamp"] = datetime.utcnow().isoformat() # Record in history instance.history.append({ "event": "reply_received", "content": reply_content, "timestamp": datetime.utcnow().isoformat() }) self.storage.save_instance(instance) # Resume if waiting for reply if instance.status == WorkflowStatus.WAITING: return self.resume_workflow(instance_id) return True def cancel_workflow(self, instance_id: str, reason: str = None) -> bool: """Cancel a workflow instance.""" instance = self.storage.get_instance(instance_id) if not instance: return False instance.status = WorkflowStatus.CANCELLED instance.completed_at = datetime.utcnow() instance.error_message = reason instance.history.append({ "event": "cancelled", "reason": reason, "timestamp": datetime.utcnow().isoformat() }) return self.storage.save_instance(instance) def _execute_next_action(self, instance: WorkflowInstance) -> bool: """Execute the next action in the workflow.""" if not instance.current_action_id: # No more actions, complete the workflow return self._complete_workflow(instance) template = self.storage.get_template(instance.template_id) if not template: return self._fail_workflow(instance, "Template not found") # Find current action action = None for a in template.actions: if a.id == instance.current_action_id: action = a break if not action: return self._fail_workflow(instance, f"Action not found: {instance.current_action_id}") # Check conditions if action.conditions: all_conditions_met = all(c.evaluate(instance.context) for c in action.conditions) if not all_conditions_met: # Skip to next action instance.current_action_id = action.next_action_id self.storage.save_instance(instance) return self._execute_next_action(instance) # Execute the action handler = self._action_handlers.get(action.action_type) if not handler: return self._fail_workflow(instance, f"No handler for action type: {action.action_type}") try: result = handler(instance, action) # Record in history instance.history.append({ "action_id": action.id, "action_type": action.action_type.value, "result": result, "timestamp": datetime.utcnow().isoformat() }) # Handle result if result.get("status") == "waiting": instance.status = WorkflowStatus.WAITING instance.scheduled_at = result.get("resume_at") self.storage.save_instance(instance) return True elif result.get("status") == "failed": if action.on_failure_action_id: instance.current_action_id = action.on_failure_action_id self.storage.save_instance(instance) return self._execute_next_action(instance) else: return self._fail_workflow(instance, result.get("error")) else: # Success, move to next action instance.current_action_id = action.next_action_id self.storage.save_instance(instance) if action.next_action_id: return self._execute_next_action(instance) else: return self._complete_workflow(instance) except Exception as e: logger.exception(f"Error executing action {action.id}") return self._fail_workflow(instance, str(e)) def _complete_workflow(self, instance: WorkflowInstance) -> bool: """Mark workflow as completed.""" instance.status = WorkflowStatus.COMPLETED instance.completed_at = datetime.utcnow() instance.current_action_id = None instance.history.append({ "event": "completed", "timestamp": datetime.utcnow().isoformat() }) return self.storage.save_instance(instance) def _fail_workflow(self, instance: WorkflowInstance, error: str) -> bool: """Mark workflow as failed.""" instance.status = WorkflowStatus.FAILED instance.completed_at = datetime.utcnow() instance.error_message = error instance.history.append({ "event": "failed", "error": error, "timestamp": datetime.utcnow().isoformat() }) return self.storage.save_instance(instance) # Action Handlers def _handle_send_sms(self, instance: WorkflowInstance, action: WorkflowAction) -> Dict[str, Any]: """Handle SMS sending action.""" message_template = action.config.get("message", "") recipient = instance.context.get("phone", instance.context.get("contact_phone")) if not recipient: return {"status": "failed", "error": "No phone number in context"} # Process template variables message = TemplateProcessor.process(message_template, instance.context) # Send the SMS result = self.sms_handler.send(recipient, message) if result.success: return {"status": "success", "message_id": result.message_id} else: return {"status": "failed", "error": result.error} def _handle_send_email(self, instance: WorkflowInstance, action: WorkflowAction) -> Dict[str, Any]: """Handle email sending action.""" subject_template = action.config.get("subject", "Notification") body_template = action.config.get("body", "") recipient = instance.context.get("email", instance.context.get("contact_email")) if not recipient: return {"status": "failed", "error": "No email in context"} # Process template variables subject = TemplateProcessor.process(subject_template, instance.context) body = TemplateProcessor.process(body_template, instance.context) # Send the email result = self.email_handler.send(recipient, body, subject=subject) if result.success: return {"status": "success", "message_id": result.message_id} else: return {"status": "failed", "error": result.error} def _handle_wait(self, instance: WorkflowInstance, action: WorkflowAction) -> Dict[str, Any]: """Handle wait action.""" wait_hours = action.config.get("hours", 0) wait_minutes = action.config.get("minutes", 0) wait_days = action.config.get("days", 0) wait_delta = timedelta( days=wait_days, hours=wait_hours, minutes=wait_minutes ) resume_at = datetime.utcnow() + wait_delta # Update context for next action instance.current_action_id = action.next_action_id return { "status": "waiting", "resume_at": resume_at, "wait_duration": str(wait_delta) } def _handle_condition(self, instance: WorkflowInstance, action: WorkflowAction) -> Dict[str, Any]: """Handle conditional branching action.""" conditions = action.config.get("conditions", []) true_action = action.config.get("true_action_id") false_action = action.config.get("false_action_id") # Evaluate all conditions all_met = True for cond in conditions: condition = WorkflowCondition(**cond) if not condition.evaluate(instance.context): all_met = False break # Set next action based on condition result if all_met: instance.current_action_id = true_action else: instance.current_action_id = false_action or action.next_action_id return {"status": "success", "condition_result": all_met} def _handle_notify_owner(self, instance: WorkflowInstance, action: WorkflowAction) -> Dict[str, Any]: """Handle owner notification action.""" notification_type = action.config.get("type", "sms") message_template = action.config.get("message", "New lead: {{first_name}} {{last_name}}") # Get owner contact from context or config owner_phone = instance.context.get("owner_phone", action.config.get("owner_phone")) owner_email = instance.context.get("owner_email", action.config.get("owner_email")) message = TemplateProcessor.process(message_template, instance.context) if notification_type == "sms" and owner_phone: result = self.sms_handler.send(owner_phone, message) elif notification_type == "email" and owner_email: result = self.email_handler.send(owner_email, message, subject="Lead Notification") else: return {"status": "failed", "error": "No owner contact for notification"} return {"status": "success" if result.success else "failed", "error": result.error} def _handle_route_to_ai(self, instance: WorkflowInstance, action: WorkflowAction) -> Dict[str, Any]: """Handle routing to AI assistant.""" # This would integrate with the AI voice/chat system logger.info(f"Routing contact {instance.contact_id} to AI assistant") instance.context["routed_to_ai"] = True instance.context["ai_routing_timestamp"] = datetime.utcnow().isoformat() return {"status": "success", "routed_to": "ai_assistant"} def _handle_update_crm(self, instance: WorkflowInstance, action: WorkflowAction) -> Dict[str, Any]: """Handle CRM update action.""" updates = action.config.get("updates", {}) # This would integrate with GHL or other CRM logger.info(f"CRM update for contact {instance.contact_id}: {updates}") return {"status": "success", "updates_applied": updates} def _handle_end_workflow(self, instance: WorkflowInstance, action: WorkflowAction) -> Dict[str, Any]: """Handle explicit workflow end.""" return {"status": "end"} def process_waiting_workflows(self) -> int: """Process all waiting workflows that are ready to resume.""" waiting = self.storage.get_waiting_instances() processed = 0 for instance in waiting: logger.info(f"Resuming workflow: {instance.id}") if self.resume_workflow(instance.id): processed += 1 return processed # ============================================================================= # PRE-BUILT WORKFLOW TEMPLATES # ============================================================================= class WorkflowTemplates: """Factory for pre-built workflow templates.""" @staticmethod def missed_call_textback() -> WorkflowTemplate: """ MISSED CALL TEXT-BACK WORKFLOW Trigger: Missed call detected Actions: 1. Send SMS "Hi {{first_name}}, sorry we missed your call. How can we help?" 2. If reply: Route to AI or notify owner """ return WorkflowTemplate( id="wft_missed_call_textback", name="Missed Call Text-Back", description="Automatically text back when a call is missed", trigger=WorkflowTrigger( trigger_type=TriggerType.MISSED_CALL, config={} ), actions=[ WorkflowAction( id="act_1_send_sms", action_type=ActionType.SEND_SMS, config={ "message": "Hi {{first_name}}, sorry we missed your call! How can we help you today? - {{business_name}}" }, next_action_id="act_2_wait_reply" ), WorkflowAction( id="act_2_wait_reply", action_type=ActionType.WAIT, config={ "minutes": 30 # Wait 30 minutes for reply }, next_action_id="act_3_check_reply" ), WorkflowAction( id="act_3_check_reply", action_type=ActionType.CONDITION, config={ "conditions": [{"field": "reply_received", "operator": "eq", "value": True}], "true_action_id": "act_4_route_ai", "false_action_id": "act_5_notify_owner" } ), WorkflowAction( id="act_4_route_ai", action_type=ActionType.ROUTE_TO_AI, config={ "ai_agent": "receptionist" }, next_action_id=None # End workflow ), WorkflowAction( id="act_5_notify_owner", action_type=ActionType.NOTIFY_OWNER, config={ "type": "sms", "message": "No reply from {{first_name}} ({{phone}}). Consider following up." }, next_action_id=None # End workflow ) ], variables={ "first_name": "", "phone": "", "business_name": "AgileAdapt Client" } ) @staticmethod def quote_followup_sequence() -> WorkflowTemplate: """ QUOTE FOLLOW-UP SEQUENCE (5 touches) Trigger: Quote sent Actions: - Day 1: "Hi {{first_name}}, just sent your quote. Any questions?" - Day 3: "Following up on your quote. Ready to proceed?" - Day 7: "Your quote expires soon. Let me know if you'd like to move forward." - Day 14: "Final reminder - your quote is expiring. Call us to lock in pricing." - Day 30: "Checking in - still interested in {{service}}?" """ return WorkflowTemplate( id="wft_quote_followup", name="Quote Follow-Up Sequence", description="5-touch follow-up sequence after sending a quote", trigger=WorkflowTrigger( trigger_type=TriggerType.QUOTE_SENT, config={} ), actions=[ # Day 1 WorkflowAction( id="act_day1_sms", action_type=ActionType.SEND_SMS, config={ "message": "Hi {{first_name}}, I just sent your quote for {{service}}. Let me know if you have any questions! - {{business_name}}" }, next_action_id="act_wait_day3" ), WorkflowAction( id="act_wait_day3", action_type=ActionType.WAIT, config={"days": 2}, # Wait until day 3 next_action_id="act_check_response1" ), WorkflowAction( id="act_check_response1", action_type=ActionType.CONDITION, config={ "conditions": [{"field": "quote_accepted", "operator": "eq", "value": True}], "true_action_id": None, # End if accepted "false_action_id": "act_day3_sms" } ), # Day 3 WorkflowAction( id="act_day3_sms", action_type=ActionType.SEND_SMS, config={ "message": "Hi {{first_name}}, following up on your quote. Ready to proceed? Reply YES to confirm or let me know if you have questions." }, next_action_id="act_wait_day7" ), WorkflowAction( id="act_wait_day7", action_type=ActionType.WAIT, config={"days": 4}, # Wait until day 7 next_action_id="act_check_response2" ), WorkflowAction( id="act_check_response2", action_type=ActionType.CONDITION, config={ "conditions": [{"field": "quote_accepted", "operator": "eq", "value": True}], "true_action_id": None, "false_action_id": "act_day7_sms" } ), # Day 7 WorkflowAction( id="act_day7_sms", action_type=ActionType.SEND_SMS, config={ "message": "Hi {{first_name}}, your quote for {{service}} expires soon. Let me know if you'd like to move forward and lock in this pricing!" }, next_action_id="act_wait_day14" ), WorkflowAction( id="act_wait_day14", action_type=ActionType.WAIT, config={"days": 7}, # Wait until day 14 next_action_id="act_check_response3" ), WorkflowAction( id="act_check_response3", action_type=ActionType.CONDITION, config={ "conditions": [{"field": "quote_accepted", "operator": "eq", "value": True}], "true_action_id": None, "false_action_id": "act_day14_sms" } ), # Day 14 WorkflowAction( id="act_day14_sms", action_type=ActionType.SEND_SMS, config={ "message": "Final reminder {{first_name}} - your quote for {{service}} is expiring. Call us at {{business_phone}} to lock in your pricing today!" }, next_action_id="act_wait_day30" ), WorkflowAction( id="act_wait_day30", action_type=ActionType.WAIT, config={"days": 16}, # Wait until day 30 next_action_id="act_check_response4" ), WorkflowAction( id="act_check_response4", action_type=ActionType.CONDITION, config={ "conditions": [{"field": "quote_accepted", "operator": "eq", "value": True}], "true_action_id": None, "false_action_id": "act_day30_sms" } ), # Day 30 WorkflowAction( id="act_day30_sms", action_type=ActionType.SEND_SMS, config={ "message": "Hi {{first_name}}, it's been a while! Just checking in - are you still interested in {{service}}? Happy to put together a fresh quote. - {{business_name}}" }, next_action_id=None ) ], variables={ "first_name": "", "service": "", "business_name": "AgileAdapt Client", "business_phone": "", "quote_accepted": False } ) @staticmethod def review_request() -> WorkflowTemplate: """ REVIEW REQUEST WORKFLOW Trigger: Job marked complete Wait: 2 hours Action: Send SMS "Thanks for choosing us! Would you mind leaving a quick review? {{review_link}}" """ return WorkflowTemplate( id="wft_review_request", name="Review Request", description="Request a review after job completion", trigger=WorkflowTrigger( trigger_type=TriggerType.JOB_COMPLETE, config={} ), actions=[ WorkflowAction( id="act_wait_2hr", action_type=ActionType.WAIT, config={"hours": 2}, next_action_id="act_send_review_request" ), WorkflowAction( id="act_send_review_request", action_type=ActionType.SEND_SMS, config={ "message": "Hi {{first_name}}! Thanks for choosing {{business_name}}! We'd love to hear how we did. Would you mind leaving a quick review? {{review_link}} - It really helps us out!" }, next_action_id="act_wait_reminder" ), WorkflowAction( id="act_wait_reminder", action_type=ActionType.WAIT, config={"days": 3}, next_action_id="act_check_review" ), WorkflowAction( id="act_check_review", action_type=ActionType.CONDITION, config={ "conditions": [{"field": "review_submitted", "operator": "eq", "value": True}], "true_action_id": None, "false_action_id": "act_reminder_sms" } ), WorkflowAction( id="act_reminder_sms", action_type=ActionType.SEND_SMS, config={ "message": "Hi {{first_name}}, just a gentle reminder - if you have a moment, we'd really appreciate a quick review: {{review_link}} Thanks! - {{business_name}}" }, next_action_id=None ) ], variables={ "first_name": "", "business_name": "AgileAdapt Client", "review_link": "" } ) @staticmethod def appointment_reminder() -> WorkflowTemplate: """ APPOINTMENT REMINDER SEQUENCE Trigger: Appointment scheduled Actions: - 24hr before: Email reminder with details - 2hr before: SMS reminder """ return WorkflowTemplate( id="wft_appointment_reminder", name="Appointment Reminder", description="Remind clients about upcoming appointments", trigger=WorkflowTrigger( trigger_type=TriggerType.APPOINTMENT_SCHEDULED, config={} ), actions=[ WorkflowAction( id="act_calc_24hr", action_type=ActionType.WAIT, config={ "until_before_appointment": True, "hours_before": 24 }, next_action_id="act_email_24hr" ), WorkflowAction( id="act_email_24hr", action_type=ActionType.SEND_EMAIL, config={ "subject": "Reminder: Your appointment tomorrow with {{business_name}}", "body": """Hi {{first_name}}, This is a friendly reminder about your upcoming appointment: Date: {{appointment_date}} Time: {{appointment_time}} Location: {{appointment_location}} Service: {{service}} If you need to reschedule, please call us at {{business_phone}} or reply to this email. See you soon! {{business_name}}""" }, next_action_id="act_wait_2hr_before" ), WorkflowAction( id="act_wait_2hr_before", action_type=ActionType.WAIT, config={ "hours": 22 # 22 hours after 24hr reminder = 2hr before appointment }, next_action_id="act_sms_2hr" ), WorkflowAction( id="act_sms_2hr", action_type=ActionType.SEND_SMS, config={ "message": "Hi {{first_name}}! Quick reminder - your appointment is in 2 hours at {{appointment_time}}. See you soon! - {{business_name}}" }, next_action_id=None ) ], variables={ "first_name": "", "appointment_date": "", "appointment_time": "", "appointment_location": "", "service": "", "business_name": "AgileAdapt Client", "business_phone": "" } ) @classmethod def get_all_templates(cls) -> List[WorkflowTemplate]: """Get all pre-built templates.""" return [ cls.missed_call_textback(), cls.quote_followup_sequence(), cls.review_request(), cls.appointment_reminder() ] @classmethod def install_templates(cls, storage: WorkflowStorage) -> int: """Install all pre-built templates into storage.""" templates = cls.get_all_templates() installed = 0 for template in templates: if storage.save_template(template): installed += 1 logger.info(f"Installed template: {template.name}") return installed # ============================================================================= # WORKFLOW TRIGGER MANAGER # ============================================================================= class WorkflowTriggerManager: """Manages workflow triggers and event routing.""" def __init__(self, engine: WorkflowEngine): self.engine = engine self._trigger_handlers: Dict[TriggerType, List[str]] = {} # trigger_type -> [template_ids] def register_triggers(self): """Register all active templates' triggers.""" templates = self.engine.storage.list_templates(active_only=True) self._trigger_handlers.clear() for template in templates: trigger_type = template.trigger.trigger_type if trigger_type not in self._trigger_handlers: self._trigger_handlers[trigger_type] = [] self._trigger_handlers[trigger_type].append(template.id) logger.info(f"Registered triggers: {self._trigger_handlers}") def handle_event( self, trigger_type: TriggerType, client_id: str, contact_id: str, context: Dict[str, Any] ) -> List[str]: """ Handle an incoming event and start relevant workflows. Returns list of created workflow instance IDs. """ template_ids = self._trigger_handlers.get(trigger_type, []) instance_ids = [] for template_id in template_ids: # Check if client has this workflow enabled client_config = self.engine.storage.get_client_config(client_id, template_id) if client_config and not client_config.get("is_enabled", True): continue # Create and start workflow instance instance = self.engine.create_instance( template_id=template_id, client_id=client_id, contact_id=contact_id, context=context ) if instance: self.engine.start_workflow(instance.id) instance_ids.append(instance.id) return instance_ids def trigger_missed_call(self, client_id: str, contact_id: str, phone: str, first_name: str = None) -> List[str]: """Convenience method for missed call trigger.""" return self.handle_event( TriggerType.MISSED_CALL, client_id, contact_id, { "phone": phone, "first_name": first_name or "there" } ) def trigger_quote_sent(self, client_id: str, contact_id: str, first_name: str, service: str, **kwargs) -> List[str]: """Convenience method for quote sent trigger.""" context = { "first_name": first_name, "service": service, **kwargs } return self.handle_event( TriggerType.QUOTE_SENT, client_id, contact_id, context ) def trigger_job_complete(self, client_id: str, contact_id: str, first_name: str, review_link: str, **kwargs) -> List[str]: """Convenience method for job complete trigger.""" context = { "first_name": first_name, "review_link": review_link, **kwargs } return self.handle_event( TriggerType.JOB_COMPLETE, client_id, contact_id, context ) def trigger_appointment_scheduled(self, client_id: str, contact_id: str, first_name: str, appointment_date: str, appointment_time: str, **kwargs) -> List[str]: """Convenience method for appointment scheduled trigger.""" context = { "first_name": first_name, "appointment_date": appointment_date, "appointment_time": appointment_time, **kwargs } return self.handle_event( TriggerType.APPOINTMENT_SCHEDULED, client_id, contact_id, context ) # ============================================================================= # MAIN ENTRY POINT # ============================================================================= def initialize_workflow_system() -> Tuple[WorkflowEngine, WorkflowTriggerManager]: """Initialize the complete workflow system.""" storage = WorkflowStorage() engine = WorkflowEngine(storage) # Install pre-built templates installed = WorkflowTemplates.install_templates(storage) logger.info(f"Installed {installed} workflow templates") # Create trigger manager trigger_manager = WorkflowTriggerManager(engine) trigger_manager.register_triggers() return engine, trigger_manager if __name__ == "__main__": # Demo/test the workflow system print("=" * 60) print("AGILEADAPT WORKFLOW BUILDER - INITIALIZATION") print("=" * 60) engine, trigger_manager = initialize_workflow_system() # List installed templates templates = engine.storage.list_templates() print(f"\nInstalled Templates ({len(templates)}):") for t in templates: print(f" - {t.name} (ID: {t.id})") print(f" Trigger: {t.trigger.trigger_type.value}") print(f" Actions: {len(t.actions)}") # Demo: Create a missed call workflow print("\n" + "=" * 60) print("DEMO: Missed Call Text-Back Workflow") print("=" * 60) instance_ids = trigger_manager.trigger_missed_call( client_id="client_demo_001", contact_id="contact_demo_001", phone="+61400000000", first_name="John" ) if instance_ids: print(f"Created workflow instances: {instance_ids}") # Get instance details instance = engine.storage.get_instance(instance_ids[0]) if instance: print(f"\nInstance Details:") print(f" Status: {instance.status.value}") print(f" Current Action: {instance.current_action_id}") print(f" History: {len(instance.history)} events") print("\n" + "=" * 60) print("WORKFLOW SYSTEM READY") print("=" * 60)