Adding A2A platform (#142)

A2A platform.
1. Updated Agent class to include A2A as a platform.
2. Implemented A2A as a platform that depedends on BAF's functionality.
3. Created server, agent-card, message router, agent and method registration.
4. Added a test script in examples to test A2A protocol.

Author: natarajan-chidambaram

README.md

@@ -69,5 +69,6 @@ Note that if you want to set your agent's language to **Luxembourgish**, you wil
 - [telegram_agent](https://github.com/BESSER-PEARL/BESSER-Agentic-Framework/blob/main/besser/agent/test/examples/telegram_agent.py): Introducing the [TelegramPlatform](https://besser-agentic-framework.readthedocs.io/latest/wiki/platforms/telegram_platform.html)
 - [github_agent](https://github.com/BESSER-PEARL/BESSER-Agentic-Framework/blob/main/besser/agent/test/examples/github_agent.py): Introducing [GitHubPlatform](https://besser-agentic-framework.readthedocs.io/latest/wiki/platforms/github_platform.html)
 - [gitlab_agent](https://github.com/BESSER-PEARL/BESSER-Agentic-Framework/blob/main/besser/agent/test/examples/gitlab_agent.py): Introducing the [GitLabPlatform](https://besser-agentic-framework.readthedocs.io/latest/wiki/platforms/gitlab_platform.html)
+- [a2a_multiagent](https://github.com/BESSER-PEARL/BESSER-Agentic-Framework/blob/main/besser/agent/test/examples/a2a_multiagent.py): Introducing the [A2APlatform](https://besser-agentic-framework.readthedocs.io/latest/wiki/platforms/a2a_platform.html)
 
 For more example agents, check out the [BAF-agent-examples](https://github.com/BESSER-PEARL/BAF-agent-examples) repository!

besser/agent/core/agent.py

@@ -30,6 +30,7 @@
 from besser.agent.platforms.websocket.websocket_platform import WebSocketPlatform
 from besser.agent.platforms.github.github_platform import GitHubPlatform
 from besser.agent.platforms.gitlab.gitlab_platform import GitLabPlatform
+from besser.agent.platforms.a2a.a2a_platform import A2APlatform
 
 
 class Agent:
@@ -528,6 +529,16 @@ def use_gitlab_platform(self) -> GitLabPlatform:
         gitlab_platform = GitLabPlatform(self)
         self._platforms.append(gitlab_platform)
         return gitlab_platform
+    
+    def use_a2a_platform(self) -> A2APlatform:
+        """Use the :class: `~besser.agent.platforms.a2a.a2a_platform.A2APlatform` on this agent.
+
+        Returns:
+            A2APlatform: the A2A platform
+        """
+        a2a_platform = A2APlatform(self)
+        self._platforms.append(a2a_platform)
+        return a2a_platform
 
     def _monitoring_db_insert_session(self, session: Session) -> None:
         """Insert a session record into the monitoring database.

besser/agent/nlp/nlp_engine.py

@@ -232,11 +232,12 @@ def text2speech(self, session: Session, text: str):
 
         Returns:
             dict: the speech synthesis as a dictionary containing 2 keys:
+
                 - audio (np.ndarray): the generated audio waveform as a numpy array with dimensions (nb_channels,
-                audio_length), where nb_channels is the number of audio channels (usually 1 for mono) and audio_length is the number
-                of samples in the audio
+                  audio_length), where nb_channels is the number of audio channels (usually 1 for mono) and audio_length is the number
+                  of samples in the audio
                 - sampling_rate (int): an integer value containing the sampling rate, eg. how many samples correspond to
-                one second of audio
+                  one second of audio
         """
 
         user_language = session.get("user_language", "en")

besser/agent/platforms/a2a/__init__.py

@@ -0,0 +1,16 @@
+"""Definition of the agent properties within the ``A2A_platform`` section:"""
+
+from besser.agent.core.property import Property
+
+SECTION_A2A = 'a2a_platform'
+
+A2A_WEBSOCKET_PORT = Property(SECTION_A2A, 'a2a.port', int, 8000)
+"""
+The server local port. This port should be exposed or proxied to make it visible by other Agents
+
+name: ``a2a.port``
+
+type: ``int``
+
+default value: ``8000``
+"""

besser/agent/platforms/a2a/a2a_platform.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel
+
+class AgentCard(BaseModel):
+    """
+    Represents the agent card that has the metadata and description about the agent.
+    """
+    # Define the fields and their type for the agent card
+    name: str
+    id: str
+    endpoints: list[str]
+    version: str
+    capabilities: list[str]
+    descriptions: list[str]
+    provider: str
+    skills: list[str]
+    examples: list[dict] | list[str] = []
+    methods: list[dict] = []
+
+    def to_json(self):
+        return self.model_dump_json(indent=4)
+    
+    @classmethod
+    def from_json(cls, json_str):
+        return cls.model_validate_json(json_str)

besser/agent/platforms/a2a/agent_card.py

@@ -0,0 +1,67 @@
+# from typing import TYPE_CHECKING
+# if TYPE_CHECKING:
+#     from besser.agent.platforms.a2a.a2a_platform import A2APlatform
+from aiohttp import web
+
+from besser.agent.exceptions.logger import logger
+from besser.agent.platforms.a2a.error_handler import AgentNotFound
+
+class AgentRegistry:
+    '''
+    Keeps track of registered A2A agents by ID.
+    Attributes:
+    _agents: dictionary of registered agents
+    '''
+
+    def __init__(self):
+        self._agents: dict[str, 'A2APlatform'] = {}
+
+    def register(self, agent_id: str, platform: 'A2APlatform') -> None:
+        """
+        Register the provided agent (through agent_id) in the given platform
+        """
+        if agent_id in self._agents:
+            logger.error(f'Agent ID "{agent_id}" already registered')
+            raise ValueError(f'Agent ID "{agent_id}" already registered')
+        logger.info(f'Registering agent {agent_id}')
+        self._agents[agent_id] = platform
+
+        # Auto-register the methods that are common to all agents.
+        if hasattr(platform, "router") and platform.router:
+            platform.router.register_task_methods(platform)
+
+    def get(self, agent_id: str) -> 'A2APlatform':
+        """
+        Get the registered agent
+        """
+        if agent_id not in self._agents:
+            raise ValueError(f'Agent ID "{agent_id}" not found')
+        return self._agents[agent_id]
+
+    def list(self) -> list:
+        '''
+        Return summary info for all registered agents.
+        '''
+        return [
+            {
+                "id": agent_id,
+                "name": platform.agent_card.name,
+                "description": platform.agent_card.descriptions,
+                "capabilities": platform.agent_card.capabilities,
+                "card_url": f"/agents/{agent_id}/agent-card"
+            }
+            for agent_id, platform in self._agents.items()
+        ]
+
+    def count(self) -> int:
+        """
+        provide total number of agents that are registered
+        """
+        return len(self._agents)
+    
+    # Used for synchronous agent orchestration calls
+    async def call_agent_method(self, target_agent_id: str, method: str, params: dict) -> web.json_response:
+        target_platform = self.get(target_agent_id)
+        if not target_platform:
+            raise AgentNotFound(f'Agent ID "{target_agent_id}" not found')
+        return await target_platform.router.handle(method, params)

besser/agent/platforms/a2a/agent_registry.py

@@ -0,0 +1,30 @@
+import requests
+import time
+import httpx
+
+class A2AClient:
+    def __init__(self, base_url: str, timeout: int = 20):
+        self.base = base_url.rstrip("/")
+        self.timeout = timeout
+
+    def call(self, method: str, params: dict | None = None, id: int | None = None):
+        payload = {"jsonrpc":"2.0","method":method,"params":params or {}, "id": id or int(time.time()*1000)}
+        r = requests.post(f"{self.base}/run", json=payload, timeout=self.timeout)
+        r.raise_for_status()
+        data = r.json()
+        if "error" in data:
+            raise RuntimeError(data["error"])
+        return data["result"]
+
+    def card(self):
+        r = requests.get(f"{self.base}/agent-card", timeout=self.timeout)
+        r.raise_for_status()
+        return r.json()
+
+    async def stream(self, method: str, params: dict | None = None):
+        async with httpx.AsyncClient(timeout=None) as client:
+            async with client.stream("POST", f"{self.base}", json={"jsonrpc":"2.0","method":method,"params":params or {}, "id":int(time.time()*1000)}) as r:
+                async for line in r.aiter_lines():
+                    if line.strip():
+                        yield line
+                        

besser/agent/platforms/a2a/client.py

@@ -0,0 +1,61 @@
+from aiohttp import web
+
+# JSON-RPC standard errors
+PARSE_ERROR = {"code": -32700, "message": "Parse error"}
+INVALID_REQUEST = {"code": -32600, "message": "Invalid Request"}
+METHOD_NOT_FOUND = {"code": -32601, "message": "Method not found"}
+INVALID_PARAMS = {"code": -32602, "message": "Invalid params"}
+INTERNAL_ERROR = {"code": -32603, "message": "Internal error"}
+
+# Custom task-related errors for A2A
+TASK_PENDING = {"code": -32000, "message": "Task is still pending"}
+TASK_FAILED = {"code": -32001, "message": "Task execution failed"}
+TASK_NOT_FOUND = {"code": -32002, "message": "Task not found"}
+
+class JSONRPCError(Exception):
+    def __init__(self, code=-32000, message="Server error", data=None):
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.data = data
+
+class AgentNotFound(JSONRPCError):
+    def __init__(self, message="Agent not found"):
+        super().__init__(-32003, message)
+
+class MethodNotFound(JSONRPCError):
+    def __init__(self, message="Method not found"):
+        super().__init__(-32601, message)
+
+class InvalidParams(JSONRPCError):
+    def __init__(self, message="Invalid params"):
+        super().__init__(-32602, message)
+
+class TaskError(Exception):
+    def __init__(self, code: str, message: str = ""):
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+def error_response(exc: Exception, request_id=None):
+    if isinstance(exc, JSONRPCError):
+        return {
+            "jsonrpc": "2.0",
+            "error": {"code": exc.code, "message": exc.message, "data": exc.data},
+            "id": request_id,
+        }
+    else:
+        return {
+            "jsonrpc": "2.0",
+            "error": {"code": -32603, "message": str(exc)},
+            "id": request_id,
+        }
+
+@web.middleware
+async def error_middleware(request, handler):
+    try:
+        return await handler(request)
+    except JSONRPCError as e:
+        return web.json_response(error_response(e))
+    except Exception as e:
+        return web.json_response(error_response(e))

besser/agent/platforms/a2a/error_handler.py

@@ -0,0 +1,125 @@
+# from typing import TYPE_CHECKING
+
+import inspect
+
+from aiohttp import web
+from aiohttp.web_request import Request
+
+from besser.agent.exceptions.logger import logger
+from besser.agent.platforms.a2a.error_handler import JSONRPCError, MethodNotFound, InvalidParams, TaskError
+from besser.agent.platforms.a2a.error_handler import INTERNAL_ERROR, PARSE_ERROR, INVALID_REQUEST, TASK_PENDING, TASK_FAILED, TASK_NOT_FOUND
+from besser.agent.platforms.a2a.agent_registry import AgentRegistry
+# if TYPE_CHECKING:
+#     from besser.agent.platforms.a2a.a2a_platform import A2APlatform
+
+class A2ARouter:
+    def __init__(self) -> None:
+        self.methods = {}
+
+    def register(self, method_name, func) -> None:
+        '''
+        Register a method (coupled to its name, also called as key) that can be called via RPC.
+        '''
+        self.methods[method_name] = func
+
+    async def handle(self, method_name: str, params: dict) -> web.json_response:
+        """
+        Execute the method given its name and parameters
+        """
+        
+        if method_name not in self.methods:
+                logger.error(f"Method '{method_name}' not found")
+                raise MethodNotFound(message=f"Method '{method_name}' not found")
+        
+        if not isinstance(params, dict):
+            logger.error(f"Params must be a dictionary")
+            raise InvalidParams()
+        
+        method = self.methods[method_name]
+
+        # for handling async tasks, else it is sync
+        if inspect.iscoroutinefunction(method):
+            return await method(**params)
+        else:
+            return method(**params)
+    
+    async def aiohttp_handler(self, request: Request) -> web.json_response:
+        """
+        Handle HTTP requests from the server
+        """
+        request_id = None
+        try:
+            body = await request.json()
+            request_id = body.get("id")
+        except Exception:
+            logger.error(PARSE_ERROR)
+            return web.json_response({
+                "jsonrpc": "2.0", 
+                "error": PARSE_ERROR,
+                "id": request_id
+                })
+        
+        if "method" not in body or not isinstance(body["method"], str):
+            logger.error(INVALID_REQUEST)
+            return web.json_response({
+                "jsonrpc": "2.0", 
+                "error": INVALID_REQUEST, 
+                "id": body.get("id")
+                })
+        
+        method = body['method']
+        params = body.get('params', {})
+
+        try:
+            result = await self.handle(method, params)
+            return web.json_response({
+            "jsonrpc": "2.0",
+            "result": result,
+            "id": request_id
+            })
+        except JSONRPCError as e:
+            return web.json_response({
+                "jsonrpc": "2.0", 
+                "error": {"code": e.code, "message": e.message}, 
+                "id": request_id
+                })
+        
+        except TaskError as e:
+            error_map = {
+            "TASK_PENDING": TASK_PENDING,
+            "TASK_FAILED": TASK_FAILED,
+            "TASK_NOT_FOUND": TASK_NOT_FOUND
+            }
+            logger.error(error_map.get(e.code, INTERNAL_ERROR))
+            return web.json_response({
+                "jsonrpc": "2.0", 
+                "error": error_map.get(e.code, INTERNAL_ERROR), 
+                "id": request_id
+                })
+        except Exception as e:
+            # print(f"Error: \n{e}")
+            logger.error(f"Internal error: {str(e)}")
+            return web.json_response({
+                "jsonrpc": "2.0",
+                "error": {**INTERNAL_ERROR, 
+                            "message": str(e)},
+                "id": request_id
+            })
+    
+    def register_task_methods(self, platform: 'A2APlatform') -> None:
+        """
+        Auto-register internal methods for creating, executing and getting task status.
+        """
+        self.register("create_task_and_run", platform.rpc_create_task)
+        self.register("task_create", platform.create_task)
+        self.register("task_status", platform.get_status)
+    
+    # 
+    def register_orchestration_methods(self, platform: 'A2APlatform', registry: AgentRegistry) -> None:
+        """
+        Register methods used for orchestration in its router. Enables one agent to call another agent.
+        """
+        async def call_agent_rpc(target_agent_id: str, method: str, params: dict):
+            return await platform.rpc_call_agent(target_agent_id, method, params, registry)
+
+        self.register("call_agent", call_agent_rpc)