#84 - adjusted A2A project passing via message.metadata
deci committed
Dec 12, 2025 at 12:22 UTC
31daa48a2a54ddc3383b8822188402e9da723165
1 file changed
+40
-50
python/helpers/fasta2a_server.py
+40
-50
@@ -60,11 +60,6 @@ except ImportError: # pragma: no cover – library not installed
60
61
_PRINTER = PrintStyle(italic=True, font_color="purple", padding=False)
62
63
-# Map message_id to project name (thread-safe, no race conditions)
64
-# Each message has unique ID that correlates request to worker
65
-_a2a_message_projects: dict[str, str] = {}
66
-_a2a_project_lock = threading.Lock()
67
-
63
64
class AgentZeroWorker(Worker): # type: ignore[misc]
65
"""Agent Zero implementation of FastA2A Worker."""
@@ -89,15 +84,11 @@ class AgentZeroWorker(Worker): # type: ignore[misc]
84
cfg = initialize_agent()
85
context = AgentContext(cfg, type=AgentContextType.BACKGROUND)
86
92
- # Retrieve project by message_id (direct lookup, no race conditions)
93
- # Note: Request has messageId (camelCase), but FastA2A converts it to message_id (snake_case)
94
- project_name = None
95
- message_id = message.get('message_id') # FastA2A converts camelCase to snake_case
96
- if message_id:
97
- with _a2a_project_lock:
98
- project_name = _a2a_message_projects.pop(message_id, None)
99
- if project_name:
100
- _PRINTER.print(f"[A2A] Retrieved project for message {message_id}: {project_name}")
87
+ # Retrieve project from message.metadata (standard A2A pattern)
88
+ metadata = message.get('metadata', {}) or {}
89
+ project_name = metadata.get('project')
90
+ if project_name:
91
+ _PRINTER.print(f"[A2A] Retrieved project from message.metadata: {project_name}")
92
93
# Activate project if specified
94
if project_name:
@@ -487,51 +478,50 @@ class DynamicA2AProxy:
478
})
479
return
480
490
- # If project specified, we need to extract message_id from request body
491
- # to correlate project with the specific message (no race conditions)
481
+ # If project specified, inject it into the request payload
482
if project_name:
493
- # Buffer all messages for replay
483
+ # Buffer messages and modify before returning the complete body
484
received_messages = []
495
- replay_index = 0
496
- message_id_extracted = False
485
+ body_modified = False
486
original_receive = receive
487
488
async def receive_wrapper():
500
- nonlocal replay_index, message_id_extracted
501
-
502
- # If replaying buffered messages, return them in order
503
- if replay_index < len(received_messages):
504
- msg = received_messages[replay_index]
505
- replay_index += 1
506
- return msg
489
+ nonlocal body_modified
490
508
- # Otherwise, receive and buffer the next message
491
+ # Receive and buffer the next message
492
message = await original_receive()
493
received_messages.append(message)
494
512
- # Parse message_id when we get the complete body
513
- if message['type'] == 'http.request':
514
- # If this is the last chunk, parse the full body
515
- if not message.get('more_body', False) and not message_id_extracted:
516
- message_id_extracted = True
517
- try:
518
- import json
519
- # Reconstruct full body from all buffered messages
520
- body_parts = [msg.get('body', b'') for msg in received_messages if msg['type'] == 'http.request']
521
- full_body = b''.join(body_parts)
522
- data = json.loads(full_body)
523
- # Handle JSON-RPC format: params.message.messageId (camelCase!)
524
- if 'params' in data and 'message' in data['params']:
525
- msg = data['params']['message']
526
- message_id = msg.get('messageId') # camelCase in raw JSON
527
- if message_id:
528
- with _a2a_project_lock:
529
- _a2a_message_projects[message_id] = project_name
530
- _PRINTER.print(f"[A2A] Stored project '{project_name}' for message {message_id}")
531
- # Reset replay index so FastA2A can replay from start
532
- replay_index = 0
533
- except Exception as e:
534
- _PRINTER.print(f"[A2A] Failed to parse message_id: {e}")
495
+ # When we get the complete body, inject project into JSON
496
+ if message['type'] == 'http.request' and not message.get('more_body', False) and not body_modified:
497
+ body_modified = True
498
+ try:
499
+ import json
500
+ # Reconstruct full body from all buffered messages
501
+ body_parts = [msg.get('body', b'') for msg in received_messages if msg['type'] == 'http.request']
502
+ full_body = b''.join(body_parts)
503
+ data = json.loads(full_body)
504
+
505
+ # INJECT project into message.metadata (standard A2A pattern)
506
+ if 'params' in data and 'message' in data['params']:
507
+ msg_data = data['params']['message']
508
+ # Initialize metadata if it doesn't exist
509
+ if 'metadata' not in msg_data or msg_data['metadata'] is None:
510
+ msg_data['metadata'] = {}
511
+ msg_data['metadata']['project'] = project_name
512
+ _PRINTER.print(f"[A2A] Injected project '{project_name}' into message.metadata")
513
+
514
+ # Serialize back to JSON
515
+ modified_body = json.dumps(data).encode('utf-8')
516
+
517
+ # Return modified message IMMEDIATELY (before FastA2A processes it)
518
+ return {
519
+ 'type': 'http.request',
520
+ 'body': modified_body,
521
+ 'more_body': False
522
+ }
523
+ except Exception as e:
524
+ _PRINTER.print(f"[A2A] Failed to inject project into payload: {e}")
525
526
return message
527