""" n8n REST API Client Handles all HTTP communication with n8n REST API. Separates API communication from business logic. """ import logging from typing import Any import aiohttp class N8nClient: """ n8n REST API client for HTTP communication. Handles authentication, request formatting, and error handling for all n8n API endpoints. Authentication: API Key via X-N8N-API-KEY header """ def __init__(self, site_url: str, api_key: str): """ Initialize n8n API client. Args: site_url: n8n instance URL (e.g., https://n8n.example.com) api_key: n8n API key for authentication """ self.site_url = site_url.rstrip("/") self.api_base = f"{self.site_url}/api/v1" self.api_key = api_key # Initialize logger self.logger = logging.getLogger(f"N8nClient.{site_url}") def _get_headers(self, additional_headers: dict | None = None) -> dict[str, str]: """ Get request headers with API key authentication. Args: additional_headers: Additional headers to include Returns: Dict: Headers with authentication """ headers = { "Content-Type": "application/json", "Accept": "application/json", "X-N8N-API-KEY": self.api_key, } # Merge additional headers if additional_headers: headers.update(additional_headers) return headers async def request( self, method: str, endpoint: str, params: dict | None = None, json_data: dict | None = None, headers_override: dict | None = None, ) -> Any: """ Make authenticated request to n8n REST API. Args: method: HTTP method (GET, POST, PUT, DELETE, PATCH) endpoint: API endpoint (without base URL) params: Query parameters json_data: JSON body data headers_override: Override default headers Returns: API response (dict, list, or None) Raises: Exception: On API errors with status code and message """ # Build full URL url = f"{self.api_base}/{endpoint.lstrip('/')}" # Setup headers headers = self._get_headers(headers_override) # Filter out None values from params if params: params = {k: v for k, v in params.items() if v is not None} # Filter None values from JSON data if json_data: json_data = {k: v for k, v in json_data.items() if v is not None} # Make request self.logger.debug(f"{method} {url}") self.logger.debug(f"Params: {params}") self.logger.debug(f"Data: {json_data}") async with ( aiohttp.ClientSession() as session, session.request( method=method, url=url, params=params, json=json_data, headers=headers ) as response, ): # Log response self.logger.debug(f"Response status: {response.status}") # Handle empty responses (e.g., 204 No Content) if response.status == 204: return {"success": True, "message": "Operation completed successfully"} # Try to parse JSON response try: response_data = await response.json() except Exception: response_text = await response.text() if response.status >= 400: raise Exception(f"n8n API error (status {response.status}): {response_text}") return {"success": True, "message": response_text} # Check for errors if response.status >= 400: error_msg = response_data.get("message", str(response_data)) raise Exception(f"n8n API error (status {response.status}): {error_msg}") return response_data # ===================== # WORKFLOW ENDPOINTS # ===================== async def list_workflows( self, active: bool | None = None, tags: str | None = None, name: str | None = None, limit: int = 50, cursor: str | None = None, ) -> dict[str, Any]: """List workflows with optional filters""" params = {"active": active, "tags": tags, "name": name, "limit": limit, "cursor": cursor} return await self.request("GET", "workflows", params=params) async def get_workflow(self, workflow_id: str) -> dict[str, Any]: """Get workflow by ID""" return await self.request("GET", f"workflows/{workflow_id}") async def create_workflow(self, data: dict[str, Any]) -> dict[str, Any]: """Create a new workflow""" return await self.request("POST", "workflows", json_data=data) async def update_workflow(self, workflow_id: str, data: dict[str, Any]) -> dict[str, Any]: """Update an existing workflow""" return await self.request("PUT", f"workflows/{workflow_id}", json_data=data) async def delete_workflow(self, workflow_id: str) -> dict[str, Any]: """Delete a workflow""" return await self.request("DELETE", f"workflows/{workflow_id}") async def activate_workflow(self, workflow_id: str) -> dict[str, Any]: """Activate a workflow""" return await self.request("POST", f"workflows/{workflow_id}/activate") async def deactivate_workflow(self, workflow_id: str) -> dict[str, Any]: """Deactivate a workflow""" return await self.request("POST", f"workflows/{workflow_id}/deactivate") async def execute_workflow( self, workflow_id: str, data: dict[str, Any] | None = None ) -> dict[str, Any]: """Execute a workflow manually""" return await self.request("POST", f"workflows/{workflow_id}/run", json_data=data or {}) async def get_workflow_tags(self, workflow_id: str) -> list[dict[str, Any]]: """Get tags assigned to a workflow""" workflow = await self.get_workflow(workflow_id) return workflow.get("tags", []) # ===================== # EXECUTION ENDPOINTS # ===================== async def list_executions( self, workflow_id: str | None = None, status: str | None = None, include_data: bool = False, limit: int = 20, cursor: str | None = None, ) -> dict[str, Any]: """List workflow executions with filters""" params = { "workflowId": workflow_id, "status": status, "includeData": str(include_data).lower(), "limit": limit, "cursor": cursor, } return await self.request("GET", "executions", params=params) async def get_execution(self, execution_id: str, include_data: bool = True) -> dict[str, Any]: """Get execution details""" params = {"includeData": str(include_data).lower()} return await self.request("GET", f"executions/{execution_id}", params=params) async def delete_execution(self, execution_id: str) -> dict[str, Any]: """Delete a single execution""" return await self.request("DELETE", f"executions/{execution_id}") async def stop_execution(self, execution_id: str) -> dict[str, Any]: """Stop a running execution""" return await self.request("POST", f"executions/{execution_id}/stop") # ===================== # CREDENTIAL ENDPOINTS # ===================== async def list_credentials(self, limit: int = 100, cursor: str | None = None) -> dict[str, Any]: """List all credentials (metadata only)""" params = {"limit": limit, "cursor": cursor} return await self.request("GET", "credentials", params=params) async def get_credential(self, credential_id: str) -> dict[str, Any]: """Get credential metadata""" return await self.request("GET", f"credentials/{credential_id}") async def create_credential(self, data: dict[str, Any]) -> dict[str, Any]: """Create a new credential""" return await self.request("POST", "credentials", json_data=data) async def delete_credential(self, credential_id: str) -> dict[str, Any]: """Delete a credential""" return await self.request("DELETE", f"credentials/{credential_id}") async def get_credential_schema(self, credential_type: str) -> dict[str, Any]: """Get schema for a credential type""" return await self.request("GET", f"credentials/schema/{credential_type}") async def transfer_credential( self, credential_id: str, destination_project_id: str ) -> dict[str, Any]: """Transfer credential to another project""" data = {"destinationProjectId": destination_project_id} return await self.request("POST", f"credentials/{credential_id}/transfer", json_data=data) # ===================== # TAG ENDPOINTS # ===================== async def list_tags(self, limit: int = 100, cursor: str | None = None) -> dict[str, Any]: """List all tags""" params = {"limit": limit, "cursor": cursor} return await self.request("GET", "tags", params=params) async def get_tag(self, tag_id: str) -> dict[str, Any]: """Get tag by ID""" return await self.request("GET", f"tags/{tag_id}") async def create_tag(self, name: str) -> dict[str, Any]: """Create a new tag""" return await self.request("POST", "tags", json_data={"name": name}) async def update_tag(self, tag_id: str, name: str) -> dict[str, Any]: """Update tag name""" return await self.request("PUT", f"tags/{tag_id}", json_data={"name": name}) async def delete_tag(self, tag_id: str) -> dict[str, Any]: """Delete a tag""" return await self.request("DELETE", f"tags/{tag_id}") # ===================== # USER ENDPOINTS # ===================== async def list_users( self, limit: int = 100, cursor: str | None = None, include_role: bool = True ) -> dict[str, Any]: """List all users""" params = {"limit": limit, "cursor": cursor, "includeRole": str(include_role).lower()} return await self.request("GET", "users", params=params) async def get_user(self, user_id: str, include_role: bool = True) -> dict[str, Any]: """Get user by ID or email""" params = {"includeRole": str(include_role).lower()} return await self.request("GET", f"users/{user_id}", params=params) async def create_user(self, users: list[dict[str, Any]]) -> dict[str, Any]: """Create/invite users""" return await self.request("POST", "users", json_data=users) async def delete_user(self, user_id: str) -> dict[str, Any]: """Delete a user""" return await self.request("DELETE", f"users/{user_id}") async def change_user_role(self, user_id: str, new_role: str) -> dict[str, Any]: """Change user's global role""" return await self.request( "PATCH", f"users/{user_id}/role", json_data={"newRoleName": new_role} ) # ===================== # PROJECT ENDPOINTS (Enterprise/Pro) # ===================== async def list_projects(self, limit: int = 100, cursor: str | None = None) -> dict[str, Any]: """List all projects""" params = {"limit": limit, "cursor": cursor} return await self.request("GET", "projects", params=params) async def get_project(self, project_id: str) -> dict[str, Any]: """Get project by ID""" return await self.request("GET", f"projects/{project_id}") async def create_project(self, name: str) -> dict[str, Any]: """Create a new project""" return await self.request("POST", "projects", json_data={"name": name}) async def update_project(self, project_id: str, name: str) -> dict[str, Any]: """Update project""" return await self.request("PUT", f"projects/{project_id}", json_data={"name": name}) async def delete_project(self, project_id: str) -> dict[str, Any]: """Delete a project""" return await self.request("DELETE", f"projects/{project_id}") async def add_project_users( self, project_id: str, relations: list[dict[str, str]] ) -> dict[str, Any]: """Add users to project with roles""" return await self.request("POST", f"projects/{project_id}/users", json_data=relations) async def change_project_user_role( self, project_id: str, user_id: str, role: str ) -> dict[str, Any]: """Change user's role in project""" return await self.request( "PUT", f"projects/{project_id}/users/{user_id}", json_data={"role": role} ) async def remove_project_user(self, project_id: str, user_id: str) -> dict[str, Any]: """Remove user from project""" return await self.request("DELETE", f"projects/{project_id}/users/{user_id}") # ===================== # VARIABLE ENDPOINTS # ===================== async def list_variables(self, limit: int = 100, cursor: str | None = None) -> dict[str, Any]: """List all variables""" params = {"limit": limit, "cursor": cursor} return await self.request("GET", "variables", params=params) async def get_variable(self, key: str) -> dict[str, Any]: """Get variable by key""" return await self.request("GET", f"variables/{key}") async def create_variable(self, key: str, value: str) -> dict[str, Any]: """Create a new variable""" return await self.request("POST", "variables", json_data={"key": key, "value": value}) async def update_variable(self, key: str, value: str) -> dict[str, Any]: """Update variable value""" return await self.request("PUT", f"variables/{key}", json_data={"value": value}) async def delete_variable(self, key: str) -> dict[str, Any]: """Delete a variable""" return await self.request("DELETE", f"variables/{key}") # ===================== # SYSTEM ENDPOINTS # ===================== async def run_audit(self, categories: list[str] | None = None) -> dict[str, Any]: """Run security audit""" data = {} if categories: data["additionalOptions"] = {"categories": categories} return await self.request("POST", "audit", json_data=data) async def source_control_pull( self, variables: dict[str, str] | None = None, force: bool = False ) -> dict[str, Any]: """Pull from source control""" data = {"force": force} if variables: data["variables"] = variables return await self.request("POST", "source-control/pull", json_data=data) async def health_check(self) -> dict[str, Any]: """Check n8n instance health""" # Use direct URL, not API base url = f"{self.site_url}/healthz" async with aiohttp.ClientSession() as session, session.get(url) as response: if response.status == 200: return {"healthy": True, "status": "ok"} else: return {"healthy": False, "status": f"unhealthy (status {response.status})"}