diff --git a/litellm/llms/a2a/common_utils.py b/litellm/llms/a2a/common_utils.py new file mode 100644 index 00000000000..899aaeaf55b --- /dev/null +++ b/litellm/llms/a2a/common_utils.py @@ -0,0 +1,82 @@ +""" +Common utilities for A2A (Agent-to-Agent) Protocol +""" +from typing import Any, Dict, List + +from litellm.llms.base_llm.chat.transformation import BaseLLMException + + +class A2AError(BaseLLMException): + """Base exception for A2A protocol errors""" + + def __init__( + self, + status_code: int, + message: str, + headers: Dict[str, Any] = {}, + ): + super().__init__( + status_code=status_code, + message=message, + headers=headers, + ) + + +def extract_text_from_a2a_message(message: Dict[str, Any]) -> str: + """ + Extract text content from A2A message parts. + + Args: + message: A2A message dict with 'parts' containing text parts + + Returns: + Concatenated text from all text parts + """ + if message is None: + return "" + + parts = message.get("parts", []) + text_parts: List[str] = [] + + for part in parts: + if part.get("kind") == "text": + text_parts.append(part.get("text", "")) + + return " ".join(text_parts) + + +def extract_text_from_a2a_response(response_dict: Dict[str, Any]) -> str: + """ + Extract text content from A2A response result. + + Args: + response_dict: A2A response dict with 'result' containing message + + Returns: + Text from response message parts + """ + result = response_dict.get("result", {}) + if not isinstance(result, dict): + return "" + + # A2A response can have different formats: + # 1. Direct message: {"result": {"kind": "message", "parts": [...]}} + # 2. Nested message: {"result": {"message": {"parts": [...]}}} + # 3. Task with artifacts: {"result": {"kind": "task", "artifacts": [{"parts": [...]}]}} + + # Check if result itself has parts (direct message) + if "parts" in result: + return extract_text_from_a2a_message(result) + + # Check for nested message + message = result.get("message") + if message: + return extract_text_from_a2a_message(message) + + # Handle task result with artifacts + artifacts = result.get("artifacts", []) + if artifacts and len(artifacts) > 0: + first_artifact = artifacts[0] + return extract_text_from_a2a_message(first_artifact) + + return ""