Merge pull request #4 from cyberholics/a2a-customer-routing

Astrodevil · web-flow · commit 47e07dbb5d41 · 2025-08-19T19:17:13.000+05:30
new changes from adk and a2a new versions
diff --git a/a2a_customer_routing/README.md b/a2a_customer_routing/README.md
@@ -143,7 +143,6 @@ Response: "I understand your frustration. A human agent will contact you shortly
 ### Core Classes
 
 - **`KB`**: Knowledge base management with LlamaIndex integration
-- **`ADKAgentExecutor`**: Wrapper for Google ADK agents to work with A2A protocol
 - **`A2AToolClient`**: Client for agent-to-agent communication
 
 ### Agent Tools
diff --git a/a2a_customer_routing/multi_agent/agent.py b/a2a_customer_routing/multi_agent/agent.py
@@ -1,28 +1,25 @@
 import uuid
 import logging
 import os
-
-from a2a.server.agent_execution import AgentExecutor, RequestContext
+import asyncio
+from typing import Any
+import httpx 
 from a2a.server.apps import A2AStarletteApplication
-from a2a.server.events import EventQueue
 from a2a.server.request_handlers import DefaultRequestHandler
-from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
+from a2a.server.tasks import InMemoryTaskStore
 from a2a.types import (
-    AgentCapabilities, AgentCard, AgentSkill, MessageSendParams, Part,
-    TaskState, TextPart, SendMessageRequest, Message
+    AgentCapabilities, AgentCard, AgentSkill, Task, TransportProtocol
 )
-from a2a.utils import new_agent_text_message, new_task
-from a2a.client import A2AClient
+from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH
+from a2a.client import ClientConfig, ClientFactory, create_text_message_object
+from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor, A2aAgentExecutorConfig
 
 from google.adk.agents import LlmAgent
 from google.adk.runners import Runner
 from google.adk.sessions import InMemorySessionService
 from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
 from google.adk.artifacts import InMemoryArtifactService
 from google.adk.models.lite_llm import LiteLlm
-from google.genai import types
-import httpx
-
 
 from .tools import resolve_query_fn, classify_fn, escalate_fn
 
@@ -32,7 +29,6 @@
 
 # --- Agent and Model Definitions ---
 def create_llm_model(model_name: str):
-    """Factory function to create LLM models with consistent configuration."""
     api_key = os.getenv("NEBIUS_API_KEY")
     return LiteLlm(model=model_name, api_key=api_key, temperature=0.1)
 
@@ -45,110 +41,65 @@ def create_llm_model(model_name: str):
 
 
 # --- A2A Server Infrastructure ---
