#!/usr/bin/env python3 """ Diagnostic script for the Dynamic Multi-Agent Router System Checks system health and identifies common issues """ import requests import json import os import sys from typing import Dict, Any, List class AgentSystemDiagnostic: def __init__(self, base_url: str = "http://localhost:5678"): self.base_url = base_url self.webhook_url = f"{base_url}/webhook/agent-task" self.issues = [] self.warnings = [] def check_n8n_availability(self) -> bool: """Check if n8n is running and accessible""" try: response = requests.get(self.base_url, timeout=5) if response.status_code == 200: print("✅ n8n is running and accessible") return True else: self.issues.append(f"n8n returned status code {response.status_code}") return False except requests.exceptions.RequestException as e: self.issues.append(f"Cannot connect to n8n: {str(e)}") return False def check_webhook_registration(self) -> bool: """Check if the webhook is registered""" try: # Try a GET request to see if webhook exists response = requests.get(self.webhook_url, timeout=5) if response.status_code == 405: # Method not allowed is expected for GET print("✅ Webhook is registered") return True elif response.status_code == 404: self.issues.append("Webhook not registered - workflow may not be active") return False else: self.warnings.append(f"Unexpected webhook response: {response.status_code}") return True except requests.exceptions.RequestException as e: self.issues.append(f"Cannot access webhook: {str(e)}") return False def check_ollama_availability(self) -> bool: """Check if Ollama is running and has the required model""" try: # Check if Ollama is running response = requests.get("http://localhost:11434/api/tags", timeout=5) if response.status_code == 200: models = response.json().get('models', []) model_names = [model.get('name', '') for model in models] if any('llama3.1' in name for name in model_names): print("✅ Ollama is running with llama3.1 model") return True else: self.issues.append("Ollama is running but llama3.1 model not found") print("Available models:", model_names) return False else: self.issues.append(f"Ollama API returned status {response.status_code}") return False except requests.exceptions.RequestException as e: self.issues.append(f"Cannot connect to Ollama: {str(e)}") return False def check_redis_availability(self) -> bool: """Check if Redis is running""" try: import redis r = redis.Redis(host='localhost', port=6379, decode_responses=True) r.ping() print("✅ Redis is running and accessible") return True except ImportError: self.warnings.append("Redis Python client not installed - cannot test Redis connection") return True except Exception as e: self.issues.append(f"Cannot connect to Redis: {str(e)}") return False def check_agent_files(self) -> bool: """Check if agent files exist and are properly formatted""" agent_dirs = [ "/home/node/.n8n/agents", "./agents", "./n8n-data/agents" ] agent_files_found = False for agent_dir in agent_dirs: if os.path.exists(agent_dir): print(f"✅ Found agent directory: {agent_dir}") agent_files_found = True # Check for .md files md_files = [] for root, dirs, files in os.walk(agent_dir): for file in files: if file.endswith('.md'): md_files.append(os.path.join(root, file)) if md_files: print(f"✅ Found {len(md_files)} agent files") # Check a few files for proper format for file_path in md_files[:3]: # Check first 3 files if not self._check_agent_file_format(file_path): self.warnings.append(f"Agent file may have format issues: {file_path}") else: self.issues.append(f"No .md files found in {agent_dir}") break if not agent_files_found: self.issues.append("No agent directory found. Checked: " + ", ".join(agent_dirs)) return False return True def _check_agent_file_format(self, file_path: str) -> bool: """Check if an agent file has proper frontmatter format""" try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() lines = content.split('\n') if len(lines) < 3: return False # Check for frontmatter markers if lines[0].strip() != '---': return False # Find end of frontmatter end_marker = -1 for i, line in enumerate(lines[1:], 1): if line.strip() == '---': end_marker = i break if end_marker == -1: return False # Check for required fields frontmatter = lines[1:end_marker] has_name = any(line.startswith('name:') for line in frontmatter) has_description = any(line.startswith('description:') for line in frontmatter) return has_name and has_description except Exception as e: self.warnings.append(f"Error reading agent file {file_path}: {str(e)}") return False def test_webhook_functionality(self) -> bool: """Test the webhook with a simple request""" try: test_payload = { "task": "Test request for system diagnostics", "context": "This is a diagnostic test" } response = requests.post( self.webhook_url, json=test_payload, headers={"Content-Type": "application/json"}, timeout=30 ) if response.status_code == 200: response_data = response.json() if response_data.get('success'): print("✅ Webhook is functioning correctly") return True else: self.issues.append(f"Webhook returned error: {response_data.get('error')}") return False else: self.issues.append(f"Webhook returned status {response.status_code}") return False except requests.exceptions.Timeout: self.issues.append("Webhook request timed out - system may be overloaded") return False except Exception as e: self.issues.append(f"Webhook test failed: {str(e)}") return False def run_full_diagnostic(self) -> Dict[str, Any]: """Run complete system diagnostic""" print("🔍 Running Dynamic Multi-Agent Router System Diagnostic") print("=" * 60) # Run all checks checks = [ ("n8n Availability", self.check_n8n_availability), ("Webhook Registration", self.check_webhook_registration), ("Ollama Availability", self.check_ollama_availability), ("Redis Availability", self.check_redis_availability), ("Agent Files", self.check_agent_files), ("Webhook Functionality", self.test_webhook_functionality) ] results = {} for check_name, check_func in checks: print(f"\n📋 {check_name}:") try: result = check_func() results[check_name.lower().replace(' ', '_')] = result except Exception as e: self.issues.append(f"Error in {check_name}: {str(e)}") results[check_name.lower().replace(' ', '_')] = False # Summary print("\n" + "=" * 60) print("📊 DIAGNOSTIC SUMMARY") print("=" * 60) if self.issues: print("❌ CRITICAL ISSUES:") for issue in self.issues: print(f" • {issue}") if self.warnings: print("\n⚠️ WARNINGS:") for warning in self.warnings: print(f" • {warning}") if not self.issues and not self.warnings: print("✅ All systems are healthy!") # Recommendations print("\n💡 RECOMMENDATIONS:") if self.issues: print(" 1. Fix all critical issues before using the system") print(" 2. Run this diagnostic again after fixes") else: print(" 1. System is ready for use") print(" 2. Run the test suite for comprehensive testing") print(" 3. Monitor system performance in production") if self.warnings: print(" 4. Address warnings to improve system reliability") return { "issues": self.issues, "warnings": self.warnings, "results": results, "healthy": len(self.issues) == 0 } def main(): """Main function to run the diagnostic""" import argparse parser = argparse.ArgumentParser(description="Diagnose the Dynamic Multi-Agent Router System") parser.add_argument("--url", default="http://localhost:5678", help="n8n base URL") parser.add_argument("--output", help="Output file for diagnostic results (JSON)") args = parser.parse_args() diagnostic = AgentSystemDiagnostic(args.url) results = diagnostic.run_full_diagnostic() if args.output: with open(args.output, 'w') as f: json.dump(results, f, indent=2) print(f"\nDiagnostic results saved to {args.output}") # Exit with appropriate code sys.exit(0 if results["healthy"] else 1) if __name__ == "__main__": main()