Merge pull request #6 from satendrakumar/develop

Code refactoring

Author: satendrakumar

README.md

@@ -1,4 +1,4 @@
-# a2a-multiple-agents-on-single-host
+# multiple-agents-on-single-a2a-server
 
 This repository demonstrates how to run **multiple A2A agents** on the **same host** using the A2A protocol.
 Each agent is served at a **unique URL path**, making it possible to host different agents without requiring multiple servers or ports.
@@ -22,8 +22,8 @@ Three agents running on the same host:
 
 1.  Clone the repository
     ```bash
-    git clone https://github.com/satendrakumar/a2a-multiple-agents-on-single-host.git
-    cd a2a-multiple-agents-on-single-host
+    git clone https://github.com/satendrakumar/multiple-agents-on-single-a2a-server.git
+    cd multiple-agents-on-single-a2a-server
     ```
 
 2.  Install dependencies (using uv)

a2a_client_app.py

@@ -8,7 +8,7 @@ async def main():
     a2a_client: A2ASimpleClient = A2ASimpleClient()
     agent_host_url = "http://localhost:8000/a2a"
 
-    trending_task = a2a_client.create_task(agent_url=f'{agent_host_url}/trending', message="What's trending today?",
+    trending_task = a2a_client.create_task(agent_url=f'{agent_host_url}/trending_topics', message="What's trending today?",
                                            context_id=str(uuid.uuid4()))
     analysis_task = a2a_client.create_task(agent_url=f"{agent_host_url}/analyzer",
                                            message="Analyze the trend AI in Social Media", context_id=str(uuid.uuid4()))

main.py

@@ -5,47 +5,53 @@
 from dotenv import load_dotenv
 from fastapi import FastAPI
 
-from src.a2a.a2a_fastapi_app import A2AFastApiApp, get_agent_request_handler
-from src.agent.analyzer_agent import analyzer_agent, get_analyzer_agent_card
-from src.agent.conversation_agent import get_conversational_agent_card, conversational_agent
-from src.agent.trending_topics_agent import trending_topics_agent, get_trending_topics_agent_card
+from src.a2a.a2a_utils import A2AUtils
+from src.agent.analyzer_agent import get_analyzer_agent_card, get_analyzer_agent
+from src.agent.conversation_agent import get_conversational_agent_card, get_conversational_agent
+from src.agent.trending_topics_agent import get_trending_topics_agent_card, get_trending_topics_agent
 
 load_dotenv()
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger()
 
 AGENT_BASE_URL = os.getenv('AGENT_BASE_URL')
+
+if not AGENT_BASE_URL:
+    raise ValueError("AGENT_BASE_URL environment variable must be set")
+
+MODEL_NAME = os.getenv('MODEL_NAME')
+
+if not MODEL_NAME:
+    raise ValueError("MODEL_NAME environment variable must be set")
 logger.info(f"AGENT BASE URL {AGENT_BASE_URL}")
 
 app: FastAPI = FastAPI(title="Run multiple agents on single host using A2A protocol.",
                        description="Run multiple agents on single host using A2A protocol.",
                        version="1.0.0",
-                       root_path=f"/a2a")
+                       root_path="/a2a")
 
 
 @app.get("/health")
 async def health_check() -> dict[str, str]:
     return {"status": "ok"}
 
 
-conversation_agent_request_handler = get_agent_request_handler(conversational_agent)
-conversational_agent_card = get_conversational_agent_card(f"{AGENT_BASE_URL}/conversation/")
-conversational_agent_server = A2AFastApiApp(fastapi_app=app, agent_card=conversational_agent_card, http_handler=conversation_agent_request_handler)
-# {path:path} added into the agent card url path to handle both 'agent-card.json' and '/.well-known/agent-card.json'
-conversational_agent_server.build(rpc_url="/conversation/", agent_card_url="/conversation/{path:path}")
+# conversation agent integration with A2A server
+A2AUtils.build(name="conversation", get_agent=get_conversational_agent, get_agent_card=get_conversational_agent_card,
+               model_name=MODEL_NAME,
+               agent_base_url=AGENT_BASE_URL, app=app)
 