-class ADKAgentExecutor(AgentExecutor):
-    def __init__(self, agent, status_message="Processing...", artifact_name="response"):
-        self.agent = agent
-        self.status_message = status_message
-        self.artifact_name = artifact_name
-        self.runner = Runner(
-            app_name=agent.name, agent=agent, artifact_service=InMemoryArtifactService(),
-            session_service=InMemorySessionService(), memory_service=InMemoryMemoryService()
-        )
-
-    async def cancel(self, task_id: str) -> None: pass
-
-    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
-        query = context.get_user_input()
-        task = context.current_task or new_task(context.message)
-        await event_queue.enqueue_event(task)
-        updater = TaskUpdater(event_queue, task.id, task.contextId)
-        try:
-            await updater.update_status(TaskState.working, new_agent_text_message(self.status_message, task.contextId, task.id))
-            session = await self.runner.session_service.create_session(app_name=self.agent.name, user_id="a2a_user", session_id=task.contextId)
-            next_message = types.Content(role='user', parts=[types.Part.from_text(text=query)])
-            final_answer = "The agent could not produce a final answer."
-            max_turns = 5
-            for i in range(max_turns):
-                logger.info(f"Agent '{self.agent.name}' Turn {i+1}/{max_turns}")
-                response_content = None
-                async for event in self.runner.run_async(user_id="a2a_user", session_id=session.id, new_message=next_message):
-                    if event.is_final_response() and event.content:
-                        response_content = event.content
-                        break
-                if not response_content or not response_content.parts:
-                    final_answer = "Agent produced an empty response."; break
-                part = response_content.parts[0]
-                if part.function_response:
-                    tool_output = f"Tool '{part.function_response.name}' returned: {part.function_response.response}"
-                    logger.info(f"Agent '{self.agent.name}' Tool call result: {tool_output[:200]}...")
-                    next_message = types.Content(role='user', parts=[types.Part.from_text(text=tool_output)])
-                    continue
-                elif part.text is not None:
-                    final_answer = part.text.strip()
-                    logger.info(f"Agent '{self.agent.name}' final answer received."); break
-                else:
-                    final_answer = "Agent produced an unexpected response type."; break
-            await updater.add_artifact([Part(root=TextPart(text=final_answer))], name=self.artifact_name)
-            await updater.complete()
-        except Exception as e:
-            logger.error(f"Error in ADKAgentExecutor for agent {self.agent.name}: {e}", exc_info=True)
-            await updater.update_status(TaskState.failed, new_agent_text_message(f"Error: {e!s}", task.contextId, task.id), final=True)
-
-# --- A2A Agent Server Creation ---
-def create_agent_a2a_server(agent, name, description, skills, host, port, status_message, artifact_name):
-    agent_card = AgentCard(
-        name=name, description=description, url=f"http://{host}:{port}", version="1.0.0",
-        defaultInputModes=["text"], defaultOutputModes=["text"],
-        capabilities=AgentCapabilities(streaming=True), skills=skills
+def create_agent_a2a_server(agent: LlmAgent, agent_card: AgentCard):
+    runner = Runner(
+        app_name=agent.name, agent=agent, artifact_service=InMemoryArtifactService(),
+        session_service=InMemorySessionService(), memory_service=InMemoryMemoryService()
     )
-    executor = ADKAgentExecutor(agent=agent, status_message=status_message, artifact_name=artifact_name)
+    config = A2aAgentExecutorConfig()
+    executor = A2aAgentExecutor(runner=runner, config=config)
     request_handler = DefaultRequestHandler(agent_executor=executor, task_store=InMemoryTaskStore())
     return A2AStarletteApplication(agent_card=agent_card, http_handler=request_handler)
 
-# --- Individual Agent Servers ---
 def create_intake_agent_server(host="127.0.0.1", port=10020):
-    return create_agent_a2a_server(agent=intake_agent, name="Sentiment Agent", description="Analyzes message sentiment.", skills=[AgentSkill(id="classify_sentiment", name="Classify Sentiment", description="Determines message sentiment.", tags=["sentiment"])], host=host, port=port, status_message="Analyzing sentiment...", artifact_name="sentiment_result")
+    card = AgentCard(name="Sentiment Agent", description="Analyzes message sentiment.", url=f"http://{host}:{port}", version="1.0.0", defaultInputModes=["text"], defaultOutputModes=["text"], capabilities=AgentCapabilities(streaming=True), skills=[AgentSkill(id="classify_sentiment", name="Classify Sentiment", description="Determines message sentiment.", tags=["sentiment"])], preferred_transport=TransportProtocol.jsonrpc)
+    return create_agent_a2a_server(agent=intake_agent, agent_card=card)
+
 def create_resolution_agent_server(host="127.0.0.1", port=10021):
-    return create_agent_a2a_server(agent=resolution_agent, name="KB Agent", description="Answers questions using a knowledge base.", skills=[AgentSkill(id="resolve_question", name="Resolve Question", description="Searches KB for answers.", tags=["knowledge", "support"])], host=host, port=port, status_message="Searching knowledge base...", artifact_name="kb_answer")
+    card = AgentCard(name="KB Agent", description="Answers questions using a knowledge base.", url=f"http://{host}:{port}", version="1.0.0", defaultInputModes=["text"], defaultOutputModes=["text"], capabilities=AgentCapabilities(streaming=True), skills=[AgentSkill(id="resolve_question", name="Resolve Question", description="Searches KB for answers.", tags=["knowledge", "support"])], preferred_transport=TransportProtocol.jsonrpc)
+    return create_agent_a2a_server(agent=resolution_agent, agent_card=card)
+
 def create_escalation_agent_server(host="127.0.0.1", port=10022):
-    return create_agent_a2a_server(agent=escalation_agent, name="Escalation Agent", description="Escalates issues to human support.", skills=[AgentSkill(id="escalate_issue", name="Escalate Issue", description="Forwards issues to humans.", tags=["escalation", "human"])], host=host, port=port, status_message="Escalating to human support...", artifact_name="escalation_result")
+    card = AgentCard(name="Escalation Agent", description="Escalates issues to human support.", url=f"http://{host}:{port}", version="1.0.0", defaultInputModes=["text"], defaultOutputModes=["text"], capabilities=AgentCapabilities(streaming=True), skills=[AgentSkill(id="escalate_issue", name="Escalate Issue", description="Forwards issues to humans.", tags=["escalation", "human"])], preferred_transport=TransportProtocol.jsonrpc)
+    return create_agent_a2a_server(agent=escalation_agent, agent_card=card)
 
 
 # --- Coordinator Agent & Client ---
 class A2AToolClient:
     def __init__(self, default_timeout: float = 120.0):
-        self._agent_info_cache: dict[str, dict[str, any] | None] = {}
+        self._agent_info_cache: dict[str, Any | None] = {}
         self.default_timeout = default_timeout
-    def add_remote_agent(self, agent_url: str):
-        normalized_url = agent_url.rstrip('/')
-        if normalized_url not in self._agent_info_cache: self._agent_info_cache[normalized_url] = None
+
     async def create_task(self, agent_url: str, message: str) -> str:
         timeout_config = httpx.Timeout(self.default_timeout)
         async with httpx.AsyncClient(timeout=timeout_config) as httpx_client:
-            agent_card_response = await httpx_client.get(f"{agent_url}/.well-known/agent.json")
+            agent_card_response = await httpx_client.get(f"{agent_url}{AGENT_CARD_WELL_KNOWN_PATH}")
             agent_card_response.raise_for_status()
             agent_card = AgentCard(**agent_card_response.json())
-            client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
-            
+
+            config = ClientConfig(httpx_client=httpx_client)
+            factory = ClientFactory(config)
+            client = factory.create(agent_card)
+
+            message_obj = create_text_message_object(content=message)
+            final_response = "Agent did not return a valid response."
+
+            async for response in client.send_message(message_obj):
+                if isinstance(response, tuple) and len(response) > 0:
+                    task: Task = response[0]
+                    if task.artifacts:
+                        try:
+                            text_response = task.artifacts[0].parts[0].root.text
+                            if text_response:
+                                final_response = text_response.strip()
+                                break
+                        except (AttributeError, IndexError):
+                            logger.warning(f"Could not extract text from task artifact for {agent_url}")
+                            final_response = f"Agent at {agent_url} returned an unreadable response."
+                else:
+                    logger.warning(f"Received an unexpected response format from {agent_url}: {response}")
             
-            message_payload = Message(
-                messageId=f"msg_{uuid.uuid4()}",
-                role='user',
-                parts=[TextPart(text=message)]
-            )
-            send_params = MessageSendParams(message=message_payload)
-            request = SendMessageRequest(
-                id=f"req_{uuid.uuid4()}",  
-                params=send_params
-            )
-
-            response = await client.send_message(request)
-            response_dict = response.model_dump(mode='json', exclude_none=True)
-            if 'result' in response_dict and 'artifacts' in response_dict['result']:
-                for artifact in response_dict['result']['artifacts']:
-                    for part in artifact.get('parts', []):
-                        if 'text' in part and part['text'].strip(): return part['text'].strip()
-            return "Agent did not return a text artifact."
+            return final_response
 
 coordinator_a2a_client = A2AToolClient()
 
@@ -157,20 +108,20 @@ def create_coordinator_agent_with_registered_agents():
         name="support_coordinator", model=llama_8b, description="Routes user messages to other agents.",
         instruction="""You are an expert support coordinator. Your job is to orchestrate other agents to resolve a user's request.
 
-Follow this exact workflow:
-1.  **Analyze Sentiment:** Use the `create_task` tool to call the Intake Agent (at http://127.0.0.1:10020) with the user's original message. This tool will return a sentiment classification.
+Follow this exact workflow step-by-step:
+1.  **Analyze Sentiment:** Use the `create_task` tool to call the Intake Agent (at http://127.0.0.1:10020) with the user's original message. The tool will return a sentiment classification.
 2.  **Route Request:**
-    *   **If the result from the Intake Agent contains the word "negative"**, use `create_task` to call the Escalation Agent (at http://127.0.0.1:10022) with the user's original message.
-    *   Otherwise (for "positive" or "neutral"), use `create_task` to call the Resolution Agent (at http://127.0.0.1:10021) with the user's original message.
-3.  **Return Final Answer:** Your final answer must be ONLY the text returned by the chosen agent (Resolution or Escalation). Do not add any of your own commentary, summaries, or phrases like "The final answer is:".
+    *   If the result from the Intake Agent contains the word "negative", use the `create_task` tool to call the Escalation Agent (at http://127.0.0.1:10022) with the user's original message.
+    *   Otherwise (for "positive" or "neutral"), use the `create_task` tool to call the Resolution Agent (at http://127.0.0.1:10021) with the user's original message.
+3.  **Finalize and Respond:** The tool used in the previous step will return the final answer. Your final job is to output that exact text as your own final answer. Do not add any of your own commentary, summaries, or phrases like "The final answer is:". Just return the text you received.
 """,
         tools=[coordinator_a2a_client.create_task]
     )
 
