@samitouri / QOSami_Agents_Zero / commits / d84d24b3

feat: A2A Client/Server Implementation

Rafael Uzarowski committed Jul 29, 2025 at 17:54 UTC d84d24b3bedea09b4880b2d888847c59df5dd83a
11 files changed +829 -15
prompts/agent.system.tool.a2a_chat.md new
+73
@@ -0,0 +1,73 @@
1 +### a2a_chat:
2 +This tool lets Agent Zero chat with any other FastA2A-compatible agent.
3 +It automatically keeps conversation **context** (so each subsequent call
4 +continues the same dialogue) and supports optional file attachments.
5 +
6 +#### What the tool can do
7 +* Start a brand-new conversation with a remote agent.
8 +* Continue an existing conversation transparently (context handled for you).
9 +* Send text plus optional file URIs (images, docs, etc.).
10 +* Receive the assistant’s reply as plain text.
11 +
12 +#### Arguments
13 +* `agent_url` (string, required) – Base URL of the *remote* agent.
14 + • Accepts `host:port`, `http://host:port`, or full path ending in `/a2a`.
15 +* `message` (string, required) – The text you want to send.
16 +* `attachments` (list[string], optional) – URIs pointing to files you want
17 + to send along with the message (can be http(s):// or file path).
18 +* `reset` (boolean, optional) – Set to `true` to start a **new** conversation
19 + with the same `agent_url` (clears stored context). Default `false`.
20 +
21 +> Leave **context_id** out – the tool handles it internally.
22 +
23 +#### Usage – first message
24 +##### Request
25 +```json
26 +{
27 + "thoughts": [
28 + "I want to ask the weather-bot for today’s forecast."
29 + ],
30 + "headline": "Ask remote agent (weather-bot)",
31 + "tool_name": "a2a_chat",
32 + "tool_args": {
33 + "agent_url": "http://weather.example.com:8000/a2a",
34 + "message": "Hello! What’s the forecast for Berlin today?",
35 + "attachments": [],
36 + "reset": false
37 + }
38 +}
39 +```
40 +##### Response (assistant-side)
41 +```plaintext
42 +☀️ It will be sunny with a high of 22 °C.
43 +```
44 +
45 +#### Usage – follow-up (context automatically preserved)
46 +##### Request
47 +```json
48 +{
49 + "thoughts": [
50 + "Need tomorrow’s forecast too."
51 + ],
52 + "headline": "Follow-up question",
53 + "tool_name": "a2a_chat",
54 + "tool_args": {
55 + "agent_url": "http://weather.example.com:8000/a2a",
56 + "message": "And tomorrow?",
57 + "attachments": [],
58 + "reset": false
59 + }
60 +}
61 +```
62 +##### Response
63 +```plaintext
64 +🌦️ Partly cloudy with showers, high 18 °C.
65 +```
66 +
67 +#### Notes
68 +1. **New conversation** – omit previous `agent_url` or use a *different* URL.
69 +2. **Attachments** – supply absolute URIs ("http://…", "file:/…").
70 +3. The tool stores session IDs per `agent_url` inside the current
71 + `AgentContext` – no manual handling required.
72 +4. Use `"reset": true` to forget previous context and start a new chat.
73 +5. The remote agent must implement FastA2A v0.2+ protocol.
python/helpers/fasta2a_client.py new
+209
@@ -0,0 +1,209 @@
1 +import uuid
2 +from typing import Any, Dict, List, Optional
3 +from python.helpers.print_style import PrintStyle
4 +
5 +try:
6 + from fasta2a.client import A2AClient # type: ignore
7 + import httpx # type: ignore
8 + FASTA2A_CLIENT_AVAILABLE = True
9 +except ImportError:
10 + FASTA2A_CLIENT_AVAILABLE = False
11 + PrintStyle.warning("FastA2A client not available. Agent-to-agent communication disabled.")
12 +
13 +_PRINTER = PrintStyle(italic=True, font_color="cyan", padding=False)
14 +
15 +
16 +class AgentConnection:
17 + """Helper class for connecting to and communicating with other Agent Zero instances via FastA2A."""
18 +
19 + def __init__(self, agent_url: str, timeout: int = 30, token: Optional[str] = None):
20 + """Initialize connection to an agent.
21 +
22 + Args:
23 + agent_url: The base URL of the agent (e.g., "https://agent.example.com")
24 + timeout: Request timeout in seconds
25 + """
26 + if not FASTA2A_CLIENT_AVAILABLE:
27 + raise RuntimeError("FastA2A client not available")
28 +
29 + # Ensure scheme is present
30 + if not agent_url.startswith(('http://', 'https://')):
31 + agent_url = 'http://' + agent_url
32 +
33 + self.agent_url = agent_url.rstrip('/')
34 + self.timeout = timeout
35 + # Auth headers
36 + if token is None:
37 + import os
38 + token = os.getenv("A2A_TOKEN")
39 + headers = {}
40 + if token:
41 + headers["Authorization"] = f"Bearer {token}"
42 + headers["X-API-KEY"] = token
43 + self._http_client = httpx.AsyncClient(timeout=timeout, headers=headers) # type: ignore
44 + self._a2a_client = A2AClient(base_url=self.agent_url, http_client=self._http_client) # type: ignore
45 + self._agent_card: Optional[Dict[str, Any]] = None
46 + # Track conversation context automatically
47 + self._context_id: Optional[str] = None
48 +
49 + async def get_agent_card(self) -> Dict[str, Any]:
50 + """Retrieve the agent card from the remote agent."""
51 + if self._agent_card is None:
52 + try:
53 + response = await self._http_client.get(f"{self.agent_url}/.well-known/agent.json")
54 + response.raise_for_status()
55 + self._agent_card = response.json()
56 + _PRINTER.print(f"Retrieved agent card from {self.agent_url}")
57 + _PRINTER.print(f"Agent: {self._agent_card.get('name', 'Unknown')}")
58 + _PRINTER.print(f"Description: {self._agent_card.get('description', 'No description')}")
59 + except Exception as e:
60 + # Fallback: if URL contains '/a2a', try root path without it
61 + if "/a2a" in self.agent_url:
62 + root_url = self.agent_url.split("/a2a", 1)[0]
63 + try:
64 + response = await self._http_client.get(f"{root_url}/.well-known/agent.json")
65 + response.raise_for_status()
66 + self._agent_card = response.json()
67 + _PRINTER.print(f"Retrieved agent card from {root_url}")
68 + except Exception:
69 + pass # swallow, will re-raise below
70 + _PRINTER.print(f"[!] Could not connect to {self.agent_url}\n → Ensure the server is running and reachable.\n → Full error: {e}")
71 + raise RuntimeError(f"Could not retrieve agent card: {e}")
72 +
73 + return self._agent_card # type: ignore
74 +
75 + async def send_message(
76 + self,
77 + message: str,
78 + attachments: Optional[List[str]] = None,
79 + context_id: Optional[str] = None,
80 + metadata: Optional[Dict[str, Any]] = None
81 + ) -> Dict[str, Any]:
82 + """Send a message to the remote agent and return task response."""
83 + if not self._agent_card:
84 + await self.get_agent_card()
85 +
86 + # Re-use context automatically if caller did not supply one
87 + if context_id is None:
88 + context_id = self._context_id
89 +
90 + # Build message parts
91 + parts = [{'kind': 'text', 'text': message}]
92 +
93 + if attachments:
94 + for attachment in attachments:
95 + file_part = {'kind': 'file', 'file': {'uri': attachment}}
96 + parts.append(file_part) # type: ignore
97 +
98 + # Construct A2A message
99 + a2a_message = {
100 + 'role': 'user',
101 + 'parts': parts,
102 + 'kind': 'message',
103 + 'message_id': str(uuid.uuid4())
104 + }
105 +
106 + if context_id is not None:
107 + a2a_message['context_id'] = context_id
108 +
109 + # Send using the message/send method (not send_task)
110 + try:
111 + response = await self._a2a_client.send_message(
112 + message=a2a_message, # type: ignore
113 + metadata=metadata,
114 + configuration={'accepted_output_modes': ['application/json', 'text/plain'], 'blocking': True} # type: ignore
115 + )
116 +
117 + # Persist context id for subsequent calls
118 + try:
119 + ctx = response.get('result', {}).get('context_id') # type: ignore[index]
120 + if isinstance(ctx, str):
121 + self._context_id = ctx
122 + except Exception:
123 + pass # ignore if structure differs
124 + return response # type: ignore
125 + except Exception as e:
126 + _PRINTER.print(f"[A2A] Error sending message: {e}")
127 + raise
128 +
129 + async def get_task(self, task_id: str) -> Dict[str, Any]:
130 + """Get the status and results of a task.
131 +
132 + Args:
133 + task_id: The ID of the task to query
134 +
135 + Returns:
136 + Dictionary containing the task information
137 + """
138 + try:
139 + response = await self._a2a_client.get_task(task_id) # type: ignore
140 + return response # type: ignore
141 + except Exception as e:
142 + _PRINTER.print(f"Failed to get task {task_id}: {e}")
143 + raise RuntimeError(f"Failed to get task: {e}")
144 +
145 + async def wait_for_completion(self, task_id: str, poll_interval: int = 2, max_wait: int = 300) -> Dict[str, Any]:
146 + """Wait for a task to complete and return the final result.
147 +
148 + Args:
149 + task_id: The ID of the task to wait for
150 + poll_interval: How often to check task status (seconds)
151 + max_wait: Maximum time to wait (seconds)
152 +
153 + Returns:
154 + Dictionary containing the completed task information
155 + """
156 + import asyncio
157 +
158 + waited = 0
159 + while waited < max_wait:
160 + task_info = await self.get_task(task_id)
161 +
162 + if 'result' in task_info:
163 + task = task_info['result']
164 + status = task.get('status', {})
165 + state = status.get('state', 'unknown')
166 +
167 + if state in ['completed', 'failed', 'canceled']:
168 + _PRINTER.print(f"Task {task_id} finished with state: {state}")
169 + return task_info
170 + else:
171 + _PRINTER.print(f"Task {task_id} status: {state}")
172 +
173 + await asyncio.sleep(poll_interval)
174 + waited += poll_interval
175 +
176 + raise TimeoutError(f"Task {task_id} did not complete within {max_wait} seconds")
177 +
178 + async def close(self):
179 + """Close the HTTP client connection."""
180 + await self._http_client.aclose()
181 +
182 + async def __aenter__(self):
183 + """Async context manager entry."""
184 + return self
185 +
186 + async def __aexit__(self, exc_type, exc_val, exc_tb):
187 + """Async context manager exit."""
188 + await self.close()
189 +
190 +
191 +async def connect_to_agent(agent_url: str, timeout: int = 30) -> AgentConnection:
192 + """Create a connection to a remote agent.
193 +
194 + Args:
195 + agent_url: The base URL of the agent
196 + timeout: Request timeout in seconds
197 +
198 + Returns:
199 + AgentConnection instance
200 + """
201 + connection = AgentConnection(agent_url, timeout)
202 + # Verify connection by retrieving agent card
203 + await connection.get_agent_card()
204 + return connection
205 +
206 +
207 +def is_client_available() -> bool:
208 + """Check if FastA2A client is available."""
209 + return FASTA2A_CLIENT_AVAILABLE
python/helpers/fasta2a_server.py new
+325
@@ -0,0 +1,325 @@
1 +# noqa: D401 (docstrings) – internal helper
2 +import asyncio
3 +import uuid
4 +import atexit
5 +from typing import Any, List
6 +import contextlib
7 +
8 +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
13 +
14 +# Local imports
15 +from python.helpers.print_style import PrintStyle
16 +from agent import AgentContext, UserMessage
17 +from initialize import initialize_agent
18 +
19 +# Attempt to import FastA2A – fall back to stubs so linters stay quiet
20 +try:
21 + from fasta2a import Worker # type: ignore
22 + from fasta2a.broker import InMemoryBroker # type: ignore
23 + from fasta2a.storage import InMemoryStorage # type: ignore
24 + from fasta2a.schema import Message, Artifact, AgentProvider, Skill # type: ignore
25 +
26 + FASTA2A_AVAILABLE = True
27 +except ImportError: # pragma: no cover – library not installed
28 + FASTA2A_AVAILABLE = False
29 +
30 + class Worker: # type: ignore
31 + """Stub so type-checkers don’t complain when FastA2A is absent."""
32 +
33 + pass
34 +
35 + # Minimal stubs for type checkers
36 + Message = Artifact = AgentProvider = Skill = Any # type: ignore
37 + InMemoryBroker = InMemoryStorage = object # type: ignore
38 +
39 +_PRINTER = PrintStyle(italic=True, font_color="purple", padding=False)
40 +
41 +if FASTA2A_AVAILABLE:
42 + class AgentZeroWorker(Worker): # type: ignore[misc]
43 + """Agent Zero implementation of FastA2A Worker."""
44 +
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 + )
72 +
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 + )
90 +
91 + _PRINTER.print(f"[A2A] Completed task {task_id}")
92 +
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 + )
99 +
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]
105 +
106 + def build_message_history(self, history: List[Any]) -> List[Message]: # type: ignore
107 + # Not used in this simplified implementation
108 + return []
109 +
110 + def build_artifacts(self, result: Any) -> List[Artifact]: # type: ignore
111 + # No artifacts for now
112 + return []
113 +
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
131 + )
132 +
133 +
134 +class DynamicA2AProxy:
135 + """Dynamic proxy for FastA2A server that allows reconfiguration."""
136 +
137 + _instance = None
138 +
139 + def __init__(self):
140 + self.app = None
141 + self._async_lock = asyncio.Lock()
142 + self._startup_done: bool = False
143 + self._worker_bg_task: asyncio.Task | None = None
144 +
145 + if FASTA2A_AVAILABLE:
146 + self._configure()
147 + self._register_shutdown()
148 +
149 + @staticmethod
150 + def get_instance():
151 + if DynamicA2AProxy._instance is None:
152 + DynamicA2AProxy._instance = DynamicA2AProxy()
153 + return DynamicA2AProxy._instance
154 +
155 + def _configure(self):
156 + """Configure the FastA2A application with Agent Zero integration."""
157 + 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 +
163 + storage = InMemoryStorage() # type: ignore[arg-type]
164 + broker = InMemoryBroker() # type: ignore[arg-type]
165 +
166 + # Define Agent Zero's skills
167 + skills: List[Skill] = [{ # type: ignore
168 + "id": "general_assistance",
169 + "name": "General AI Assistant",
170 + "description": "Provides general AI assistance including code execution, file management, web browsing, and problem solving",
171 + "tags": ["ai", "assistant", "code", "files", "web", "automation"],
172 + "examples": [
173 + "Write and execute Python code",
174 + "Manage files and directories",
175 + "Browse the web and extract information",
176 + "Solve complex problems step by step",
177 + "Install software and manage systems"
178 + ],
179 + "input_modes": ["text/plain", "application/octet-stream"],
180 + "output_modes": ["text/plain", "application/json"]
181 + }]
182 +
183 + provider: AgentProvider = { # type: ignore
184 + "organization": "Agent Zero",
185 + "url": "https://github.com/frdel/agent-zero"
186 + }
187 +
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
214 + storage=storage,
215 + broker=broker,
216 + name="Agent Zero",
217 + description=(
218 + "A general AI assistant that can execute code, manage files, browse the web, and "
219 + "solve complex problems in an isolated Linux environment."
220 + ),
221 + version="1.0.0",
222 + provider=provider,
223 + skills=skills,
224 + lifespan=None, # We manage lifespan manually
225 + middleware=middleware_list,
226 + )
227 +
228 + # Store for later lazy startup (needs active event-loop)
229 + self._storage = storage # type: ignore[attr-defined]
230 + self._broker = broker # type: ignore[attr-defined]
231 + self._worker = AgentZeroWorker(broker=broker, storage=storage) # type: ignore[attr-defined]
232 +
233 + _PRINTER.print("[A2A] FastA2A server configured successfully")
234 +
235 + except Exception as e:
236 + _PRINTER.print(f"[A2A] Failed to configure FastA2A server: {e}")
237 + self.app = None
238 +
239 + # ---------------------------------------------------------------------
240 + # Shutdown handling
241 + # ---------------------------------------------------------------------
242 +
243 + def _register_shutdown(self):
244 + """Register an atexit hook to gracefully stop worker & task manager."""
245 +
246 + def _sync_shutdown():
247 + try:
248 + if not self._startup_done or not FASTA2A_AVAILABLE:
249 + return
250 + loop = asyncio.new_event_loop()
251 + loop.run_until_complete(self._async_shutdown())
252 + loop.close()
253 + except Exception:
254 + pass # ignore errors during interpreter shutdown
255 +
256 + atexit.register(_sync_shutdown)
257 +
258 + async def _async_shutdown(self):
259 + """Async shutdown: cancel worker task & close task manager."""
260 + if self._worker_bg_task and not self._worker_bg_task.done():
261 + self._worker_bg_task.cancel()
262 + with contextlib.suppress(asyncio.CancelledError):
263 + await self._worker_bg_task
264 + try:
265 + await self.app.task_manager.__aexit__(None, None, None) # type: ignore[attr-defined]
266 + except Exception:
267 + pass
268 +
269 + async def _startup(self):
270 + """Ensure TaskManager and Worker are running inside current event-loop."""
271 + if self._startup_done or not FASTA2A_AVAILABLE:
272 + return
273 + self._startup_done = True
274 +
275 + # Start task manager
276 + await self.app.task_manager.__aenter__() # type: ignore[attr-defined]
277 +
278 + async def _worker_loop():
279 + async with self._worker.run(): # type: ignore[attr-defined]
280 + await asyncio.Event().wait()
281 +
282 + # fire-and-forget background task – keep reference
283 + self._worker_bg_task = asyncio.create_task(_worker_loop())
284 + _PRINTER.print("[A2A] Worker & TaskManager started")
285 +
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 +
292 + async def __call__(self, scope, receive, send):
293 + """ASGI application interface."""
294 + if self.app is None:
295 + # FastA2A not available, return 503
296 + response = b'HTTP/1.1 503 Service Unavailable\r\n\r\nFastA2A not available'
297 + await send({
298 + 'type': 'http.response.start',
299 + 'status': 503,
300 + 'headers': [[b'content-type', b'text/plain']],
301 + })
302 + await send({
303 + 'type': 'http.response.body',
304 + 'body': response,
305 + })
306 + return
307 +
308 + # Lazy-start background components the first time we get a request
309 + if FASTA2A_AVAILABLE and not self._startup_done:
310 + await self._startup()
311 +
312 + # Delegate to FastA2A app
313 + await self.app(scope, receive, send)
314 +
315 +
316 +def is_available():
317 + """Check if FastA2A is available and properly configured."""
318 + return FASTA2A_AVAILABLE and DynamicA2AProxy.get_instance().app is not None
319 +
320 +
321 +def get_proxy():
322 + """Get the FastA2A proxy instance."""
323 + if not FASTA2A_AVAILABLE:
324 + return None
325 + return DynamicA2AProxy.get_instance()
python/helpers/settings.py
+22 -2
@@ -67,7 +67,7 @@ class Settings(TypedDict):
67 memory_memorize_enabled: bool
68 memory_memorize_consolidation: bool
69 memory_memorize_replace_threshold: float
70 -
70 +
71
72 api_keys: dict[str, str]
73
@@ -493,6 +493,25 @@ def convert_out(settings: Settings) -> SettingsOutput:
493 }
494 )
495
496 + # -------- A2A Section --------
497 + a2a_fields: list[SettingsField] = [
498 + {
499 + "id": "show_a2a_connection",
500 + "title": "Show A2A connection info",
501 + "description": "Display the URL (including token) other agents can use to connect via FastA2A.",
502 + "type": "button",
503 + "value": "Show",
504 + }
505 + ]
506 +
507 + a2a_section: SettingsSection = {
508 + "id": "a2a_server",
509 + "title": "A2A Connection",
510 + "description": "Share this connection string with other agents.",
511 + "fields": a2a_fields,
512 + "tab": "external",
513 + }
514 +
515 if runtime.is_dockerized():
516 auth_fields.append(
517 {
@@ -880,7 +899,7 @@ def convert_out(settings: Settings) -> SettingsOutput:
899
900 # TTS fields
901 tts_fields: list[SettingsField] = []
883 -
902 +
903 tts_fields.append(
904 {
905 "id": "tts_kokoro",
@@ -1028,6 +1047,7 @@ def convert_out(settings: Settings) -> SettingsOutput:
1047 speech_section,
1048 api_keys_section,
1049 auth_section,
1050 + a2a_section,
1051 mcp_client_section,
1052 mcp_server_section,
1053 backup_section,
python/tools/a2a_chat.py new
+57
@@ -0,0 +1,57 @@
1 +from python.helpers.tool import Tool, Response
2 +from python.helpers.print_style import PrintStyle
3 +
4 +try:
5 + from python.helpers.fasta2a_client import connect_to_agent, is_client_available # type: ignore
6 +except ImportError: # pragma: no cover – client helper missing
7 + is_client_available = lambda: False # type: ignore
8 +
9 +
10 +class A2AChatTool(Tool):
11 + """Communicate with another FastA2A-compatible agent."""
12 +
13 + async def execute(self, **kwargs):
14 + if not is_client_available():
15 + return Response(message="FastA2A client not available on this instance.", break_loop=False)
16 +
17 + agent_url: str | None = kwargs.get("agent_url") # required
18 + user_message: str | None = kwargs.get("message") # required
19 + attachments = kwargs.get("attachments", None) # optional list[str]
20 + reset = bool(kwargs.get("reset", False))
21 + if not agent_url or not isinstance(agent_url, str):
22 + return Response(message="agent_url argument missing", break_loop=False)
23 + if not user_message or not isinstance(user_message, str):
24 + return Response(message="message argument missing", break_loop=False)
25 +
26 + # Retrieve or create session cache on the Agent instance
27 + sessions: dict[str, str] = self.agent.get_data("_a2a_sessions") or {}
28 +
29 + # Handle reset flag – start fresh conversation
30 + if reset and agent_url in sessions:
31 + sessions.pop(agent_url, None)
32 +
33 + context_id = None if reset else sessions.get(agent_url)
34 + try:
35 + async with await connect_to_agent(agent_url) as conn:
36 + task_resp = await conn.send_message(user_message, attachments=attachments, context_id=context_id)
37 + task_id = task_resp.get("result", {}).get("id") # type: ignore[index]
38 + if not task_id:
39 + return Response(message="Remote agent failed to create task.", break_loop=False)
40 + final = await conn.wait_for_completion(task_id)
41 + new_context_id = final["result"].get("context_id") # type: ignore[index]
42 + if isinstance(new_context_id, str):
43 + sessions[agent_url] = new_context_id
44 + # persist back to agent data
45 + self.agent.set_data("_a2a_sessions", sessions)
46 + # Extract latest assistant text
47 + history = final["result"].get("history", [])
48 + assistant_text = ""
49 + if history:
50 + last_parts = history[-1].get("parts", [])
51 + assistant_text = "\n".join(
52 + p.get("text", "") for p in last_parts if p.get("kind") == "text"
53 + )
54 + return Response(message=assistant_text or "(no response)", break_loop=False)
55 + except Exception as e:
56 + PrintStyle.error(f"A2A chat error: {e}")
57 + return Response(message=f"A2A chat error: {e}", break_loop=False)
requirements.txt
+1
@@ -5,6 +5,7 @@ docker==7.1.0
5 duckduckgo-search==6.1.12
6 faiss-cpu==1.11.0
7 fastmcp==2.3.4
8 +fasta2a==0.5.0
9 flask[async]==3.0.3
10 flask-basicauth==0.2.0
11 flaredantic==0.1.4
run_ui.py
+32 -12
@@ -1,27 +1,33 @@
1 from datetime import timedelta
2 import os
3 import secrets
4 -import sys
4 import time
5 import socket
6 import struct
7 from functools import wraps
8 import threading
10 -import signal
11 -from typing import override
9 from flask import Flask, request, Response, session
10 from flask_basicauth import BasicAuth
11 import initialize
15 -from python.helpers import errors, files, git, mcp_server
12 +from python.helpers import files, git, mcp_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 +
27
28 # Set the new timezone to 'UTC'
29 os.environ["TZ"] = "UTC"
30 +os.environ["TOKENIZERS_PARALLELISM"] = "false"
31 # Apply the timezone change
32 if hasattr(time, 'tzset'):
33 time.tzset()
@@ -171,7 +177,7 @@ def run():
177 from werkzeug.serving import WSGIRequestHandler
178 from werkzeug.serving import make_server
179 from werkzeug.middleware.dispatcher import DispatcherMiddleware
174 - from a2wsgi import ASGIMiddleware, WSGIMiddleware
180 + from a2wsgi import ASGIMiddleware
181
182 PrintStyle().print("Starting server...")
183
@@ -214,13 +220,27 @@ def run():
220 for handler in handlers:
221 register_api_handler(webapp, handler)
222
217 - # add the webapp and mcp to the app
218 - app = DispatcherMiddleware(
219 - webapp,
220 - {
221 - "/mcp": ASGIMiddleware(app=mcp_server.DynamicMcpProxy.get_instance()), # type: ignore
222 - },
223 - )
223 + # add the webapp, mcp, and a2a to the app
224 + middleware_routes = {
225 + "/mcp": ASGIMiddleware(app=mcp_server.DynamicMcpProxy.get_instance()), # type: ignore
226 + }
227 +
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 +
243 + app = DispatcherMiddleware(webapp, middleware_routes) # type: ignore
244
245 PrintStyle().debug(f"Starting server at http://{host}:{port} ...")
246
test_fasta2a_client.py new
+56
@@ -0,0 +1,56 @@
1 +import asyncio
2 +import sys
3 +from typing import Any, Dict
4 +
5 +from python.helpers.fasta2a_client import (
6 + connect_to_agent,
7 + is_client_available,
8 +)
9 +
10 +
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.")
14 + return
15 +
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.')
20 +
21 + while True:
22 + try:
23 + user_msg = input("You: ").strip()
24 + except (EOFError, KeyboardInterrupt):
25 + print()
26 + break
27 +
28 + if user_msg.lower() in {"exit", "quit"}:
29 + break
30 +
31 + 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")
47 + except Exception as e:
48 + print(f"[!] Error: {e}")
49 +
50 +
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)
55 +
56 + asyncio.run(chat_loop(sys.argv[1]))
webui/components/settings/external/a2a-connection.html new
+47
@@ -0,0 +1,47 @@
1 +<html>
2 +
3 +<head>
4 + <title>Connection to A0 A2A Server</title>
5 +</head>
6 +
7 +<body>
8 + <div x-data>
9 + <p>Agent Zero A2A Server enables FastA2A protocol communication with other agents.</p>
10 + <p>Other agents can connect using the URL below (replace host if needed):</p>
11 +
12 + <h3>A2A Connection URL</h3>
13 + <div id="a2a-connection-example"></div>
14 +
15 + <script>
16 + setTimeout(() => {
17 + const url = window.location.origin;
18 + // Try to get a2a_token first, fallback to mcp_server_token
19 + let tokenField = null;
20 + try {
21 + const allFields = settingsModalProxy.settings.sections.flatMap(s => s.fields);
22 + tokenField = allFields.find(f => f.id === 'a2a_token') || allFields.find(f => f.id === 'mcp_server_token');
23 + } catch (e) { }
24 + const token = tokenField ? tokenField.value : '';
25 + const connectionUrl = `${url}/a2a/t-${token}`;
26 +
27 + const editor = ace.edit("a2a-connection-example");
28 + const dark = localStorage.getItem("darkMode");
29 + editor.setTheme(dark !== "false" ? "ace/theme/github_dark" : "ace/theme/tomorrow");
30 + editor.session.setMode("ace/mode/text");
31 + editor.setValue(connectionUrl);
32 + editor.clearSelection();
33 + editor.setReadOnly(true);
34 + }, 0);
35 + </script>
36 + </div>
37 +
38 + <style>
39 + #a2a-connection-example {
40 + width: 100%;
41 + height: 3em;
42 + }
43 + </style>
44 +
45 +</body>
46 +
47 +</html>
webui/js/settings.js
+3 -1
@@ -1,4 +1,3 @@
1 -
1 const settingsModalProxy = {
2 isOpen: false,
3 settings: {},
@@ -291,6 +290,9 @@ const settingsModalProxy = {
290 openModal("settings/backup/backup.html");
291 } else if (field.id === "backup_restore") {
292 openModal("settings/backup/restore.html");
293 + } else if (field.id === "show_a2a_connection") {
294 + console.log('Opening A2A connection modal...');
295 + openModal("settings/external/a2a-connection.html");
296 }
297 }
298 };
webui/public/a2a_server.svg new
+4
@@ -0,0 +1,4 @@
1 +<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
2 + <path d="M10 13a5 5 0 0 0 7.54.54l3-3a5 5 0 0 0-7.07-7.07l-1.72 1.72"/>
3 + <path d="M14 11a5 5 0 0 0-7.54-.54l-3 3a5 5 0 0 0 7.07 7.07l1.72-1.72"/>
4 +</svg>