feat(a2a): add file support and multimodal content processing for A2A protocol

Author: Davidson Gomes

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Add Task Agent for structured single-task execution
 - Improve context management in agent execution
+- Add file support for A2A protocol (Agent-to-Agent) endpoints
+- Implement multimodal content processing in A2A messages
 
 ## [0.0.9] - 2025-05-13
 

src/api/a2a_routes.py

@@ -31,6 +31,7 @@
 Routes for the A2A (Agent-to-Agent) protocol.
 
 This module implements the standard A2A routes according to the specification.
+Supports both text messages and file uploads through the message parts mechanism.
 """
 
 import uuid
@@ -92,18 +93,100 @@ async def process_a2a_request(
     db: Session = Depends(get_db),
     a2a_service: A2AService = Depends(get_a2a_service),
 ):
-    """Processes an A2A request."""
+    """
+    Processes an A2A request.
+
+    Supports both text messages and file uploads. For file uploads,
+    include file parts in the message following the A2A protocol format:
+
+    {
+        "jsonrpc": "2.0",
+        "id": "request-id",
+        "method": "tasks/send",
+        "params": {
+            "id": "task-id",
+            "sessionId": "session-id",
+            "message": {
+                "role": "user",
+                "parts": [
+                    {
+                        "type": "text",
+                        "text": "Analyze this image"
+                    },
+                    {
+                        "type": "file",
+                        "file": {
+                            "name": "example.jpg",
+                            "mimeType": "image/jpeg",
+                            "bytes": "base64-encoded-content"
+                        }
+                    }
+                ]
+            }
+        }
+    }
+    """
     # Verify the API key
     if not verify_api_key(db, x_api_key):
         raise HTTPException(status_code=401, detail="Invalid API key")
 
     # Process the request
     try:
         request_body = await request.json()
+
+        debug_request_body = {}
+        if "method" in request_body:
+            debug_request_body["method"] = request_body["method"]
+        if "id" in request_body:
+            debug_request_body["id"] = request_body["id"]
+
+        logger.info(f"A2A request received: {debug_request_body}")
+
+        # Log if request contains file parts for debugging
+        if isinstance(request_body, dict) and "params" in request_body:
+            params = request_body.get("params", {})
+            message = params.get("message", {})
+            parts = message.get("parts", [])
+
+            logger.info(f"A2A message contains {len(parts)} parts")
+            for i, part in enumerate(parts):
+                if not isinstance(part, dict):
+                    logger.warning(f"Part {i+1} is not a dictionary: {type(part)}")
+                    continue
+
+                part_type = part.get("type")
+                logger.info(f"Part {i+1} type: {part_type}")
+
+                if part_type == "file":
+                    file_info = part.get("file", {})
+                    logger.info(
+                        f"File part found: {file_info.get('name')} ({file_info.get('mimeType')})"
+                    )
+                    if "bytes" in file_info:
+                        bytes_data = file_info.get("bytes", "")
+                        bytes_size = len(bytes_data) * 0.75
+                        logger.info(f"File size: ~{bytes_size/1024:.2f} KB")
+                        if bytes_data:
+                            sample = (
+                                bytes_data[:10] + "..."
+                                if len(bytes_data) > 10
+                                else bytes_data
+                            )
+                            logger.info(f"Sample of base64 data: {sample}")
+                elif part_type == "text":
+                    text_content = part.get("text", "")
+                    preview = (
+                        text_content[:30] + "..."
+                        if len(text_content) > 30
+                        else text_content
+                    )
+                    logger.info(f"Text part found: '{preview}'")
+
         result = await a2a_service.process_request(agent_id, request_body)
 
         # If the response is a streaming response, return as EventSourceResponse
         if hasattr(result, "__aiter__"):
+            logger.info("Returning streaming response")
 
             async def event_generator():
                 async for item in result:
@@ -115,11 +198,15 @@ async def event_generator():
             return EventSourceResponse(event_generator())
 
         # Otherwise, return as JSONResponse
+        logger.info("Returning standard JSON response")
         if hasattr(result, "model_dump"):
             return JSONResponse(result.model_dump(exclude_none=True))
         return JSONResponse(result)
     except Exception as e:
         logger.error(f"Error processing A2A request: {e}")
+        import traceback
+
+        logger.error(f"Full traceback: {traceback.format_exc()}")
         return JSONResponse(
             status_code=500,
             content={

src/api/chat_routes.py

@@ -28,6 +28,7 @@
 """
 
 import uuid
+import base64
 from fastapi import (
     APIRouter,
     Depends,
@@ -47,7 +48,7 @@
 from src.services import (
     agent_service,
 )
-from src.schemas.chat import ChatRequest, ChatResponse, ErrorResponse
+from src.schemas.chat import ChatRequest, ChatResponse, ErrorResponse, FileData
 from src.services.agent_runner import run_agent, run_agent_stream
 from src.core.exceptions import AgentNotFoundError
 from src.services.service_providers import (
@@ -59,7 +60,7 @@
 from datetime import datetime
 import logging
 import json
-from typing import Optional, Dict
+from typing import Optional, Dict, List, Any
 
 logger = logging.getLogger(__name__)
 
@@ -195,6 +196,29 @@ async def websocket_chat(
                     if not message:
                         continue
 
+                    files = None
+                    if data.get("files") and isinstance(data.get("files"), list):
+                        try:
+                            files = []
+                            for file_data in data.get("files"):
+                                if (
+                                    isinstance(file_data, dict)
+                                    and file_data.get("filename")
+                                    and file_data.get("content_type")
+                                    and file_data.get("data")
+                                ):
+                                    files.append(
+                                        FileData(
+                                            filename=file_data.get("filename"),
+                                            content_type=file_data.get("content_type"),
+                                            data=file_data.get("data"),
+                                        )
+                                    )
+                            logger.info(f"Processed {len(files)} files via WebSocket")
+                        except Exception as e:
+                            logger.error(f"Error processing files: {str(e)}")
+                            files = None
+
                     async for chunk in run_agent_stream(
                         agent_id=agent_id,
                         external_id=external_id,
@@ -203,6 +227,7 @@ async def websocket_chat(
                         artifacts_service=artifacts_service,
                         memory_service=memory_service,
                         db=db,
+                        files=files,
                     ):
                         await websocket.send_json(
                             {"message": json.loads(chunk), "turn_complete": False}
@@ -259,6 +284,7 @@ async def chat(
             artifacts_service,
             memory_service,
             db,
+            files=request.files,
         )
 
         return {