-
 coordinator_agent = None
 
 def create_coordinator_agent_server(host="127.0.0.1", port=10023):
     global coordinator_agent
     if coordinator_agent is None: raise ValueError("Coordinator agent not initialized.")
-    return create_agent_a2a_server(agent=coordinator_agent, name="Support Coordinator", description="Orchestrates customer support.", skills=[AgentSkill(id="coordinate_support", name="Coordinate Support", description="Routes customer message to the right agent.", tags=["routing", "sentiment"])], host=host, port=port, status_message="Coordinating request...", artifact_name="support_response")
+    card = AgentCard(name="Support Coordinator", description="Orchestrates customer support.", url=f"http://{host}:{port}", version="1.0.0", defaultInputModes=["text"], defaultOutputModes=["text"], capabilities=AgentCapabilities(streaming=True), skills=[AgentSkill(id="coordinate_support", name="Coordinate Support", description="Routes customer message to the right agent.", tags=["routing", "sentiment"])], preferred_transport=TransportProtocol.jsonrpc)
+    return create_agent_a2a_server(agent=coordinator_agent, agent_card=card)
diff --git a/a2a_customer_routing/multi_agent/run_agents.py b/a2a_customer_routing/multi_agent/run_agents.py
@@ -6,6 +6,8 @@
 import requests
 from . import agent as agent_module
 
