@samitouri / QOSami_Agents_Zero-1 / commits / f3605eb5

fix: token auth and reconfiguration

Rafael Uzarowski committed Jul 31, 2025 at 19:12 UTC f3605eb53bd358700ccd4774c593eef78498413f
4 files changed +509 -204
python/helpers/fasta2a_server.py
+298 -139
@@ -4,132 +4,157 @@ import uuid
4 import atexit
5 from typing import Any, List
6 import contextlib
7 +import threading
8
9 from python.helpers import settings
9 -from starlette.middleware.base import BaseHTTPMiddleware
10 from starlette.requests import Request
11 -from starlette.responses import PlainTextResponse
12 -from starlette.middleware import Middleware
11
12 # Local imports
13 from python.helpers.print_style import PrintStyle
14 from agent import AgentContext, UserMessage
15 from initialize import initialize_agent
16
19 -# Attempt to import FastA2A – fall back to stubs so linters stay quiet
17 +# Import FastA2A
18 try:
21 - from fasta2a import Worker # type: ignore
19 + from fasta2a import Worker, FastA2A # type: ignore
20 from fasta2a.broker import InMemoryBroker # type: ignore
21 from fasta2a.storage import InMemoryStorage # type: ignore
22 from fasta2a.schema import Message, Artifact, AgentProvider, Skill # type: ignore
25 -
23 FASTA2A_AVAILABLE = True
24 except ImportError: # pragma: no cover – library not installed
25 FASTA2A_AVAILABLE = False
26 + # Minimal stubs for type checkers when FastA2A is not available
27
28 class Worker: # type: ignore
31 - """Stub so type-checkers don’t complain when FastA2A is absent."""
29 + def __init__(self, **kwargs):
30 + pass
31 +
32 + async def run_task(self, params):
33 + pass
34 +
35 + async def cancel_task(self, params):
36 + pass
37
38 + def build_message_history(self, history):
39 + return []
40 +
41 + def build_artifacts(self, result):
42 + return []
43 +
44 + class FastA2A: # type: ignore
45 + def __init__(self, **kwargs):
46 + pass
47 +
48 + async def __call__(self, scope, receive, send):
49 + pass
50 +
51 + class InMemoryBroker: # type: ignore
52 pass
53
35 - # Minimal stubs for type checkers
54 + class InMemoryStorage: # type: ignore
55 + async def update_task(self, **kwargs):
56 + pass
57 +
58 Message = Artifact = AgentProvider = Skill = Any # type: ignore
37 - InMemoryBroker = InMemoryStorage = object # type: ignore
59
60 _PRINTER = PrintStyle(italic=True, font_color="purple", padding=False)
61
41 -if FASTA2A_AVAILABLE:
42 - class AgentZeroWorker(Worker): # type: ignore[misc]
43 - """Agent Zero implementation of FastA2A Worker."""
62
45 - async def run_task(self, params: Any) -> None: # params: TaskSendParams
46 - """Execute a task by processing the message through Agent Zero."""
47 - try:
48 - task_id = params['id']
49 - context_id = params['context_id']
50 - message = params['message']
51 -
52 - _PRINTER.print(f"[A2A] Processing task {task_id} in context {context_id}")
53 -
54 - # Convert A2A message to Agent Zero format
55 - agent_message = self._convert_message(message)
56 -
57 - # Get or create Agent Zero context
58 - context = AgentContext.get(context_id)
59 - if not context:
60 - # Create new context for this A2A conversation
61 - cfg = initialize_agent()
62 - context = AgentContext(cfg, id=context_id)
63 -
64 - # Log user message so it appears instantly in UI chat window
65 - context.log.log(
66 - type="user", # type: ignore[arg-type]
67 - heading="Remote user message",
68 - content=agent_message.message,
69 - kvps={"from": "A2A"},
70 - temp=False,
71 - )
63 +class AgentZeroWorker(Worker): # type: ignore[misc]
64 + """Agent Zero implementation of FastA2A Worker."""
65
73 - # Process message through Agent Zero (includes response)
74 - task = context.communicate(agent_message)
75 - result_text = await task.result()
76 -
77 - # Build A2A message from result
78 - response_message: Message = { # type: ignore
79 - 'role': 'agent',
80 - 'parts': [{'kind': 'text', 'text': str(result_text)}],
81 - 'kind': 'message',
82 - 'message_id': str(uuid.uuid4())
83 - }
84 -
85 - await self.storage.update_task( # type: ignore[attr-defined]
86 - task_id=task_id,
87 - state='completed',
88 - new_messages=[response_message]
89 - )
66 + def __init__(self, broker, storage):
67 + super().__init__(broker=broker, storage=storage)
68 + self.storage = storage
69
91 - _PRINTER.print(f"[A2A] Completed task {task_id}")
70 + async def run_task(self, params: Any) -> None: # params: TaskSendParams
71 + """Execute a task by processing the message through Agent Zero."""
72 + try:
73 + task_id = params['id']
74 + context_id = params['context_id']
75 + message = params['message']
76 +
77 + _PRINTER.print(f"[A2A] Processing task {task_id} in context {context_id}")
78 +
79 + # Convert A2A message to Agent Zero format
80 + agent_message = self._convert_message(message)
81 +
82 + # Get or create Agent Zero context
83 + context = AgentContext.get(context_id)
84 + if not context:
85 + # Create new context for this A2A conversation
86 + cfg = initialize_agent()
87 + context = AgentContext(cfg, id=context_id)
88 +
89 + # Log user message so it appears instantly in UI chat window
90 + context.log.log(
91 + type="user", # type: ignore[arg-type]
92 + heading="Remote user message",
93 + content=agent_message.message,
94 + kvps={"from": "A2A"},
95 + temp=False,
96 + )
97
93 - except Exception as e:
94 - _PRINTER.print(f"[A2A] Error processing task {params.get('id', 'unknown')}: {e}")
95 - await self.storage.update_task(
96 - task_id=params.get('id', 'unknown'),
97 - state='failed'
98 - )
98 + # Process message through Agent Zero (includes response)
99 + task = context.communicate(agent_message)
100 + result_text = await task.result()
101
100 - async def cancel_task(self, params: Any) -> None: # params: TaskIdParams
101 - """Cancel a running task."""
102 - task_id = params['id']
103 - _PRINTER.print(f"[A2A] Cancelling task {task_id}")
104 - await self.storage.update_task(task_id=task_id, state='canceled') # type: ignore[attr-defined]
102 + # Build A2A message from result
103 + response_message: Message = { # type: ignore
104 + 'role': 'agent',
105 + 'parts': [{'kind': 'text', 'text': str(result_text)}],
106 + 'kind': 'message',
107 + 'message_id': str(uuid.uuid4())
108 + }
109
106 - def build_message_history(self, history: List[Any]) -> List[Message]: # type: ignore
107 - # Not used in this simplified implementation
108 - return []
110 + await self.storage.update_task( # type: ignore[attr-defined]
111 + task_id=task_id,
112 + state='completed',
113 + new_messages=[response_message]
114 + )
115
110 - def build_artifacts(self, result: Any) -> List[Artifact]: # type: ignore
111 - # No artifacts for now
112 - return []
116 + _PRINTER.print(f"[A2A] Completed task {task_id}")
117
114 - def _convert_message(self, a2a_message: Message) -> UserMessage: # type: ignore
115 - """Convert A2A message to Agent Zero UserMessage."""
116 - # Extract text from message parts
117 - text_parts = [part.get('text', '') for part in a2a_message.get('parts', []) if part.get('kind') == 'text']
118 - message_text = '\n'.join(text_parts)
119 -
120 - # Extract file attachments
121 - attachments = []
122 - for part in a2a_message.get('parts', []):
123 - if part.get('kind') == 'file':
124 - file_info = part.get('file', {})
125 - if 'uri' in file_info:
126 - attachments.append(file_info['uri'])
127 -
128 - return UserMessage(
129 - message=message_text,
130 - attachments=attachments
118 + except Exception as e:
119 + _PRINTER.print(f"[A2A] Error processing task {params.get('id', 'unknown')}: {e}")
120 + await self.storage.update_task(
121 + task_id=params.get('id', 'unknown'),
122 + state='failed'
123 )
124
125 + async def cancel_task(self, params: Any) -> None: # params: TaskIdParams
126 + """Cancel a running task."""
127 + task_id = params['id']
128 + _PRINTER.print(f"[A2A] Cancelling task {task_id}")
129 + await self.storage.update_task(task_id=task_id, state='canceled') # type: ignore[attr-defined]
130 +
131 + def build_message_history(self, history: List[Any]) -> List[Message]: # type: ignore
132 + # Not used in this simplified implementation
133 + return []
134 +
135 + def build_artifacts(self, result: Any) -> List[Artifact]: # type: ignore
136 + # No artifacts for now
137 + return []
138 +
139 + def _convert_message(self, a2a_message: Message) -> UserMessage: # type: ignore
140 + """Convert A2A message to Agent Zero UserMessage."""
141 + # Extract text from message parts
142 + text_parts = [part.get('text', '') for part in a2a_message.get('parts', []) if part.get('kind') == 'text']
143 + message_text = '\n'.join(text_parts)
144 +
145 + # Extract file attachments
146 + attachments = []
147 + for part in a2a_message.get('parts', []):
148 + if part.get('kind') == 'file':
149 + file_info = part.get('file', {})
150 + if 'uri' in file_info:
151 + attachments.append(file_info['uri'])
152 +
153 + return UserMessage(
154 + message=message_text,
155 + attachments=attachments
156 + )
157 +
158
159 class DynamicA2AProxy:
160 """Dynamic proxy for FastA2A server that allows reconfiguration."""
@@ -138,13 +163,20 @@ class DynamicA2AProxy:
163
164 def __init__(self):
165 self.app = None
141 - self._async_lock = asyncio.Lock()
166 + self.token = ""
167 + self._lock = threading.Lock() # Use threading.Lock instead of asyncio.Lock
168 self._startup_done: bool = False
169 self._worker_bg_task: asyncio.Task | None = None
170 + self._reconfigure_needed: bool = False # Flag for deferred reconfiguration
171
172 if FASTA2A_AVAILABLE:
173 + # Initialize with default token
174 + cfg = settings.get_settings()
175 + self.token = cfg.get("mcp_server_token", "")
176 self._configure()
177 self._register_shutdown()
178 + else:
179 + _PRINTER.print("[A2A] FastA2A not available, server will return 503")
180
181 @staticmethod
182 def get_instance():
@@ -152,14 +184,19 @@ class DynamicA2AProxy:
184 DynamicA2AProxy._instance = DynamicA2AProxy()
185 return DynamicA2AProxy._instance
186
187 + def reconfigure(self, token: str):
188 + """Reconfigure the FastA2A server with new token."""
189 + self.token = token
190 + if FASTA2A_AVAILABLE:
191 + with self._lock:
192 + # Mark that reconfiguration is needed - will be done on next request
193 + self._reconfigure_needed = True
194 + self._startup_done = False # Force restart on next request
195 + _PRINTER.print("[A2A] Reconfiguration scheduled for next request")
196 +
197 def _configure(self):
198 """Configure the FastA2A application with Agent Zero integration."""
199 try:
158 - # Import inside the method to handle missing dependencies gracefully
159 - from fasta2a import FastA2A # type: ignore
160 - from fasta2a.broker import InMemoryBroker # type: ignore
161 - from fasta2a.storage import InMemoryStorage # type: ignore
162 -
200 storage = InMemoryStorage() # type: ignore[arg-type]
201 broker = InMemoryBroker() # type: ignore[arg-type]
202
@@ -185,32 +222,8 @@ class DynamicA2AProxy:
222 "url": "https://github.com/frdel/agent-zero"
223 }
224
188 - # Authentication middleware (Bearer or X-API-KEY)
189 - class A2AAuthMiddleware(BaseHTTPMiddleware):
190 - async def dispatch(self, request: Request, call_next): # type: ignore[override]
191 - cfg = settings.get_settings()
192 - expected = cfg.get("a2a_token") or cfg.get("mcp_server_token")
193 - if not expected:
194 - return await call_next(request) # no auth configured
195 -
196 - auth_header = request.headers.get("Authorization", "")
197 - if auth_header.startswith("Bearer ") and auth_header.split(" ", 1)[1] == expected:
198 - return await call_next(request)
199 -
200 - api_key = request.headers.get("X-API-KEY") or request.query_params.get("api_key")
201 - if api_key == expected:
202 - return await call_next(request)
203 -
204 - # Fallback: check token in mount path (root_path contains the prefix stripped by DispatcherMiddleware)
205 - root_path = request.scope.get("root_path", "") # type: ignore[attr-defined]
206 - if root_path and f"/t-{expected}" in root_path:
207 - return await call_next(request)
208 -
209 - return PlainTextResponse("Unauthorized", 401)
210 -
211 - middleware_list = [Middleware(A2AAuthMiddleware)]
212 -
213 - self.app = FastA2A( # type: ignore
225 + # Create new FastA2A app with proper thread safety
226 + new_app = FastA2A( # type: ignore
227 storage=storage,
228 broker=broker,
229 name="Agent Zero",
@@ -222,7 +235,7 @@ class DynamicA2AProxy:
235 provider=provider,
236 skills=skills,
237 lifespan=None, # We manage lifespan manually
225 - middleware=middleware_list,
238 + middleware=[], # No middleware - we handle auth in wrapper
239 )
240
241 # Store for later lazy startup (needs active event-loop)
@@ -230,11 +243,15 @@ class DynamicA2AProxy:
243 self._broker = broker # type: ignore[attr-defined]
244 self._worker = AgentZeroWorker(broker=broker, storage=storage) # type: ignore[attr-defined]
245
246 + # Atomic update of the app
247 + self.app = new_app
248 +
249 _PRINTER.print("[A2A] FastA2A server configured successfully")
250
251 except Exception as e:
252 _PRINTER.print(f"[A2A] Failed to configure FastA2A server: {e}")
253 self.app = None
254 + raise
255
256 # ---------------------------------------------------------------------
257 # Shutdown handling
@@ -262,10 +279,33 @@ class DynamicA2AProxy:
279 with contextlib.suppress(asyncio.CancelledError):
280 await self._worker_bg_task
281 try:
265 - await self.app.task_manager.__aexit__(None, None, None) # type: ignore[attr-defined]
282 + if hasattr(self, 'app') and self.app:
283 + await self.app.task_manager.__aexit__(None, None, None) # type: ignore[attr-defined]
284 except Exception:
285 pass
286
287 + async def _async_reconfigure(self):
288 + """Perform async reconfiguration with proper lifecycle management."""
289 + _PRINTER.print("[A2A] Starting async reconfiguration")
290 +
291 + # Shutdown existing components
292 + await self._async_shutdown()
293 +
294 + # Reset startup state
295 + self._startup_done = False
296 + self._worker_bg_task = None
297 +
298 + # Reconfigure with new token
299 + self._configure()
300 +
301 + # Restart components
302 + await self._startup()
303 +
304 + # Clear reconfiguration flag
305 + self._reconfigure_needed = False
306 +
307 + _PRINTER.print("[A2A] Async reconfiguration completed")
308 +
309 async def _startup(self):
310 """Ensure TaskManager and Worker are running inside current event-loop."""
311 if self._startup_done or not FASTA2A_AVAILABLE:
@@ -283,15 +323,9 @@ class DynamicA2AProxy:
323 self._worker_bg_task = asyncio.create_task(_worker_loop())
324 _PRINTER.print("[A2A] Worker & TaskManager started")
325
286 - async def reconfigure(self):
287 - """Reconfigure the FastA2A server (placeholder for future use)."""
288 - async with self._async_lock:
289 - if FASTA2A_AVAILABLE:
290 - self._configure()
291 -
326 async def __call__(self, scope, receive, send):
293 - """ASGI application interface."""
294 - if self.app is None:
327 + """ASGI application interface with token-based routing."""
328 + if not FASTA2A_AVAILABLE:
329 # FastA2A not available, return 503
330 response = b'HTTP/1.1 503 Service Unavailable\r\n\r\nFastA2A not available'
331 await send({
@@ -305,12 +339,139 @@ class DynamicA2AProxy:
339 })
340 return
341
342 + # Check if reconfiguration is needed
343 + if self._reconfigure_needed:
344 + try:
345 + await self._async_reconfigure()
346 + except Exception as e:
347 + _PRINTER.print(f"[A2A] Error during reconfiguration: {e}")
348 + # Return 503 if reconfiguration failed
349 + await send({
350 + 'type': 'http.response.start',
351 + 'status': 503,
352 + 'headers': [[b'content-type', b'text/plain']],
353 + })
354 + await send({
355 + 'type': 'http.response.body',
356 + 'body': b'FastA2A reconfiguration failed',
357 + })
358 + return
359 +
360 + if self.app is None:
361 + # FastA2A not configured, return 503
362 + response = b'HTTP/1.1 503 Service Unavailable\r\n\r\nFastA2A not configured'
363 + await send({
364 + 'type': 'http.response.start',
365 + 'status': 503,
366 + 'headers': [[b'content-type', b'text/plain']],
367 + })
368 + await send({
369 + 'type': 'http.response.body',
370 + 'body': response,
371 + })
372 + return
373 +
374 # Lazy-start background components the first time we get a request
309 - if FASTA2A_AVAILABLE and not self._startup_done:
310 - await self._startup()
375 + if not self._startup_done:
376 + try:
377 + _PRINTER.print("[A2A] Starting up FastA2A components")
378 + await self._startup()
379 + except Exception as e:
380 + _PRINTER.print(f"[A2A] Error during startup: {e}")
381 + # Return 503 if startup failed
382 + await send({
383 + 'type': 'http.response.start',
384 + 'status': 503,
385 + 'headers': [[b'content-type', b'text/plain']],
386 + })
387 + await send({
388 + 'type': 'http.response.body',
389 + 'body': b'FastA2A startup failed',
390 + })
391 + return
392 +
393 + # Handle token-based routing: /a2a/t-{token}/... or /t-{token}/...
394 + path = scope.get('path', '')
395 +
396 + # Strip /a2a prefix if present (DispatcherMiddleware doesn't always strip it)
397 + if path.startswith('/a2a'):
398 + path = path[4:] # Remove '/a2a' prefix
399 +
400 + # Check if path matches token pattern /t-{token}/
401 + if path.startswith('/t-') and '/' in path[3:]:
402 + # Extract token from path
403 + path_parts = path[3:].split('/', 1) # Remove '/t-' prefix
404 + request_token = path_parts[0]
405 + remaining_path = '/' + path_parts[1] if len(path_parts) > 1 else '/'
406 +
407 + # Validate token
408 + cfg = settings.get_settings()
409 + expected_token = cfg.get("mcp_server_token")
410 +
411 + if expected_token and request_token != expected_token:
412 + # Invalid token, return 401
413 + await send({
414 + 'type': 'http.response.start',
415 + 'status': 401,
416 + 'headers': [[b'content-type', b'text/plain']],
417 + })
418 + await send({
419 + 'type': 'http.response.body',
420 + 'body': b'Unauthorized',
421 + })
422 + return
423 +
424 + # Update scope with cleaned path
425 + scope = dict(scope)
426 + scope['path'] = remaining_path
427 + else:
428 + # No token in path, check other auth methods
429 + request = Request(scope, receive=receive)
430 +
431 + cfg = settings.get_settings()
432 + expected = cfg.get("mcp_server_token")
433 +
434 + if expected:
435 + auth_header = request.headers.get("Authorization", "")
436 + api_key = request.headers.get("X-API-KEY") or request.query_params.get("api_key")
437 +
438 + is_authorized = (
439 + (auth_header.startswith("Bearer ") and auth_header.split(" ", 1)[1] == expected) or
440 + (api_key == expected)
441 + )
442
312 - # Delegate to FastA2A app
313 - await self.app(scope, receive, send)
443 + if not is_authorized:
444 + # No valid auth, return 401
445 + await send({
446 + 'type': 'http.response.start',
447 + 'status': 401,
448 + 'headers': [[b'content-type', b'text/plain']],
449 + })
450 + await send({
451 + 'type': 'http.response.body',
452 + 'body': b'Unauthorized',
453 + })
454 + return
455 + else:
456 + _PRINTER.print("[A2A] No expected token found in settings")
457 +
458 + # Delegate to FastA2A app with cleaned scope
459 + with self._lock:
460 + app = self.app
461 + if app:
462 + await app(scope, receive, send)
463 + else:
464 + # App not configured, return 503
465 + await send({
466 + 'type': 'http.response.start',
467 + 'status': 503,
468 + 'headers': [[b'content-type', b'text/plain']],
469 + })
470 + await send({
471 + 'type': 'http.response.body',
472 + 'body': b'FastA2A app not configured',
473 + })
474 + return
475
476
477 def is_available():
@@ -320,6 +481,4 @@ def is_available():
481
482 def get_proxy():
483 """Get the FastA2A proxy instance."""
323 - if not FASTA2A_AVAILABLE:
324 - return None
484 return DynamicA2AProxy.get_instance()
python/helpers/settings.py
+12
@@ -1353,6 +1353,18 @@ def _apply_settings(previous: Settings | None):
1353 update_mcp_token, current_token
1354 ) # TODO overkill, replace with background task
1355
1356 + # update token in a2a server
1357 + if not previous or current_token != previous["mcp_server_token"]:
1358 +
1359 + async def update_a2a_token(token: str):
1360 + from python.helpers.fasta2a_server import DynamicA2AProxy
1361 +
1362 + DynamicA2AProxy.get_instance().reconfigure(token=token)
1363 +
1364 + task4 = defer.DeferredTask().start_task(
1365 + update_a2a_token, current_token
1366 + ) # TODO overkill, replace with background task
1367 +
1368
1369 def _env_to_dict(data: str):
1370 env_dict = {}
run_ui.py
+3 -24
@@ -9,21 +9,13 @@ import threading
9 from flask import Flask, request, Response, session
10 from flask_basicauth import BasicAuth
11 import initialize
12 -from python.helpers import files, git, mcp_server
12 +from python.helpers import files, git, mcp_server, fasta2a_server
13 from python.helpers.files import get_abs_path
14 from python.helpers import runtime, dotenv, process
15 from python.helpers.extract_tools import load_classes_from_folder
16 from python.helpers.api import ApiHandler
17 from python.helpers.print_style import PrintStyle
18
19 -# Try to import fasta2a_server, but handle gracefully if it fails
20 -try:
21 - from python.helpers import fasta2a_server
22 - FASTA2A_AVAILABLE = True
23 -except ImportError:
24 - FASTA2A_AVAILABLE = False
25 - fasta2a_server = None
26 -
19
20 # Set the new timezone to 'UTC'
21 os.environ["TZ"] = "UTC"
@@ -223,26 +215,13 @@ def run():
215 # add the webapp, mcp, and a2a to the app
216 middleware_routes = {
217 "/mcp": ASGIMiddleware(app=mcp_server.DynamicMcpProxy.get_instance()), # type: ignore
218 + "/a2a": ASGIMiddleware(app=fasta2a_server.DynamicA2AProxy.get_instance()), # type: ignore
219 }
220
228 - # Add A2A server if available
229 - if FASTA2A_AVAILABLE and fasta2a_server and fasta2a_server.is_available():
230 - a2a_proxy = fasta2a_server.get_proxy()
231 - if a2a_proxy:
232 - from python.helpers import settings as _s
233 - cfg = _s.get_settings()
234 - token = cfg.get("a2a_token") or cfg.get("mcp_server_token")
235 - mount_path = f"/a2a/t-{token}"
236 - middleware_routes[mount_path] = ASGIMiddleware(app=a2a_proxy) # type: ignore
237 - # Also mount at /a2a for agent card and convenience
238 - middleware_routes["/a2a"] = ASGIMiddleware(app=a2a_proxy) # type: ignore
239 - PrintStyle().print(f"FastA2A server enabled at {mount_path} and /a2a")
240 - else:
241 - PrintStyle().debug("FastA2A server not available")
242 -
221 app = DispatcherMiddleware(webapp, middleware_routes) # type: ignore
222
223 PrintStyle().debug(f"Starting server at http://{host}:{port} ...")
224 + PrintStyle().debug("FastA2A mounted at: /a2a")
225
226 server = make_server(
227 host=host,
test_fasta2a_client.py
+196 -41
@@ -1,56 +1,211 @@
1 +#!/usr/bin/env python3
2 +"""
3 +Test script to verify FastA2A agent card routing and authentication.
4 +"""
5 +
6 import asyncio
7 import sys
3 -from typing import Any, Dict
8 +from python.helpers import settings
9 +
10 +
11 +def get_test_urls():
12 + """Get the URLs to test based on current settings."""
13 + try:
14 + cfg = settings.get_settings()
15 + token = cfg.get("mcp_server_token", "")
16 +
17 + if not token:
18 + print("❌ No mcp_server_token found in settings")
19 + return None
20 +
21 + base_url = "http://localhost:50101"
22 +
23 + urls = {
24 + "token_based": f"{base_url}/a2a/t-{token}/.well-known/agent.json",
25 + "bearer_auth": f"{base_url}/a2a/.well-known/agent.json",
26 + "api_key_header": f"{base_url}/a2a/.well-known/agent.json",
27 + "api_key_query": f"{base_url}/a2a/.well-known/agent.json?api_key={token}"
28 + }
29
5 -from python.helpers.fasta2a_client import (
6 - connect_to_agent,
7 - is_client_available,
8 -)
30 + return {"token": token, "urls": urls}
31
32 + except Exception as e:
33 + print(f"❌ Error getting settings: {e}")
34 + return None
35
11 -async def chat_loop(base_url: str):
12 - if not is_client_available():
13 - print("FastA2A client library not available – install `fasta2a httpx` first.")
36 +
37 +def print_test_commands():
38 + """Print curl commands to test FastA2A authentication."""
39 + data = get_test_urls()
40 + if not data:
41 return
42
16 - # Establish connection (agent card is fetched internally)
17 - async with await connect_to_agent(base_url) as conn:
18 - print(f"[✓] Connected to {base_url}")
19 - print('Type "exit" to quit.')
43 + token = data["token"]
44 + urls = data["urls"]
45
21 - while True:
22 - try:
23 - user_msg = input("You: ").strip()
24 - except (EOFError, KeyboardInterrupt):
25 - print()
26 - break
46 + print("🚀 FastA2A Agent Card Testing Commands")
47 + print("=" * 60)
48 + print(f"Current token: {token}")
49 + print()
50 +
51 + print("1️⃣ Token-based URL (recommended):")
52 + print(f" curl -v '{urls['token_based']}'")
53 + print()
54 +
55 + print("2️⃣ Bearer authentication:")
56 + print(f" curl -v -H 'Authorization: Bearer {token}' '{urls['bearer_auth']}'")
57 + print()
58 +
59 + print("3️⃣ API key header:")
60 + print(f" curl -v -H 'X-API-KEY: {token}' '{urls['api_key_header']}'")
61 + print()
62 +
63 + print("4️⃣ API key query parameter:")
64 + print(f" curl -v '{urls['api_key_query']}'")
65 + print()
66 +
67 + print("Expected response (if working):")
68 + print(" HTTP/1.1 200 OK")
69 + print(" Content-Type: application/json")
70 + print(" {")
71 + print(' "name": "Agent Zero",')
72 + print(' "version": "1.0.0",')
73 + print(' "skills": [...]')
74 + print(" }")
75 + print()
76 +
77 + print("Expected error (if auth fails):")
78 + print(" HTTP/1.1 401 Unauthorized")
79 + print(" Unauthorized")
80 + print()
81 +
82 +
83 +def print_troubleshooting():
84 + """Print troubleshooting information."""
85 + print("🔧 Troubleshooting FastA2A Issues")
86 + print("=" * 40)
87 + print()
88 + print("1. Server not running:")
89 + print(" - Make sure Agent Zero is running: python run_ui.py")
90 + print(" - Check the correct port (default: 50101)")
91 + print()
92 +
93 + print("2. Authentication failures:")
94 + print(" - Verify token matches in settings")
95 + print(" - Check token format (should be 16 characters)")
96 + print(" - Try different auth methods")
97 + print()
98 +
99 + print("3. FastA2A not available:")
100 + print(" - Install FastA2A: pip install fasta2a")
101 + print(" - Check server logs for FastA2A configuration errors")
102 + print()
103 +
104 + print("4. Routing issues:")
105 + print(" - Verify /a2a prefix is working")
106 + print(" - Check DispatcherMiddleware configuration")
107 + print(" - Look for FastA2A startup messages in logs")
108 + print()
109 +
110 +
111 +def validate_token_format():
112 + """Validate that the token format is correct."""
113 + try:
114 + cfg = settings.get_settings()
115 + token = cfg.get("mcp_server_token", "")
116
28 - if user_msg.lower() in {"exit", "quit"}:
29 - break
117 + print("🔍 Token Validation")
118 + print("=" * 25)
119
120 + if not token:
121 + print("❌ No token found")
122 + return False
123 +
124 + print(f"✅ Token found: {token}")
125 + print(f"✅ Token length: {len(token)} characters")
126 +
127 + if len(token) != 16:
128 + print("⚠️ Warning: Expected token length is 16 characters")
129 +
130 + # Check token characters
131 + if token.isalnum():
132 + print("✅ Token contains only alphanumeric characters")
133 + else:
134 + print("⚠️ Warning: Token contains non-alphanumeric characters")
135 +
136 + return True
137 +
138 + except Exception as e:
139 + print(f"❌ Error validating token: {e}")
140 + return False
141 +
142 +
143 +async def test_server_connectivity():
144 + """Test basic server connectivity."""
145 + try:
146 + import httpx
147 +
148 + print("🌐 Server Connectivity Test")
149 + print("=" * 30)
150 +
151 + async with httpx.AsyncClient() as client:
152 try:
32 - # Send message & get immediate task response
33 - task_response: Dict[str, Any] = await conn.send_message(message=user_msg)
34 - task_id = task_response.get("result", {}).get("id")
35 - if not task_id:
36 - print("[!] No task ID returned – response: ", task_response)
37 - continue
38 -
39 - # Wait for the task to complete and fetch final result
40 - final = await conn.wait_for_completion(task_id)
41 - latest_history = final["result"].get("history", [])
42 - assistant_parts = latest_history[-1].get("parts", []) if latest_history else []
43 - assistant_text = "\n".join(
44 - p.get("text", "") for p in assistant_parts if p.get("kind") == "text"
45 - )
46 - print(f"Assistant: {assistant_text}\n")
153 + # Test basic server
154 + await client.get("http://localhost:50101/", timeout=5.0)
155 + print("✅ Agent Zero server is running")
156 + return True
157 + except httpx.ConnectError:
158 + print("❌ Cannot connect to Agent Zero server")
159 + print(" Make sure the server is running: python run_ui.py")
160 + return False
161 except Exception as e:
48 - print(f"[!] Error: {e}")
162 + print(f"❌ Server connectivity error: {e}")
163 + return False
164
165 + except ImportError:
166 + print("ℹ️ httpx not available, skipping connectivity test")
167 + print(" Install with: pip install httpx")
168 + return None
169
51 -if __name__ == "__main__":
52 - if len(sys.argv) < 2:
53 - print("Usage: python test_fasta2a_client.py <AGENT_BASE_URL>")
54 - sys.exit(1)
170
56 - asyncio.run(chat_loop(sys.argv[1]))
171 +def main():
172 + """Main test function."""
173 + print("🧪 FastA2A Agent Card Testing Utility")
174 + print("=" * 45)
175 + print()
176 +
177 + # Validate token
178 + if not validate_token_format():
179 + print()
180 + print_troubleshooting()
181 + return 1
182 +
183 + print()
184 +
185 + # Test connectivity if possible
186 + try:
187 + connectivity = asyncio.run(test_server_connectivity())
188 + print()
189 +
190 + if connectivity is False:
191 + print_troubleshooting()
192 + return 1
193 +
194 + except Exception as e:
195 + print(f"Error testing connectivity: {e}")
196 + print()
197 +
198 + # Print test commands
199 + print_test_commands()
200 +
201 + print("📋 Next Steps:")
202 + print("1. Start Agent Zero server if not running")
203 + print("2. Run one of the curl commands above")
204 + print("3. Check for successful 200 response with agent card JSON")
205 + print("4. If issues occur, see troubleshooting section")
206 +
207 + return 0
208 +
209 +
210 +if __name__ == "__main__":
211 + sys.exit(main())