# Workflow Actions RBAC Integration Concept ## Übersicht Dieses Dokument beschreibt das Konzept für die Umstrukturierung der Workflow Actions, um: 1. **RBAC-Integration** zu ermöglichen (Schutz von Actions über RESOURCE-Context) 2. **Strukturierte Parameter-Definitionen** statt Docstrings zu verwenden 3. **UI-Rendering-Typen** für Parameter zu definieren 4. **Keine Duplikation** von Parameter-Definitionen zu haben 5. **Plug-and-Play** Funktionalität beizubehalten ## Architektur-Konzept ### Grundprinzip: Deklarative Action-Definition Ähnlich wie bei `aicore` Models, wo eine Struktur definiert wird und die Funktion ein Attribut ist, werden Actions jetzt deklarativ definiert: ```python class MethodOutlook(MethodBase): def __init__(self, services): super().__init__(services) self.name = "outlook" self.description = "Handle Microsoft Outlook email operations" # Actions werden deklarativ definiert self._actions = { "readEmails": WorkflowActionDefinition( actionId="outlook.readEmails", # Für RBAC: RESOURCE context description="Read emails from Outlook mailbox", parameters={ "connectionReference": WorkflowActionParameter( name="connectionReference", type="str", frontendType=FrontendType.USER_CONNECTION, required=True, description="Microsoft connection label" ), "query": WorkflowActionParameter( name="query", type="str", frontendType=FrontendType.TEXT, required=False, description="Search query for emails" ), "folder": WorkflowActionParameter( name="folder", type="str", frontendType=FrontendType.SELECT, frontendOptions="outlook.folder", required=False, description="Folder name (e.g., 'Inbox', 'Drafts')" ), "limit": WorkflowActionParameter( name="limit", type="int", frontendType=FrontendType.NUMBER, required=False, default=50, description="Maximum number of emails to return" ) }, execute=self._executeReadEmails # Funktion als Attribut ), "sendEmail": WorkflowActionDefinition( actionId="outlook.sendEmail", description="Send email via Outlook", parameters={...}, execute=self._executeSendEmail ) } async def _executeReadEmails(self, parameters: Dict[str, Any]) -> ActionResult: """Execute function - keine Parameter-Definition mehr hier""" # Implementation... ``` ## Globale Frontend-Type-Definition **WICHTIG**: Frontend-Types werden zentral in `modules/shared/frontendTypes.py` definiert, nicht redundant pro Action. Die globale `FrontendType` Enum enthält: - **Standard Types**: `text`, `textarea`, `number`, `select`, `multiselect`, `checkbox`, `date`, `datetime`, `email`, `timestamp`, `json`, `multilingual`, `file` - **Custom Types für Actions**: `userConnection`, `documentReference`, `workflowAction` Custom-Types unterstützen dynamische Option-Listen über API-Endpoints: - `userConnection` → `/api/options/user.connection` (Connections des aktuellen Users) - `documentReference` → `/api/options/workflow.documentReference` (Document-Referenzen aus Workflow-Context) - `workflowAction` → `/api/options/workflow.action` (Verfügbare Actions aus Workflow-Context) ## Datenmodelle ### WorkflowActionParameter **WICHTIG**: - Frontend-Types werden global definiert in `modules/shared/frontendTypes.py` und nicht redundant in Actions - Diese Klasse heißt `WorkflowActionParameter` (nicht `ActionParameter`) um Konflikte mit `ActionParameters` aus `datamodelChat.py` zu vermeiden ```python from typing import Optional, Any, Union, List, Dict from pydantic import BaseModel, Field from modules.shared.frontendTypes import FrontendType # Globale Definition class WorkflowActionParameter(BaseModel): """ Parameter schema definition for a workflow action. This defines the structure and UI rendering for a single action parameter, NOT the actual parameter values (those are in ActionDefinition.parameters). """ name: str = Field(description="Parameter name") type: str = Field(description="Python type as string (e.g., 'str', 'int', 'bool', 'List[str]')") frontendType: FrontendType = Field(description="UI rendering type (from global FrontendType enum)") frontendOptions: Optional[Union[str, List[Dict[str, Any]]]] = Field( None, description="Options for select/multiselect/custom types. String reference (e.g., 'user.connection') or static list. For custom types like userConnection, this is automatically set to the API endpoint." ) required: bool = Field(False, description="Whether parameter is required") default: Optional[Any] = Field(None, description="Default value") description: str = Field("", description="Parameter description") validation: Optional[Dict[str, Any]] = Field( None, description="Validation rules (e.g., {'min': 1, 'max': 100})" ) ``` **Custom Frontend Types**: - `FrontendType.USER_CONNECTION`: User connection selector - dynamische Options von `/api/options/user.connection` - `FrontendType.DOCUMENT_REFERENCE`: Document reference selector - dynamische Options aus Workflow-Context - `FrontendType.WORKFLOW_ACTION`: Workflow action selector - dynamische Options aus verfügbaren Actions Für Custom-Types wird `frontendOptions` automatisch auf den entsprechenden API-Endpoint gesetzt (z.B. `"user.connection"`). ### WorkflowActionDefinition **WICHTIG**: Diese Klasse heißt `WorkflowActionDefinition` (nicht `ActionDefinition`) um Konflikte mit der bestehenden `ActionDefinition` aus `datamodelWorkflow.py` zu vermeiden: - **Bestehende `ActionDefinition`**: Für Workflow-Execution-Planning (enthält konkrete Werte: `action`, `actionObjective`, `parameters` mit Werten) - **Neue `WorkflowActionDefinition`**: Für Action-Schema-Definitionen (enthält Metadaten: `actionId`, `description`, `parameters` als Schemas) ```python from typing import Dict, Callable, Awaitable, Optional, List from pydantic import BaseModel, Field from modules.datamodels.datamodelChat import ActionResult class WorkflowActionDefinition(BaseModel): """ Complete schema definition of a workflow action. This defines the metadata, parameters, and execution function for an action. This is different from datamodelWorkflow.ActionDefinition which contains actual execution values (action, actionObjective, parameters with values). This class defines the ACTION SCHEMA, not the execution plan. """ actionId: str = Field( description="Unique action identifier for RBAC (format: 'module.actionName', e.g., 'outlook.readEmails')" ) description: str = Field(description="Action description") parameters: Dict[str, WorkflowActionParameter] = Field( default_factory=dict, description="Parameter schema definitions" ) execute: Optional[Callable[[Dict[str, Any]], Awaitable[ActionResult]]] = Field( None, description="Execution function - async function that takes parameters dict and returns ActionResult. Set dynamically." ) category: Optional[str] = Field(None, description="Action category for grouping") tags: List[str] = Field(default_factory=list, description="Tags for search/filtering") ``` ## MethodBase Erweiterung ### Neue MethodBase Struktur ```python class MethodBase: """Base class for all methods""" def __init__(self, services: Any): self.services = services self.name: str self.description: str self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") # Actions werden als Dictionary definiert self._actions: Dict[str, WorkflowActionDefinition] = {} # Nach Initialisierung: Actions registrieren self._registerActions() def _registerActions(self): """Register all actions defined in _actions""" # Kann überschrieben werden für dynamische Registrierung pass @property def actions(self) -> Dict[str, Dict[str, Any]]: """ Dynamically collect all actions from _actions dictionary. Returns format compatible with existing system. """ result = {} for actionName, actionDef in self._actions.items(): # RBAC-Check: Prüfe ob Action für aktuellen User verfügbar ist if not self._checkActionPermission(actionDef.actionId): continue # Skip if user doesn't have permission # Konvertiere WorkflowActionDefinition zu altem Format für Kompatibilität result[actionName] = { 'description': actionDef.description, 'parameters': self._convertParametersToOldFormat(actionDef.parameters), 'method': self._createActionWrapper(actionDef) } return result def _checkActionPermission(self, actionId: str) -> bool: """ Check if current user has permission to execute this action. Uses RBAC RESOURCE context. """ if not hasattr(self.services, 'rbac') or not self.services.rbac: # Fallback: Allow if RBAC not available (backward compatibility) return True currentUser = self.services.chat.getCurrentUser() if not currentUser: return False # RBAC-Check: RESOURCE context, item = actionId permissions = self.services.rbac.getUserPermissions( user=currentUser, context=AccessRuleContext.RESOURCE, item=actionId ) return permissions.view def _convertParametersToOldFormat(self, parameters: Dict[str, WorkflowActionParameter]) -> Dict[str, Dict[str, Any]]: """Convert WorkflowActionParameter dict to old format for compatibility""" result = {} for paramName, param in parameters.items(): result[paramName] = { 'type': param.type, 'required': param.required, 'description': param.description, 'default': param.default, 'frontendType': param.frontendType.value, 'frontendOptions': param.frontendOptions } return result def _createActionWrapper(self, actionDef: WorkflowActionDefinition): """Create wrapper function that matches old action signature""" async def wrapper(parameters: Dict[str, Any], *args, **kwargs): # Parameter-Validierung basierend auf WorkflowActionParameter definitions validatedParams = self._validateParameters(parameters, actionDef.parameters) # Execute action return await actionDef.execute(validatedParams, *args, **kwargs) wrapper.is_action = True return wrapper def _validateParameters(self, parameters: Dict[str, Any], paramDefs: Dict[str, WorkflowActionParameter]) -> Dict[str, Any]: """Validate parameters against definitions""" validated = {} for paramName, paramDef in paramDefs.items(): value = parameters.get(paramName) # Check required if paramDef.required and value is None: raise ValueError(f"Required parameter '{paramName}' is missing") # Use default if not provided if value is None and paramDef.default is not None: value = paramDef.default # Type validation if value is not None: value = self._validateType(value, paramDef.type) # Custom validation rules if paramDef.validation and value is not None: self._applyValidationRules(value, paramDef.validation) validated[paramName] = value return validated def _validateType(self, value: Any, expectedType: type) -> Any: """Validate and convert value to expected type""" # Type validation logic... if expectedType == int: return int(value) elif expectedType == str: return str(value) # ... weitere Typen return value def _applyValidationRules(self, value: Any, rules: Dict[str, Any]): """Apply custom validation rules""" if 'min' in rules and value < rules['min']: raise ValueError(f"Value must be >= {rules['min']}") if 'max' in rules and value > rules['max']: raise ValueError(f"Value must be <= {rules['max']}") # ... weitere Validierungsregeln ``` ## Migrationsstrategie ### Schritt 1: Neue Datenmodelle erstellen **WICHTIG**: Die bestehenden Klassen `ActionDefinition` (in `datamodelWorkflow.py`) und `ActionParameters` (in `datamodelChat.py`) haben einen anderen Zweck: - `ActionDefinition` (existing): Für Workflow-Execution-Planning (enthält konkrete Werte) - `ActionParameters` (existing): Einfacher Parameter-Wrapper **Lösung**: Neue Klassen mit klaren Namen für Action-Schema-Definitionen erstellen. **Datei**: `gateway/modules/datamodels/datamodelWorkflowActions.py` ```python from typing import Optional, Any, Union, List, Dict, Callable, Awaitable from pydantic import BaseModel, Field from modules.datamodels.datamodelChat import ActionResult from modules.shared.frontendTypes import FrontendType # Globale Definition verwenden from modules.shared.attributeUtils import registerModelLabels class WorkflowActionParameter(BaseModel): """ Parameter schema definition for a workflow action. This defines the structure and UI rendering for a single action parameter, NOT the actual parameter values (those are in ActionDefinition.parameters). """ name: str = Field(description="Parameter name") type: str = Field(description="Python type as string: 'str', 'int', 'bool', 'List[str]', etc.") frontendType: FrontendType = Field(description="UI rendering type (from global FrontendType enum)") frontendOptions: Optional[Union[str, List[Dict[str, Any]]]] = Field( None, description="Options for select/multiselect/custom types. String reference (e.g., 'user.connection') or static list. For custom types, this is automatically set to the API endpoint." ) required: bool = Field(False, description="Whether parameter is required") default: Optional[Any] = Field(None, description="Default value") description: str = Field("", description="Parameter description") validation: Optional[Dict[str, Any]] = Field( None, description="Validation rules (e.g., {'min': 1, 'max': 100})" ) class WorkflowActionDefinition(BaseModel): """ Complete schema definition of a workflow action. This defines the metadata, parameters, and execution function for an action. This is different from datamodelWorkflow.ActionDefinition which contains actual execution values (action, actionObjective, parameters with values). This class defines the ACTION SCHEMA, not the execution plan. """ actionId: str = Field( description="Unique action identifier for RBAC (format: 'module.actionName', e.g., 'outlook.readEmails')" ) description: str = Field(description="Action description") parameters: Dict[str, WorkflowActionParameter] = Field( default_factory=dict, description="Parameter schema definitions" ) execute: Optional[Callable] = Field( None, description="Execution function - async function that takes parameters dict and returns ActionResult. Set dynamically." ) category: Optional[str] = Field(None, description="Action category for grouping") tags: List[str] = Field(default_factory=list, description="Tags for search/filtering") # Register model labels for UI registerModelLabels( "WorkflowActionDefinition", {"en": "Workflow Action Definition", "fr": "Définition d'action de workflow"}, { "actionId": {"en": "Action ID", "fr": "ID d'action"}, "description": {"en": "Description", "fr": "Description"}, "parameters": {"en": "Parameters", "fr": "Paramètres"}, "category": {"en": "Category", "fr": "Catégorie"}, "tags": {"en": "Tags", "fr": "Étiquettes"}, }, ) registerModelLabels( "WorkflowActionParameter", {"en": "Workflow Action Parameter", "fr": "Paramètre d'action de workflow"}, { "name": {"en": "Name", "fr": "Nom"}, "type": {"en": "Type", "fr": "Type"}, "frontendType": {"en": "Frontend Type", "fr": "Type frontend"}, "frontendOptions": {"en": "Frontend Options", "fr": "Options frontend"}, "required": {"en": "Required", "fr": "Requis"}, "default": {"en": "Default", "fr": "Par défaut"}, "description": {"en": "Description", "fr": "Description"}, "validation": {"en": "Validation", "fr": "Validation"}, }, ) ``` ### Schritt 2: MethodBase erweitern **Datei**: `gateway/modules/workflows/methods/methodBase.py` - Neue `_actions` Dictionary Property - RBAC-Check Integration - Parameter-Validierung - Kompatibilität mit bestehendem System ### Schritt 3: Beispiel-Migration **Vorher** (methodOutlook.py): ```python @action async def readEmails(self, parameters: Dict[str, Any]) -> ActionResult: """ GENERAL: - Purpose: Read emails from Outlook mailbox Parameters: - connectionReference (str, required): Microsoft connection label. - query (str, optional): Search query for emails. - folder (str, optional): Folder name. - limit (int, optional): Maximum number of emails. Default: 50. """ # Implementation... ``` **Nachher** (methodOutlook.py): ```python def __init__(self, services): super().__init__(services) self.name = "outlook" self.description = "Handle Microsoft Outlook email operations" self._actions = { "readEmails": WorkflowActionDefinition( actionId="outlook.readEmails", description="Read emails from Outlook mailbox", parameters={ "connectionReference": WorkflowActionParameter( name="connectionReference", type="str", frontendType=FrontendType.USER_CONNECTION, # Custom type - automatisch API-Endpoint required=True, description="Microsoft connection label" ), "query": WorkflowActionParameter( name="query", type="str", frontendType=FrontendType.TEXT, required=False, description="Search query for emails" ), "folder": WorkflowActionParameter( name="folder", type="str", frontendType=FrontendType.SELECT, frontendOptions="outlook.folder", required=False, description="Folder name (e.g., 'Inbox', 'Drafts')" ), "limit": WorkflowActionParameter( name="limit", type="int", frontendType=FrontendType.NUMBER, required=False, default=50, description="Maximum number of emails to return", validation={"min": 1, "max": 1000} ) }, execute=self._executeReadEmails ) } async def _executeReadEmails(self, parameters: Dict[str, Any]) -> ActionResult: """Execute function - keine Parameter-Definition mehr hier""" # Implementation bleibt gleich... ``` ## RBAC-Integration ### Action-IDs Format Actions werden im RBAC-System als RESOURCE-Context Items behandelt: - **Format**: `{moduleName}.{actionName}` - **Beispiele**: - `outlook.readEmails` - `outlook.sendEmail` - `sharepoint.uploadDocument` - `ai.process` ### RBAC-Regeln für Actions ```json { "roleLabel": "user", "context": "RESOURCE", "item": "outlook.readEmails", "view": true } ``` ```json { "roleLabel": "admin", "context": "RESOURCE", "item": "outlook", "view": true } ``` **Hierarchie**: Spezifische Action-Regeln überschreiben generische Module-Regeln. ### Bootstrap: Default RBAC Rules für Actions In `interfaceBootstrap.py`: ```python def initRbacRules(db: DatabaseConnector) -> None: # ... existing rules ... # Action Rules (RESOURCE context) createActionRules(db) def createActionRules(db: DatabaseConnector): """Create default RBAC rules for workflow actions""" # SysAdmin: Access to all actions db.recordCreate(AccessRule( roleLabel="sysadmin", context=AccessRuleContext.RESOURCE, item=None, # All resources view=True )) # Admin: Access to all actions db.recordCreate(AccessRule( roleLabel="admin", context=AccessRuleContext.RESOURCE, item=None, view=True )) # User: Access to specific actions only userActions = [ "outlook.readEmails", "outlook.sendEmail", "sharepoint.readDocuments", "ai.process" ] for actionId in userActions: db.recordCreate(AccessRule( roleLabel="user", context=AccessRuleContext.RESOURCE, item=actionId, view=True )) # Viewer: Read-only actions viewerActions = [ "outlook.readEmails", "sharepoint.readDocuments" ] for actionId in viewerActions: db.recordCreate(AccessRule( roleLabel="viewer", context=AccessRuleContext.RESOURCE, item=actionId, view=True )) ``` ## Vorteile ### 1. Keine Duplikation - Parameter werden nur einmal definiert (in `WorkflowActionDefinition`) - Keine Docstring-Parsing mehr nötig - Type-Safety durch Pydantic Models ### 2. RBAC-Integration - Jede Action hat eine eindeutige ID für RBAC - Granulare Kontrolle pro Action möglich - Hierarchische Regeln (Module → Action) ### 3. UI-Rendering - Frontend-Typen explizit definiert - Options-Referenzen für dynamische Optionen - Validierung auf Backend-Ebene ### 4. Plug-and-Play - Actions bleiben als separate Method-Klassen - Einfache Erweiterung durch neue Method-Klassen - Kompatibilität mit bestehendem System ### 5. Type Safety - Pydantic Models für Validierung - Type-Hints für bessere IDE-Unterstützung - Runtime-Validierung ## Migration Timeline ### Phase 1: Foundation (Woche 1) - ✅ Datenmodelle erstellen (`datamodelWorkflowActions.py`) - ✅ MethodBase erweitern - ✅ RBAC-Integration in MethodBase ### Phase 2: Beispiel-Migration (Woche 2) - 📝 Ein Method-Beispiel migrieren (z.B. `methodAi.py`) - 📝 Tests schreiben - 📝 Dokumentation aktualisieren ### Phase 3: Vollständige Migration (Woche 3-4) - 📝 Alle Methods migrieren - 📝 RBAC-Regeln in Bootstrap erstellen - 📝 Frontend-Integration ### Phase 4: Testing & Cleanup (Woche 5) - 📝 Unit Tests - 📝 Integration Tests - 📝 Performance Tests - 📝 Alte Docstring-Parsing-Logik entfernen ## Offene Fragen 1. **Backward Compatibility**: Sollen alte Actions ohne `_actions` Dictionary weiterhin funktionieren? - **Antwort**: Ja, MethodBase prüft zuerst `_actions`, dann fallback auf `@action` Decorator 2. **Parameter-Validierung**: Soll Validierung strikt sein oder tolerant? - **Antwort**: Konfigurierbar pro Action 3. **Action-Discovery**: Sollen Actions zur Laufzeit registriert werden können? - **Antwort**: Ja, über `_registerActions()` Methode 4. **Frontend-Integration**: Wie werden Actions im Frontend angezeigt? - **Antwort**: API-Endpoint `/api/workflows/actions` liefert strukturierte Action-Definitionen ## API-Endpunkte ### GET /api/workflows/actions Liefert alle verfügbaren Actions für den aktuellen User (gefiltert nach RBAC): ```json { "actions": [ { "module": "outlook", "actionId": "outlook.readEmails", "name": "readEmails", "description": "Read emails from Outlook mailbox", "parameters": { "connectionReference": { "type": "str", "frontendType": "userConnection", "frontendOptions": "user.connection", # Automatisch für Custom-Types "required": true, "description": "Microsoft connection label" }, "documentList": { "type": "List[str]", "frontendType": "documentReference", "frontendOptions": "workflow.documentReference", # Automatisch für Custom-Types "required": false, "description": "Document list reference(s) from previous actions" }, ... } }, ... ] } ``` ### GET /api/workflows/actions/{module} Liefert Actions für ein spezifisches Modul. ### POST /api/workflows/actions/{module}/{action}/execute Führt eine Action aus (mit RBAC-Check). ## Custom Frontend Types für Actions ### Verfügbare Custom Types 1. **`FrontendType.USER_CONNECTION`** - **API-Endpoint**: `/api/options/user.connection` - **Beschreibung**: Zeigt alle aktiven Connections des aktuellen Users - **Verwendung**: Für Parameter wie `connectionReference` in Outlook/SharePoint Actions - **Beispiel**: ```python WorkflowActionParameter( name="connectionReference", type="str", frontendType=FrontendType.USER_CONNECTION, required=True, description="Microsoft connection label" ) ``` 2. **`FrontendType.DOCUMENT_REFERENCE`** - **API-Endpoint**: `/api/options/workflow.documentReference` (zu implementieren) - **Beschreibung**: Zeigt verfügbare Document-Referenzen aus dem aktuellen Workflow-Context - **Verwendung**: Für Parameter wie `documentList` in Actions, die auf vorherige Action-Ergebnisse verweisen - **Beispiel**: ```python WorkflowActionParameter( name="documentList", type="List[str]", frontendType=FrontendType.DOCUMENT_REFERENCE, required=False, description="Document list reference(s) from previous actions" ) ``` 3. **`FrontendType.WORKFLOW_ACTION`** - **API-Endpoint**: `/api/options/workflow.action` (zu implementieren) - **Beschreibung**: Zeigt verfügbare Actions aus dem Workflow-Context - **Verwendung**: Für Parameter, die auf andere Actions verweisen ### Custom Types erweitern Neue Custom-Types können über `frontendTypes.py` registriert werden: ```python from modules.shared.frontendTypes import FrontendType, registerCustomType # Neuer Custom-Type hinzufügen FrontendType.SHAREPOINT_FOLDER = "sharepointFolder" # Registrieren registerCustomType( frontendType=FrontendType.SHAREPOINT_FOLDER, optionsApiEndpoint="sharepoint.folder", description={ "en": "SharePoint Folder", "fr": "Dossier SharePoint", "de": "SharePoint-Ordner" } ) ``` ### Frontend-Integration Das Frontend muss: 1. Custom-Types erkennen (z.B. `frontendType === "userConnection"`) 2. Automatisch Options von `/api/options/{optionsName}` laden 3. Die Options als Select/Multiselect rendern **Beispiel Frontend-Logik**: ```typescript if (param.frontendType === 'userConnection') { // Automatisch Options von /api/options/user.connection laden const options = await fetch(`/api/options/${param.frontendOptions}`); // Als Select rendern } ```