rejoin with offical google a2a

Author: daltonnyx

AgentCrew/modules/a2a/adapters.py

@@ -4,16 +4,18 @@
 
 import base64
 from typing import Dict, Any, List, Optional
-from .common.types import (
+from a2a.types import (
     GetTaskResponse,
     Message,
+    Task,
     TextPart,
     FilePart,
     FileWithBytes,
     FileWithUri,
     Artifact,
     Part,
     SendMessageResponse,
+    JSONRPCErrorResponse,
     DataPart,
     Role,
 )
@@ -34,14 +36,14 @@ def convert_a2a_message_to_agent(message: Message) -> Dict[str, Any]:
     content = []
 
     for part in message.parts:
-        part_data = part.root if hasattr(part, "root") else part
+        part_data = part.root
 
         if part_data.kind == "text":
             content.append({"type": "text", "text": part_data.text})
         elif part_data.kind == "file":
             # Handle file content
             file_data = part_data.file
-            if hasattr(file_data, "bytes") and file_data.bytes:
+            if isinstance(file_data, FileWithBytes) and file_data.bytes:
                 # Base64 encoded file
                 content.append(
                     {
@@ -51,7 +53,7 @@ def convert_a2a_message_to_agent(message: Message) -> Dict[str, Any]:
                         "mime_type": file_data.mimeType or "application/octet-stream",
                     }
                 )
-            elif hasattr(file_data, "uri") and file_data.uri:
+            elif isinstance(file_data, FileWithUri) and file_data.uri:
                 # File URI
                 content.append(
                     {
@@ -70,7 +72,7 @@ def convert_a2a_message_to_agent(message: Message) -> Dict[str, Any]:
 
 # TODO: cover all of cases for images
 def convert_agent_message_to_a2a(
-    message: Dict[str, Any], message_id: str = None
+    message: Dict[str, Any], message_id: Optional[str] = None
 ) -> Message:
     """
     Convert a SwissKnife message to A2A format.
@@ -104,7 +106,7 @@ def convert_agent_message_to_a2a(
                             file=FileWithBytes(
                                 name=part.get("file_name"),
                                 mimeType=part.get("mime_type"),
-                                bytes=part.get("file_data"),
+                                bytes=part.get("file_data", ""),
                             )
                         )
                     )
@@ -115,7 +117,7 @@ def convert_agent_message_to_a2a(
                             file=FileWithUri(
                                 name=part.get("file_name"),
                                 mimeType=part.get("mime_type"),
-                                uri=part.get("uri"),
+                                uri=part.get("uri", ""),
                             )
                         )
                     )
@@ -134,7 +136,7 @@ def convert_agent_message_to_a2a(
 def convert_agent_response_to_a2a(
     response: str,
     tool_uses: Optional[List[Dict[str, Any]]] = None,
-    artifact_id: str = None,
+    artifact_id: Optional[str] = None,
 ) -> Artifact:
     """
     Convert a SwissKnife response to an A2A artifact.
@@ -147,7 +149,7 @@ def convert_agent_response_to_a2a(
     Returns:
         The response as an A2A artifact
     """
-    parts = [TextPart(text=response)]
+    parts = [Part(root=TextPart(text=response))]
 
     # If there were tool uses, we could add them as metadata
     metadata = None
@@ -188,8 +190,10 @@ def convert_file_to_a2a_part(
     # Encode file content as base64
     base64_content = base64.b64encode(file_content).decode("utf-8")
 
-    return FilePart(
-        file=FileWithBytes(name=file_name, mimeType=mime_type, bytes=base64_content)
+    return Part(
+        root=FilePart(
+            file=FileWithBytes(name=file_name, mimeType=mime_type, bytes=base64_content)
+        )
     )
 
 
@@ -200,25 +204,26 @@ def convert_a2a_send_task_response_to_agent_message(
     if not response or not hasattr(response, "root"):
         return None
 
-    result = response.root.result if hasattr(response.root, "result") else None
-    if not result:
+    if isinstance(response.root, JSONRPCErrorResponse):
         return None
 
+    result = response.root.result if hasattr(response.root, "result") else None
+
     # Handle both Task and Message results
-    if hasattr(result, "artifacts") and result.artifacts:
+    if isinstance(result, Task) and result.artifacts:
         # Task result with artifacts
         latest_artifact = result.artifacts[-1]
         content_parts = []
         for part in latest_artifact.parts:
-            part_data = part.root if hasattr(part, "root") else part
+            part_data = part.root
             if part_data.kind == "text":
                 content_parts.append(part_data.text)
         return "\n".join(content_parts)
-    elif hasattr(result, "parts"):
+    elif isinstance(result, Message) and result.parts:
         # Direct message result
         content_parts = []
         for part in result.parts:
-            part_data = part.root if hasattr(part, "root") else part
+            part_data = part.root
             if part_data.kind == "text":
                 content_parts.append(part_data.text)
         return "\n".join(content_parts)

AgentCrew/modules/a2a/agent_cards.py

@@ -4,7 +4,7 @@
 
 from typing import List
 from AgentCrew.modules.agents import LocalAgent
-from .common.types import AgentCard, AgentCapabilities, AgentSkill, AgentProvider
+from a2a.types import AgentCard, AgentCapabilities, AgentSkill, AgentProvider
 
 
 def map_tool_to_skill(tool_name: str, tool_def) -> AgentSkill:

AgentCrew/modules/a2a/common/client/card_resolver.py

@@ -2,10 +2,7 @@
 
 import httpx
 
-from AgentCrew.modules.a2a.common.types import (
-    A2AClientJSONError,
-    AgentCard,
-)
+from a2a.types import AgentCard
 
 
 class A2ACardResolver:
@@ -20,4 +17,4 @@ def get_agent_card(self) -> AgentCard:
             try:
                 return AgentCard(**response.json())
             except json.JSONDecodeError as e:
-                raise A2AClientJSONError(str(e)) from e
+                raise httpx.RequestError(str(e)) from e

AgentCrew/modules/a2a/common/client/client.py

@@ -9,9 +9,7 @@
 from httpx._types import TimeoutTypes
 from httpx_sse import aconnect_sse
 
-from AgentCrew.modules.a2a.common.types import (
-    A2AClientHTTPError,
-    A2AClientJSONError,
+from a2a.types import (
     A2ARequest,
     AgentCard,
     CancelTaskRequest,
@@ -70,9 +68,7 @@ async def send_message_streaming(
                             json.loads(sse.data)
                         )
                 except json.JSONDecodeError as e:
-                    raise A2AClientJSONError(str(e)) from e
-                except httpx.RequestError as e:
-                    raise A2AClientHTTPError(400, str(e)) from e
+                    raise httpx.DecodingError(str(e)) from e
 
     async def _send_request(self, request: A2ARequest) -> dict[str, Any]:
         async with httpx.AsyncClient() as client:
@@ -83,10 +79,8 @@ async def _send_request(self, request: A2ARequest) -> dict[str, Any]:
                 )
                 response.raise_for_status()
                 return response.json()
-            except httpx.HTTPStatusError as e:
-                raise A2AClientHTTPError(e.response.status_code, str(e)) from e
             except json.JSONDecodeError as e:
-                raise A2AClientJSONError(str(e)) from e
+                raise httpx.DecodingError(str(e)) from e
 
     async def get_task(self, payload: TaskQueryParams) -> GetTaskResponse:
         request = GetTaskRequest(id=str(uuid4()), params=payload)

AgentCrew/modules/a2a/common/server/task_manager.py

@@ -5,16 +5,21 @@
 from collections.abc import AsyncIterable
 
 from AgentCrew.modules.a2a.common.server.utils import new_not_implemented_error
-from AgentCrew.modules.a2a.common.types import (
+from a2a.types import (
     Artifact,
     CancelTaskRequest,
     CancelTaskResponse,
     GetTaskPushNotificationConfigRequest,
     GetTaskPushNotificationConfigResponse,
     GetTaskRequest,
     GetTaskResponse,
+    SetTaskPushNotificationConfigSuccessResponse,
+    GetTaskPushNotificationConfigSuccessResponse,
+    SendStreamingMessageSuccessResponse,
     InternalError,
     JSONRPCError,
+    GetTaskSuccessResponse,
+    JSONRPCErrorResponse,
     JSONRPCResponse,
     PushNotificationConfig,
     SendMessageRequest,
@@ -104,13 +109,17 @@ async def on_get_task(self, request: GetTaskRequest) -> GetTaskResponse:
         async with self.lock:
             task = self.tasks.get(task_query_params.id)
             if task is None:
-                return GetTaskResponse(error=TaskNotFoundError(), id=request.id)
+                return GetTaskResponse(
+                    root=JSONRPCErrorResponse(error=TaskNotFoundError(), id=request.id)
+                )
 
             task_result = self.append_task_history(
                 task, task_query_params.historyLength
             )
 
-        return GetTaskResponse(result=task_result, id=request.id)
+        return GetTaskResponse(
+            root=GetTaskSuccessResponse(result=task_result, id=request.id)
+        )
 
     async def on_cancel_task(self, request: CancelTaskRequest) -> CancelTaskResponse:
         logger.info(f"Cancelling task {request.params.id}")
@@ -119,9 +128,13 @@ async def on_cancel_task(self, request: CancelTaskRequest) -> CancelTaskResponse
         async with self.lock:
             task = self.tasks.get(task_id_params.id)
             if task is None:
-                return CancelTaskResponse(error=TaskNotFoundError(), id=request.id)
+                return CancelTaskResponse(
+                    root=JSONRPCErrorResponse(error=TaskNotFoundError(), id=request.id)
+                )
 
-        return CancelTaskResponse(error=TaskNotCancelableError(), id=request.id)
+        return CancelTaskResponse(
+            root=JSONRPCErrorResponse(error=TaskNotCancelableError(), id=request.id)
+        )
 
     @abstractmethod
     async def on_send_message(self, request: SendMessageRequest) -> SendMessageResponse:
@@ -168,15 +181,19 @@ async def on_set_task_push_notification(
             )
         except Exception as e:
             logger.error(f"Error while setting push notification info: {e}")
-            return JSONRPCResponse(
-                id=request.id,
-                error=InternalError(
-                    message="An error occurred while setting push notification info"
-                ),
+            return SetTaskPushNotificationConfigResponse(
+                root=JSONRPCErrorResponse(
+                    id=request.id,
+                    error=InternalError(
+                        message="An error occurred while setting push notification info"
+                    ),
+                )
             )
 
         return SetTaskPushNotificationConfigResponse(
-            id=request.id, result=task_notification_params
+            root=SetTaskPushNotificationConfigSuccessResponse(
+                id=request.id, result=task_notification_params
+            )
         )
 
     async def on_get_task_push_notification(
@@ -190,17 +207,21 @@ async def on_get_task_push_notification(
         except Exception as e:
             logger.error(f"Error while getting push notification info: {e}")
             return GetTaskPushNotificationConfigResponse(
-                id=request.id,
-                error=InternalError(
-                    message="An error occurred while getting push notification info"
-                ),
+                root=JSONRPCErrorResponse(
+                    id=request.id,
+                    error=InternalError(
+                        message="An error occurred while getting push notification info"
+                    ),
+                )
             )
 
         return GetTaskPushNotificationConfigResponse(
-            id=request.id,
-            result=TaskPushNotificationConfig(
-                taskId=task_params.id, pushNotificationConfig=notification_info
-            ),
+            root=GetTaskPushNotificationConfigSuccessResponse(
+                id=request.id,
+                result=TaskPushNotificationConfig(
+                    taskId=task_params.id, pushNotificationConfig=notification_info
+                ),
+            )
         )
 
     async def upsert_task(self, message_send_params: MessageSendParams) -> Task:
@@ -296,10 +317,16 @@ async def dequeue_events_for_sse(
             while True:
                 event = await sse_event_queue.get()
                 if isinstance(event, JSONRPCError):
-                    yield SendStreamingMessageResponse(id=request_id, error=event)
+                    yield SendStreamingMessageResponse(
+                        root=JSONRPCErrorResponse(id=request_id, error=event)
+                    )
                     break
 
-                yield SendStreamingMessageResponse(id=request_id, result=event)
+                yield SendStreamingMessageResponse(
+                    root=SendStreamingMessageSuccessResponse(
+                        id=request_id, result=event
+                    )
+                )
                 if isinstance(event, TaskStatusUpdateEvent) and event.final:
                     break
         finally:

AgentCrew/modules/a2a/common/server/utils.py

@@ -1,5 +1,6 @@
-from AgentCrew.modules.a2a.common.types import (
+from a2a.types import (
     ContentTypeNotSupportedError,
+    JSONRPCErrorResponse,
     JSONRPCResponse,
     UnsupportedOperationError,
 )
@@ -21,8 +22,12 @@ def are_modalities_compatible(
 
 
 def new_incompatible_types_error(request_id):
-    return JSONRPCResponse(id=request_id, error=ContentTypeNotSupportedError())
+    return JSONRPCResponse(
+        root=JSONRPCErrorResponse(id=request_id, error=ContentTypeNotSupportedError())
+    )
 
 
 def new_not_implemented_error(request_id):
-    return JSONRPCResponse(id=request_id, error=UnsupportedOperationError())
+    return JSONRPCResponse(
+        root=JSONRPCErrorResponse(id=request_id, error=UnsupportedOperationError())
+    )