import asyncio from dataclasses import dataclass from datetime import datetime from enum import Enum import json import logging import os from typing import Any, Dict, List, Optional, Set import uuid from services.agent_service import agent_service from core.oauth_handler import SLACK_OAUTH_CONFIG from integrations.gmail_service import get_gmail_service from integrations.slack_enhanced_service import SlackEnhancedService # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class TriggerType(Enum): """Types of automation triggers""" SCHEDULED = "scheduled" EVENT_BASED = "event_based" MANUAL = "manual" API_CALL = "api_call" class ActionType(Enum): """Types of automation actions""" CREATE = "create" UPDATE = "update" DELETE = "delete" NOTIFY = "notify" SEARCH = "search" SYNC = "sync" TRANSFORM = "transform" class PlatformType(Enum): """Supported platform types for automation""" SLACK = "slack" TEAMS = "teams" DISCORD = "discord" GMAIL = "gmail" GOOGLE_CHAT = "google_chat" TELEGRAM = "telegram" WHATSAPP = "whatsapp" ZOOM = "zoom" GOOGLE_DRIVE = "google_drive" DROPBOX = "dropbox" BOX = "box" ONEDRIVE = "onedrive" GITHUB = "github" ASANA = "asana" NOTION = "notion" LINEAR = "linear" MONDAY = "monday" TRELLO = "trello" JIRA = "jira" GITLAB = "gitlab" SALESFORCE = "salesforce" HUBSPOT = "hubspot" INTERCOM = "intercom" FRESHDESK = "freshdesk" ZENDESK = "zendesk" STRIPE = "stripe" QUICKBOOKS = "quickbooks" XERO = "xero" MAILCHIMP = "mailchimp" HUBSPOT_MARKETING = "hubspot_marketing" TABLEAU = "tableau" GOOGLE_ANALYTICS = "google_analytics" FIGMA = "figma" SHOPIFY = "shopify" @dataclass class AutomationTrigger: """Definition of an automation trigger""" trigger_id: str trigger_type: TriggerType platform: PlatformType event_name: str conditions: Dict[str, Any] description: str is_active: bool = True @dataclass class AutomationAction: """Definition of an automation action""" action_id: str action_type: ActionType platform: PlatformType target_entity: str parameters: Dict[str, Any] description: str @dataclass class AutomationWorkflow: """Complete automation workflow definition""" workflow_id: str name: str description: str trigger: AutomationTrigger actions: List[AutomationAction] conditions: List[Dict[str, Any]] is_active: bool = True created_at: datetime = None updated_at: datetime = None @dataclass class WorkflowExecution: """Record of workflow execution""" execution_id: str workflow_id: str trigger_data: Dict[str, Any] start_time: datetime end_time: Optional[datetime] = None status: str = "running" actions_executed: List[str] = None errors: List[str] = None results: Dict[str, Any] = None duration_ms: float = 0.0 def __post_init__(self): if self.actions_executed is None: self.actions_executed = [] if self.errors is None: self.errors = [] if self.results is None: self.results = {} def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for serialization""" return { "execution_id": self.execution_id, "workflow_id": self.workflow_id, "trigger_data": self.trigger_data, "start_time": self.start_time.isoformat() if self.start_time else None, "end_time": self.end_time.isoformat() if self.end_time else None, "status": self.status, "actions_executed": self.actions_executed, "errors": self.errors, "results": self.results, "duration_ms": self.duration_ms } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'WorkflowExecution': """Create from dictionary""" execution = cls( execution_id=data["execution_id"], workflow_id=data["workflow_id"], trigger_data=data.get("trigger_data", {}), start_time=datetime.fromisoformat(data["start_time"]) if data.get("start_time") else datetime.now(), status=data.get("status", "unknown") ) execution.end_time = datetime.fromisoformat(data["end_time"]) if data.get("end_time") else None execution.actions_executed = data.get("actions_executed", []) execution.errors = data.get("errors", []) execution.results = data.get("results", {}) execution.duration_ms = data.get("duration_ms", 0.0) return execution class AutomationEngine: """Cross-Platform Automation Engine for ATOM Platform""" def __init__(self): self.workflows: Dict[str, AutomationWorkflow] = {} self.executions: Dict[str, WorkflowExecution] = {} self.executions_file = "executions.json" self._load_executions() self.slack_service = SlackEnhancedService({ "client_id": SLACK_OAUTH_CONFIG.client_id, "client_secret": SLACK_OAUTH_CONFIG.client_secret, "signing_secret": "dummy", # Not needed for sending messages "redirect_uri": SLACK_OAUTH_CONFIG.redirect_uri }) self.platform_connectors = self._initialize_platform_connectors() self.action_handlers = self._initialize_action_handlers() def _load_executions(self): """Load executions from file""" try: if os.path.exists(self.executions_file): with open(self.executions_file, 'r') as f: data = json.load(f) for exec_data in data: execution = WorkflowExecution.from_dict(exec_data) self.executions[execution.execution_id] = execution logger.info(f"Loaded {len(self.executions)} executions from {self.executions_file}") except Exception as e: logger.error(f"Error loading executions: {e}") def _save_execution(self, execution: WorkflowExecution): """Save execution to file""" try: self.executions[execution.execution_id] = execution # Convert all executions to dict list data = [e.to_dict() for e in self.executions.values()] with open(self.executions_file, 'w') as f: json.dump(data, f, indent=2) except Exception as e: logger.error(f"Error saving execution: {e}") def _initialize_platform_connectors(self) -> Dict[PlatformType, callable]: """Initialize platform action connectors""" # In production, these would be actual API connectors connectors = {platform: self._mock_platform_connector for platform in PlatformType} # Override with real connectors where available connectors[PlatformType.SLACK] = self._slack_connector # Gmail is not in PlatformType enum explicitly but might be mapped from GOOGLE_DRIVE or added # Assuming we use a generic google connector or add GMAIL to enum if needed. # For now, let's add a specific check in the mock connector or just use _gmail_connector if we add GMAIL type. # But wait, PlatformType doesn't have GMAIL. It has GOOGLE_CHAT, GOOGLE_DRIVE. # I should probably add GMAIL to PlatformType or just map it. # Let's assume we can use a custom string or just add it. # For this task, I'll add GMAIL to PlatformType enum first. # Override with real connectors where available connectors[PlatformType.SLACK] = self._slack_connector connectors[PlatformType.GMAIL] = self._gmail_connector return connectors def _initialize_action_handlers(self) -> Dict[ActionType, callable]: """Initialize action handler functions""" return { ActionType.CREATE: self._handle_create_action, ActionType.UPDATE: self._handle_update_action, ActionType.DELETE: self._handle_delete_action, ActionType.NOTIFY: self._handle_notify_action, ActionType.SEARCH: self._handle_search_action, ActionType.SYNC: self._handle_sync_action, ActionType.TRANSFORM: self._handle_transform_action, } def create_workflow(self, workflow_data: Dict[str, Any]) -> AutomationWorkflow: """Create a new automation workflow""" workflow_id = str(uuid.uuid4()) # Create trigger trigger = AutomationTrigger( trigger_id=str(uuid.uuid4()), trigger_type=TriggerType(workflow_data["trigger"]["type"]), platform=PlatformType(workflow_data["trigger"]["platform"]), event_name=workflow_data["trigger"]["event_name"], conditions=workflow_data["trigger"].get("conditions", {}), description=workflow_data["trigger"]["description"], ) # Create actions actions = [] for action_data in workflow_data["actions"]: action = AutomationAction( action_id=str(uuid.uuid4()), action_type=ActionType(action_data["type"]), platform=PlatformType(action_data["platform"]), target_entity=action_data["target_entity"], parameters=action_data.get("parameters", {}), description=action_data["description"], ) actions.append(action) # Create workflow workflow = AutomationWorkflow( workflow_id=workflow_id, name=workflow_data["name"], description=workflow_data["description"], trigger=trigger, actions=actions, conditions=workflow_data.get("conditions", []), created_at=datetime.now(), updated_at=datetime.now(), ) self.workflows[workflow_id] = workflow logger.info(f"Created workflow: {workflow.name} (ID: {workflow_id})") return workflow async def execute_workflow( self, workflow_id: str, trigger_data: Dict[str, Any] ) -> WorkflowExecution: """Execute an automation workflow""" workflow = self.workflows.get(workflow_id) if not workflow: raise ValueError(f"Workflow {workflow_id} not found") if not workflow.is_active: raise ValueError(f"Workflow {workflow_id} is not active") # Create execution record execution = WorkflowExecution( execution_id=str(uuid.uuid4()), workflow_id=workflow_id, trigger_data=trigger_data, start_time=datetime.now(), ) self.executions[execution.execution_id] = execution logger.info(f"Starting workflow execution: {workflow.name}") try: # Check conditions if not await self._check_conditions(workflow.conditions, trigger_data): execution.status = "skipped" execution.end_time = datetime.now() execution.errors.append("Conditions not met") return execution # Execute actions in sequence for action in workflow.actions: try: result = await self._execute_action(action, trigger_data) execution.actions_executed.append(action.action_id) execution.results[action.action_id] = result logger.info(f"Executed action: {action.description}") except Exception as e: error_msg = f"Action {action.action_id} failed: {str(e)}" execution.errors.append(error_msg) logger.error(f"Error executing {action.action_type.value} action on {action.platform.value}: {str(e)}") execution.errors.append(f"{action.action_id}: {str(e)}") # Continue with next action (configurable behavior) execution.status = "completed" execution.end_time = datetime.now() logger.info(f"Workflow {workflow.workflow_id} completed with status: {execution.status}") except Exception as e: execution.status = "failed" execution.end_time = datetime.now() execution.errors.append(f"Workflow execution failed: {str(e)}") logger.error(f"Workflow execution failed: {str(e)}") # Calculate duration if execution.end_time and execution.start_time: execution.duration_ms = (execution.end_time - execution.start_time).total_seconds() * 1000 self._save_execution(execution) return execution async def _check_conditions( self, conditions: List[Dict[str, Any]], trigger_data: Dict[str, Any] ) -> bool: """Check if all conditions are met""" for condition in conditions: condition_type = condition.get("type") field = condition.get("field") operator = condition.get("operator") value = condition.get("value") # Get field value from trigger data field_value = trigger_data.get(field) if not self._evaluate_condition(field_value, operator, value): return False return True def _evaluate_condition( self, field_value: Any, operator: str, expected_value: Any ) -> bool: """Evaluate a single condition""" if operator == "equals": return field_value == expected_value elif operator == "not_equals": return field_value != expected_value elif operator == "contains": return expected_value in str(field_value) elif operator == "greater_than": return float(field_value) > float(expected_value) elif operator == "less_than": return float(field_value) < float(expected_value) elif operator == "exists": return field_value is not None elif operator == "not_exists": return field_value is None else: logger.warning(f"Unknown operator: {operator}") return True # Default to true for unknown operators async def _execute_action( self, action: AutomationAction, trigger_data: Dict[str, Any] ) -> Dict[str, Any]: """Execute a single automation action""" handler = self.action_handlers.get(action.action_type) if not handler: raise ValueError(f"No handler for action type: {action.action_type}") # Merge trigger data with action parameters execution_data = {**trigger_data, **action.parameters} result = await handler(action, execution_data) return result async def _handle_create_action( self, action: AutomationAction, data: Dict[str, Any] ) -> Dict[str, Any]: """Handle create actions""" platform_connector = self.platform_connectors.get(action.platform) if not platform_connector: raise ValueError(f"No connector for platform: {action.platform}") # Mock implementation - in production, this would call actual APIs result = await platform_connector("create", action.target_entity, data) return {"success": True, "created_id": str(uuid.uuid4()), "data": result} async def _handle_update_action( self, action: AutomationAction, data: Dict[str, Any] ) -> Dict[str, Any]: """Handle update actions""" platform_connector = self.platform_connectors.get(action.platform) if not platform_connector: raise ValueError(f"No connector for platform: {action.platform}") # Mock implementation result = await platform_connector("update", action.target_entity, data) return {"success": True, "updated_id": data.get("id"), "data": result} async def _handle_delete_action( self, action: AutomationAction, data: Dict[str, Any] ) -> Dict[str, Any]: """Handle delete actions""" platform_connector = self.platform_connectors.get(action.platform) if not platform_connector: raise ValueError(f"No connector for platform: {action.platform}") # Mock implementation result = await platform_connector("delete", action.target_entity, data) return {"success": True, "deleted_id": data.get("id"), "data": result} async def _handle_notify_action( self, action: AutomationAction, data: Dict[str, Any] ) -> Dict[str, Any]: """Handle notification actions""" platform_connector = self.platform_connectors.get(action.platform) if not platform_connector: raise ValueError(f"No connector for platform: {action.platform}") # Mock implementation result = await platform_connector("notify", action.target_entity, data) return {"success": True, "notification_sent": True, "data": result} async def _handle_search_action( self, action: AutomationAction, data: Dict[str, Any] ) -> Dict[str, Any]: """Handle search actions""" platform_connector = self.platform_connectors.get(action.platform) if not platform_connector: raise ValueError(f"No connector for platform: {action.platform}") # Mock implementation result = await platform_connector("search", action.target_entity, data) return {"success": True, "results": result, "count": len(result)} async def _handle_sync_action( self, action: AutomationAction, data: Dict[str, Any] ) -> Dict[str, Any]: """Handle sync actions between platforms""" # This would synchronize data between different platforms source_platform = data.get("source_platform") target_platform = action.platform # Mock implementation return { "success": True, "synced_items": 5, "input_data": data, "output_data": {"transformed": True, **data}, } async def _handle_transform_action( self, action: AutomationAction, data: Dict[str, Any] ) -> Dict[str, Any]: """Handle data transformation actions""" # This would transform data from one format to another transformation_type = data.get("transformation_type", "default") # Mock implementation return { "success": True, "transformation_type": transformation_type, "input_data": data, "output_data": {"transformed": True, **data}, } async def _mock_platform_connector( self, operation: str, entity: str, data: Dict[str, Any] ) -> Dict[str, Any]: """Mock connector for platforms without real implementation""" logger.info( f"Mock execution for platform: {operation} on {entity}" ) return { "operation": operation, "entity": entity, "platform": "mock", "timestamp": datetime.now().isoformat(), "data": data, } async def _slack_connector( self, operation: str, entity: str, data: Dict[str, Any] ) -> Dict[str, Any]: """Real Slack connector""" if operation == "notify": channel = data.get("channel") message = data.get("message") # We need a workspace_id. For MVP, we might need to look it up or pass it in data. # If not provided, we might default to the first available workspace in token storage? # Or just fail if not provided. # Let's try to get it from data or token storage. workspace_id = data.get("workspace_id") # If no workspace_id, try to find one from token storage (hack for MVP) if not workspace_id: from core.token_storage import token_storage token = token_storage.get_token("slack") if token: workspace_id = token.get("team", {}).get("id") if workspace_id and channel and message: result = await self.slack_service.send_message(workspace_id, channel, message) return {"success": result.get("ok", False), "data": result} else: raise ValueError("Missing workspace_id, channel, or message for Slack notification") return await self._mock_platform_connector(operation, entity, data) async def _gmail_connector( self, operation: str, entity: str, data: Dict[str, Any] ) -> Dict[str, Any]: """Real Gmail connector""" service = get_gmail_service() if operation == "notify" or operation == "create": to = data.get("to") subject = data.get("subject") body = data.get("body") or data.get("message") if to and subject and body: result = service.send_message(to, subject, body) return {"success": bool(result), "data": result} else: raise ValueError("Missing to, subject, or body for Gmail message") elif operation == "search": query = data.get("query", "") messages = service.search_messages(query) return {"success": True, "data": messages, "count": len(messages)} return await self._mock_platform_connector(operation, entity, data) def get_workflow(self, workflow_id: str) -> Optional[AutomationWorkflow]: """Get workflow by ID""" return self.workflows.get(workflow_id) def list_workflows(self, active_only: bool = True) -> List[AutomationWorkflow]: """List all workflows""" workflows = list(self.workflows.values()) if active_only: workflows = [w for w in workflows if w.is_active] return workflows def update_workflow( self, workflow_id: str, updates: Dict[str, Any] ) -> AutomationWorkflow: """Update an existing workflow""" workflow = self.workflows.get(workflow_id) if not workflow: raise ValueError(f"Workflow {workflow_id} not found") # Update fields if "name" in updates: workflow.name = updates["name"] if "description" in updates: workflow.description = updates["description"] if "is_active" in updates: workflow.is_active = updates["is_active"] if "conditions" in updates: workflow.conditions = updates["conditions"] workflow.updated_at = datetime.now() logger.info(f"Updated workflow: {workflow.name}") return workflow def delete_workflow(self, workflow_id: str) -> bool: """Delete a workflow""" if workflow_id in self.workflows: del self.workflows[workflow_id] logger.info(f"Deleted workflow: {workflow_id}") return True return False async def execute_workflow_definition(self, workflow_def: Dict[str, Any], input_data: Dict[str, Any] = None, execution_id: str = None) -> Dict[str, Any]: """ Execute a workflow from its definition (as stored in workflows.json) Args: workflow_def: Workflow definition with nodes and connections input_data: Optional input data for the workflow execution_id: Optional ID for this execution Returns: Execution results and metadata """ results = [] input_data = input_data or {} execution_id = execution_id or str(uuid.uuid4()) logger.info(f"Executing workflow: {workflow_def.get('name')} (ID: {execution_id})") # Create execution record execution = WorkflowExecution( execution_id=execution_id, workflow_id=workflow_def.get('id'), trigger_data=input_data, start_time=datetime.now(), status="running" ) self.executions[execution_id] = execution # Execute each node in order for node in workflow_def.get('nodes', []): node_result = { "node_id": node['id'], "node_type": node['type'], "node_title": node['title'], "status": "pending", "output": None, "error": None } try: if node['type'] == 'action': # Get node configuration config = node.get('config', {}) action_type = config.get('actionType') integration_id = config.get('integrationId') logger.info(f"Executing action node: {node['title']} (action: {action_type}, integration: {integration_id})") # Execute based on action type and integration if action_type == 'send_email' and integration_id == 'gmail': # Execute Gmail send email gmail_service = get_gmail_service() result = gmail_service.send_message( to=config.get('to', ''), subject=config.get('subject', 'No Subject'), body=config.get('body', '') ) node_result['output'] = result node_result['status'] = "success" elif action_type == 'notify' and integration_id == 'slack': # Execute Slack notification result = await self.slack_service.send_message( channel=config.get('channel', '#general'), message=config.get('message', '') ) node_result['output'] = result node_result['status'] = "success" elif action_type == 'run_agent_task': # Execute Computer Use Agent Task goal = config.get('goal', '') mode = config.get('mode', 'thinker') logger.info(f"Starting agent task: {goal} ({mode})") # Start agent task param_result = await agent_service.execute_task(goal, mode) node_result['output'] = param_result node_result['status'] = "success" else: # Unsupported action type node_result['status'] = "skipped" node_result['output'] = f"Action type '{action_type}' with integration '{integration_id}' not yet implemented" elif node['type'] == 'trigger': # Trigger nodes don't execute, they just define when the workflow runs node_result['status'] = "success" node_result['output'] = "Trigger node (manual execution)" else: # Other node types (condition, delay, etc.) node_result['status'] = "skipped" node_result['output'] = f"Node type '{node['type']}' not yet implemented" except Exception as e: logger.error(f"Error executing node {node['id']}: {e}") node_result['status'] = "failed" node_result['error'] = str(e) execution.errors.append(f"Node {node['id']}: {str(e)}") results.append(node_result) execution.actions_executed.append(node['id']) execution.results[node['id']] = node_result # If any node fails, mark execution as failed (or continue based on policy) if node_result['status'] == 'failed': execution.status = "failed" # Finalize execution record if execution.status == "running": execution.status = "completed" execution.end_time = datetime.now() if execution.start_time: execution.duration_ms = (execution.end_time - execution.start_time).total_seconds() * 1000 self._save_execution(execution) logger.info(f"Workflow execution complete with {len(results)} nodes processed") return results def get_execution_history( self, workflow_id: str, limit: int = 10 ) -> List[WorkflowExecution]: """Get execution history for a workflow""" executions = [ e for e in self.executions.values() if e.workflow_id == workflow_id ] executions.sort(key=lambda x: x.start_time, reverse=True) return executions[:limit] # Example usage and testing async def main(): """Test the automation engine""" engine = AutomationEngine() # Create a sample workflow workflow_data = { "name": "Daily Team Update", "description": "Send daily team updates and create follow-up tasks", "trigger": { "type": "scheduled", "platform": "slack", "event_name": "daily_reminder", "conditions": {"time": "09:00", "weekday": "mon-fri"}, "description": "Triggered every weekday at 9 AM", }, "actions": [ { "type": "search", "platform": "asana", "target_entity": "tasks", "parameters": {"status": "today", "assignee": "team"}, "description": "Find today's tasks for the team", }, { "type": "notify", "platform": "slack", "target_entity": "channel", "parameters": { "channel": "#team-updates", "message": "Daily update ready", }, "description": "Send notification to Slack channel", }, { "type": "create", "platform": "asana", "target_entity": "task", "parameters": { "name": "Follow up on daily update", "assignee": "manager", }, "description": "Create follow-up task", }, ], "conditions": [ { "type": "business_hours", "field": "time", "operator": "greater_than", "value": "08:00", } ], } # Create the workflow workflow = engine.create_workflow(workflow_data) print(f"Created workflow: {workflow.name}") # Execute the workflow trigger_data = {"time": "09:00", "weekday": "monday", "team": "engineering"} execution = await engine.execute_workflow(workflow.workflow_id, trigger_data) print(f"Execution completed with status: {execution.status}") print if __name__ == "__main__": asyncio.run(main())