-trending_agent_request_handler = get_agent_request_handler(trending_topics_agent)
-trending_topics_agent_card = get_trending_topics_agent_card(f"{AGENT_BASE_URL}/trending/")
-trending_agent_server = A2AFastApiApp(fastapi_app=app, agent_card=trending_topics_agent_card,
-                                      http_handler=trending_agent_request_handler)
-trending_agent_server.build(rpc_url="/trending/", agent_card_url="/trending/{path:path}")
+# trending_topics agent integration with A2A server
+A2AUtils.build(name="trending_topics", get_agent=get_trending_topics_agent,
+               get_agent_card=get_trending_topics_agent_card,
+               model_name=MODEL_NAME,
+               agent_base_url=AGENT_BASE_URL, app=app)
 
-analyzer_agent_request_handler = get_agent_request_handler(analyzer_agent)
-analyzer_agent_card = get_analyzer_agent_card(f"{AGENT_BASE_URL}/analyzer/")
-analyzer_agent_server = A2AFastApiApp(fastapi_app=app, agent_card=analyzer_agent_card,
-                                      http_handler=analyzer_agent_request_handler)
-analyzer_agent_server.build(rpc_url="/analyzer/", agent_card_url="/analyzer/{path:path}")
+# analyzer agent integration with A2A server
+A2AUtils.build(name="analyzer", get_agent=get_analyzer_agent, get_agent_card=get_analyzer_agent_card,
+               model_name=MODEL_NAME,
+               agent_base_url=AGENT_BASE_URL, app=app)
 
 if __name__ == '__main__':
     uvicorn.run(app, host="0.0.0.0", port=8000)

pyproject.toml

@@ -1,5 +1,5 @@
 [project]
-name = "a2a-multiple-agents-on-single-host"
+name = "multiple-agents-on-single-a2a-server"
 version = "0.1.0"
 description = "Add your description here"
 readme = "README.md"

src/a2a/a2a_client.py

@@ -1,24 +1,21 @@
 from typing import Any
-
-
-import httpx
-from a2a.client import ClientConfig, ClientFactory, ClientCallContext
-from a2a.types import (AgentCard, TransportProtocol, )
-from a2a.types import Message, Part, Role, TextPart
-
 from uuid import uuid4
 
+import httpx
+from a2a.client import ClientConfig, ClientFactory
+from a2a.types import AgentCard, TransportProtocol
 from a2a.types import Message, Part, Role, TextPart
 
 AGENT_CARD_PATH = '/agent-card.json'
 
+
 class A2ASimpleClient:
     def __init__(self, default_timeout: float = 240.0):
         # Cache for agent metadata
         self._agent_info_cache: dict[str, dict[str, Any] | None] = {}
         self.default_timeout = default_timeout
 
