IRIS Legacy

#!/usr/bin/env python3 """ Test specifico per IRIS MCP Server Simula esattamente il protocollo MCP che userebbe un agente """ import requests import json import time BASE_URL = "https://trustypa.brainaihub.tech" def test_tools_list(): """Test 1: Lista tools disponibili""" print("\n" + "="*80) print("TEST 1: GET /mcp/tools") print("="*80) url = f"{BASE_URL}/mcp/tools" response = requests.get(url, verify=False) print(f"Status: {response.status_code}") print(f"Content-Type: {response.headers.get('Content-Type')}") if response.status_code == 200: data = response.json() tools_count = len(data.get('tools', [])) print(f"✅ Tools disponibili: {tools_count}") # Mostra primi 5 tools print("\n📋 Primi 5 tools:") for tool in data['tools'][:5]: print(f" - {tool['name']}") return True else: print(f"❌ FAIL: {response.text}") return False def test_sse_connection(): """Test 2: Connessione SSE""" print("\n" + "="*80) print("TEST 2: GET /mcp/sse (SSE Connection)") print("="*80) url = f"{BASE_URL}/mcp/sse" try: response = requests.get( url, stream=True, timeout=3, verify=False, headers={'Accept': 'text/event-stream'} ) print(f"Status: {response.status_code}") print(f"Content-Type: {response.headers.get('Content-Type')}") if response.status_code == 200: # Leggi primo evento SSE (endpoint) print("\n📡 SSE Events:") count = 0 for line in response.iter_lines(decode_unicode=True): if line: print(f" {line}") count += 1 if count >= 2: # event + data break print("✅ SSE Connection OK (ricevuto endpoint event)") return True else: print(f"❌ FAIL: HTTP {response.status_code}") return False except requests.exceptions.Timeout: # Timeout è OK - SSE rimane aperto print("✅ SSE Connection OK (stream aperto, timeout normale)") return True except Exception as e: print(f"❌ FAIL: {str(e)}") return False def test_mcp_initialize(): """Test 3: MCP Initialize handshake""" print("\n" + "="*80) print("TEST 3: POST /mcp/message (MCP Initialize)") print("="*80) url = f"{BASE_URL}/mcp/message" # Richiesta MCP initialize (come fa l'agente reale) mcp_request = { "jsonrpc": "2.0", "id": 0, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": { "name": "test-mcp-client", "version": "1.0.0" } } } print(f"\n📤 Request:") print(json.dumps(mcp_request, indent=2)) response = requests.post( url, json=mcp_request, verify=False, headers={'Content-Type': 'application/json'} ) print(f"\nStatus: {response.status_code}") print(f"Content-Type: {response.headers.get('Content-Type')}") print(f"\n📥 Response:") try: data = response.json() print(json.dumps(data, indent=2)) # IRIS risponde 202 e manda response via SSE if response.status_code == 202: print("✅ Initialize accepted (risposta via SSE)") return True elif response.status_code == 200: print("✅ Initialize successful") return True else: print(f"❌ FAIL: HTTP {response.status_code}") return False except: print(response.text) return False def test_mcp_tools_list(): """Test 4: MCP Tools List via POST""" print("\n" + "="*80) print("TEST 4: POST /mcp/message (tools/list)") print("="*80) url = f"{BASE_URL}/mcp/message" # Richiesta MCP tools/list mcp_request = { "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {} } print(f"\n📤 Request:") print(json.dumps(mcp_request, indent=2)) response = requests.post( url, json=mcp_request, verify=False, headers={'Content-Type': 'application/json'} ) print(f"\nStatus: {response.status_code}") # IRIS risponde 202 e manda response via SSE if response.status_code == 202: print("✅ Tools/list accepted (risposta via SSE)") return True elif response.status_code == 200: print("✅ Tools/list successful") return True else: print(f"❌ FAIL: HTTP {response.status_code}") print(response.text) return False def test_cors_headers(): """Test 5: CORS Headers""" print("\n" + "="*80) print("TEST 5: CORS Headers") print("="*80) url = f"{BASE_URL}/mcp/tools" # Prova prima con GET (sempre supportato) print("Test GET request per verificare CORS...") response = requests.get(url, verify=False) print(f"GET Status: {response.status_code}") cors_headers = { 'Access-Control-Allow-Origin': response.headers.get('Access-Control-Allow-Origin'), 'Access-Control-Allow-Credentials': response.headers.get('Access-Control-Allow-Credentials'), } print("\n📋 CORS Headers (da GET):") for key, value in cors_headers.items(): status = "✅" if value else "❌" print(f" {status} {key}: {value}") has_cors = cors_headers['Access-Control-Allow-Origin'] is not None if has_cors: print("\n✅ CORS configurato correttamente") return True else: print("\n❌ CORS non configurato") return False def run_all_tests(): """Esegui tutti i test""" print("\n" + "="*80) print("🧪 TEST IRIS MCP SERVER") print("="*80) print(f"Base URL: {BASE_URL}") # Disable SSL warnings import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) tests = [ ("Tools List (GET /mcp/tools)", test_tools_list), ("SSE Connection (GET /mcp/sse)", test_sse_connection), ("MCP Initialize (POST /mcp/message)", test_mcp_initialize), ("MCP Tools List (POST /mcp/message)", test_mcp_tools_list), ("CORS Headers", test_cors_headers), ] results = {} for test_name, test_func in tests: try: result = test_func() results[test_name] = result except Exception as e: print(f"\n❌ EXCEPTION: {str(e)}") results[test_name] = False # Summary print("\n" + "="*80) print("📊 SUMMARY") print("="*80) for test_name, result in results.items(): status = "✅ PASS" if result else "❌ FAIL" print(f"{status} - {test_name}") passed = sum(1 for v in results.values() if v) total = len(results) print(f"\nTotal: {passed}/{total} tests passed") return 0 if passed == total else 1 if __name__ == "__main__": import sys exit_code = run_all_tests() sys.exit(exit_code)