# Gateway Development Framework: Connectors → Interfaces → Services → Workflows This document explains the gateway's code logic and development framework to build market customer journey features. It focuses on how connectors, interfaces, services, and workflows compose a standardized services landscape that can be consumed by routes, features, and agent models to perform tasks and actions. --- ## Purpose - **Unify external tools**: Combine many third‑party APIs and utilities behind a consistent interface. - **Standardize service design**: Model capabilities as reusable services with clear contracts. - **Enable workflow automation**: Let agent models orchestrate multi‑step tasks using the centralized services. - **Abstract complexity**: Hide implementation details behind clean, well-defined APIs. - **Enforce security and governance**: Apply consistent access control, audit trails, and data isolation across all layers. --- ## High‑Level Architecture The Gateway follows a layered architecture pattern with clear separation of concerns: 1. **Connectors**: Vendor-specific adapters for external systems (databases, APIs, cloud services) handling auth, transport, retries, and basic mapping. 2. **Interfaces**: Normalization layer exposing common contracts independent of any single vendor. Provides CRUD operations, access control, and data transformation. 3. **Services**: Business‑level capabilities built on interfaces, composed into feature‑ready functions. Orchestrate multiple interfaces and apply business rules. 4. **Service Center**: Central registry/factory (`Services` class) that instantiates and exposes services with consistent configuration, user context, and lifecycle management. 5. **Workflows & Methods**: Orchestration engine that calls services to perform tasks/actions. Methods provide extensible, plugin-like actions that workflows can invoke. **Data/control flow**: Client or Workflow → Service Center → Service → Interface → Connector → External Tool/Database --- ## Directory Overview (gateway) ``` gateway/ ├── modules/ │ ├── connectors/ # Vendor-specific adapters │ │ ├── connectorDbPostgre.py # PostgreSQL database │ │ ├── connectorDbJson.py # JSON file-based database │ │ ├── connectorVoiceGoogle.py # Google Cloud Speech services │ │ ├── connectorTicketsJira.py # JIRA integration │ │ └── connectorTicketsClickup.py # ClickUp integration │ │ │ ├── datamodels/ # Pydantic models defining data structures │ │ ├── datamodelRealEstate.py │ │ ├── datamodelChat.py │ │ ├── datamodelAi.py │ │ ├── datamodelUam.py # User & Mandate models │ │ └── ... │ │ │ ├── interfaces/ # Data access layer │ │ ├── interfaceDbRealEstateObjects.py # CRUD operations │ │ ├── interfaceDbRealEstateAccess.py # Access control │ │ ├── interfaceDbChatObjects.py │ │ ├── interfaceDbChatAccess.py │ │ ├── interfaceDbAppObjects.py │ │ ├── interfaceDbComponentObjects.py │ │ ├── interfaceAiObjects.py # AI operations │ │ ├── interfaceTicketObjects.py # Ticket systems │ │ └── interfaceVoiceObjects.py # Voice operations │ │ │ ├── services/ # Business-level capabilities │ │ ├── __init__.py # Services container (Service Center) │ │ ├── serviceAi/ # AI operations │ │ ├── serviceChat/ # Workflow & document management │ │ ├── serviceExtraction/ # Content extraction │ │ ├── serviceGeneration/ # Document generation │ │ ├── serviceNeutralization/ # Data anonymization │ │ ├── serviceSharepoint/ # SharePoint integration │ │ ├── serviceTicket/ # Ticket system integration │ │ └── serviceUtils/ # Common utilities │ │ │ ├── workflows/ # Orchestration engine │ │ ├── workflowManager.py # Main orchestration controller │ │ ├── processing/ # Processing logic │ │ │ ├── workflowProcessor.py │ │ │ ├── core/ # Core components │ │ │ ├── modes/ # Execution modes │ │ │ └── shared/ # Shared utilities │ │ └── methods/ # Extensible action methods │ │ ├── methodBase.py │ │ ├── methodAi.py │ │ └── ... │ │ │ ├── routes/ # HTTP endpoints exposing capabilities │ │ ├── routeChatPlayground.py │ │ ├── routeWorkflows.py │ │ └── ... │ │ │ ├── features/ # Domain-specific business logic │ │ └── mainChatPlayground.py │ │ │ ├── security/ # Authentication, authorization, token management │ │ ├── auth.py │ │ ├── jwtService.py │ │ ├── tokenManager.py │ │ └── ... │ │ │ └── shared/ # Cross-cutting utilities │ ├── config.py │ ├── logging.py │ └── ... ``` --- ## 1) Connectors: Many External Tools, One Adapter Shape **Role**: Provide the lowest-level integration with external systems (databases, APIs, SDKs, auth, retries). **Responsibility**: - **Authentication and credential handling**: Manage API keys, OAuth tokens, database credentials - **Transport**: HTTP/WebSocket clients, connection pooling, retry logic, circuit breaking - **Response normalization**: Map vendor-specific responses to minimal internal shapes - **Error handling**: Transform external errors into consistent internal error structures **Output**: Vendor‑flavored data mapped to connector models, not directly used by workflows or services. **Key Guidelines**: - Keep connectors vendor‑specific and replaceable (e.g., `connectorDbPostgre.py` vs `connectorDbJson.py`) - No business logic; only integration concerns and basic mapping - Use duck typing (no formal interfaces) for flexibility - Handle retries, timeouts, and connection management internally - Return structured error responses, never raise exceptions to application layer **Example Connector Types**: - **Database Connectors**: PostgreSQL (`connectorDbPostgre.py`), JSON file-based (`connectorDbJson.py`) - **Voice Connectors**: Google Cloud Speech (`connectorVoiceGoogle.py`) - **Ticket Connectors**: JIRA (`connectorTicketsJira.py`), ClickUp (`connectorTicketsClickup.py`) --- ## 2) Interfaces: Stable Contracts Over Connectors **Role**: Define capability‑oriented contracts (e.g., `ChatObjects`, `AppObjects`, `AiObjects`) and map connector outputs into interface DTOs. **Responsibility**: - **Normalize differing vendors**: Convert vendor-specific data into consistent domain objects - **Hide vendor peculiarities**: Abstract away implementation details behind clean, typed DTOs - **Provide CRUD operations**: Create, Read, Update, Delete methods for domain entities - **Enforce access control**: Apply user privilege checks and mandate-based filtering - **Offer capability toggles**: Sensible defaults and configuration options **Output**: Clean, stable methods used by services (e.g., `getWorkflow()`, `createMessage()`, `call()`). **Interface Structure**: Interfaces are split into two file types: - **Objects Files** (`interface*Objects.py`): CRUD operations and business logic - **Access Files** (`interface*Access.py`): Permission checking and data filtering **Key Guidelines**: - Prefer capability names over vendor names (e.g., `ChatObjects` not `PostgreChatObjects`) - Keep interfaces small, cohesive, and testable with mocks - Always require user context for database interfaces (enables access control) - Use Pydantic models (datamodels) for type safety - Apply Unified Access Management (UAM) for all database queries **Example Interface Types**: - **Database Interfaces**: `interfaceDbChatObjects`, `interfaceDbAppObjects`, `interfaceDbRealEstateObjects` - **External System Interfaces**: `interfaceAiObjects`, `interfaceTicketObjects`, `interfaceVoiceObjects` --- ## 3) Services: Business‑Level Capabilities **Role**: Compose one or more interfaces to implement feature‑ready operations (e.g., "answer question with web grounding", "extract and analyze documents"). **Responsibility**: - **Apply business rules**: Validation, guardrails, transformations, data enrichment - **Orchestrate multiple interfaces**: Coordinate between interfaces and other services - **Emit domain events/metrics**: Track operations, costs, performance - **Enforce security policies**: Apply additional security checks beyond interface layer - **Handle complex workflows**: Multi-step operations with error recovery **Output**: High‑level operations that workflows and routes can call atomically. **Service Container Pattern**: All services are initialized through the `Services` container. Initialize with user context using `Services(user=current_user, workflow=current_workflow)`, then access services via `services.ai.callAiDocuments()`, `services.chat.storeMessageWithDocuments()`, etc. **Key Guidelines**: - Services depend on interfaces, not connectors directly - Keep input/output DTOs explicit and versioned when necessary - Services can call other services via `self.services` - Use `PublicService` wrapper to expose only public methods - Keep services stateless (no session state, use database for persistence) **Core Services**: - **AI Service**: AI model operations, planning, document processing - **Chat Service**: Workflow management, message handling, document resolution - **Extraction Service**: Multi-format document extraction and processing - **Generation Service**: Document rendering in various formats - **Neutralization Service**: Data anonymization for GDPR compliance - **SharePoint Service**: SharePoint integration - **Ticket Service**: Ticket system integration (Jira, ClickUp) - **Utils Service**: Common utilities (config, events, time, debug) --- ## 4) Centralized Service Center **Role**: A registry/factory (`Services` class) that instantiates and exposes services with consistent configuration and lifecycle. **Responsibility**: - **Discoverability**: List/get services by capability key (e.g., `services.ai`, `services.chat`) - **Configuration**: Environment, credentials, routing to specific vendors - **Cross‑cutting**: User context, workflow context, interface access - **Lifecycle management**: Initialize services with proper dependencies - **Access control**: Provide user context to all services and interfaces **Usage Pattern**: 1. Route receives request with authenticated user (via `getCurrentUser` dependency) 2. Create Services container with user context using `Services(user=currentUser)` 3. Call service method with typed input (e.g., `services.ai.callAiDocuments()`) 4. Receive typed output **Service Center Structure**: The `Services` class initializes with user and optional workflow context. It initializes interfaces via `getChatInterface()`, `getAppInterface()`, `getComponentInterface()`, and wraps all services in `PublicService` wrappers (e.g., `PublicService(AiService(self))`). **Key Features**: - **User Context**: Every service has access to `self.services.user` for access control - **Workflow Context**: Services can access `self.services.workflow` for workflow-aware operations - **Interface Access**: Services access interfaces via `self.services.interfaceDbChat`, etc. - **Service Composition**: Services call other services via `self.services.otherService.method()` --- ## 5) Workflows & Agent Models **Role**: Coordinate tasks and actions by invoking services in sequence/branches/loops. **Responsibility**: - **Maintain execution state**: Track workflow progress, round/task/action counters - **Choose actions**: Use agent models (AI) or predefined plans to determine next steps - **Handle retries/compensation**: Retry failed tasks with improvements, rollback on failure - **Record audit logs**: Track all workflow steps, decisions, and outcomes - **Manage document flow**: Resolve document references, track document lineage **Typical Pattern**: 1. **Ingest user intent/context**: Analyze user input, extract documents, detect language 2. **Plan next action**: Use AI to generate task plan or follow predefined JSON plan 3. **Call services via Service Center**: Invoke services to perform operations 4. **Persist outputs**: Store results, update state, decide next step 5. **Generate feedback**: Create completion messages, summarize results **Workflow Modes**: - **Actionplan Mode**: Batch planning with quality review and intelligent retry - **Dynamic Mode**: Iterative, just-in-time action generation - **Automation Mode**: Predefined JSON-based deterministic execution **Method System**: Workflows invoke actions through an extensible method system: - **Methods**: Plugin-like classes that expose actions via `@action` decorator - **Actions**: Async methods that perform specific operations (e.g., `methodAi.process()`, `methodSharepoint.search()`) - **Automatic Discovery**: Methods are discovered at runtime via introspection - **Signature Generation**: Action signatures are generated for AI prompt generation --- ## Standardized Interface Example (Actual Implementation) Interfaces like `ChatObjects` provide methods such as `getWorkflow()` and `createMessage()`. The `AiObjects` interface provides `call()` for AI model operations. Vendors like OpenAI/Anthropic implement `AiObjects` through connectors; database connectors implement `ChatObjects`. Services compose these interfaces. --- ## Example Service Composition (Actual Implementation) The `AiService.callAiDocuments()` method demonstrates service composition: **Steps**: 1. `ExtractionService.extractContent()` → extracts content from documents 2. `AiObjects.call()` → processes with AI model 3. `ChatService.storeWorkflowStat()` → records statistics **Outputs**: AI-generated content, processing statistics, cost tracking --- ## Adding a New Capability ### Step 1: Create Connector (if needed) Add vendor adapter in `modules/connectors/` (e.g., `connectorNewVendor.py`). The connector class should initialize with configuration, handle API calls, and return structured responses with `{"success": True/False, "data": ...}` format. ### Step 2: Create Interface Implement capability contract in `modules/interfaces/` (e.g., `interfaceNewCapabilityObjects.py`). The interface class should initialize with user context, use the connector, and provide normalized methods like `performOperation()` that return domain objects. ### Step 3: Create Service Compose the interface in `modules/services/serviceNewCapability/mainServiceNewCapability.py`. The service class should initialize with the services container, access the interface, and provide business-level methods like `performBusinessOperation()` that apply validation, call the interface, and enrich results. ### Step 4: Register in Service Center Wire into `Services` class in `modules/services/__init__.py`. Import the service class and wrap it in `PublicService()` (e.g., `self.newCapability = PublicService(NewCapabilityService(self))`). ### Step 5: Expose via Route (if needed) Add HTTP endpoint in `modules/routes/routeNewCapability.py`. Create a route handler that uses `getCurrentUser` dependency, creates a `Services` instance, calls the service method, and returns the result. ### Step 6: Use in Workflows (if needed) Create method action in `modules/workflows/methods/methodNewCapability.py`. Inherit from `MethodBase`, use the `@action` decorator on async methods, and return `ActionResult` objects with success status and documents. --- ## Adding a New Database Domain Adding a completely new database domain (like Real Estate, Projects, Inventory) requires creating datamodels, database interfaces, and access control. This section covers creating a new domain from scratch. ### Overview A new database domain consists of: 1. **Datamodels**: Pydantic models defining data structures 2. **Database Interface Objects**: CRUD operations for domain entities 3. **Database Interface Access**: Access control and permission checking 4. **Database Configuration**: Connection settings for the new database 5. **Service Integration**: Optional service layer for business logic ### Step 1: Create Datamodels Create a new datamodel file in `modules/datamodels/datamodel[Domain].py` (e.g., `datamodelRealEstate.py`, `datamodelProject.py`). **Structure**: - Define Pydantic models inheriting from `BaseModel` - Include enums for status fields and categories - Add helper models for complex nested structures - Use Field() with frontend metadata for UI generation - Include standard fields: `id`, `mandateId`, `_createdBy`, `_createdAt`, `_modifiedBy`, `_modifiedAt` **Example Structure**: ``` datamodel[Domain].py ├── Enums (StatusEnum, CategoryEnum, etc.) ├── Helper Models (GeoPoint, Address, etc.) ├── Main Entity Models │ ├── Entity1 (id, mandateId, fields, timestamps) │ ├── Entity2 (id, mandateId, fields, timestamps) │ └── Entity3 (id, mandateId, fields, timestamps) └── Relationship Models (if needed) ``` **Key Requirements**: - All main entities must have `id: str` (UUID) - All main entities must have `mandateId: str` for multi-tenant isolation - Include audit fields: `_createdBy`, `_createdAt`, `_modifiedBy`, `_modifiedAt` - Use `Field()` with `frontend_type`, `frontend_readonly`, `frontend_required` for UI metadata - Define relationships using ForwardRef if models reference each other **Example**: ``` from pydantic import BaseModel, Field from enum import Enum import uuid class StatusEnum(str, Enum): ACTIVE = "active" INACTIVE = "inactive" ARCHIVED = "archived" class Project(BaseModel): id: str = Field(default_factory=lambda: str(uuid.uuid4())) mandateId: str name: str = Field(..., frontend_type="text", frontend_required=True) status: StatusEnum = Field(..., frontend_type="select") description: Optional[str] = Field(None, frontend_type="textarea") _createdBy: Optional[str] = None _createdAt: Optional[int] = None _modifiedBy: Optional[str] = None _modifiedAt: Optional[int] = None ``` ### Step 2: Create Database Interface Objects Create `modules/interfaces/interfaceDb[Domain]Objects.py` (e.g., `interfaceDbRealEstateObjects.py`). **Structure**: - `[Domain]Objects` class that initializes database connector - CRUD methods for each entity: `create[Entity]()`, `get[Entity]()`, `get[Entities]()`, `update[Entity]()`, `delete[Entity]()` - Query execution method: `executeQuery()` for custom SQL queries - User context management: `setUserContext()` **Key Components**: **1. Database Initialization**: ``` def _initializeDatabase(self): dbHost = APP_CONFIG.get("DB_[DOMAIN]_HOST", "localhost") dbDatabase = APP_CONFIG.get("DB_[DOMAIN]_DATABASE", "poweron_[domain]") dbUser = APP_CONFIG.get("DB_[DOMAIN]_USER") dbPassword = APP_CONFIG.get("DB_[DOMAIN]_PASSWORD_SECRET") dbPort = int(APP_CONFIG.get("DB_[DOMAIN]_PORT", 5432)) self.db = DatabaseConnector( dbHost=dbHost, dbDatabase=dbDatabase, dbUser=dbUser, dbPassword=dbPassword, dbPort=dbPort, userId=self.userId if self.userId else None, ) self.db.initDbSystem() ``` **2. CRUD Pattern**: ``` def create[Entity](self, entity: [Entity]) -> [Entity]: # Ensure mandateId is set if not entity.mandateId: entity.mandateId = self.mandateId # Apply access control self.access.uam([Entity], []) # Save to database self.db.recordCreate([Entity], entity.model_dump()) return entity def get[Entity](self, entityId: str) -> Optional[[Entity]]: records = self.db.getRecordset( [Entity], recordFilter={"id": entityId} ) if not records: return None # Apply access control filtered = self.access.uam([Entity], records) if not filtered: return None return [Entity](**filtered[0]) def get[Entities](self, filters: Optional[Dict] = None) -> List[[Entity]]: records = self.db.getRecordset([Entity], recordFilter=filters or {}) filtered = self.access.uam([Entity], records) return [Entity](**r) for r in filtered] def update[Entity](self, entityId: str, updates: Dict) -> Optional[[Entity]]: # Check access control self.access.canModify([Entity], entityId) # Update in database self.db.recordUpdate([Entity], entityId, updates) # Return updated entity return self.get[Entity](entityId) def delete[Entity](self, entityId: str) -> bool: # Check access control self.access.canModify([Entity], entityId) # Delete from database self.db.recordDelete([Entity], entityId) return True ``` **3. Singleton Factory Pattern**: ``` _[domain]Interfaces = {} def get[Domain]Interface(currentUser: Optional[User] = None) -> [Domain]Objects: """Factory function to get or create interface instance.""" userId = currentUser.id if currentUser else None if userId not in _[domain]Interfaces: _[domain]Interfaces[userId] = [Domain]Objects(currentUser) return _[domain]Interfaces[userId] ``` ### Step 3: Create Database Interface Access Create `modules/interfaces/interfaceDb[Domain]Access.py` (e.g., `interfaceDbRealEstateAccess.py`). **Structure**: - `[Domain]Access` class that handles permission checking - `uam()` method for Unified Access Management (filtering and flagging) - `canModify()` method for write permission checking **Key Components**: **1. Access Control Class**: ``` class [Domain]Access: def __init__(self, currentUser: User, db: DatabaseConnector): self.currentUser = currentUser self.userId = currentUser.id self.mandateId = currentUser.mandateId self.userRole = currentUser.role self.db = db def uam(self, model: Type[BaseModel], records: List[Dict]) -> List[Dict]: """Unified Access Management: Filter records and add access flags.""" if self.userRole == "SYSADMIN": # SYSADMIN sees all records filtered = records elif self.userRole == "ADMIN": # ADMIN sees all records in their mandate filtered = [r for r in records if r.get("mandateId") == self.mandateId] else: # USER sees only their own records filtered = [r for r in records if r.get("_createdBy") == self.userId] # Add access flags for record in filtered: record["_hideView"] = False record["_hideEdit"] = not self.canModify(model, record.get("id")) record["_hideDelete"] = not self.canModify(model, record.get("id")) return filtered def canModify(self, model: Type[BaseModel], recordId: str) -> bool: """Check if user can modify a record.""" if self.userRole == "SYSADMIN": return True # Get record to check ownership records = self.db.getRecordset(model, recordFilter={"id": recordId}) if not records: return False record = records[0] if self.userRole == "ADMIN": # ADMIN can modify records in their mandate return record.get("mandateId") == self.mandateId else: # USER can only modify their own records return record.get("_createdBy") == self.userId ``` **2. Import in Objects File**: ``` from modules.interfaces.interfaceDb[Domain]Access import [Domain]Access ``` ### Step 4: Configure Database Connection Add database configuration to `config.ini`: ``` [Database] DB_[DOMAIN]_HOST=localhost DB_[DOMAIN]_DATABASE=poweron_[domain] DB_[DOMAIN]_USER=postgres DB_[DOMAIN]_PASSWORD_SECRET=your_password DB_[DOMAIN]_PORT=5432 ``` **Database Creation**: - The `DatabaseConnector.initDbSystem()` method automatically creates the database if it doesn't exist - Tables are created on-demand when first accessed via `_ensureTableExists()` - No manual database schema creation needed ### Step 5: Register Interface in Services (Optional) If you need business logic, create a service that uses the interface: **Create Service** (`modules/services/service[Domain]/mainService[Domain].py`): ``` class [Domain]Service: def __init__(self, services: 'Services'): self.services = services def get[Domain]Interface(self) -> [Domain]Objects: """Get interface instance with current user context.""" return get[Domain]Interface(self.services.workflow.currentUser) def performBusinessOperation(self, ...): """Business-level method that uses interface.""" interface = self.get[Domain]Interface() # Apply business logic # Call interface methods # Return enriched results ``` **Register in Service Center** (`modules/services/__init__.py`): ``` from modules.services.service[Domain].mainService[Domain] import [Domain]Service class Services: def __init__(self, ...): ... self.[domain] = PublicService([Domain]Service(self)) ``` ### Step 6: Create Routes (Optional) If you need HTTP endpoints, create `modules/routes/route[Domain].py`: ``` from fastapi import APIRouter, Depends from modules.features.shared.dependencies import getCurrentUser from modules.datamodels.datamodelUam import User from modules.services import Services from modules.interfaces.interfaceDb[Domain]Objects import get[Domain]Interface router = APIRouter() @router.get("/[domain]/entities") async def getEntities( currentUser: User = Depends(getCurrentUser) ): """Get all entities.""" interface = get[Domain]Interface(currentUser) entities = interface.get[Entities]() return {"success": True, "data": [e.model_dump() for e in entities]} ``` ### Step 7: Use in Workflows (Optional) If you need workflow actions, create `modules/workflows/methods/method[Domain].py`: ``` from modules.workflows.methods.methodBase import MethodBase from modules.workflows.methods.methodBase import action from modules.workflows.methods.methodBase import ActionResult class Method[Domain](MethodBase): name = "[domain]" description = "[Domain] operations" def __init__(self, services): super().__init__(services) @action async def performOperation(self, parameters: Dict[str, Any]) -> ActionResult: """Perform domain operation.""" interface = get[Domain]Interface(self.services.workflow.currentUser) # Perform operation # Return ActionResult ``` ### Complete Example: Real Estate Domain **Datamodels** (`datamodelRealEstate.py`): - `Projekt`, `Parzelle`, `Dokument`, `Kanton`, `Gemeinde`, `Land` - `GeoPunkt`, `GeoPolylinie` (geographic data) - `Kontext` (context/notes) - Enums: `StatusProzess`, `DokumentTyp`, `GeoTag` **Interface Objects** (`interfaceDbRealEstateObjects.py`): - `RealEstateObjects` class - CRUD methods for all entities - `executeQuery()` for custom SQL - Database initialization with `DB_REALESTATE_*` config **Interface Access** (`interfaceDbRealEstateAccess.py`): - `RealEstateAccess` class - `uam()` method for filtering - `canModify()` method for permissions **Configuration**: ``` DB_REALESTATE_HOST=localhost DB_REALESTATE_DATABASE=poweron_realestate DB_REALESTATE_USER=postgres DB_REALESTATE_PASSWORD_SECRET=... DB_REALESTATE_PORT=5432 ``` ### Best Practices **1. Naming Conventions**: - Datamodel file: `datamodel[Domain].py` (PascalCase domain name) - Interface Objects: `interfaceDb[Domain]Objects.py` - Interface Access: `interfaceDb[Domain]Access.py` - Database config: `DB_[DOMAIN]_*` (uppercase with underscores) **2. Mandate Isolation**: - Always set `mandateId` on create operations - Filter by `mandateId` in access control - Never expose data across mandates **3. Access Control**: - Always call `self.access.uam()` before returning records - Always call `self.access.canModify()` before write operations - Respect role hierarchy: SYSADMIN > ADMIN > USER **4. Error Handling**: - Validate user context before operations - Handle missing records gracefully (return None, not raise) - Log errors with context (user ID, mandate ID, operation) **5. Database Management**: - Let `DatabaseConnector` handle table creation automatically - Use `_ensureTableExists()` for supporting tables with foreign keys - Don't manually create database schemas **6. Testing**: - Test CRUD operations with different user roles - Test mandate isolation (users can't see other mandates' data) - Test access control (users can't modify others' records) ### Common Patterns **Pattern 1: Simple Domain (Single Entity)** - One main entity model - Basic CRUD operations - Standard access control **Pattern 2: Hierarchical Domain (Parent-Child)** - Multiple related entities - Foreign key relationships - Cascade operations (delete children when parent deleted) **Pattern 3: Complex Domain (Multiple Entities + Relationships)** - Multiple entities with relationships - Supporting tables (lookup tables, reference data) - Custom query methods for complex operations --- ## Security & Governance ### Access Control - **RBAC**: Role-based access control enforced at Interface layer (`interface*Access.py`) - **SYSADMIN**: Full system access, all mandates - **ADMIN**: Full access within mandate - **USER**: Access to own records only - **UAM**: Unified Access Management filters recordsets by privilege and adds access flags (`_hideView`, `_hideEdit`, `_hideDelete`) ### Secrets Management - **Centralized Configuration**: Credentials stored in `config.ini` with encryption - **Interface-Level Access**: Connectors receive credentials through interfaces, not directly - **No Leakage**: Credentials never exposed to workflows or services ### Audit - **Automatic Tracking**: All database operations include `_createdBy`, `_modifiedBy`, `_createdAt`, `_modifiedAt` - **Workflow Logging**: Workflow steps logged via `ChatService.storeLog()` - **Security Events**: Authentication events logged via `auditLogger.logSecurityEvent()` ### Quotas - **Rate Limiting**: Applied at route level using `slowapi.Limiter` - **Token Refresh Limits**: OAuth token refresh limited to 3 attempts per hour per connection - **Cost Tracking**: AI operations track costs via `ChatService.storeWorkflowStat()` --- ## Observability ### Structured Logging - **Layer-Specific Loggers**: Each layer uses module-specific loggers (e.g., `logging.getLogger("modules.services.serviceAi")`) - **Context Information**: Logs include user ID, workflow ID, operation context - **Error Details**: Exceptions logged with full stack traces and context ### Tracing - **Operation IDs**: Long-running operations use unique operation IDs for tracking - **Progress Logging**: `ChatService.progressLogStart()`, `progressLogUpdate()`, `progressLogFinish()` - **Workflow State**: Workflow state persisted to database for debugging ### Metrics - **Per-Capability Tracking**: Services track operation counts, costs, processing time - **Workflow Statistics**: `ChatStat` records track bytes sent/received, error counts, prices - **Performance Monitoring**: Processing time tracked for all AI calls and service operations --- ## Minimal Request Lifecycle ```mermaid sequenceDiagram participant Client participant Route participant Services as Service Center participant Service participant Interface participant Connector participant External as External System/DB Client->>Route: HTTP Request Route->>Route: Authenticate (getCurrentUser) Route->>Services: Create(user, workflow) Services->>Services: Initialize interfaces Services->>Services: Initialize services Services-->>Route: services instance Route->>Service: services.capability.operation() Service->>Interface: interface.method(params) Interface->>Interface: Apply access control (UAM) Interface->>Connector: connector.operation(params) Connector->>External: API call / DB query External-->>Connector: Response Connector-->>Interface: Normalized data Interface-->>Service: Domain object Service-->>Services: Business result Services-->>Route: Result Route-->>Client: HTTP Response ``` **Steps**: 1. Route receives request or workflow triggers an action 2. Service Center resolves service instance and validates user context 3. Service executes using interfaces; interfaces call connectors 4. Results propagate back; logs/metrics recorded; workflow advances state --- ## Benefits - **Replace vendors without breaking services**: Interfaces shield changes (e.g., swap PostgreSQL for JSON connector) - **Accelerate feature delivery**: Services are reusable building blocks - **Improve reliability and security**: Centralized policies and observability - **Empower workflows/agents**: Perform complex tasks with simple, typed calls - **Type safety**: Pydantic models ensure data consistency - **Testability**: Clear boundaries enable mocking and unit testing - **Maintainability**: Separation of concerns makes code easier to understand and modify --- ## Quick Map to Code (for orientation) - `gateway/modules/connectors/` → Vendor adapters (e.g., `connectorDbPostgre.py`, `connectorVoiceGoogle.py`) - `gateway/modules/interfaces/` → Capability contracts (e.g., `interfaceDbChatObjects.py`, `interfaceAiObjects.py`) - `gateway/modules/services/` → Composed capabilities (e.g., `serviceAi/mainServiceAi.py`, `serviceChat/mainServiceChat.py`) - `gateway/modules/workflows/` → Orchestrations/agents (e.g., `workflowManager.py`, `methods/methodAi.py`) - `gateway/modules/routes/` → HTTP endpoints (e.g., `routeChatPlayground.py`, `routeWorkflows.py`) This framework is the backbone for market customer journey features: build once as services, reuse everywhere in workflows. --- ## Visuals ### Layered Architecture ```mermaid flowchart TB subgraph ClientOrWorkflow[Client / Workflow Engine] C[Feature or Agent Task] end subgraph ServiceCenter[Service Center] SC[Services Container\nUser Context, Interfaces, Services] end subgraph Services[Services] S1[AI Service] S2[Chat Service] S3[Extraction Service] S4[Generation Service] end subgraph Interfaces[Interfaces] I1[ChatObjects] I2[AppObjects] I3[AiObjects] I4[ComponentObjects] end subgraph Connectors[Connectors] K1[PostgreSQL Connector] K2[JSON Connector] K3[Google Speech Connector] K4[AI Provider Connectors] end subgraph External[External Systems] E1[(PostgreSQL Database)] E2[Google Cloud APIs] E3[AI APIs\nOpenAI, Anthropic] end C --> SC --> S1 & S2 & S3 & S4 S1 --> I3 S2 --> I1 S3 --> I4 S4 --> I1 & I4 I1 --> K1 I2 --> K1 I3 --> K4 I4 --> K1 K1 --> E1 K2 --> E1 K3 --> E2 K4 --> E3 ``` ### Request / Action Sequence ```mermaid sequenceDiagram participant Client as Client / Workflow participant SC as Service Center participant S as Service participant I as Interface participant AC as Access Control participant K as Connector participant EXT as External Tool/DB Client->>SC: Request capability (e.g., services.ai.callAiDocuments) SC->>SC: Initialize with user context SC->>S: Get service instance S->>I: Call normalized method (e.g., aiObjects.call) I->>AC: Check permissions (UAM) AC-->>I: Permission granted I->>K: Prepare vendor-specific request K->>EXT: API/DB call (auth, retries) EXT-->>K: Response K-->>I: Map to normalized DTO I-->>S: Return normalized result S->>S: Apply business logic S-->>SC: Business output (validated, enriched) SC-->>Client: Typed response, telemetry recorded ``` ### Service Center Components ```mermaid graph LR subgraph SC[Service Center - Services Class] REG[Service Registry] CTX[User Context] WF[Workflow Context] INT[Interface Access] FAC[Service Factory] end REG --> FAC CTX --> FAC WF --> FAC INT --> FAC FAC -->|builds| Svc[(Service Instances)] subgraph Layers[Below Services] IF[Interfaces] CON[Connectors] end Svc --> IF --> CON subgraph Services[Services] AI[AI Service] Chat[Chat Service] Extract[Extraction Service] Gen[Generation Service] end Svc --> AI & Chat & Extract & Gen ``` ### Workflow State Machine (Conceptual) ```mermaid stateDiagram-v2 [*] --> Plan Plan: Decide next action (AI or rules) Plan --> CallService: needs external capability Plan --> Done: no more steps CallService: Invoke via Service Center CallService --> HandleResult HandleResult: Persist, evaluate, log HandleResult --> Plan: more work HandleResult --> Done: goal achieved Done --> [*] ``` ### Interface Access Control Flow ```mermaid sequenceDiagram participant Service participant Interface as Interface Objects participant Access as Access Control participant Connector participant DB as Database Service->>Interface: CRUD Operation Interface->>Access: Check permissions (uam) Access->>Access: Check user privilege Access->>Access: Filter by mandateId Access->>Access: Check ownership (_createdBy) Access->>Access: Add access flags Access-->>Interface: Filtered data + flags Interface->>Connector: Execute query Connector->>DB: SQL Query DB-->>Connector: Results Connector-->>Interface: Raw data Interface->>Interface: Transform to datamodel Interface-->>Service: Domain objects with access flags ``` --- ## Development Best Practices ### 1. Always Use Service Center ✅ **GOOD**: Use Service Center via `Services(user=current_user)` and call `services.ai.callAiDocuments()`, `services.chat.storeMessageWithDocuments()`, etc. ❌ **BAD**: Direct interface access bypasses the service layer (e.g., calling `getChatInterface(user).getWorkflow()` directly). ### 2. Keep Services Stateless ✅ **GOOD**: Stateless services use the database for persistence (e.g., `self.services.interfaceDbApp.getCache()`). ❌ **BAD**: Stateful services store data in instance variables (e.g., `self.cache = {}`). ### 3. Use Datamodels for Type Safety ✅ **GOOD**: Use Pydantic models like `ChatWorkflow`, `ChatMessage` from `modules.datamodels.datamodelChat`. Create instances with `ChatWorkflow(**data)` and return typed results. ❌ **BAD**: Use raw dictionaries without type safety. ### 4. Apply Access Control ✅ **GOOD**: Interfaces apply UAM automatically (e.g., `self.interfaceDbChat.getWorkflows()` filters by user privilege). ❌ **BAD**: Bypass access control by calling connectors directly (e.g., `self.connector.getRecordset()` has no filtering). ### 5. Handle Errors Gracefully ✅ **GOOD**: Return structured errors with `{"success": True/False, "data": ..., "error": ...}` format. Log exceptions with context. ❌ **BAD**: Let exceptions propagate to callers without handling. --- ## Workflow Engineering Workflow engineering is the process of designing, building, and maintaining workflows that orchestrate multi-step tasks using the gateway's service layer. Workflows transform user requests into structured execution plans, coordinate action execution, and manage state throughout the process. ### Understanding Workflow Architecture Workflows operate at the highest level of the gateway architecture, orchestrating services to accomplish complex goals. They provide: - **Intelligent Planning**: AI-powered task breakdown and action generation - **State Management**: Track progress, maintain context, and handle errors - **Document Flow**: Manage document references and lineage throughout execution - **Adaptive Execution**: Retry failed tasks, learn from results, improve over time - **Multi-Mode Support**: Different execution strategies for different use cases ### Workflow Components **WorkflowManager**: Main orchestration controller that manages workflow lifecycle (`workflowStart()`, `workflowStop()`, `_workflowProcess()`) **WorkflowProcessor**: Delegates to mode-specific implementations (Actionplan, Dynamic, Automation) **TaskPlanner**: Generates structured task plans from user input using AI **ActionExecutor**: Executes individual actions by invoking methods from the global methods catalog **MessageCreator**: Creates and persists workflow messages with document associations **Method System**: Extensible plugin framework for defining reusable actions ### Workflow Execution Pipeline Every workflow follows a four-stage pipeline: 1. **Send First Message**: Analyze user intent, extract documents, detect language, normalize request 2. **Plan Tasks**: Generate structured task plan with objectives, success criteria, and dependencies 3. **Execute Tasks**: Execute each task sequentially, maintaining context between tasks 4. **Process Results**: Generate feedback, create completion message, update workflow status --- ## Workflow Modes The gateway supports three distinct workflow modes, each optimized for different use cases: ### Actionplan Mode **Strategy**: Batch planning with quality review and intelligent retry **Characteristics**: - Plans all actions upfront before execution begins - Reviews results against success criteria after execution - Retries failed tasks up to 3 times with cumulative improvements - Best for complex multi-step workflows with specific requirements **Use Cases**: Data processing pipelines, document analysis with requirements, complex transformations **Execution Flow**: 1. Generate complete action plan for entire task 2. Execute all actions sequentially 3. Review results against success criteria 4. Retry with improvements if criteria not met 5. Return final result ### Dynamic Mode **Strategy**: Iterative, just-in-time action generation **Characteristics**: - Generates one action at a time based on current state - Each action's result influences the next action - Workflow path emerges organically based on findings - Limited by `maxSteps` (default: 5) to prevent infinite loops **Use Cases**: Research workflows, exploratory data analysis, iterative problem solving, uncertain paths **Execution Flow**: 1. Generate single next action based on current context 2. Execute action immediately 3. Evaluate if task objective is met 4. Continue if objective not met and under max steps 5. Return result when objective met or max steps reached ### Automation Mode **Strategy**: Predefined JSON-based deterministic execution **Characteristics**: - No AI planning or action generation - User provides complete task and action plan in JSON format - Deterministic execution (same input always produces same sequence) - Fastest execution time (no planning overhead) **Use Cases**: Repeated workflows, automated jobs, batch processing, template execution, routine operations **Execution Flow**: 1. Parse predefined JSON plan from user input 2. Execute actions in order specified in JSON 3. Collect results without review 4. Return execution summary --- ## Building New Workflows New workflows are typically built using Actionplan or Dynamic modes, where AI generates the execution plan based on user input. This section covers how to create workflows that adapt to user requests. ### Starting a New Workflow **Entry Point**: `WorkflowManager.workflowStart()` **Required Parameters**: - `userInput`: UserInputRequest containing prompt, file IDs, and language - `workflowMode`: WorkflowModeEnum (WORKFLOW_ACTIONPLAN, WORKFLOW_DYNAMIC, or WORKFLOW_AUTOMATION) - `workflowId`: Optional ID to continue existing workflow **Process**: 1. Create or load `ChatWorkflow` record in database 2. Initialize workflow state (status="running", currentRound=1, counters=0) 3. Discover and update method instances with current services 4. Launch asynchronous processing pipeline 5. Return workflow object immediately (non-blocking) **Example Flow**: ``` Route → chatStart() → WorkflowManager.workflowStart() → _workflowProcess() ``` ### Workflow Input Processing The first stage (`_sendFirstMessage()`) processes user input: **Intent Analysis**: AI analyzes user input to extract: - Detected language (ISO 639-1 code) - Normalized request (full, explicit restatement) - Core intent (primary goals and requirements) - Bulky context items (large data blocks extracted as separate documents) **Document Management**: - Processes user-uploaded files (converts file IDs to ChatDocument objects) - Extracts large content blocks from prompt (code snippets, tables, lists) - Creates document records in component database - Applies neutralization if enabled in user settings - Associates documents with labels (e.g., "round1_usercontext") **Message Creation**: Creates first message with role="user", status="first", and all associated documents ### Task Planning The second stage (`_planTasks()`) generates structured task plans: **Planning Process**: 1. Uses cleaned user intent from previous stage 2. Calls `WorkflowProcessor.generateTaskPlan()` which delegates to mode-specific implementation 3. For Actionplan/Dynamic modes: Uses `TaskPlanner.generateTaskPlan()` with AI 4. For Automation mode: Parses predefined JSON plan from user input **TaskPlan Structure**: - `overview`: High-level description of the plan - `tasks`: Array of TaskStep objects - `userMessage`: Original user request **TaskStep Structure**: - `id`: Unique task identifier - `objective`: What the task should accomplish - `dependencies`: Array of task IDs this task depends on - `successCriteria`: Array of measurable criteria for task completion - `estimatedComplexity`: Complexity estimate (simple, medium, complex) - `userMessage`: User-facing description of the task **AI Planning**: Uses `services.ai.callAiPlanning()` with quality settings to generate detailed task breakdown. The AI receives: - User prompt and normalized intent - Available methods and actions (from method discovery) - Available documents and connections - Workflow context and history ### Task Execution The third stage (`_executeTasks()`) executes each task sequentially: **For Each Task**: 1. Build `TaskContext` containing: - Task details (objective, success criteria, dependencies) - Workflow state (current round, task, action numbers) - Available documents (from current and previous rounds) - Available connections (user's OAuth connections) - Previous task results (for context and dependencies) 2. Call `WorkflowProcessor.executeTask()` which delegates to mode-specific execution 3. Receive `TaskResult` with: - `success`: Boolean indicating task completion status - `feedback`: Human-readable summary of what was accomplished - `documents`: List of ChatDocument objects created during task execution - `reviewResult`: Optional ReviewResult if quality review was performed 4. Prepare task handover data for subsequent tasks 5. Accumulate results for use by dependent tasks **Mode-Specific Execution**: **Actionplan Mode**: - Generates complete action plan for entire task upfront - Executes all actions sequentially - Reviews results against success criteria - Retries with improvements if criteria not met (max 3 attempts) **Dynamic Mode**: - Generates single next action based on current state - Executes action immediately - Evaluates if task objective is met - Continues generating actions until objective met or max steps reached **Automation Mode**: - Uses predefined action list from JSON plan - Executes actions in order specified - No retry logic or quality review ### Action Execution Actions are executed by `ActionExecutor.executeSingleAction()`: **Process**: 1. Resolve parameters (document references, connections, etc.) 2. Look up method in global methods catalog 3. Validate action exists within method 4. Invoke action method with parameters 5. Extract result text from ActionDocument objects 6. Convert ActionDocuments to ChatDocuments for persistence 7. Create action completion message 8. Return ActionResult with success status and documents **Action Invocation**: Actions are invoked using compound names (e.g., "ai.process", "sharepoint.search") or separate method/action names. **Document References**: Actions receive document references in three formats: - `docItem::`: Single document by ID - `docList: