feat: use a2a sdk

Author: ducphamle2

openhands/a2a/A2AManager.py

@@ -3,18 +3,18 @@
 from abc import ABC
 from typing import AsyncGenerator, List
 
-from openhands.a2a.client.card_resolver import A2ACardResolver
-from openhands.a2a.client.client import A2AClient
-from openhands.a2a.common.types import (
-    A2AClientHTTPError,
-    A2AClientJSONError,
+from a2a.client import A2ACardResolver, A2AClient
+from a2a.client.errors import A2AClientHTTPError, A2AClientJSONError
+from a2a.types import (
     AgentCard,
     Message,
-    SendTaskResponse,
-    SendTaskStreamingResponse,
-    TaskSendParams,
+    MessageSendParams,
+    SendMessageRequest,
+    SendMessageResponse,
+    SendStreamingMessageResponse,
     TextPart,
 )
+
 from openhands.core.logger import openhands_logger as logger
 
 
@@ -73,9 +73,9 @@ def list_remote_agents(self):
             )
         return remote_agent_info
 
-    async def send_task(
-        self, agent_name: str, message: str, sid: str
-    ) -> AsyncGenerator[SendTaskStreamingResponse | SendTaskResponse, None]:
+    async def send_message(
+        self, agent_name: str, message: str, sid: str, role: str = 'user'
+    ) -> AsyncGenerator[SendStreamingMessageResponse | SendMessageResponse, None]:
         """Send a task to a remote agent and yield task responses.
 
         Args:
@@ -91,25 +91,29 @@ async def send_task(
 
         card = self.list_remote_agent_cards[agent_name]
         client = A2AClient(card)
-        request: TaskSendParams = TaskSendParams(
-            id=str(uuid.uuid4()),
+        params: MessageSendParams = MessageSendParams(
             sessionId=sid,
             message=Message(
-                role='user',
+                role=role,
                 parts=[TextPart(text=message)],
                 metadata={},
             ),
             acceptedOutputModes=['text', 'text/plain', 'image/png'],
-            metadata={'conversation_id': sid},
+            metadata={'conversation_id': sid, 'session_id': sid},
+        )
+        request: SendMessageRequest = SendMessageRequest(
+            id=str(uuid.uuid4()),
+            sessionId=sid,
+            params=params,
         )
 
         logger.info(f'Sending task to {agent_name} with message: {message}')
         logger.info(f'Card capabilities: {card.capabilities}')
         if card.capabilities.streaming:
-            async for response in client.send_task_streaming(request):
+            async for response in client.send_message_streaming(request=request):
                 yield response
         else:
-            response = await client.send_task(request)
+            response = await client.send_message(request=request)
             yield response
 
     # async def send_cancel_task(self, task_id: str, sid: str):

openhands/a2a/a2a.py

@@ -2,8 +2,11 @@
 import asyncio
 from uuid import uuid4
 
-from openhands.a2a.client import A2ACardResolver, A2AClient
-from openhands.a2a.common.types import TaskState, TaskStatusUpdateEvent
+from a2a.client import A2ACardResolver, A2AClient
+from a2a.types import AgentCard, JSONRPCErrorResponse, TaskState, TaskStatusUpdateEvent
+
+from containers.runtime.code.openhands.utils.async_utils import call_async_from_sync
+from openhands.core.logger import openhands_logger as logger
 from openhands.core.message import TextContent
 
 
@@ -13,10 +16,7 @@ def __init__(self, a2a_server_url: str, session: str = None, history: bool = Fal
         self.history = history
 
         self.card_resolver = A2ACardResolver(a2a_server_url)
-        self.card = self.card_resolver.get_agent_card()
-
-        print('======= Agent Card ========')
-        print(self.card.model_dump_json(exclude_none=True))
+        self.card: AgentCard = call_async_from_sync(self.card_resolver.get_agent_card())
 
         self.client = A2AClient(agent_card=self.card)
         if session:
@@ -61,21 +61,24 @@ async def completeTask(self, streaming, taskId, sessionId, messages: list[str]):
 
         taskResult = None
         if streaming:
-            response_stream = self.client.send_task_streaming(payload)
+            response_stream = self.client.send_message_streaming(payload)
             async for result in response_stream:
                 print(f'stream event => {result.model_dump_json(exclude_none=True)}')
                 if (
-                    result.result
-                    and isinstance(result.result, TaskStatusUpdateEvent)
-                    and result.result.final
+                    result.root
+                    and isinstance(result.root.result, TaskStatusUpdateEvent)
+                    and result.root.result.final
                 ):
                     return False
         else:
-            taskResult = await self.client.send_task(payload)
+            taskResult = await self.client.send_message(payload)
             print(f'\ntask result => {taskResult.model_dump_json(exclude_none=True)}')
             ## if the result is that more input is required, loop again.
-            state = TaskState(taskResult.result.status.state)
-            if state.name == TaskState.INPUT_REQUIRED.name:
+            if isinstance(taskResult.root, JSONRPCErrorResponse):
+                logger.error(f'Error sending message to agent A2A: {taskResult.root}')
+                return False
+            state = TaskState(taskResult.root.result.status.state)
+            if state.name == TaskState.input_required.name:
                 return await self.completeTask(streaming, taskId, sessionId)
             else:
                 ## task is complete