| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | Test script to verify FastA2A agent card routing and authentication. |
| 4 | """ |
| 5 | |
| 6 | import sys, os |
| 7 | sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
| 8 | |
| 9 | |
| 10 | import asyncio |
| 11 | import pytest |
| 12 | from helpers import settings |
| 13 | |
| 14 | |
| 15 | def get_test_urls(): |
| 16 | """Get the URLs to test based on current settings.""" |
| 17 | try: |
| 18 | cfg = settings.get_settings() |
| 19 | token = cfg.get("mcp_server_token", "") |
| 20 | |
| 21 | if not token: |
| 22 | print("❌ No mcp_server_token found in settings") |
| 23 | return None |
| 24 | |
| 25 | base_url = "http://localhost:50101" |
| 26 | |
| 27 | urls = { |
| 28 | "token_based": f"{base_url}/a2a/t-{token}/.well-known/agent.json", |
| 29 | "bearer_auth": f"{base_url}/a2a/.well-known/agent.json", |
| 30 | "api_key_header": f"{base_url}/a2a/.well-known/agent.json", |
| 31 | "api_key_query": f"{base_url}/a2a/.well-known/agent.json?api_key={token}" |
| 32 | } |
| 33 | |
| 34 | return {"token": token, "urls": urls} |
| 35 | |
| 36 | except Exception as e: |
| 37 | print(f"❌ Error getting settings: {e}") |
| 38 | return None |
| 39 | |
| 40 | |
| 41 | def print_test_commands(): |
| 42 | """Print curl commands to test FastA2A authentication.""" |
| 43 | data = get_test_urls() |
| 44 | if not data: |
| 45 | return |
| 46 | |
| 47 | token = data["token"] |
| 48 | urls = data["urls"] |
| 49 | |
| 50 | print("🚀 FastA2A Agent Card Testing Commands") |
| 51 | print("=" * 60) |
| 52 | print(f"Current token: {token}") |
| 53 | print() |
| 54 | |
| 55 | print("1️⃣ Token-based URL (recommended):") |
| 56 | print(f" curl -v '{urls['token_based']}'") |
| 57 | print() |
| 58 | |
| 59 | print("2️⃣ Bearer authentication:") |
| 60 | print(f" curl -v -H 'Authorization: Bearer {token}' '{urls['bearer_auth']}'") |
| 61 | print() |
| 62 | |
| 63 | print("3️⃣ API key header:") |
| 64 | print(f" curl -v -H 'X-API-KEY: {token}' '{urls['api_key_header']}'") |
| 65 | print() |
| 66 | |
| 67 | print("4️⃣ API key query parameter:") |
| 68 | print(f" curl -v '{urls['api_key_query']}'") |
| 69 | print() |
| 70 | |
| 71 | print("Expected response (if working):") |
| 72 | print(" HTTP/1.1 200 OK") |
| 73 | print(" Content-Type: application/json") |
| 74 | print(" {") |
| 75 | print(' "name": "Agent Zero",') |
| 76 | print(' "version": "1.0.0",') |
| 77 | print(' "skills": [...]') |
| 78 | print(" }") |
| 79 | print() |
| 80 | |
| 81 | print("Expected error (if auth fails):") |
| 82 | print(" HTTP/1.1 401 Unauthorized") |
| 83 | print(" Unauthorized") |
| 84 | print() |
| 85 | |
| 86 | |
| 87 | def print_troubleshooting(): |
| 88 | """Print troubleshooting information.""" |
| 89 | print("🔧 Troubleshooting FastA2A Issues") |
| 90 | print("=" * 40) |
| 91 | print() |
| 92 | print("1. Server not running:") |
| 93 | print(" - Make sure Agent Zero is running: python run_ui.py") |
| 94 | print(" - Check the correct port (default: 50101)") |
| 95 | print() |
| 96 | |
| 97 | print("2. Authentication failures:") |
| 98 | print(" - Verify token matches in settings") |
| 99 | print(" - Check token format (should be 16 characters)") |
| 100 | print(" - Try different auth methods") |
| 101 | print() |
| 102 | |
| 103 | print("3. FastA2A not available:") |
| 104 | print(" - Install FastA2A: pip install fasta2a") |
| 105 | print(" - Check server logs for FastA2A configuration errors") |
| 106 | print() |
| 107 | |
| 108 | print("4. Routing issues:") |
| 109 | print(" - Verify /a2a prefix is working") |
| 110 | print(" - Check DispatcherMiddleware configuration") |
| 111 | print(" - Look for FastA2A startup messages in logs") |
| 112 | print() |
| 113 | |
| 114 | |
| 115 | def validate_token_format(): |
| 116 | """Validate that the token format is correct.""" |
| 117 | try: |
| 118 | cfg = settings.get_settings() |
| 119 | token = cfg.get("mcp_server_token", "") |
| 120 | |
| 121 | print("🔍 Token Validation") |
| 122 | print("=" * 25) |
| 123 | |
| 124 | if not token: |
| 125 | print("❌ No token found") |
| 126 | return False |
| 127 | |
| 128 | print(f"✅ Token found: {token}") |
| 129 | print(f"✅ Token length: {len(token)} characters") |
| 130 | |
| 131 | if len(token) != 16: |
| 132 | print("⚠️ Warning: Expected token length is 16 characters") |
| 133 | |
| 134 | # Check token characters |
| 135 | if token.isalnum(): |
| 136 | print("✅ Token contains only alphanumeric characters") |
| 137 | else: |
| 138 | print("⚠️ Warning: Token contains non-alphanumeric characters") |
| 139 | |
| 140 | return True |
| 141 | |
| 142 | except Exception as e: |
| 143 | print(f"❌ Error validating token: {e}") |
| 144 | return False |
| 145 | |
| 146 | |
| 147 | @pytest.mark.asyncio |
| 148 | async def test_server_connectivity(): |
| 149 | """Test basic server connectivity.""" |
| 150 | try: |
| 151 | import httpx |
| 152 | |
| 153 | print("🌐 Server Connectivity Test") |
| 154 | print("=" * 30) |
| 155 | |
| 156 | async with httpx.AsyncClient() as client: |
| 157 | try: |
| 158 | # Test basic server |
| 159 | await client.get("http://localhost:50101/", timeout=5.0) |
| 160 | print("✅ Agent Zero server is running") |
| 161 | return True |
| 162 | except httpx.ConnectError: |
| 163 | print("❌ Cannot connect to Agent Zero server") |
| 164 | print(" Make sure the server is running: python run_ui.py") |
| 165 | return False |
| 166 | except Exception as e: |
| 167 | print(f"❌ Server connectivity error: {e}") |
| 168 | return False |
| 169 | |
| 170 | except ImportError: |
| 171 | print("ℹ️ httpx not available, skipping connectivity test") |
| 172 | print(" Install with: pip install httpx") |
| 173 | return None |
| 174 | |
| 175 | |
| 176 | def main(): |
| 177 | """Main test function.""" |
| 178 | print("🧪 FastA2A Agent Card Testing Utility") |
| 179 | print("=" * 45) |
| 180 | print() |
| 181 | |
| 182 | # Validate token |
| 183 | if not validate_token_format(): |
| 184 | print() |
| 185 | print_troubleshooting() |
| 186 | return 1 |
| 187 | |
| 188 | print() |
| 189 | |
| 190 | # Test connectivity if possible |
| 191 | try: |
| 192 | connectivity = asyncio.run(test_server_connectivity()) |
| 193 | print() |
| 194 | |
| 195 | if connectivity is False: |
| 196 | print_troubleshooting() |
| 197 | return 1 |
| 198 | |
| 199 | except Exception as e: |
| 200 | print(f"Error testing connectivity: {e}") |
| 201 | print() |
| 202 | |
| 203 | # Print test commands |
| 204 | print_test_commands() |
| 205 | |
| 206 | print("📋 Next Steps:") |
| 207 | print("1. Start Agent Zero server if not running") |
| 208 | print("2. Run one of the curl commands above") |
| 209 | print("3. Check for successful 200 response with agent card JSON") |
| 210 | print("4. If issues occur, see troubleshooting section") |
| 211 | |
| 212 | return 0 |
| 213 | |
| 214 | |
| 215 | if __name__ == "__main__": |
| 216 | sys.exit(main()) |