techm-2025 · ssyechuri · Oct 17, 2025
diff --git a/A2A-MAS-no-SLIM/__init__.py b/A2A-MAS-no-SLIM/__init__.py
diff --git a/A2A-MAS-no-SLIM/agent.py 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
diff --git a/A2A-MAS-no-SLIM/client.py 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
@@ -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())
diff --git a/A2A-MAS-no-SLIM/requirements.txt 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