-    async def create_task(self, agent_url: str, message: str, context_id:str) -> str:
+    async def create_task(self, agent_url: str, message: str, context_id: str) -> str:
         """Send a message following the official A2A SDK pattern."""
         # Configure httpx client with timeout
         timeout_config = httpx.Timeout(
@@ -60,7 +57,8 @@ async def create_task(self, agent_url: str, message: str, context_id:str) -> str
 
             factory = ClientFactory(config)
             client = factory.create(agent_card)
-            message_obj= Message(role=Role.user, parts=[Part(TextPart(text=message))], message_id=str(uuid4()), context_id=context_id)
+            message_obj = Message(role=Role.user, parts=[Part(TextPart(text=message))], message_id=str(uuid4()),
+                                  context_id=context_id)
             responses = []
             async for response in client.send_message(message_obj):
                 responses.append(response)

src/a2a/a2a_utils.py

@@ -0,0 +1,25 @@
+from typing import Callable
+
+from a2a.types import AgentCard
+from fastapi import FastAPI
+from google.adk.agents import LlmAgent
+
+from src.a2a.a2a_fastapi_app import get_agent_request_handler, A2AFastApiApp
+
+
+class A2AUtils:
+    @staticmethod
+    def build(
+            name: str,
+            get_agent: Callable[[str], LlmAgent],
+            get_agent_card: Callable[[str], AgentCard],
+            model_name: str,
+            agent_base_url: str,
+            app: FastAPI
+    ) -> None:
+        agent = get_agent(model_name)
+        agent_request_handler = get_agent_request_handler(agent)
+        agent_card = get_agent_card(f"{agent_base_url}/{name}/")
+        agent_server = A2AFastApiApp(fastapi_app=app, agent_card=agent_card,
+                                     http_handler=agent_request_handler)
+        agent_server.build(rpc_url=f"/{name}/", agent_card_url=f"/{name}/{{path:path}}")

src/agent/analyzer_agent.py

@@ -1,32 +1,34 @@
 from a2a.types import AgentCard, AgentCapabilities, TransportProtocol, AgentSkill
-from google.adk import Agent
+from google.adk.agents import LlmAgent
 from google.adk.tools import google_search
 
-analyzer_agent = Agent(
-    model='gemini-2.5-flash',
-    name='trend_analyzer_agent',
-    instruction="""
-    You are a data analyst specializing in trend analysis. When given a trending topic,
-    perform deep research to find quantitative data and insights.
 
-    For each trend you analyze:
-    1. Search for statistics, numbers, and metrics related to the trend
-    2. Look for:
-       - Engagement metrics (views, shares, mentions)
-       - Growth rates and timeline
-       - Geographic distribution
-       - Related hashtags or keywords
-    3. Provide concrete numbers and data points
-
-    Keep it somehow concise
-
-    Always prioritize quantitative information over qualitative descriptions.
-    """,
-    tools=[google_search],
-)
+def get_analyzer_agent(model: str) -> LlmAgent:
+    return LlmAgent(
+        model=model,
+        name='trend_analyzer_agent',
+        instruction="""
+        You are a data analyst specializing in trend analysis. When given a trending topic,
+        perform deep research to find quantitative data and insights.
+    
+        For each trend you analyze:
+        1. Search for statistics, numbers, and metrics related to the trend
+        2. Look for:
+           - Engagement metrics (views, shares, mentions)
+           - Growth rates and timeline
+           - Geographic distribution
+           - Related hashtags or keywords
+        3. Provide concrete numbers and data points
+    
+        Keep it somehow concise
+    
+        Always prioritize quantitative information over qualitative descriptions.
+        """,
+        tools=[google_search],
+    )
 
 
-def get_analyzer_agent_card(agent_url):
+def get_analyzer_agent_card(agent_url: str) -> AgentCard:
     return AgentCard(
         name='Trend Analyzer Agent',
         url=agent_url,

src/agent/conversation_agent.py

@@ -1,5 +1,4 @@
 from a2a.types import AgentCard, AgentCapabilities, TransportProtocol, AgentSkill
-from google.adk import Agent
 from google.adk.agents import LlmAgent
 from google.adk.tools import google_search
 
@@ -62,16 +61,18 @@
 - End with an offer to assist further
 """
 
-conversational_agent: LlmAgent = Agent(
-    model="gemini-2.5-flash",
-    name="conversational_agent",
-    description="An AI assistant that enhances conversations with live web search when needed.",
-    instruction=CONVERSATION_AGENT_INSTRUCTIONS,
-    tools=[google_search],
-)
+
+def get_conversational_agent(model: str) -> LlmAgent:
+    return LlmAgent(
+        model=model,
+        name="conversational_agent",
+        description="An AI assistant that enhances conversations with live web search when needed.",
+        instruction=CONVERSATION_AGENT_INSTRUCTIONS,
+        tools=[google_search],
+    )
 
 
-def get_conversational_agent_card(agent_url: str):
+def get_conversational_agent_card(agent_url: str) -> AgentCard:
     return AgentCard(
         name="Conversational Agent",
         description="Smart Conversational Agent Enhanced with Web Search Capabilities",

src/agent/trending_topics_agent.py

@@ -1,11 +1,13 @@
 from a2a.types import AgentCard, AgentCapabilities, TransportProtocol, AgentSkill
-from google.adk import Agent
+from google.adk.agents import LlmAgent
 from google.adk.tools import google_search
 
-trending_topics_agent = Agent(
-    model='gemini-2.5-flash',
-    name='trending_topics_agent',
-    instruction="""
+
+def get_trending_topics_agent(model: str) -> LlmAgent:
+    return LlmAgent(
+        model=model,
+        name='trending_topics_agent',
+        instruction="""
     You are a social media trends analyst. Your job is to search the web for current trending topics,
     particularly from social platforms.
 
@@ -39,11 +41,11 @@
 
     Only return the JSON object, no additional text.
     """,
-    tools=[google_search],
-)
+        tools=[google_search],
+    )
 
 
-def get_trending_topics_agent_card(agent_url: str):
+def get_trending_topics_agent_card(agent_url: str) -> AgentCard:
     return AgentCard(
         name='Trending Topics Agent',
         url=agent_url,