+from a2a.utils.constants import AGENT_CARD_WELL_KNOWN_PATH
+
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
@@ -40,7 +42,8 @@ def wait_for_agents(agent_urls, timeout=45):
         for url in agent_urls:
             if url in ready_agents: continue
             try:
-                if requests.get(f"{url}/.well-known/agent.json", timeout=1).status_code == 200:
+                health_check_url = f"{url}{AGENT_CARD_WELL_KNOWN_PATH}"
+                if requests.get(health_check_url, timeout=1).status_code == 200:
                     logger.info(f"  ✅ Agent at {url} is ready.")
                     ready_agents.add(url)
             except requests.ConnectionError: pass
@@ -54,7 +57,6 @@ def wait_for_agents(agent_urls, timeout=45):
 def start_all_agents():
     """Start all support agents and the coordinator. This is the main entry point."""
     
-
     support_agents_to_start = {
         "Intake": (agent_module.create_intake_agent_server, 10020),
         "Resolution": (agent_module.create_resolution_agent_server, 10021),
@@ -64,9 +66,6 @@ def start_all_agents():
     support_agent_urls = [f"http://127.0.0.1:{port}" for _, port in support_agents_to_start.values()]
     wait_for_agents(support_agent_urls)
 
-    
-    for url in support_agent_urls:
-        agent_module.coordinator_a2a_client.add_remote_agent(url)
 
     agent_module.coordinator_agent = agent_module.create_coordinator_agent_with_registered_agents()
     threads["Coordinator"] = run_agent_in_background(agent_module.create_coordinator_agent_server, 10023, "Coordinator")
diff --git a/a2a_customer_routing/requirements.txt b/a2a_customer_routing/requirements.txt
@@ -1,5 +1,5 @@
-a2a-sdk==0.2.8
-google-adk==1.2.1
+a2a-sdk==0.3.0
+google-adk==1.11.0
 google-genai==1.19.0
 httpx==0.28.1
 litellm==1.72.2
diff --git a/a2a_customer_routing/streamlit_app.py b/a2a_customer_routing/streamlit_app.py
