diff --git a/A2A-MAS-no-SLIM/__init__.py b/A2A-MAS-no-SLIM/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/A2A-MAS-no-SLIM/agent.py b/A2A-MAS-no-SLIM/agent.py new file mode 100644 index 0000000..f6b2234 --- /dev/null +++ b/A2A-MAS-no-SLIM/agent.py @@ -0,0 +1,92 @@ +"""A2A Agent Server""" +import asyncio +import uvicorn +from a2a.server.apps import A2AStarletteApplication +from a2a.server.request_handlers import DefaultRequestHandler +from a2a.server.tasks import InMemoryTaskStore +from a2a.server.agent_execution import AgentExecutor, RequestContext +from a2a.server.events.event_queue import EventQueue +from a2a.types import AgentCard, AgentSkill, Message, TextPart, AgentCapabilities + + +class EchoAgent(AgentExecutor): + """Simple echo agent""" + + async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: + # Get incoming message + incoming = context.message + + # Extract text properly + if incoming.parts: + part = incoming.parts[0] + # part dict/object + if hasattr(part, 'text'): + text = part.text + elif isinstance(part, dict): + text = part.get('text', 'empty') + else: + text = str(part) + else: + text = "empty" + + # Create response + response = Message( + messageId=f"resp-{incoming.message_id}", + role="agent", + parts=[TextPart(type="text", text=f"Echo: {text}")] + ) + + # Send it back + await event_queue.enqueue_event(response) + + async def cancel(self, task_id: str, context_id: str | None) -> None: + """ cancel task - not used in echo agent """ + pass + + +async def start_agent(agent_id: int, port: int): + """Start A2A agent server""" + + # Agent card + card = AgentCard( + name=f'Agent-{agent_id}', + description=f'Echo agent {agent_id}', + url=f'http://localhost:{port}/', + version='1.0.0', + defaultInputModes=['text'], + defaultOutputModes=['text'], + skills=[AgentSkill( + id=f'echo_{agent_id}', + name='Echo', + description='Echoes back messages', + tags=['echo', 'test'], + examples=['hello', 'test'] + )], + capabilities=AgentCapabilities() + ) + + # Request handler + handler = DefaultRequestHandler( + agent_executor=EchoAgent(), + task_store=InMemoryTaskStore() + ) + + # Starlette app + app = A2AStarletteApplication( + agent_card=card, + http_handler=handler + ) + + # Start server + config = uvicorn.Config( + app=app.build(), + host="127.0.0.1", + port=port, + log_level="error", + access_log=False + ) + + server = uvicorn.Server(config) + asyncio.create_task(server.serve()) + + return server \ No newline at end of file diff --git a/A2A-MAS-no-SLIM/client.py b/A2A-MAS-no-SLIM/client.py new file mode 100644 index 0000000..d86081d --- /dev/null +++ b/A2A-MAS-no-SLIM/client.py @@ -0,0 +1,59 @@ +"""A2A Client""" +import time +import uuid +import httpx +from a2a.types import Message, TextPart + + +# Single shared client for all requests +client = httpx.AsyncClient( + timeout=30.0, + limits=httpx.Limits(max_connections=100, max_keepalive_connections=20) +) + + +async def send_message(sender_id: int, receiver_id: int, sender_port: int, receiver_port: int, msg_num: int): + """Send a message and measure latency""" + try: + # Create A2A message + message = Message( + messageId=str(uuid.uuid4()), + role="user", + parts=[TextPart( + type="text", + text=f"Message {msg_num} from Agent-{sender_id}" + )] + ) + + # JSON-RPC 2.0 request + request = { + "jsonrpc": "2.0", + "method": "message/send", + "params": { + "message": message.model_dump(mode='json'), + "metadata": {} + }, + "id": str(uuid.uuid4()) + } + + # Send to receiver's port + url = f"http://localhost:{receiver_port}/" + + t0 = time.perf_counter() + response = await client.post(url, json=request) + t1 = time.perf_counter() + + response.raise_for_status() + latency_ms = (t1 - t0) * 1000 + + return { + 'sender_id': sender_id, + 'receiver_id': receiver_id, + 'sender_port': sender_port, + 'receiver_port': receiver_port, + 'message_num': msg_num, + 'latency_ms': round(latency_ms, 3) + } + + except Exception: + return None diff --git a/A2A-MAS-no-SLIM/main.py b/A2A-MAS-no-SLIM/main.py new file mode 100644 index 0000000..0151ac9 --- /dev/null +++ b/A2A-MAS-no-SLIM/main.py @@ -0,0 +1,116 @@ +"""A2A Multi-Agent Latency Testing""" +import asyncio +import time +import random +from datetime import datetime +import pandas as pd +from agent import start_agent +from client import send_message + +random.seed() + +async def main(): + # Config + NUM_AGENTS = 10 + MESSAGES_PER_SENDER = 100 + BASE_PORT = 8000 + + # 1. Start agents (agent i runs on port BASE_PORT + i) + print(f"Starting {NUM_AGENTS} agents...") + servers = [] + for i in range(NUM_AGENTS): + server = await start_agent(agent_id=i, port=BASE_PORT + i) + servers.append(server) + + await asyncio.sleep(2) + print(f"Running on ports {BASE_PORT} to {BASE_PORT + NUM_AGENTS - 1}\n") + + # 2. Randomly assign senders/receivers + num_senders = random.randint(1, NUM_AGENTS - 1) + sender_ids = random.sample(range(NUM_AGENTS), num_senders) + receiver_ids = [i for i in range(NUM_AGENTS) if i not in sender_ids] + + print(f"Senders ({len(sender_ids)}): {sender_ids}") + print(f"Receivers ({len(receiver_ids)}): {receiver_ids}") + print(f"Total messages: {len(sender_ids) * MESSAGES_PER_SENDER}\n") + + # 3. Send messages and measure latency + print("Sending messages...") + tasks = [] + for sender_id in sender_ids: + for msg_num in range(MESSAGES_PER_SENDER): + receiver_id = random.choice(receiver_ids) + sender_port = BASE_PORT + sender_id + receiver_port = BASE_PORT + receiver_id + tasks.append(send_message(sender_id, receiver_id, sender_port, receiver_port, msg_num)) + + start_time = time.time() + results = await asyncio.gather(*tasks, return_exceptions=True) + elapsed = time.time() - start_time + + # Filter successful results + results = [r for r in results if r and not isinstance(r, Exception)] + print(f"Completed in {elapsed:.2f}s ({len(results)}/{len(tasks)} successful)\n") + + # 4. Compute statistics + df = pd.DataFrame(results) + + print("LATENCY STATISTICS") + print(f"Messages: {len(df)}") + print(f"Mean: {df['latency_ms'].mean():.2f} ms") + print(f"Median: {df['latency_ms'].median():.2f} ms") + print(f"Std Dev: {df['latency_ms'].std():.2f} ms") + print(f"Min: {df['latency_ms'].min():.2f} ms") + print(f"Max: {df['latency_ms'].max():.2f} ms") + print(f"P50: {df['latency_ms'].quantile(0.50):.2f} ms") + print(f"P90: {df['latency_ms'].quantile(0.90):.2f} ms") + print(f"P95: {df['latency_ms'].quantile(0.95):.2f} ms") + print(f"P99: {df['latency_ms'].quantile(0.99):.2f} ms") + + # 5. Export to Excel + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + filename = f"latency_report_{timestamp}.xlsx" + + with pd.ExcelWriter(filename, engine='openpyxl') as writer: + # Overall stats + overall = pd.DataFrame([{ + 'total_messages': len(df), + 'mean_ms': df['latency_ms'].mean(), + 'median_ms': df['latency_ms'].median(), + 'std_ms': df['latency_ms'].std(), + 'min_ms': df['latency_ms'].min(), + 'max_ms': df['latency_ms'].max(), + 'p50_ms': df['latency_ms'].quantile(0.50), + 'p90_ms': df['latency_ms'].quantile(0.90), + 'p95_ms': df['latency_ms'].quantile(0.95), + 'p99_ms': df['latency_ms'].quantile(0.99), + }]) + overall.to_excel(writer, sheet_name='Overall', index=False) + + # All messages + df.to_excel(writer, sheet_name='Messages', index=False) + + # Per-sender stats + sender_stats = df.groupby('sender_id')['latency_ms'].agg([ + 'count', 'mean', 'median', 'std', 'min', 'max' + ]).round(2) + sender_stats.to_excel(writer, sheet_name='Senders') + + # Per-receiver stats + receiver_stats = df.groupby('receiver_id')['latency_ms'].agg([ + 'count', 'mean', 'median', 'std', 'min', 'max' + ]).round(2) + receiver_stats.to_excel(writer, sheet_name='Receivers') + + print(f"✓ Excel report saved: {filename}\n") + + # 6. Shutdown + print("Shutting down agents...") + for server in servers: + server.should_exit = True + await asyncio.sleep(0.5) + print("Done!!\n") + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/A2A-MAS-no-SLIM/requirements.txt b/A2A-MAS-no-SLIM/requirements.txt new file mode 100644 index 0000000..44be7ff --- /dev/null +++ b/A2A-MAS-no-SLIM/requirements.txt @@ -0,0 +1,14 @@ +# A2A Multi-Agent Latency Testing +# Python 3.10+ + +# Official A2A SDK +"a2a-sdk[all]" +"a2a-sdk[http-server]" + +# HTTP client +httpx>=0.25.0 + +# Data export +pandas>=2.0.0 +openpyxl>=3.1.0 +numpy>=1.24.0 \ No newline at end of file