mirror of
https://github.com/thecyberlearn/quantum-ai.git
synced 2026-08-18 10:13:00 +00:00
2c129708d9
- Add N8N workflow directory structure for webhook-based agents - Create workflow management scripts (import/export/sync/backup) - Add comprehensive documentation for each agent's workflow setup - Clarify N8N deployment architecture (separate hosting from Django) - Update deployment guides with clear separation warnings - Add workflow deployment automation scripts Architecture clarification: - Django app deploys to Railway - N8N runs separately (N8N Cloud, separate Railway project, or self-hosted) - Communication via HTTP webhook URLs only 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
268 lines
10 KiB
Python
Executable File
268 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
N8N Workflow Management Script for Quantum Tasks AI
|
|
|
|
This script helps manage N8N workflows for webhook-based agents:
|
|
- Import workflows to N8N instance
|
|
- Export workflows from N8N instance
|
|
- Sync workflows between local files and N8N
|
|
- Backup and restore workflows
|
|
|
|
Usage:
|
|
python manage_n8n_workflows.py import [agent_name]
|
|
python manage_n8n_workflows.py export [agent_name]
|
|
python manage_n8n_workflows.py sync
|
|
python manage_n8n_workflows.py backup
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import requests
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
import argparse
|
|
|
|
# Configuration
|
|
N8N_BASE_URL = os.getenv('N8N_BASE_URL', 'http://localhost:5678')
|
|
N8N_API_KEY = os.getenv('N8N_API_KEY', '')
|
|
|
|
# Webhook-based agents that need N8N workflows
|
|
WEBHOOK_AGENTS = [
|
|
'data_analyzer',
|
|
'social_ads_generator',
|
|
'job_posting_generator',
|
|
'five_whys_analyzer'
|
|
]
|
|
|
|
class N8NWorkflowManager:
|
|
def __init__(self):
|
|
self.base_url = N8N_BASE_URL
|
|
self.api_key = N8N_API_KEY
|
|
self.headers = {
|
|
'Content-Type': 'application/json',
|
|
'X-N8N-API-KEY': self.api_key
|
|
} if self.api_key else {'Content-Type': 'application/json'}
|
|
|
|
def get_workflow_path(self, agent_name):
|
|
"""Get the workflow directory path for an agent"""
|
|
return Path(f"{agent_name}/n8n_workflows")
|
|
|
|
def load_workflow_json(self, agent_name, filename='workflow.json'):
|
|
"""Load workflow JSON from agent directory"""
|
|
workflow_path = self.get_workflow_path(agent_name) / filename
|
|
if not workflow_path.exists():
|
|
print(f"❌ Workflow file not found: {workflow_path}")
|
|
return None
|
|
|
|
try:
|
|
with open(workflow_path, 'r') as f:
|
|
return json.load(f)
|
|
except json.JSONDecodeError as e:
|
|
print(f"❌ Invalid JSON in {workflow_path}: {e}")
|
|
return None
|
|
|
|
def save_workflow_json(self, agent_name, workflow_data, filename='workflow.json'):
|
|
"""Save workflow JSON to agent directory"""
|
|
workflow_path = self.get_workflow_path(agent_name)
|
|
workflow_path.mkdir(exist_ok=True)
|
|
|
|
filepath = workflow_path / filename
|
|
with open(filepath, 'w') as f:
|
|
json.dump(workflow_data, f, indent=2)
|
|
|
|
print(f"✅ Workflow saved: {filepath}")
|
|
|
|
def import_workflow_to_n8n(self, agent_name):
|
|
"""Import workflow from local file to N8N instance"""
|
|
print(f"📥 Importing workflow for {agent_name}...")
|
|
|
|
workflow_data = self.load_workflow_json(agent_name)
|
|
if not workflow_data:
|
|
return False
|
|
|
|
# Create workflow in N8N
|
|
try:
|
|
response = requests.post(
|
|
f"{self.base_url}/api/v1/workflows",
|
|
headers=self.headers,
|
|
json=workflow_data
|
|
)
|
|
|
|
if response.status_code == 201:
|
|
workflow_id = response.json().get('id')
|
|
print(f"✅ Workflow imported successfully: ID {workflow_id}")
|
|
|
|
# Activate the workflow
|
|
activate_response = requests.post(
|
|
f"{self.base_url}/api/v1/workflows/{workflow_id}/activate",
|
|
headers=self.headers
|
|
)
|
|
|
|
if activate_response.status_code == 200:
|
|
print(f"✅ Workflow activated successfully")
|
|
else:
|
|
print(f"⚠️ Workflow imported but activation failed: {activate_response.text}")
|
|
|
|
return True
|
|
else:
|
|
print(f"❌ Import failed: {response.status_code} - {response.text}")
|
|
return False
|
|
|
|
except requests.RequestException as e:
|
|
print(f"❌ Connection error: {e}")
|
|
return False
|
|
|
|
def export_workflow_from_n8n(self, agent_name, workflow_name=None):
|
|
"""Export workflow from N8N instance to local file"""
|
|
print(f"📤 Exporting workflow for {agent_name}...")
|
|
|
|
if not workflow_name:
|
|
workflow_name = f"{agent_name.replace('_', ' ').title()} Agent"
|
|
|
|
try:
|
|
# Get all workflows
|
|
response = requests.get(
|
|
f"{self.base_url}/api/v1/workflows",
|
|
headers=self.headers
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
print(f"❌ Failed to fetch workflows: {response.text}")
|
|
return False
|
|
|
|
workflows = response.json()
|
|
|
|
# Find workflow by name
|
|
target_workflow = None
|
|
for workflow in workflows:
|
|
if workflow.get('name', '').lower() == workflow_name.lower():
|
|
target_workflow = workflow
|
|
break
|
|
|
|
if not target_workflow:
|
|
print(f"❌ Workflow '{workflow_name}' not found in N8N")
|
|
print("Available workflows:")
|
|
for wf in workflows:
|
|
print(f" - {wf.get('name', 'Unnamed')}")
|
|
return False
|
|
|
|
# Save workflow with timestamp
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
self.save_workflow_json(agent_name, target_workflow, f"workflow_exported_{timestamp}.json")
|
|
|
|
# Also save as main workflow file
|
|
self.save_workflow_json(agent_name, target_workflow, "workflow.json")
|
|
|
|
return True
|
|
|
|
except requests.RequestException as e:
|
|
print(f"❌ Connection error: {e}")
|
|
return False
|
|
|
|
def backup_all_workflows(self):
|
|
"""Backup all workflows to timestamped files"""
|
|
print("🔄 Backing up all workflows...")
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
for agent_name in WEBHOOK_AGENTS:
|
|
workflow_path = self.get_workflow_path(agent_name)
|
|
if (workflow_path / "workflow.json").exists():
|
|
workflow_data = self.load_workflow_json(agent_name)
|
|
if workflow_data:
|
|
self.save_workflow_json(agent_name, workflow_data, f"workflow_backup_{timestamp}.json")
|
|
print(f"✅ Backed up {agent_name} workflow")
|
|
|
|
def sync_workflows(self):
|
|
"""Sync workflows between local files and N8N instance"""
|
|
print("🔄 Syncing all workflows...")
|
|
|
|
for agent_name in WEBHOOK_AGENTS:
|
|
print(f"\n--- {agent_name} ---")
|
|
|
|
# Check if local workflow exists
|
|
if (self.get_workflow_path(agent_name) / "workflow.json").exists():
|
|
print(f"📁 Local workflow found for {agent_name}")
|
|
|
|
# Try to import to N8N
|
|
success = self.import_workflow_to_n8n(agent_name)
|
|
if not success:
|
|
print(f"⚠️ Failed to sync {agent_name} to N8N")
|
|
else:
|
|
print(f"❌ No local workflow found for {agent_name}")
|
|
print(f"💡 Place your workflow JSON file at: {self.get_workflow_path(agent_name)}/workflow.json")
|
|
|
|
def list_workflows(self):
|
|
"""List all workflows in N8N and local directories"""
|
|
print("📋 Listing all workflows...\n")
|
|
|
|
# List N8N workflows
|
|
try:
|
|
response = requests.get(f"{self.base_url}/api/v1/workflows", headers=self.headers)
|
|
if response.status_code == 200:
|
|
workflows = response.json()
|
|
print(f"🌐 N8N Instance ({len(workflows)} workflows):")
|
|
for wf in workflows:
|
|
status = "🟢 Active" if wf.get('active') else "🔴 Inactive"
|
|
print(f" - {wf.get('name', 'Unnamed')} ({status})")
|
|
else:
|
|
print("❌ Could not connect to N8N instance")
|
|
except requests.RequestException:
|
|
print("❌ Could not connect to N8N instance")
|
|
|
|
print()
|
|
|
|
# List local workflows
|
|
print("📁 Local Workflows:")
|
|
for agent_name in WEBHOOK_AGENTS:
|
|
workflow_path = self.get_workflow_path(agent_name)
|
|
if workflow_path.exists():
|
|
files = list(workflow_path.glob("*.json"))
|
|
if files:
|
|
print(f" {agent_name}: {len(files)} files")
|
|
for file in files:
|
|
print(f" - {file.name}")
|
|
else:
|
|
print(f" {agent_name}: No workflow files")
|
|
else:
|
|
print(f" {agent_name}: Directory not found")
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='N8N Workflow Management for Quantum Tasks AI')
|
|
parser.add_argument('action', choices=['import', 'export', 'sync', 'backup', 'list'],
|
|
help='Action to perform')
|
|
parser.add_argument('agent', nargs='?', choices=WEBHOOK_AGENTS,
|
|
help='Specific agent to operate on (for import/export)')
|
|
parser.add_argument('--workflow-name', help='Workflow name in N8N (for export)')
|
|
|
|
args = parser.parse_args()
|
|
|
|
manager = N8NWorkflowManager()
|
|
|
|
if args.action == 'import':
|
|
if not args.agent:
|
|
print("❌ Please specify an agent name for import")
|
|
print(f"Available agents: {', '.join(WEBHOOK_AGENTS)}")
|
|
sys.exit(1)
|
|
success = manager.import_workflow_to_n8n(args.agent)
|
|
sys.exit(0 if success else 1)
|
|
|
|
elif args.action == 'export':
|
|
if not args.agent:
|
|
print("❌ Please specify an agent name for export")
|
|
print(f"Available agents: {', '.join(WEBHOOK_AGENTS)}")
|
|
sys.exit(1)
|
|
success = manager.export_workflow_from_n8n(args.agent, args.workflow_name)
|
|
sys.exit(0 if success else 1)
|
|
|
|
elif args.action == 'sync':
|
|
manager.sync_workflows()
|
|
|
|
elif args.action == 'backup':
|
|
manager.backup_all_workflows()
|
|
|
|
elif args.action == 'list':
|
|
manager.list_workflows()
|
|
|
|
if __name__ == '__main__':
|
|
main() |