#!/usr/bin/env python3 """ N8N API Client - Core utility for interacting with N8N workflows Provides comprehensive workflow management capabilities for Claude Code CLI """ import json import requests import time from typing import Dict, List, Optional, Any from dataclasses import dataclass from datetime import datetime @dataclass class N8NConfig: """Configuration for N8N API connection""" api_url: str api_key: str headers: Dict[str, str] class N8NClient: """Main client for N8N API operations""" def __init__(self, config_path: str = "n8n_api_credentials.json"): """Initialize N8N client with configuration""" self.config = self._load_config(config_path) self.session = requests.Session() self.session.headers.update(self.config.headers) def _load_config(self, config_path: str) -> N8NConfig: """Load N8N configuration from JSON file""" try: with open(config_path, 'r') as f: config_data = json.load(f) return N8NConfig( api_url=config_data['api_url'], api_key=config_data['api_key'], headers=config_data['headers'] ) except Exception as e: raise Exception(f"Failed to load N8N configuration: {e}") def _make_request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Dict: """Make authenticated request to N8N API""" url = f"{self.config.api_url.rstrip('/')}/{endpoint.lstrip('/')}" try: if method.upper() == 'GET': response = self.session.get(url, params=data) elif method.upper() == 'POST': response = self.session.post(url, json=data) elif method.upper() == 'PUT': response = self.session.put(url, json=data) elif method.upper() == 'DELETE': response = self.session.delete(url) else: raise ValueError(f"Unsupported HTTP method: {method}") response.raise_for_status() return response.json() if response.content else {} except requests.exceptions.RequestException as e: raise Exception(f"N8N API request failed: {e}") # Workflow Management Methods def list_workflows(self) -> List[Dict]: """Get list of all workflows""" response = self._make_request('GET', '/workflows') # N8N API returns workflows in a 'data' property return response.get('data', response) if isinstance(response, dict) else response def get_workflow(self, workflow_id: str) -> Dict: """Get specific workflow by ID""" return self._make_request('GET', f'/workflows/{workflow_id}') def create_workflow(self, workflow_data: Dict) -> Dict: """Create new workflow""" return self._make_request('POST', '/workflows', workflow_data) def update_workflow(self, workflow_id: str, workflow_data: Dict) -> Dict: """Update existing workflow""" # Clean payload to only include API-allowed fields clean_payload = { 'name': workflow_data['name'], 'nodes': workflow_data['nodes'], 'connections': workflow_data['connections'], 'settings': workflow_data.get('settings', {}), 'staticData': workflow_data.get('staticData', {}) } return self._make_request('PUT', f'/workflows/{workflow_id}', clean_payload) def delete_workflow(self, workflow_id: str) -> Dict: """Delete workflow""" return self._make_request('DELETE', f'/workflows/{workflow_id}') def activate_workflow(self, workflow_id: str) -> Dict: """Activate workflow""" return self._make_request('POST', f'/workflows/{workflow_id}/activate') def deactivate_workflow(self, workflow_id: str) -> Dict: """Deactivate workflow""" return self._make_request('POST', f'/workflows/{workflow_id}/deactivate') # Execution Methods def execute_workflow(self, workflow_id: str, test_data: Optional[Dict] = None) -> Dict: """Execute workflow with optional test data""" payload = {"data": test_data} if test_data else {} return self._make_request('POST', f'/workflows/{workflow_id}/execute', payload) def get_executions(self, workflow_id: Optional[str] = None, limit: int = 20) -> List[Dict]: """Get workflow executions""" params = {"limit": limit} if workflow_id: params["workflowId"] = workflow_id return self._make_request('GET', '/executions', params) def get_execution(self, execution_id: str, include_data: bool = True) -> Dict: """Get specific execution details with full data""" params = {'includeData': 'true'} if include_data else {} return self._make_request('GET', f'/executions/{execution_id}', params) def delete_execution(self, execution_id: str) -> Dict: """Delete execution""" return self._make_request('DELETE', f'/executions/{execution_id}') # Utility Methods def find_workflow_by_name(self, name: str) -> Optional[Dict]: """Find workflow by name""" workflows = self.list_workflows() for workflow in workflows: if workflow.get('name') == name: return workflow return None def wait_for_execution(self, execution_id: str, timeout: int = 300, poll_interval: int = 5) -> Dict: """Wait for execution to complete""" start_time = time.time() while time.time() - start_time < timeout: execution = self.get_execution(execution_id) status = execution.get('status') if status in ['success', 'error', 'cancelled']: return execution time.sleep(poll_interval) raise TimeoutError(f"Execution {execution_id} did not complete within {timeout} seconds") def get_workflow_health(self, workflow_id: str, days: int = 7) -> Dict: """Get workflow health statistics""" executions = self.get_executions(workflow_id, limit=100) recent_executions = [] cutoff_time = datetime.now().timestamp() - (days * 24 * 3600) for execution in executions: if execution.get('startedAt'): exec_time = datetime.fromisoformat(execution['startedAt'].replace('Z', '+00:00')).timestamp() if exec_time > cutoff_time: recent_executions.append(execution) total = len(recent_executions) success = len([e for e in recent_executions if e.get('status') == 'success']) errors = len([e for e in recent_executions if e.get('status') == 'error']) return { 'total_executions': total, 'success_count': success, 'error_count': errors, 'success_rate': (success / total * 100) if total > 0 else 0, 'recent_executions': recent_executions } if __name__ == "__main__": # Quick test of the client try: client = N8NClient() workflows = client.list_workflows() print(f"Connected to N8N successfully. Found {len(workflows)} workflows.") except Exception as e: print(f"Failed to connect to N8N: {e}")